Attached every service

Author: ldecheverz-split

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

@@ -4,6 +4,8 @@
 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;
@@ -49,7 +51,7 @@ public class EventClientImpl implements EventClient {
     private final CloseableHttpClient _httpclient;
     private final URI _target;
     private final int _waitBeforeShutdown;
-    private final TelemetryRuntimeProducer telemetryRuntimeProducer;
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
     ThreadFactory eventClientThreadFactory(final String name) {
         return new ThreadFactory() {
@@ -67,18 +69,18 @@ public void run() {
     }
 
 
-    public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsRootTarget, int maxQueueSize, long flushIntervalMillis, int waitBeforeShutdown, TelemetryEvaluationProducer telemetryEvaluationProducer) throws URISyntaxException {
+    public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsRootTarget, int maxQueueSize, long flushIntervalMillis, int waitBeforeShutdown, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
         return new EventClientImpl(new LinkedBlockingQueue<WrappedEvent>(),
                 httpclient,
                 Utils.appendPath(eventsRootTarget, "api/events/bulk"),
                 maxQueueSize,
                 flushIntervalMillis,
                 waitBeforeShutdown,
-                telemetryEvaluationProducer);
+                telemetryRuntimeProducer);
     }
 
     EventClientImpl(BlockingQueue<WrappedEvent> eventQueue, CloseableHttpClient httpclient, URI target, int maxQueueSize,
-                    long flushIntervalMillis, int waitBeforeShutdown, TelemetryEvaluationProducer telemetryEvaluationProducer) throws URISyntaxException {
+                    long flushIntervalMillis, int waitBeforeShutdown, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
 
         _httpclient = httpclient;
 
@@ -87,9 +89,9 @@ public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsR
         _eventQueue = eventQueue;
         _waitBeforeShutdown = waitBeforeShutdown;
 
-        _maxQueueSize = maxQueueSize;
+        _maxQueueSize = 1;
         _flushIntervalMillis = flushIntervalMillis;
-        telemetryRuntimeProducer = checkNotNull(telemetryEvaluationProducer);
+        _telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
 
         _senderExecutor = new ThreadPoolExecutor(
                 1,
@@ -129,9 +131,12 @@ public boolean track(Event event, int eventSize) {
             if (event == null) {
                 return false;
             }
-            _eventQueue.put(new WrappedEvent(event, eventSize));
+            WrappedEvent we = new WrappedEvent(event, eventSize);
+            _eventQueue.put(we);
+            _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
 
-        } catch (InterruptedException e) {
+        } catch (ClassCastException | NullPointerException | InterruptedException e) {
+            _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
             _log.warn("Interruption when adding event withed while adding message %s.", event);
             return false;
         }
@@ -191,7 +196,8 @@ public void run() {
                         events = new ArrayList<>();
                         accumulated = 0;
                         long endTime = System.currentTimeMillis();
-                        telemetryRuntimeProducer.recordSuccessfulSync(LastSynchronizationRecordsEnum.EVENTS, endTime-initTime);
+                        _telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.EVENTS, endTime-initTime);
+                        _telemetryRuntimeProducer.recordSuccessfulSync(LastSynchronizationRecordsEnum.EVENTS, endTime);
                     }
                 }
             } catch (InterruptedException e) {

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

@@ -6,6 +6,10 @@
 import io.split.client.utils.Utils;
 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;
@@ -33,21 +37,17 @@ public final class HttpSegmentChangeFetcher implements SegmentChangeFetcher {
 
     private final CloseableHttpClient _client;
     private final URI _target;
-    private final Metrics _metrics;
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
-    public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root) throws URISyntaxException {
-        return create(client, root, new Metrics.NoopMetrics());
+    public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
+        return new HttpSegmentChangeFetcher(client, Utils.appendPath(root, "api/segmentChanges"), telemetryRuntimeProducer);
     }
 
-    public static HttpSegmentChangeFetcher create(CloseableHttpClient client, URI root, Metrics metrics) throws URISyntaxException {
-        return new HttpSegmentChangeFetcher(client, Utils.appendPath(root, "api/segmentChanges"), metrics);
-    }
-
-    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
@@ -68,16 +68,18 @@ public SegmentChange fetch(String segmentName, long since, boolean addCacheHeade
             int statusCode = response.getCode();
 
             if (statusCode < 200 || statusCode >= 300) {
+                _telemetryRuntimeProducer.recordSyncError(ResourceEnum.SEGMENT_SYNC, statusCode);
                 _log.error("Response status was: " + statusCode);
                 if (statusCode == 403) {
                     _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);
             }
 
-
+            long endTime = System.currentTimeMillis();
+            _telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.SEGMENTS, endTime-start);
+            _telemetryRuntimeProducer.recordSuccessfulSync(LastSynchronizationRecordsEnum.SEGMENTS, endTime);
 
             String json = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
             if (_log.isDebugEnabled()) {
@@ -86,11 +88,9 @@ 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 {
             Utils.forceClose(response);
-            _metrics.time(PREFIX + ".time", System.currentTimeMillis() - start);
         }
 
 

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

@@ -6,6 +6,10 @@
 import io.split.client.utils.Utils;
 import io.split.engine.experiments.SplitChangeFetcher;
 import io.split.engine.metrics.Metrics;
+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;
@@ -33,26 +37,21 @@ public final class HttpSplitChangeFetcher implements SplitChangeFetcher {
 
     private final CloseableHttpClient _client;
     private final URI _target;
-    private final Metrics _metrics;
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
-    public static HttpSplitChangeFetcher create(CloseableHttpClient client, URI root) throws URISyntaxException {
-        return create(client, root, new Metrics.NoopMetrics());
+    public static HttpSplitChangeFetcher create(CloseableHttpClient client, URI root, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
+        return new HttpSplitChangeFetcher(client, Utils.appendPath(root, "api/splitChanges"), telemetryRuntimeProducer);
     }
 
-    public static HttpSplitChangeFetcher create(CloseableHttpClient client, URI root, Metrics metrics) throws URISyntaxException {
-        return new HttpSplitChangeFetcher(client, Utils.appendPath(root, "api/splitChanges"), metrics);
-    }
-
-    private HttpSplitChangeFetcher(CloseableHttpClient client, URI uri, Metrics metrics) {
+    private HttpSplitChangeFetcher(CloseableHttpClient client, URI uri, TelemetryRuntimeProducer telemetryRuntimeProducer) {
         _client = client;
         _target = uri;
-        _metrics = metrics;
         checkNotNull(_target);
+        _telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
     }
 
     @Override
     public SplitChange fetch(long since, boolean addCacheHeader) {
-
         long start = System.currentTimeMillis();
 
         CloseableHttpResponse response = null;
@@ -69,9 +68,11 @@ public SplitChange fetch(long since, boolean addCacheHeader) {
             int statusCode = response.getCode();
 
             if (statusCode < 200 || statusCode >= 300) {
-                _metrics.count(PREFIX + ".status." + statusCode, 1);
+                _telemetryRuntimeProducer.recordSyncError(ResourceEnum.SPLIT_SYNC, statusCode);
                 throw new IllegalStateException("Could not retrieve splitChanges; http return code " + statusCode);
             }
+            long endtime = System.currentTimeMillis();
+            _telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.SPLITS, endtime-start);
 
 
             String json = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
@@ -81,11 +82,9 @@ public SplitChange fetch(long since, boolean addCacheHeader) {
 
             return Json.fromJson(json, SplitChange.class);
         } catch (Throwable t) {
-            _metrics.count(PREFIX + ".exception", 1);
             throw new IllegalStateException("Problem fetching splitChanges: " + t.getMessage(), t);
         } finally {
             Utils.forceClose(response);
-            _metrics.time(PREFIX + ".time", System.currentTimeMillis() - start);
         }
     }
 

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

@@ -6,6 +6,8 @@
 import io.split.engine.SDKReadinessGates;
 import io.split.engine.evaluator.EvaluatorImp;
 import io.split.engine.metrics.Metrics;
+import io.split.telemetry.storage.InMemoryTelemetryStorage;
+import io.split.telemetry.storage.NoopTelemetryStorage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,7 +61,7 @@ public LocalhostSplitFactory(String directory, String file) throws IOException {
         _cacheUpdaterService.updateCache(splitAndKeyToTreatment);
         _client = new SplitClientImpl(this, splitCache,
                 new ImpressionsManager.NoOpImpressionsManager(), new NoopEventClient(),
-                SplitClientConfig.builder().setBlockUntilReadyTimeout(1).build(), sdkReadinessGates, new EvaluatorImp(splitCache));
+                SplitClientConfig.builder().setBlockUntilReadyTimeout(1).build(), sdkReadinessGates, new EvaluatorImp(splitCache), new NoopTelemetryStorage(), new NoopTelemetryStorage());
         _manager = LocalhostSplitManager.of(splitAndKeyToTreatment);
 
         _splitFile.registerWatcher();

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

@@ -21,7 +21,7 @@ public class SplitClientConfig {
     public static final String EVENTS_ENDPOINT = "https://events.split.io";
     public static final String AUTH_ENDPOINT = "https://auth.split.io/api/auth";
     public static final String STREAMING_ENDPOINT = "https://streaming.split.io/sse";
-    public static final String TELEMETRY_ENDPOINT = "https://telemetry.split.io/api/v1";
+    public static final String TELEMETRY_ENDPOINT = "https://telemetry.split-stage.io/api/v1/";
 
     private final String _endpoint;
     private final String _eventsEndpoint;

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

@@ -15,7 +15,9 @@
 import io.split.inputValidation.KeyValidator;
 import io.split.inputValidation.SplitNameValidator;
 import io.split.inputValidation.TrafficTypeValidator;
+import io.split.telemetry.domain.enums.LastSynchronizationRecordsEnum;
 import io.split.telemetry.domain.enums.MethodEnum;
+import io.split.telemetry.storage.TelemetryConfigProducer;
 import io.split.telemetry.storage.TelemetryEvaluationProducer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -49,6 +51,7 @@ public final class SplitClientImpl implements SplitClient {
     private final SDKReadinessGates _gates;
     private final Evaluator _evaluator;
     private final TelemetryEvaluationProducer _telemetryEvaluationProducer;
+    private final TelemetryConfigProducer _telemetryConfigProducer;
 
     public SplitClientImpl(SplitFactory container,
                            SplitCache splitCache,
@@ -57,7 +60,8 @@ public SplitClientImpl(SplitFactory container,
                            SplitClientConfig config,
                            SDKReadinessGates gates,
                            Evaluator evaluator,
-                           TelemetryEvaluationProducer telemetryEvaluationProducer) {
+                           TelemetryEvaluationProducer telemetryEvaluationProducer,
+                           TelemetryConfigProducer telemetryConfigProducer) {
         _container = container;
         _splitCache = checkNotNull(splitCache);
         _impressionManager = checkNotNull(impressionManager);
@@ -66,6 +70,7 @@ public SplitClientImpl(SplitFactory container,
         _gates = checkNotNull(gates);
         _evaluator = checkNotNull(evaluator);
         _telemetryEvaluationProducer = checkNotNull(telemetryEvaluationProducer);
+        _telemetryConfigProducer = checkNotNull(telemetryConfigProducer);
     }
 
     @Override
@@ -145,6 +150,7 @@ public void destroy() {
     }
 
     private boolean track(Event event) {
+        long initTime = System.currentTimeMillis();
         if (_container.isDestroyed()) {
             _log.error("Client has already been destroyed - no calls possible");
             return false;
@@ -174,15 +180,19 @@ private boolean track(Event event) {
         }
 
         event.properties = propertiesResult.getValue();
+        _telemetryEvaluationProducer.recordLatency(MethodEnum.TRACK, System.currentTimeMillis() - initTime);
 
         return _eventClient.track(event, propertiesResult.getEventSize());
     }
 
     private SplitResult getTreatmentWithConfigInternal(String method, String matchingKey, String bucketingKey, String split, Map<String, Object> attributes, MethodEnum methodEnum) {
         long initTime = System.currentTimeMillis();
         try {
+            if(!_gates.isSDKReadyNow()){
+                _log.warn(method + ": the SDK is not ready, results may be incorrect. Make sure to wait for SDK readiness before using this method");
+                _telemetryConfigProducer.recordNonReadyUsage();
+            }
             if (_container.isDestroyed()) {
-                _telemetryEvaluationProducer.recordException(methodEnum);
                 _log.error("Client has already been destroyed - no calls possible");
                 return SPLIT_RESULT_CONTROL;
             }