add retries and cdn bypass logic to segment fetchers

Author: mredolatti

client/src/main/java/io/split/client/HttpSegmentChangeFetcher.java

@@ -4,11 +4,13 @@
 import io.split.client.dtos.SegmentChange;
 import io.split.client.utils.Json;
 import io.split.client.utils.Utils;
+import io.split.engine.common.FetchOptions;
 import io.split.engine.metrics.Metrics;
 import io.split.engine.segments.SegmentChangeFetcher;
 import org.apache.hc.client5.http.classic.methods.HttpGet;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
+import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.io.entity.EntityUtils;
 import org.apache.hc.core5.net.URIBuilder;
 import org.slf4j.Logger;
@@ -17,6 +19,8 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
@@ -27,9 +31,13 @@ public final class HttpSegmentChangeFetcher implements SegmentChangeFetcher {
     private static final Logger _log = LoggerFactory.getLogger(HttpSegmentChangeFetcher.class);
 
     private static final String SINCE = "since";
+    private static final String TILL = "till";
     private static final String PREFIX = "segmentChangeFetcher";
-    private static final String NAME_CACHE = "Cache-Control";
-    private static final String VALUE_CACHE = "no-cache";
+    private static final String CACHE_CONTROL_HEADER_NAME = "Cache-Control";
+    private static final String CACHE_CONTROL_HEADER_VALUE = "no-cache";
+
+    private static final String HEADER_FASTLY_DEBUG_NAME = "Fastly-Debug";
+    private static final String HEADER_FASTLY_DEBUG_VALUE = "1";
 
     private final CloseableHttpClient _client;
     private final URI _target;
@@ -51,19 +59,34 @@ private HttpSegmentChangeFetcher(CloseableHttpClient client, URI uri, Metrics me
     }
 
     @Override
-    public SegmentChange fetch(String segmentName, long since, boolean addCacheHeader) {
+    public SegmentChange fetch(String segmentName, long since, FetchOptions options) {
         long start = System.currentTimeMillis();
 
         CloseableHttpResponse response = null;
 
         try {
             String path = _target.getPath() + "/" + segmentName;
-            URI uri = new URIBuilder(_target).setPath(path).addParameter(SINCE, "" + since).build();
+            URIBuilder uriBuilder = new URIBuilder(_target)
+                    .setPath(path)
+                    .addParameter(SINCE, "" + since);
+            if (options.hasCustomCN()) {
+                uriBuilder.addParameter(TILL, "" + options.targetCN());
+            }
+
+            URI uri = uriBuilder.build();
             HttpGet request = new HttpGet(uri);
-            if(addCacheHeader) {
-                request.setHeader(NAME_CACHE, VALUE_CACHE);
+
+            if(options.cacheControlHeadersEnabled()) {
+                request.setHeader(CACHE_CONTROL_HEADER_NAME, CACHE_CONTROL_HEADER_VALUE);
             }
+
+            if (options.fastlyDebugHeaderEnabled()) {
+                request.addHeader(HEADER_FASTLY_DEBUG_NAME, HEADER_FASTLY_DEBUG_VALUE);
+            }
+
             response = _client.execute(request);
+            options.handleResponseHeaders(Arrays.stream(response.getHeaders())
+                    .collect(Collectors.toMap(Header::getName, Header::getValue)));
 
             int statusCode = response.getCode();
 

client/src/main/java/io/split/client/jmx/SplitJmxMonitor.java

@@ -44,7 +44,7 @@ public boolean forceSyncFeatures() {
     public boolean forceSyncSegment(String segmentName) {
         SegmentFetcher fetcher = _segmentSynchronizationTask.getFetcher(segmentName);
         try{
-            fetcher.fetch(true);
+            fetcher.fetch(new FetchOptions.Builder().build());
         }
         //We are sure this will never happen because getFetcher firts initiate the segment. This try/catch is for safe only.
         catch (NullPointerException np){

client/src/main/java/io/split/engine/common/SynchronizerImp.java

@@ -103,10 +103,10 @@ private static class SyncResult {
         private final int _remainingAttempts;
     }
 
-    private SyncResult attemptSync(long targetChangeNumber,
-                                   FetchOptions opts,
-                                   Function<Void, Long> nextWaitMs,
-                                   int maxRetries) {
+    private SyncResult attemptSplitsSync(long targetChangeNumber,
+                                         FetchOptions opts,
+                                         Function<Void, Long> nextWaitMs,
+                                         int maxRetries) {
         int remainingAttempts = maxRetries;
         while(true) {
             remainingAttempts--;
@@ -126,9 +126,9 @@ private SyncResult attemptSync(long targetChangeNumber,
         }
     }
 
-    private void logCdnHeaders(int maxRetries, int remainingAttempts, List<Map<String, String>> headers) {
+    private void logCdnHeaders(String prefix, int maxRetries, int remainingAttempts, List<Map<String, String>> headers) {
         if (maxRetries - remainingAttempts > _failedAttemptsBeforeLogging) {
-            _log.info(String.format("CDN Debug headers: %s", gson.toJson(headers)));
+            _log.info(String.format("%s: CDN Debug headers: %s", prefix, gson.toJson(headers)));
         }
     }
 
@@ -146,22 +146,22 @@ public void refreshSplits(long targetChangeNumber) {
                 .responseHeadersCallback(_cdnResponseHeadersLogging ? captor::handle : null)
                 .build();
 
-        SyncResult regularResult = attemptSync(targetChangeNumber, opts,
+        SyncResult regularResult = attemptSplitsSync(targetChangeNumber, opts,
                 (discard) -> (long) _onDemandFetchRetryDelayMs, _onDemandFetchMaxRetries);
 
         int attempts =  _onDemandFetchMaxRetries - regularResult.remainingAttempts();
         if (regularResult.success()) {
             _log.debug(String.format("Refresh completed in %s attempts.", attempts));
             if (_cdnResponseHeadersLogging) {
-                logCdnHeaders(_onDemandFetchMaxRetries , regularResult.remainingAttempts(), captor.get());
+                logCdnHeaders("[splits]", _onDemandFetchMaxRetries , regularResult.remainingAttempts(), captor.get());
             }
             return;
         }
 
         _log.info(String.format("No changes fetched after %s attempts. Will retry bypassing CDN.", attempts));
         FetchOptions withCdnBypass = new FetchOptions.Builder(opts).targetChangeNumber(targetChangeNumber).build();
         Backoff backoff = new Backoff(ON_DEMAND_FETCH_BACKOFF_BASE_MS, ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_MS);
-        SyncResult withCDNBypassed = attemptSync(targetChangeNumber, withCdnBypass,
+        SyncResult withCDNBypassed = attemptSplitsSync(targetChangeNumber, withCdnBypass,
                 (discard) -> backoff.interval(), ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES);
 
         int withoutCDNAttempts = ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - withCDNBypassed._remainingAttempts;
@@ -172,7 +172,7 @@ public void refreshSplits(long targetChangeNumber) {
         }
 
         if (_cdnResponseHeadersLogging) {
-            logCdnHeaders(_onDemandFetchMaxRetries + ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES,
+            logCdnHeaders("[splits]", _onDemandFetchMaxRetries + ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES,
                     withCDNBypassed.remainingAttempts(), captor.get());
         }
     }
@@ -185,19 +185,76 @@ public void localKillSplit(String splitName, String defaultTreatment, long newCh
         }
     }
 
-    @Override
-    public void refreshSegment(String segmentName, long changeNumber) {
-        int retries = 1;
-        while(changeNumber > _segmentCache.getChangeNumber(segmentName) && retries <= _onDemandFetchMaxRetries) {
-            SegmentFetcher fetcher = _segmentSynchronizationTaskImp.getFetcher(segmentName);
-            try{
-                fetcher.fetch(true);
+    public SyncResult attemptSegmentSync(String segmentName,
+                                         long targetChangeNumber,
+                                         FetchOptions opts,
+                                         Function<Void, Long> nextWaitMs,
+                                         int maxRetries) {
+
+        int remainingAttempts = maxRetries;
+        SegmentFetcher fetcher = _segmentSynchronizationTaskImp.getFetcher(segmentName);
+        checkNotNull(fetcher);
+
+        while(true) {
+            remainingAttempts--;
+            fetcher.fetch(opts);
+            if (targetChangeNumber <= _segmentCache.getChangeNumber(segmentName)) {
+                return new SyncResult(true, remainingAttempts);
+            } else if (remainingAttempts <= 0) {
+                return new SyncResult(false, remainingAttempts);
             }
-            //We are sure this will never happen because getFetcher firts initiate the segment. This try/catch is for safe only.
-            catch (NullPointerException np){
-                throw new NullPointerException();
+            try {
+                long howLong = nextWaitMs.apply(null);
+                Thread.sleep(howLong);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                _log.debug("Error trying to sleep current Thread.");
+            }
+        }
+    }
+
+    @Override
+    public void refreshSegment(String segmentName, long targetChangeNumber) {
+
+        if (targetChangeNumber <= _segmentCache.getChangeNumber(segmentName)) {
+            return;
+        }
+
+        FastlyHeadersCaptor captor = new FastlyHeadersCaptor();
+        FetchOptions opts = new FetchOptions.Builder()
+                .cacheControlHeaders(true)
+                .fastlyDebugHeader(_cdnResponseHeadersLogging)
+                .responseHeadersCallback(_cdnResponseHeadersLogging ? captor::handle : null)
+                .build();
+
+        SyncResult regularResult = attemptSegmentSync(segmentName, targetChangeNumber, opts,
+                (discard) -> (long) _onDemandFetchRetryDelayMs, _onDemandFetchMaxRetries);
+
+        int attempts =  _onDemandFetchMaxRetries - regularResult.remainingAttempts();
+        if (regularResult.success()) {
+            _log.debug(String.format("Segment %s refresh completed in %s attempts.", segmentName, attempts));
+            if (_cdnResponseHeadersLogging) {
+                logCdnHeaders(String.format("[segment/%s]", segmentName), _onDemandFetchMaxRetries , regularResult.remainingAttempts(), captor.get());
             }
-            retries++;
+            return;
+        }
+
+        _log.info(String.format("No changes fetched for segment %s after %s attempts. Will retry bypassing CDN.", segmentName, attempts));
+        FetchOptions withCdnBypass = new FetchOptions.Builder(opts).targetChangeNumber(targetChangeNumber).build();
+        Backoff backoff = new Backoff(ON_DEMAND_FETCH_BACKOFF_BASE_MS, ON_DEMAND_FETCH_BACKOFF_MAX_WAIT_MS);
+        SyncResult withCDNBypassed = attemptSegmentSync(segmentName, targetChangeNumber, withCdnBypass,
+                (discard) -> backoff.interval(), ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES);
+
+        int withoutCDNAttempts = ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES - withCDNBypassed._remainingAttempts;
+        if (withCDNBypassed.success()) {
+            _log.debug(String.format("Segment %s refresh completed bypassing the CDN in %s attempts.", segmentName, withoutCDNAttempts));
+        } else {
+            _log.debug(String.format("No changes fetched for segment %s after %s attempts with CDN bypassed.", segmentName, withoutCDNAttempts));
+        }
+
+        if (_cdnResponseHeadersLogging) {
+            logCdnHeaders(String.format("[segment/%s]", segmentName), _onDemandFetchMaxRetries + ON_DEMAND_FETCH_BACKOFF_MAX_RETRIES,
+                    withCDNBypassed.remainingAttempts(), captor.get());
         }
     }
 }

client/src/main/java/io/split/engine/experiments/SplitFetcherImp.java

@@ -46,7 +46,7 @@ public SplitFetcherImp(SplitChangeFetcher splitChangeFetcher, SplitParser parser
     @Override
     public void forceRefresh(FetchOptions options) {
         _log.debug("Force Refresh splits starting ...");
-        final long initialCN = _splitCache.getChangeNumber();
+        final long INITIAL_CN = _splitCache.getChangeNumber();
         try {
             while (true) {
                 long start = _splitCache.getChangeNumber();
@@ -56,7 +56,7 @@ public void forceRefresh(FetchOptions options) {
                 // If the previous execution was the first one, clear the `cdnBypass` flag
                 // for the next fetches. (This will clear a local copy of the fetch options,
                 // not the original object that was passed to this method).
-                if (initialCN == start) {
+                if (INITIAL_CN == start) {
                     options = new FetchOptions.Builder(options).targetChangeNumber(FetchOptions.DEFAULT_TARGET_CHANGENUMBER).build();
                 }
 

client/src/main/java/io/split/engine/segments/SegmentChangeFetcher.java

@@ -1,6 +1,7 @@
 package io.split.engine.segments;
 
 import io.split.client.dtos.SegmentChange;
+import io.split.engine.common.FetchOptions;
 
 /**
  * Fetches changes in the segment since a reference point.
@@ -22,8 +23,9 @@ public interface SegmentChangeFetcher {
      * @param segmentName                  the name of the segment to fetch.
      * @param changesSinceThisChangeNumber a value less than zero implies that the client is
      *                                     requesting information on this segment for the first time.
+     * @param options
      * @return SegmentChange
      * @throws java.lang.RuntimeException if there was a problem fetching segment changes
      */
-    SegmentChange fetch(String segmentName, long changesSinceThisChangeNumber, boolean addCacheHeader);
+    SegmentChange fetch(String segmentName, long changesSinceThisChangeNumber, FetchOptions options);
 }

client/src/main/java/io/split/engine/segments/SegmentFetcher.java

@@ -1,13 +1,15 @@
 package io.split.engine.segments;
 
+import io.split.engine.common.FetchOptions;
+
 /**
  * Created by adilaijaz on 5/7/15.
  */
 public interface SegmentFetcher {
     /**
      * fetch
      */
-    void fetch(boolean addCacheHeader);
+    void fetch(FetchOptions opts);
 
     void runWhitCacheHeader();
 

client/src/main/java/io/split/engine/segments/SegmentFetcherImp.java

@@ -3,6 +3,7 @@
 import io.split.cache.SegmentCache;
 import io.split.client.dtos.SegmentChange;
 import io.split.engine.SDKReadinessGates;
+import io.split.engine.common.FetchOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -31,9 +32,9 @@ public SegmentFetcherImp(String segmentName, SegmentChangeFetcher segmentChangeF
     }
 
     @Override
-    public void fetch(boolean addCacheHeader){
+    public void fetch(FetchOptions opts){
         try {
-            callLoopRun(false, addCacheHeader);
+            callLoopRun(opts);
         } catch (Throwable t) {
             _log.error("RefreshableSegmentFetcher failed: " + t.getMessage());
             if (_log.isDebugEnabled()) {
@@ -42,8 +43,8 @@ public void fetch(boolean addCacheHeader){
         }
     }
 
-    private void runWithoutExceptionHandling(boolean addCacheHeader) {
-        SegmentChange change = _segmentChangeFetcher.fetch(_segmentName, _segmentCache.getChangeNumber(_segmentName), addCacheHeader);
+    private void runWithoutExceptionHandling(FetchOptions options) {
+        SegmentChange change = _segmentChangeFetcher.fetch(_segmentName, _segmentCache.getChangeNumber(_segmentName), options);
 
         if (change == null) {
             throw new IllegalStateException("SegmentChange was null");
@@ -109,14 +110,15 @@ private String summarize(List<String> changes) {
         return bldr.toString();
     }
 
-    private void callLoopRun(boolean isFetch, boolean addCacheHeader){
+    private void callLoopRun(FetchOptions opts){
+        final long INITIAL_CN = _segmentCache.getChangeNumber(_segmentName);
         while (true) {
             long start = _segmentCache.getChangeNumber(_segmentName);
-            runWithoutExceptionHandling(addCacheHeader);
-            long end = _segmentCache.getChangeNumber(_segmentName);
-            if (isFetch && _log.isDebugEnabled()) {
-                _log.debug(_segmentName + " segment fetch before: " + start + ", after: " + _segmentCache.getChangeNumber(_segmentName) /*+ " size: " + _concurrentKeySet.size()*/);
+            runWithoutExceptionHandling(opts);
+            if (INITIAL_CN == start) {
+                opts = new FetchOptions.Builder(opts).targetChangeNumber(FetchOptions.DEFAULT_TARGET_CHANGENUMBER).build();
             }
+            long end = _segmentCache.getChangeNumber(_segmentName);
             if (start >= end) {
                 break;
             }
@@ -125,18 +127,18 @@ private void callLoopRun(boolean isFetch, boolean addCacheHeader){
 
     @Override
     public void runWhitCacheHeader(){
-        this.fetchAndUpdate(true);
+        this.fetchAndUpdate(new FetchOptions.Builder().cacheControlHeaders(true).build());
     }
 
     /**
      * Calls callLoopRun and after fetchs segment.
-     * @param addCacheHeader indicates if CacheHeader is required
+     * @param opts contains all soft of options used when issuing the fetch request
      */
-    private void fetchAndUpdate(boolean addCacheHeader) {
+    private void fetchAndUpdate(FetchOptions opts) {
         try {
             // Do this again in case the previous call errored out.
             _gates.registerSegment(_segmentName);
-            callLoopRun(true, addCacheHeader);
+            callLoopRun(opts);
 
             _gates.segmentIsReady(_segmentName);
 
@@ -150,7 +152,7 @@ private void fetchAndUpdate(boolean addCacheHeader) {
 
     @Override
     public void fetchAll() {
-        this.fetchAndUpdate(false);
+        this.fetchAndUpdate(new FetchOptions.Builder().build());
     }
 
 

client/src/test/java/io/split/client/HttpSegmentChangeFetcherTest.java

@@ -2,6 +2,7 @@
 
 import io.split.TestHelper;
 import io.split.client.dtos.SegmentChange;
+import io.split.engine.common.FetchOptions;
 import io.split.engine.metrics.Metrics;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.HttpClients;
@@ -61,7 +62,7 @@ public void testFetcherWithSpecialCharacters() throws URISyntaxException, IOExce
         Metrics.NoopMetrics metrics = new Metrics.NoopMetrics();
         HttpSegmentChangeFetcher fetcher = HttpSegmentChangeFetcher.create(httpClientMock, rootTarget, metrics);
 
-        SegmentChange change = fetcher.fetch("some_segment", 1234567, true);
+        SegmentChange change = fetcher.fetch("some_segment", 1234567, new FetchOptions.Builder().build());
 
         Assert.assertNotNull(change);
         Assert.assertEquals(1, change.added.size());