Merge pull request #233 from splitio/development

Telemetry

Author: ldecheverz-split

README.md

@@ -64,7 +64,7 @@ Split has built and maintains SDKs for:
 * Java [Github](https://github.com/splitio/java-client) [Docs](https://help.split.io/hc/en-us/articles/360020405151-Java-SDK)
 * Javascript [Github](https://github.com/splitio/javascript-client) [Docs](https://help.split.io/hc/en-us/articles/360020448791-JavaScript-SDK)
 * Node [Github](https://github.com/splitio/javascript-client) [Docs](https://help.split.io/hc/en-us/articles/360020564931-Node-js-SDK)
-* .NET [Github](https://github.com/splitio/.net-core-client) [Docs](https://help.split.io/hc/en-us/articles/360020240172--NET-SDK)
+* .NET [Github](https://github.com/splitio/dotnet-client) [Docs](https://help.split.io/hc/en-us/articles/360020240172--NET-SDK)
 * Ruby [Github](https://github.com/splitio/ruby-client) [Docs](https://help.split.io/hc/en-us/articles/360020673251-Ruby-SDK)
 * PHP [Github](https://github.com/splitio/php-client) [Docs](https://help.split.io/hc/en-us/articles/360020350372-PHP-SDK)
 * Python [Github](https://github.com/splitio/python-client) [Docs](https://help.split.io/hc/en-us/articles/360020359652-Python-SDK)

client/CHANGES.txt

@@ -1,10 +1,15 @@
 CHANGES
 
+4.2.0 (Jun 7, 2021)
+- Updated SDK telemetry storage, metrics and updater to be more effective and send less often.
+- Improved the synchronization flow to be more reliable in the event of an edge case generating delay in cache purge propagation, keeping the SDK cache properly synced.
+- Fixed issue where the SDK was validating no Split had over 50 conditions (legacy code).
+
 4.1.6 (Apr 15, 2021)
--Updated log level and message in some messages.
+- Updated log level and message in some messages.
 
 4.1.5 (Apr 6, 2021)
--Updated: Streaming retry fix.
+- Updated streaming logic to use limited fetch retry attempts.
 
 4.1.4 (Mar 19, 2021)
 - Updated: Internal cache structure refactor.

client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>io.split.client</groupId>
         <artifactId>java-client-parent</artifactId>
-        <version>4.1.6</version>
+        <version>4.2.0</version>
     </parent>
     <artifactId>java-client</artifactId>
     <packaging>jar</packaging>
@@ -121,7 +121,7 @@
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
-            <version>29.0-jre</version>
+            <version>30.0-jre</version>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>

client/src/main/java/io/split/cache/SegmentCache.java

@@ -1,6 +1,9 @@
 package io.split.cache;
 
+import io.split.engine.segments.SegmentImp;
+
 import java.util.List;
+import java.util.Set;
 
 /**
  * Memory for segments
@@ -42,4 +45,16 @@ public interface SegmentCache {
      * clear all segments
      */
     void clear();
+
+    /**
+     * return every segment
+     * @return
+     */
+    List<SegmentImp> getAll();
+
+    /**
+     * return key count
+     * @return
+     */
+    long getKeyCount();
 }

client/src/main/java/io/split/cache/SegmentCacheInMemoryImpl.java

@@ -6,7 +6,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ConcurrentMap;
+import java.util.stream.Collectors;
 
 /**
  * InMemoryCache Implementation
@@ -59,4 +61,14 @@ public long getChangeNumber(String segmentName) {
     public void clear() {
         _segments.clear();
     }
+
+    @Override
+    public List<SegmentImp> getAll() {
+        return _segments.values().stream().collect(Collectors.toList());
+    }
+
+    @Override
+    public long getKeyCount() {
+        return _segments.values().stream().mapToLong(SegmentImp::getKeysSize).sum();
+    }
 }

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

@@ -6,6 +6,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
+import java.util.Map;
+
 public class ApiKeyCounter {
 
     private static final Logger _log = LoggerFactory.getLogger(ApiKeyCounter.class);
@@ -63,4 +66,18 @@ boolean isApiKeyPresent(String apiKey) {
     int getCount(String apiKey) {
         return USED_API_KEYS.count(apiKey);
     }
+
+    public Map<String, Long> getFactoryInstances() {
+        Map<String, Long> factoryInstances = new HashMap<>();
+        for (String factory :USED_API_KEYS) {
+            factoryInstances.putIfAbsent(factory, new Long(getCount(factory)));
+        }
+
+        return factoryInstances;
+    }
+
+    @VisibleForTesting
+    void clearApiKeys() {
+        USED_API_KEYS.clear();
+    }
 }

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

@@ -4,6 +4,11 @@
 import io.split.client.dtos.Event;
 import io.split.client.utils.GenericClientUtil;
 import io.split.client.utils.Utils;
+import io.split.telemetry.domain.enums.EventsDataRecordsEnum;
+import io.split.telemetry.domain.enums.HTTPLatenciesEnum;
+import io.split.telemetry.domain.enums.LastSynchronizationRecordsEnum;
+import io.split.telemetry.storage.TelemetryEvaluationProducer;
+import io.split.telemetry.storage.TelemetryRuntimeProducer;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -23,6 +28,7 @@
 import java.util.concurrent.TimeUnit;
 
 import static java.lang.Thread.MIN_PRIORITY;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
  * Responsible for sending events added via .track() to Split collection services
@@ -45,34 +51,28 @@ public class EventClientImpl implements EventClient {
     private final CloseableHttpClient _httpclient;
     private final URI _target;
     private final int _waitBeforeShutdown;
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
     ThreadFactory eventClientThreadFactory(final String name) {
-        return new ThreadFactory() {
-            @Override
-            public Thread newThread(final Runnable r) {
-                return new Thread(new Runnable() {
-                    @Override
-                    public void run() {
-                        Thread.currentThread().setPriority(MIN_PRIORITY);
-                        r.run();
-                    }
-                }, name);
-            }
-        };
+        return r -> new Thread(() -> {
+            Thread.currentThread().setPriority(MIN_PRIORITY);
+            r.run();
+        }, name);
     }
 
 
-    public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsRootTarget, int maxQueueSize, long flushIntervalMillis, int waitBeforeShutdown) throws URISyntaxException {
-        return new EventClientImpl(new LinkedBlockingQueue<WrappedEvent>(),
+    public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsRootTarget, int maxQueueSize, long flushIntervalMillis, int waitBeforeShutdown, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
+        return new EventClientImpl(new LinkedBlockingQueue<>(maxQueueSize),
                 httpclient,
                 Utils.appendPath(eventsRootTarget, "api/events/bulk"),
                 maxQueueSize,
                 flushIntervalMillis,
-                waitBeforeShutdown);
+                waitBeforeShutdown,
+                telemetryRuntimeProducer);
     }
 
     EventClientImpl(BlockingQueue<WrappedEvent> eventQueue, CloseableHttpClient httpclient, URI target, int maxQueueSize,
-                    long flushIntervalMillis, int waitBeforeShutdown) throws URISyntaxException {
+                    long flushIntervalMillis, int waitBeforeShutdown, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
 
         _httpclient = httpclient;
 
@@ -83,6 +83,7 @@ public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsR
 
         _maxQueueSize = maxQueueSize;
         _flushIntervalMillis = flushIntervalMillis;
+        _telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
 
         _senderExecutor = new ThreadPoolExecutor(
                 1,
@@ -122,9 +123,16 @@ public boolean track(Event event, int eventSize) {
             if (event == null) {
                 return false;
             }
-            _eventQueue.put(new WrappedEvent(event, eventSize));
+            if(_eventQueue.offer(new WrappedEvent(event, eventSize))) {
+                _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
+            }
+            else {
+                _log.warn("Event dropped.");
+                _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
+            }
 
-        } catch (InterruptedException e) {
+        } catch (ClassCastException | NullPointerException | IllegalArgumentException e) {
+            _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
             _log.warn("Interruption when adding event withed while adding message %s.", event);
             return false;
         }
@@ -153,7 +161,7 @@ public void run() {
             List<Event> events = new ArrayList<>();
             long accumulated = 0;
             try {
-                while (true) {
+                while (!Thread.currentThread().isInterrupted()) {
                     WrappedEvent data = _eventQueue.take();
                     Event event = data.event();
                     Long size = data.size();
@@ -169,7 +177,7 @@ public void run() {
 
                         continue;
                     }
-
+                    long initTime = System.currentTimeMillis();
                     if (events.size() >= _maxQueueSize ||  accumulated >= MAX_SIZE_BYTES || event == CENTINEL) {
 
                         // Send over the network
@@ -183,6 +191,8 @@ public void run() {
                         // Clear the queue of events for the next batch.
                         events = new ArrayList<>();
                         accumulated = 0;
+                        _telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.EVENTS, System.currentTimeMillis()-initTime);
+                        _telemetryRuntimeProducer.recordSuccessfulSync(LastSynchronizationRecordsEnum.EVENTS, System.currentTimeMillis());
                     }
                 }
             } catch (InterruptedException e) {

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

@@ -4,11 +4,18 @@
 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 io.split.telemetry.domain.enums.HTTPLatenciesEnum;
+import io.split.telemetry.domain.enums.LastSynchronizationRecordsEnum;
+import io.split.telemetry.domain.enums.ResourceEnum;
+import io.split.telemetry.storage.TelemetryRuntimeProducer;
 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.HttpStatus;
+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 +24,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,57 +36,72 @@ 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;
-    private final Metrics _metrics;
-
-    public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root) throws URISyntaxException {
-        return create(client, root, new Metrics.NoopMetrics());
-    }
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
-    public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root, Metrics metrics) throws URISyntaxException {
-        return new HttpSegmentChangeFetcher(client, Utils.appendPath(root, "api/segmentChanges"), metrics);
+    public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
+        return new HttpSegmentChangeFetcher(client, Utils.appendPath(root, "api/segmentChanges"), telemetryRuntimeProducer);
     }
 
-    private HttpSegmentChangeFetcher(CloseableHttpClient client, URI uri, Metrics metrics) {
+    private HttpSegmentChangeFetcher(CloseableHttpClient client, URI uri, TelemetryRuntimeProducer telemetryRuntimeProducer) {
         _client = client;
         _target = uri;
-        _metrics = metrics;
         checkNotNull(_target);
+        _telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
     }
 
     @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();
 
-            if (statusCode < 200 || statusCode >= 300) {
+            if (statusCode < HttpStatus.SC_OK || statusCode >= HttpStatus.SC_MULTIPLE_CHOICES) {
+                _telemetryRuntimeProducer.recordSyncError(ResourceEnum.SEGMENT_SYNC, statusCode);
                 _log.error("Response status was: " + statusCode);
-                if (statusCode == 403) {
+                if (statusCode == HttpStatus.SC_FORBIDDEN) {
                     _log.error("factory instantiation: you passed a browser type api_key, " +
                             "please grab an api key from the Split console that is of type sdk");
                 }
-                _metrics.count(PREFIX + ".status." + statusCode, 1);
                 throw new IllegalStateException("Could not retrieve segment changes for " + segmentName + "; http return code " + statusCode);
             }
 
-
+            _telemetryRuntimeProducer.recordSuccessfulSync(LastSynchronizationRecordsEnum.SEGMENTS, System.currentTimeMillis());
 
             String json = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
             if (_log.isDebugEnabled()) {
@@ -86,11 +110,10 @@ public SegmentChange fetch(String segmentName, long since, boolean addCacheHeade
 
             return Json.fromJson(json, SegmentChange.class);
         } catch (Throwable t) {
-            _metrics.count(PREFIX + ".exception", 1);
             throw new IllegalStateException("Problem fetching segmentChanges: " + t.getMessage(), t);
         } finally {
+            _telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.SEGMENTS, System.currentTimeMillis()-start);
             Utils.forceClose(response);
-            _metrics.time(PREFIX + ".time", System.currentTimeMillis() - start);
         }