Fixing PR Comments

Author: ldecheverz-split

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

@@ -124,7 +124,7 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         // Cache Initialisations
         _segmentCache = new SegmentCacheInMemoryImpl();
         _splitCache = new InMemoryCacheImp();
-        _telemetrySynchronizer = new SynchronizerMemory(_httpclient, URI.create(config.get_telemetryURL()), _telemetryStorage, _splitCache, _segmentCache);
+        _telemetrySynchronizer = new SynchronizerMemory(_httpclient, URI.create(config.get_telemetryURL()), _telemetryStorage, _splitCache, _segmentCache, _telemetryStorage);
         _telemetrySyncTask = new TelemetrySyncTask(config.get_telemetryRefreshRate(), _telemetrySynchronizer);
 
         // Segments
@@ -185,14 +185,14 @@ public synchronized void destroy() {
                 _log.info("Successful shutdown of splits");
                 _impressionsManager.close();
                 _log.info("Successful shutdown of impressions manager");
-                _httpclient.close();
-                _log.info("Successful shutdown of httpclient");
                 _eventClient.close();
                 _log.info("Successful shutdown of eventClient");
                 _syncManager.shutdown();
                 _log.info("Successful shutdown of syncManager");
                 _telemetryStorage.recordSessionLength(System.currentTimeMillis() - _startTime);
                 _telemetrySyncTask.stopScheduledTask();
+                _httpclient.close();
+                _log.info("Successful shutdown of httpclient");
             } catch (IOException e) {
                 _log.error("We could not shutdown split", e);
             }

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

@@ -16,6 +16,8 @@
 import io.split.engine.sse.workers.SplitsWorkerImp;
 import io.split.engine.sse.workers.Worker;
 
+import io.split.telemetry.domain.StreamingEvent;
+import io.split.telemetry.domain.enums.StreamEventsEnum;
 import io.split.telemetry.storage.TelemetryRuntimeProducer;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.slf4j.Logger;
@@ -42,13 +44,15 @@ public class PushManagerImp implements PushManager {
     private Future<?> _nextTokenRefreshTask;
     private final ScheduledExecutorService _scheduledExecutorService;
     private AtomicLong _expirationTime;
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
     @VisibleForTesting
     /* package private */ PushManagerImp(AuthApiClient authApiClient,
                                              EventSourceClient eventSourceClient,
                                              SplitsWorker splitsWorker,
                                              Worker<SegmentQueueDto> segmentWorker,
-                                             PushStatusTracker pushStatusTracker) {
+                                             PushStatusTracker pushStatusTracker,
+                                            TelemetryRuntimeProducer telemetryRuntimeProducer) {
 
         _authApiClient = checkNotNull(authApiClient);
         _eventSourceClient = checkNotNull(eventSourceClient);
@@ -60,6 +64,7 @@ public class PushManagerImp implements PushManager {
                 .setDaemon(true)
                 .setNameFormat("Split-SSERefreshToken-%d")
                 .build());
+        _telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
     }
 
     public static PushManagerImp build(Synchronizer synchronizer,
@@ -71,12 +76,12 @@ public static PushManagerImp build(Synchronizer synchronizer,
                                        TelemetryRuntimeProducer telemetryRuntimeProducer) {
         SplitsWorker splitsWorker = new SplitsWorkerImp(synchronizer);
         Worker<SegmentQueueDto> segmentWorker = new SegmentsWorkerImp(synchronizer);
-        PushStatusTracker pushStatusTracker = new PushStatusTrackerImp(statusMessages);
+        PushStatusTracker pushStatusTracker = new PushStatusTrackerImp(statusMessages, telemetryRuntimeProducer);
         return new PushManagerImp(new AuthApiClientImp(authUrl, httpClient, telemetryRuntimeProducer),
-                EventSourceClientImp.build(streamingUrl, splitsWorker, segmentWorker, pushStatusTracker, sseHttpClient),
+                EventSourceClientImp.build(streamingUrl, splitsWorker, segmentWorker, pushStatusTracker, sseHttpClient, telemetryRuntimeProducer),
                 splitsWorker,
                 segmentWorker,
-                pushStatusTracker);
+                pushStatusTracker, telemetryRuntimeProducer);
     }
 
     @Override
@@ -85,6 +90,7 @@ public synchronized void start() {
         _log.debug(String.format("Auth service response pushEnabled: %s", response.isPushEnabled()));
         if (response.isPushEnabled() && startSse(response.getToken(), response.getChannels())) {
             _expirationTime.set(response.getExpiration());
+            _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.TOKEN_REFRESH.get_type(), response.getExpiration(), System.currentTimeMillis()));
             return;
         }
 

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

@@ -8,6 +8,8 @@
 import io.split.engine.experiments.SplitFetcher;
 import io.split.engine.experiments.SplitSynchronizationTask;
 import io.split.engine.segments.SegmentSynchronizationTaskImp;
+import io.split.telemetry.domain.StreamingEvent;
+import io.split.telemetry.domain.enums.StreamEventsEnum;
 import io.split.telemetry.storage.TelemetryRuntimeProducer;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.slf4j.Logger;
@@ -23,6 +25,8 @@
 
 public class SyncManagerImp implements SyncManager {
     private static final Logger _log = LoggerFactory.getLogger(SyncManager.class);
+    private static final int STREAMING_STREAMING_EVENT = 0;
+    private static final int POLLING_STREAMING_EVENT = 1;
 
     private final AtomicBoolean _streamingEnabledConfig;
     private final Synchronizer _synchronizer;
@@ -34,14 +38,15 @@ public class SyncManagerImp implements SyncManager {
     private final SDKReadinessGates _gates;
     private Future<?> _pushStatusMonitorTask;
     private Backoff _backoff;
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
     @VisibleForTesting
     /* package private */ SyncManagerImp(boolean streamingEnabledConfig,
                                          Synchronizer synchronizer,
                                          PushManager pushManager,
                                          LinkedBlockingQueue<PushManager.Status> pushMessages,
                                          int authRetryBackOffBase,
-                                         SDKReadinessGates gates) {
+                                         SDKReadinessGates gates, TelemetryRuntimeProducer telemetryRuntimeProducer) {
         _streamingEnabledConfig = new AtomicBoolean(streamingEnabledConfig);
         _synchronizer = checkNotNull(synchronizer);
         _pushManager = checkNotNull(pushManager);
@@ -57,6 +62,7 @@ public class SyncManagerImp implements SyncManager {
                 .build());
         _backoff = new Backoff(authRetryBackOffBase);
         _gates = checkNotNull(gates);
+        _telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
     }
 
     public static SyncManagerImp build(boolean streamingEnabledConfig,
@@ -76,13 +82,12 @@ public static SyncManagerImp build(boolean streamingEnabledConfig,
         LinkedBlockingQueue<PushManager.Status> pushMessages = new LinkedBlockingQueue<>();
         Synchronizer synchronizer = new SynchronizerImp(splitSynchronizationTask, splitFetcher, segmentSynchronizationTaskImp, splitCache, segmentCache, streamingRetryDelay, gates);
         PushManager pushManager = PushManagerImp.build(synchronizer, streamingServiceUrl, authUrl, httpClient, pushMessages, sseHttpClient, telemetryRuntimeProducer);
-        return new SyncManagerImp(streamingEnabledConfig, synchronizer, pushManager, pushMessages, authRetryBackOffBase, gates);
+        return new SyncManagerImp(streamingEnabledConfig, synchronizer, pushManager, pushMessages, authRetryBackOffBase, gates, telemetryRuntimeProducer);
     }
 
     @Override
     public void start() {
         _synchronizer.syncAll();
-
         if (_streamingEnabledConfig.get()) {
             startStreamingMode();
         } else {
@@ -103,6 +108,7 @@ private void startStreamingMode() {
             _pushStatusMonitorTask = _executorService.submit(this::incomingPushStatusHandler);
         }
         _pushManager.start();
+        _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.SYNC_MODE_UPDATE.get_type(), STREAMING_STREAMING_EVENT, System.currentTimeMillis()));
     }
 
     private void startPollingMode() {
@@ -114,6 +120,7 @@ private void startPollingMode() {
 
         _log.debug("Starting in polling mode ...");
         _synchronizer.startPeriodicFetching();
+        _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.SYNC_MODE_UPDATE.get_type(), POLLING_STREAMING_EVENT, System.currentTimeMillis()));
     }
 
     @VisibleForTesting

client/src/main/java/io/split/engine/sse/EventSourceClientImp.java

@@ -7,6 +7,7 @@
 import io.split.engine.sse.exceptions.EventParsingException;
 import io.split.engine.sse.workers.SplitsWorker;
 import io.split.engine.sse.workers.Worker;
+import io.split.telemetry.storage.TelemetryRuntimeProducer;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.core5.net.URIBuilder;
 import org.slf4j.Logger;
@@ -35,7 +36,8 @@ public class EventSourceClientImp implements EventSourceClient {
                                                NotificationParser notificationParser,
                                                NotificationProcessor notificationProcessor,
                                                PushStatusTracker pushStatusTracker,
-                                               CloseableHttpClient sseHttpClient) {
+                                               CloseableHttpClient sseHttpClient,
+                                               TelemetryRuntimeProducer telemetryRuntimeProducer) {
         _baseStreamingUrl = checkNotNull(baseStreamingUrl);
         _notificationParser = checkNotNull(notificationParser);
         _notificationProcessor = checkNotNull(notificationProcessor);
@@ -44,20 +46,21 @@ public class EventSourceClientImp implements EventSourceClient {
         _sseClient = new SSEClient(
                 inboundEvent -> { onMessage(inboundEvent); return null; },
                 status -> { _pushStatusTracker.handleSseStatus(status); return null; },
-                sseHttpClient);
+                sseHttpClient, telemetryRuntimeProducer);
         _firstEvent = new AtomicBoolean();
     }
 
     public static EventSourceClientImp build(String baseStreamingUrl,
                                              SplitsWorker splitsWorker,
                                              Worker<SegmentQueueDto> segmentWorker,
                                              PushStatusTracker pushStatusTracker,
-                                             CloseableHttpClient sseHttpClient) {
+                                             CloseableHttpClient sseHttpClient, TelemetryRuntimeProducer telemetryRuntimeProducer) {
         return new EventSourceClientImp(baseStreamingUrl,
                 new NotificationParserImp(),
                 NotificationProcessorImp.build(splitsWorker, segmentWorker, pushStatusTracker),
                 pushStatusTracker,
-                sseHttpClient);
+                sseHttpClient,
+                telemetryRuntimeProducer);
     }
 
     @Override

client/src/main/java/io/split/engine/sse/PushStatusTrackerImp.java

@@ -7,25 +7,39 @@
 import io.split.engine.sse.dtos.ControlType;
 import io.split.engine.sse.dtos.ErrorNotification;
 import io.split.engine.sse.dtos.OccupancyNotification;
+import io.split.telemetry.domain.StreamingEvent;
+import io.split.telemetry.domain.enums.StreamEventsEnum;
+import io.split.telemetry.storage.TelemetryRuntimeProducer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.TimeZone;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+
 public class PushStatusTrackerImp implements PushStatusTracker {
     private static final Logger _log = LoggerFactory.getLogger(PushStatusTracker.class);
+    private static final String CONTROL_PRI_CHANNEL = "control_pri";
+    private static final String CONTROL_SEC_CHANNEL = "control_sec";
+    private static final int STREAMING_DISABLED = 0;
+    private static final int STREAMING_ENABLED = 1;
+    private static final int STREAMING_PAUSED = 2;
 
     private final AtomicBoolean _publishersOnline = new AtomicBoolean(true);
     private final AtomicReference<SSEClient.StatusMessage> _sseStatus = new AtomicReference<>(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS);
     private final AtomicReference<ControlType> _backendStatus = new AtomicReference<>(ControlType.STREAMING_RESUMED);
     private final LinkedBlockingQueue<PushManager.Status> _statusMessages;
     private final ConcurrentMap<String, Integer> regions = Maps.newConcurrentMap();
 
-    public PushStatusTrackerImp(LinkedBlockingQueue<PushManager.Status> statusMessages) {
+    private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
+
+    public PushStatusTrackerImp(LinkedBlockingQueue<PushManager.Status> statusMessages, TelemetryRuntimeProducer telemetryRuntimeProducer) {
         _statusMessages = statusMessages;
+        _telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
     }
 
     private synchronized void reset() {
@@ -42,6 +56,8 @@ public void handleSseStatus(SSEClient.StatusMessage newStatus) {
             case FIRST_EVENT:
                 if (SSEClient.StatusMessage.CONNECTED.equals(_sseStatus.get())) {
                     _statusMessages.offer(PushManager.Status.STREAMING_READY);
+                    _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.STREAMING_STATUS.get_type(), STREAMING_ENABLED, System.currentTimeMillis()));
+                    _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.CONNECTION_ESTABLISHED.get_type(),0l, System.currentTimeMillis()));
                 }
             case CONNECTED:
                 _sseStatus.compareAndSet(SSEClient.StatusMessage.INITIALIZATION_IN_PROGRESS, SSEClient.StatusMessage.CONNECTED);
@@ -85,12 +101,14 @@ public void handleIncomingControlEvent(ControlNotification controlNotification)
                 }
                 break;
             case STREAMING_PAUSED:
+                _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.STREAMING_STATUS.get_type(), STREAMING_PAUSED, System.currentTimeMillis()));
                 if (_backendStatus.compareAndSet(ControlType.STREAMING_RESUMED, ControlType.STREAMING_PAUSED) && _publishersOnline.get()) {
                     // If there are no publishers online, the STREAMING_DOWN message should have already been sent
                     _statusMessages.offer(PushManager.Status.STREAMING_DOWN);
                 }
                 break;
             case STREAMING_DISABLED:
+                _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.STREAMING_STATUS.get_type(), STREAMING_DISABLED, System.currentTimeMillis()));
                 _backendStatus.set(ControlType.STREAMING_DISABLED);
                 _statusMessages.offer(PushManager.Status.STREAMING_OFF);
                 break;
@@ -102,6 +120,7 @@ public void handleIncomingOccupancyEvent(OccupancyNotification occupancyNotifica
         _log.debug(String.format("handleIncomingOccupancyEvent: publishers=%d", occupancyNotification.getMetrics().getPublishers()));
 
         int publishers = occupancyNotification.getMetrics().getPublishers();
+        recordTelemetryOcuppancy(occupancyNotification, publishers);
         regions.put(occupancyNotification.getChannel(), publishers);
         boolean isPublishers = isPublishers();
         if (!isPublishers && _publishersOnline.compareAndSet(true, false) && _backendStatus.get().equals(ControlType.STREAMING_RESUMED)) {
@@ -114,7 +133,7 @@ public void handleIncomingOccupancyEvent(OccupancyNotification occupancyNotifica
     @Override
     public void handleIncomingAblyError(ErrorNotification notification) {
         _log.debug(String.format("handleIncomingAblyError: %s", notification.getMessage()));
-
+        _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.ABLY_ERROR.get_type(), notification.getCode(), System.currentTimeMillis()));
         if (_backendStatus.get().equals(ControlType.STREAMING_DISABLED)) {
             return; // Ignore
         }
@@ -145,4 +164,14 @@ private boolean isPublishers() {
         }
         return false;
     }
+
+    private void recordTelemetryOcuppancy(OccupancyNotification occupancyNotification, int publishers) {
+        if (CONTROL_PRI_CHANNEL.equals(occupancyNotification.getChannel())) {
+            _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.OCCUPANCY_PRI.get_type(), publishers, System.currentTimeMillis()));
+        }
+        else if (CONTROL_SEC_CHANNEL.equals(occupancyNotification.getChannel())){
+            _telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.OCCUPANCY_SEC.get_type(), publishers, System.currentTimeMillis()));
+        }
+
+    }
 }