FFM-5251 - Merge FFM-5149 (flags out of sync) from 1.0.5.X to main (#117)

FFM-5251 - Merge FFM-5149 (flags out of sync) from 1.0.5.x to main

###What
SSE client should avoid using a read timeout of 0 since if the socket is not gracefully shutdown we have no way to detect that the socket is dead. Instead enforce a default read time out > 0 to force the connection to refresh and remove any stale socket. This change integrates the original bug fix FFM-5149 from 1.0.5.X into main. Main has previsouly been refactored and diverged so a 1:1 merge is not possible so a re-implementation has been done.

Additional changes:

- okhttp bumped from 3.9.1 to 4.9.0 to match 1.0.5.X version
- sseTimeout is now defined in HarnessConfig since Config was deprecated
- Fixed an issue where Poller would not restart on SSE disconnect due to the starttime not being set
- Added additional logging around socket events and ScheduledService states

###Why
This will allow the client to self heal in the event of a broken connection and avoid a situtation where the client gets out of sync with the FF server.

###Testing
Manual testing, checked client was able to recover its connection even after the socket was broken on purpose.

Author: andybharness

pom.xml

@@ -81,12 +81,12 @@
         <dependency>
             <groupId>com.squareup.okhttp3</groupId>
             <artifactId>okhttp</artifactId>
-            <version>3.9.1</version>
+            <version>4.9.0</version>
         </dependency>
         <dependency>
             <groupId>com.squareup.okhttp3</groupId>
             <artifactId>logging-interceptor</artifactId>
-            <version>3.9.1</version>
+            <version>4.9.0</version>
         </dependency>
         <dependency>
             <groupId>org.apache.maven</groupId>

src/main/java/io/harness/cf/client/api/InnerClient.java

@@ -111,6 +111,8 @@ protected void setUp(@NonNull final Connector connector, @NonNull final BaseConf
     metricsProcessor = new MetricsProcessor(this.connector, this.options, this);
     updateProcessor = new UpdateProcessor(this.connector, this.repository, this);
 
+    pollerStartedAt = new Date();
+
     // start with authentication
     authService.startAsync();
   }
@@ -205,13 +207,15 @@ public synchronized void onMetricsFailure() {
 
   @Override
   public void onConnected() {
+    log.info("onConnected triggered");
     if (pollProcessor.state() == Service.State.RUNNING) {
       pollProcessor.stop();
     }
   }
 
   @Override
   public void onDisconnected() {
+    log.info("onDisconnected triggered, starting poller to get latest flags");
     // onDisconnected can be called multiple times from updater because of retries
     // and we cannot create many poller instances so we need to check if
     // on closing the client, state of the poller and when poller is last time started
@@ -227,32 +231,44 @@ public void onDisconnected() {
           new PollingProcessor(connector, repository, options.getPollIntervalInSeconds(), this);
       pollProcessor.start();
       pollerStartedAt = new Date();
+    } else {
+      log.warn(
+          "Poller was not restarted [closing={} terminated={} pollStartTime+interval={} now={} ]",
+          closing,
+          pollProcessor.state() == Service.State.TERMINATED,
+          instant,
+          now);
     }
   }
 
   @Override
   public void onReady() {
+    log.info("onReady triggered");
     initialize(Processor.STREAM);
   }
 
   @Override
   public void onError() {
+    log.info("onError triggered");
     // when error happens on updater (stream)
     onDisconnected();
   }
 
   @Override
   public synchronized void onFailure(@NonNull final String error) {
+    log.info("onFailure triggered [error={}] ", error);
     failure = true;
     notifyAll();
   }
 
   @Override
   public void update(@NonNull final Message message) {
+    log.info("update triggered [event={}] ", message.getEvent());
     updateProcessor.update(message);
   }
 
   public void update(@NonNull final Message message, final boolean manual) {
+    log.info("update triggered [event={} manual={}] ", message.getEvent(), manual);
     if (options.isStreamEnabled() && manual) {
       log.warn(
           "You have run update method manually with the stream enabled. Please turn off the stream in this case.");

src/main/java/io/harness/cf/client/api/PollingProcessor.java

@@ -1,6 +1,8 @@
 package io.harness.cf.client.api;
 
 import com.google.common.util.concurrent.AbstractScheduledService;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.harness.cf.client.common.ScheduledServiceStateLogger;
 import io.harness.cf.client.connector.Connector;
 import io.harness.cf.client.connector.ConnectorException;
 import io.harness.cf.model.FeatureConfig;
@@ -29,6 +31,9 @@ public PollingProcessor(
     this.pollIntervalSeconds = pollIntervalSeconds;
     this.repository = repository;
     this.callback = callback;
+    this.addListener(
+        new ScheduledServiceStateLogger(PollingProcessor.class.getSimpleName()),
+        MoreExecutors.directExecutor());
   }
 
   public CompletableFuture<List<FeatureConfig>> retrieveFlags() {

src/main/java/io/harness/cf/client/common/ScheduledServiceStateLogger.java

@@ -0,0 +1,37 @@
+package io.harness.cf.client.common;
+
+import com.google.common.util.concurrent.Service;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@AllArgsConstructor
+public class ScheduledServiceStateLogger extends Service.Listener {
+  private final String name;
+
+  @Override
+  public void starting() {
+    log.info("{}: ScheduledService starting", name);
+  }
+
+  @Override
+  public void running() {
+    log.info("{}: ScheduledService running", name);
+  }
+
+  @Override
+  public void stopping(Service.State from) {
+    log.info("{}: ScheduledService stopping [from={}]", name, from);
+  }
+
+  @Override
+  public void terminated(Service.State from) {
+    log.info("{}: ScheduledService terminated [from={}]", name, from);
+  }
+
+  @Override
+  public void failed(Service.State from, Throwable failure) {
+    log.warn("{}: ScheduledService failed [from={} message={}]", name, from, failure.getMessage());
+    log.warn(name + ": ScheduledService failed exception", failure);
+  }
+}

src/main/java/io/harness/cf/client/connector/EventSource.java

@@ -32,19 +32,25 @@ public class EventSource implements ServerSentEvent.Listener, AutoCloseable, Ser
     LogUtil.setSystemProps();
   }
 
-  public EventSource(@NonNull String url, Map<String, String> headers, @NonNull Updater updater) {
+  public EventSource(
+      @NonNull String url,
+      Map<String, String> headers,
+      @NonNull Updater updater,
+      long sseReadTimeoutMins) {
     this.updater = updater;
-    okSse = new OkSse(makeStreamClient());
+    okSse = new OkSse(makeStreamClient(sseReadTimeoutMins));
     builder = new Request.Builder().url(url);
     headers.put("User-Agent", "JavaSDK " + io.harness.cf.Version.VERSION);
     headers.forEach(builder::header);
     updater.onReady();
     log.info("EventSource initialized with url {} and headers {}", url, headers);
   }
 
-  protected OkHttpClient makeStreamClient() {
+  protected OkHttpClient makeStreamClient(long sseReadTimeoutMins) {
     OkHttpClient.Builder httpClientBuilder =
-        new OkHttpClient.Builder().readTimeout(0L, TimeUnit.SECONDS).retryOnConnectionFailure(true);
+        new OkHttpClient.Builder()
+            .readTimeout(sseReadTimeoutMins, TimeUnit.MINUTES)
+            .retryOnConnectionFailure(true);
     if (log.isDebugEnabled()) {
       loggingInterceptor = new HttpLoggingInterceptor();
       loggingInterceptor.setLevel(HttpLoggingInterceptor.Level.BODY);
@@ -71,6 +77,7 @@ protected OkHttpClient makeStreamClient() {
 
   @Override
   public void onOpen(ServerSentEvent serverSentEvent, Response response) {
+    log.info("EventSource onOpen");
     if (updater != null) {
       log.info("EventSource connected!");
       updater.onConnected();
@@ -79,7 +86,7 @@ public void onOpen(ServerSentEvent serverSentEvent, Response response) {
 
   @Override
   public void onMessage(ServerSentEvent sse, String id, String event, String message) {
-    log.info("EventSource message received {}", message);
+    log.info("EventSource onMessage {}", message);
     Message msg = gson.fromJson(message, Message.class);
     updater.update(msg);
   }
@@ -98,7 +105,11 @@ public boolean onRetryTime(ServerSentEvent serverSentEvent, long l) {
   @Override
   public boolean onRetryError(
       ServerSentEvent serverSentEvent, Throwable throwable, Response response) {
-    log.warn("EventSource onRetryError");
+    log.warn(
+        "EventSource onRetryError [throwable={} message={}]",
+        throwable.getClass().getSimpleName(),
+        throwable.getMessage());
+    log.trace("onRetryError exception", throwable);
     updater.onError();
     if (response != null) {
       return response.code() == 429 || response.code() >= 500;
@@ -108,7 +119,7 @@ public boolean onRetryError(
 
   @Override
   public void onClosed(ServerSentEvent serverSentEvent) {
-    log.info("EventSource disconnected");
+    log.info("EventSource onClosed - disconnected");
     updater.onDisconnected();
   }
 

src/main/java/io/harness/cf/client/connector/HarnessConfig.java

@@ -20,4 +20,7 @@ public class HarnessConfig {
   @Builder.Default int readTimeout = 30000;
   /** timeout in milliseconds for writing data to CF Server */
   @Builder.Default int writeTimeout = 10000;
+
+  /** read timeout in minutes for SSE connections */
+  @Builder.Default long sseReadTimeout = 1;
 }

src/main/java/io/harness/cf/client/connector/HarnessConnector.java

@@ -221,6 +221,7 @@ public List<FeatureConfig> getFlags() throws ConnectorException {
           featureConfig.size(),
           this.environment,
           this.cluster);
+      log.info("Got the following features: " + featureConfig);
       return featureConfig;
     } catch (ApiException e) {
       log.error(
@@ -355,7 +356,7 @@ public Service stream(@NonNull final Updater updater) {
     map.put("Authorization", "Bearer " + token);
     map.put("API-Key", apiKey);
     log.info("Initialize new EventSource instance");
-    eventSource = new EventSource(sseUrl, map, updater);
+    eventSource = new EventSource(sseUrl, map, updater, Math.max(options.getSseReadTimeout(), 1));
     return eventSource;
   }