add logging for cdn headers

Author: mredolatti

client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.1.7-rc1</version>
+        <version>4.1.7-rc2</version>
     </parent>
     <artifactId>java-client</artifactId>
     <packaging>jar</packaging>

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

@@ -256,6 +256,9 @@ public String streamingServiceURL() {
 
     public int streamingRetryDelay() {return _onDemandFetchRetryDelayMs;}
 
+    public boolean cdnDebugLogging() { return _cdnDebugLogging; }
+
+
     public static final class Builder {
 
         private String _endpoint = "https://sdk.split.io";

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

@@ -129,7 +129,9 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         _splitFetcher = buildSplitFetcher();
 
         // SplitSynchronizationTask
-        _splitSynchronizationTask = new SplitSynchronizationTask(_splitFetcher, _splitCache, findPollingPeriod(RANDOM, config.featuresRefreshRate()));
+        _splitSynchronizationTask = new SplitSynchronizationTask(_splitFetcher,
+                _splitCache,
+                findPollingPeriod(RANDOM, config.featuresRefreshRate()));
 
         // Impressions
         _impressionsManager = buildImpressionsManager(config);
@@ -138,27 +140,52 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         _cachedFireAndForgetMetrics = buildCachedFireAndForgetMetrics(config);
 
         // EventClient
-        _eventClient = EventClientImpl.create(_httpclient, _eventsRootTarget, config.eventsQueueSize(), config.eventFlushIntervalInMillis(), config.waitBeforeShutdown());
+        _eventClient = EventClientImpl.create(_httpclient,
+                _eventsRootTarget,
+                config.eventsQueueSize(),
+                config.eventFlushIntervalInMillis(),
+                config.waitBeforeShutdown());
 
         // SyncManager
-        _syncManager = SyncManagerImp.build(config.streamingEnabled(), _splitSynchronizationTask, _splitFetcher, _segmentSynchronizationTaskImp, _splitCache, config.authServiceURL(), _httpclient, config.streamingServiceURL(), config.authRetryBackoffBase(), buildSSEdHttpClient(config), _segmentCache, config.streamingRetryDelay());
+        _syncManager = SyncManagerImp.build(config.streamingEnabled(),
+                _splitSynchronizationTask,
+                _splitFetcher,
+                _segmentSynchronizationTaskImp,
+                _splitCache,
+                config.authServiceURL(),
+                _httpclient,
+                config.streamingServiceURL(),
+                config.authRetryBackoffBase(),
+                buildSSEdHttpClient(config),
+                _segmentCache,
+                config.streamingRetryDelay(),
+                config.cdnDebugLogging());
         _syncManager.start();
 
         // Evaluator
         _evaluator = new EvaluatorImp(_splitCache);
 
         // SplitClient
-        _client = new SplitClientImpl(this, _splitCache, _impressionsManager, _cachedFireAndForgetMetrics, _eventClient, config, _gates, _evaluator);
+        _client = new SplitClientImpl(this,
+                _splitCache,
+                _impressionsManager,
+                _cachedFireAndForgetMetrics,
+                _eventClient,
+                config,
+                _gates,
+                _evaluator);
 
         // SplitManager
         _manager = new SplitManagerImpl(_splitCache, config, _gates);
 
         // DestroyOnShutDown
         if (config.destroyOnShutDown()) {
-            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            Thread shutdown = new Thread(() -> {
                 // Using the full path to avoid conflicting with Thread.destroy()
                 SplitFactoryImpl.this.destroy();
-            }));
+            });
+            shutdown.setName("split-destroy-worker");
+            Runtime.getRuntime().addShutdownHook(shutdown);
         }
     }
 

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

@@ -5,7 +5,7 @@
 
 public class FastlyHeadersCaptor {
 
-    public static Set<String> FIELDS_TO_CAPTURE = new HashSet<>(Arrays.asList(
+    private static final Set<String> HEADERS_TO_CAPTURE = new HashSet<>(Arrays.asList(
             "Fastly-Debug-Path",
             "Fastly-Debug-TTL",
             "Fastly-Debug-Digest",
@@ -24,7 +24,7 @@ public class FastlyHeadersCaptor {
 
     public Void handle(Map<String, String> responseHeaders) {
         _headers.add(responseHeaders.entrySet().stream()
-                .filter(e -> FIELDS_TO_CAPTURE.contains(e.getKey()))
+                .filter(e -> HEADERS_TO_CAPTURE.contains(e.getKey()))
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
         return null;
     }

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

@@ -1,5 +1,8 @@
 package io.split.engine.common;
 
+import io.split.engine.matchers.AttributeMatcher;
+import org.checkerframework.checker.units.qual.A;
+
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.Function;
@@ -55,6 +58,18 @@ private FetchOptions(boolean cacheControlHeaders, Function<Map<String, String>,
         _fastlyDebugHeader = fastlyDebugHeader;
     }
 
+    public boolean equals(Object obj) {
+        if (null == obj) return false;
+        if (this == obj) return true;
+        if (!(obj instanceof FetchOptions)) return false;
+
+        FetchOptions other = (FetchOptions) obj;
+
+        return Objects.equals(_cacheControlHeaders, other._cacheControlHeaders)
+                && Objects.equals(_fastlyDebugHeader, other._fastlyDebugHeader)
+                && Objects.equals(_responseHeadersCallback, other._responseHeadersCallback);
+    }
+
     private final boolean _cacheControlHeaders;
     private final boolean _fastlyDebugHeader;
     private final Function<Map<String, String>, Void> _responseHeadersCallback;

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

@@ -60,9 +60,10 @@ public static SyncManagerImp build(boolean streamingEnabledConfig,
                                        int authRetryBackOffBase,
                                        CloseableHttpClient sseHttpClient,
                                        SegmentCache segmentCache,
-                                       int streamingRetryDelay) {
+                                       int streamingRetryDelay,
+                                       boolean cdnDebugLogging) {
         LinkedBlockingQueue<PushManager.Status> pushMessages = new LinkedBlockingQueue<>();
-        Synchronizer synchronizer = new SynchronizerImp(splitSynchronizationTask, splitFetcher, segmentSynchronizationTaskImp, splitCache, segmentCache, streamingRetryDelay);
+        Synchronizer synchronizer = new SynchronizerImp(splitSynchronizationTask, splitFetcher, segmentSynchronizationTaskImp, splitCache, segmentCache, streamingRetryDelay, cdnDebugLogging);
         PushManager pushManager = PushManagerImp.build(synchronizer, streamingServiceUrl, authUrl, httpClient, pushMessages, sseHttpClient);
         return new SyncManagerImp(streamingEnabledConfig, synchronizer, pushManager, pushMessages, authRetryBackOffBase);
     }
@@ -111,14 +112,16 @@ private void startPollingMode() {
                         _pushManager.startWorkers();
                         _pushManager.scheduleConnectionReset();
                         _backoff.reset();
+                        _log.info("Streaming up and running.");
                         break;
                     case STREAMING_DOWN:
+                        _log.info("Streaming service temporarily unavailable, working in polling mode.");
                         _pushManager.stopWorkers();
                         _synchronizer.startPeriodicFetching();
                         break;
                     case STREAMING_BACKOFF:
                         long howLong = _backoff.interval() * 1000;
-                        _log.error(String.format("Retryable error in streaming subsystem. Switching to polling and retrying in %d seconds", howLong/1000));
+                        _log.info(String.format("Retryable error in streaming subsystem. Switching to polling and retrying in %d seconds", howLong/1000));
                         _synchronizer.startPeriodicFetching();
                         _pushManager.stopWorkers();
                         _pushManager.stop();
@@ -127,6 +130,7 @@ private void startPollingMode() {
                         _pushManager.start();
                         break;
                     case STREAMING_OFF:
+                        _log.info("Unrecoverable error in streaming subsystem. SDK will work in polling-mode and will not retry an SSE connection.");
                         _pushManager.stop();
                         _synchronizer.startPeriodicFetching();
                         if (null != _pushStatusMonitorTask) {

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

@@ -12,9 +12,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
@@ -23,31 +20,34 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 public class SynchronizerImp implements Synchronizer {
-    private static final Logger _log = LoggerFactory.getLogger(Synchronizer.class);
-    private static final int RETRIES_NUMBER = 10;
 
-    private final boolean _cdnResponseHeadersLogging = true;
+    private static final int MAX_ATTEMPTS = 10;
+
+    private static final Logger _log = LoggerFactory.getLogger(Synchronizer.class);
     private final SplitSynchronizationTask _splitSynchronizationTask;
     private final SplitFetcher _splitFetcher;
     private final SegmentSynchronizationTask _segmentSynchronizationTaskImp;
     private final ScheduledExecutorService _syncAllScheduledExecutorService;
     private final SplitCache _splitCache;
     private final SegmentCache _segmentCache;
     private final int _onDemandFetchRetryDelayMs;
+    private final boolean _cdnResponseHeadersLogging;
     private final Gson gson = new GsonBuilder().create();
 
     public SynchronizerImp(SplitSynchronizationTask splitSynchronizationTask,
                            SplitFetcher splitFetcher,
                            SegmentSynchronizationTask segmentSynchronizationTaskImp,
                            SplitCache splitCache,
                            SegmentCache segmentCache,
-                           int onDemandFetchRetryDelayMs) {
+                           int onDemandFetchRetryDelayMs,
+                           boolean cdnResponseHeadersLogging) {
         _splitSynchronizationTask = checkNotNull(splitSynchronizationTask);
         _splitFetcher = checkNotNull(splitFetcher);
         _segmentSynchronizationTaskImp = checkNotNull(segmentSynchronizationTaskImp);
         _splitCache = checkNotNull(splitCache);
         _segmentCache = checkNotNull(segmentCache);
         _onDemandFetchRetryDelayMs = checkNotNull(onDemandFetchRetryDelayMs);
+        _cdnResponseHeadersLogging = cdnResponseHeadersLogging;
 
         ThreadFactory splitsThreadFactory = new ThreadFactoryBuilder()
                 .setDaemon(true)
@@ -80,7 +80,7 @@ public void stopPeriodicFetching() {
 
     @Override
     public void refreshSplits(long targetChangeNumber) {
-        int retries = RETRIES_NUMBER;
+
         if (targetChangeNumber <= _splitCache.getChangeNumber()) {
             return;
         }
@@ -92,18 +92,16 @@ public void refreshSplits(long targetChangeNumber) {
                 .responseHeadersCallback(_cdnResponseHeadersLogging ? captor::handle : null)
                 .build();
 
+        int remainingAttempts = MAX_ATTEMPTS;
         while(true) {
-            retries--;
+            remainingAttempts--;
             _splitFetcher.forceRefresh(opts);
             if (targetChangeNumber <= _splitCache.getChangeNumber()) {
-                _log.debug("Refresh completed in %s attempts.", RETRIES_NUMBER - retries);
-                return;
-            } else if (retries <= 0) {
-                _log.warn("No changes fetched after %s attempts.", RETRIES_NUMBER);
-                if (_cdnResponseHeadersLogging) {
-                    _log.debug("CDN Debug headers: ", gson.toJson(captor.get()));
-                }
-                return;
+                _log.debug(String.format("Refresh completed in %s attempts.", MAX_ATTEMPTS - remainingAttempts));
+                break;
+            } else if (remainingAttempts <= 0) {
+                _log.info(String.format("No changes fetched after %s attempts.", MAX_ATTEMPTS));
+                break;
             }
             try {
                 Thread.sleep(_onDemandFetchRetryDelayMs);
@@ -112,6 +110,10 @@ public void refreshSplits(long targetChangeNumber) {
                 _log.debug("Error trying to sleep current Thread.");
             }
         }
+
+        if (_cdnResponseHeadersLogging && remainingAttempts <= (MAX_ATTEMPTS / 2)) {
+            _log.info(String.format("CDN Debug headers: %s", gson.toJson(captor.get())));
+        }
     }
 
     @Override
@@ -125,7 +127,7 @@ 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 <= RETRIES_NUMBER) {
+        while(changeNumber > _segmentCache.getChangeNumber(segmentName) && retries <= MAX_ATTEMPTS) {
             SegmentFetcher fetcher = _segmentSynchronizationTaskImp.getFetcher(segmentName);
             try{
                 fetcher.fetch(true);

client/src/main/java/io/split/engine/sse/workers/Worker.java

@@ -25,6 +25,7 @@ public void start() {
             _log.debug(String.format("%s Worker starting ...", _workerName));
             _queue.clear();
             _thread = new Thread( this);
+            _thread.setName(String.format("%s-worker", _workerName));
             _thread.start();
         } else {
             _log.debug(String.format("%s Worker already running.", _workerName));

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

@@ -3,6 +3,7 @@
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
 import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.HttpEntity;
 import org.mockito.Mockito;
 
@@ -18,7 +19,7 @@ public static CloseableHttpClient mockHttpClient(String jsonName, int httpStatus
         ClassicHttpResponse httpResponseMock = Mockito.mock(ClassicHttpResponse.class);
         Mockito.when(httpResponseMock.getEntity()).thenReturn(entityMock);
         Mockito.when(httpResponseMock.getCode()).thenReturn(httpStatus);
-
+        Mockito.when(httpResponseMock.getHeaders()).thenReturn(new Header[0]);
         CloseableHttpClient httpClientMock = Mockito.mock(CloseableHttpClient.class);
         Mockito.when(httpClientMock.execute(Mockito.anyObject())).thenReturn(classicResponseToCloseableMock(httpResponseMock));
 

client/src/test/java/io/split/engine/common/FastlyHeadersCaptorTest.java

@@ -0,0 +1,62 @@
+package io.split.engine.common;
+
+import org.junit.Test;
+
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class FastlyHeadersCaptorTest {
+
+    @Test
+    public void filterWorks() {
+        FastlyHeadersCaptor captor = new FastlyHeadersCaptor();
+        captor.handle(Stream.of(new String[][] {
+                {"Fastly-Debug-Path", "something"},
+                {"Fastly-Debug-TTL", "something"},
+                {"Fastly-Debug-Digest", "something"},
+                {"X-Served-By", "something"},
+                {"X-Cache", "something"},
+                {"X-Cache-Hits", "something"},
+                {"X-Timer", "something"},
+                {"Surrogate-Key", "something"},
+                {"ETag", "something"},
+                {"Cache-Control", "something"},
+                {"X-Request-ID", "something"},
+                {"Last-Modified", "something"},
+                {"NON_IMPORTANT_1", "something"},
+                {"ANOTHER_NON_IMPORTANT", "something"}
+        }).collect(Collectors.toMap(d -> d[0], d -> d[1])));
+
+        assertEquals(captor.get().size(), 1);
+        assertEquals(captor.get().get(0).size(), 12);
+        assertFalse(captor.get().get(0).containsKey("NON_IMPORTANT_1"));
+        assertFalse(captor.get().get(0).containsKey("ANOTHER_NON_IMPORTANT"));
+    }
+
+    @Test
+    public void orderIsPreserved() {
+        FastlyHeadersCaptor captor = new FastlyHeadersCaptor();
+        captor.handle(Stream.of(new String[][]{
+                {"Fastly-Debug-Path", "first"},
+        }).collect(Collectors.toMap(d -> d[0], d -> d[1])));
+
+        captor.handle(Stream.of(new String[][]{
+                {"Fastly-Debug-Path", "second"},
+        }).collect(Collectors.toMap(d -> d[0], d -> d[1])));
+
+        captor.handle(Stream.of(new String[][]{
+                {"Fastly-Debug-Path", "third"},
+        }).collect(Collectors.toMap(d -> d[0], d -> d[1])));
+
+        assertEquals(captor.get().size(), 3);
+        assertEquals(captor.get().get(0).size(), 1);
+        assertEquals(captor.get().get(1).size(), 1);
+        assertEquals(captor.get().get(2).size(), 1);
+        assertEquals(captor.get().get(0).get("Fastly-Debug-Path"), "first");
+        assertEquals(captor.get().get(1).get("Fastly-Debug-Path"), "second");
+        assertEquals(captor.get().get(2).get("Fastly-Debug-Path"), "third");
+    }
+}