fix: improves FDv2 DataSource logging

Author: tanderson-ld

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/DataSourceSynchronizerAdapter.java

@@ -61,6 +61,11 @@ public DataSourceSynchronizerAdapter(DataSourceFactory dataSourceFactory) {
         this.dataSource = dataSourceFactory.create(convertingSink);
     }
 
+    @Override
+    public String name() {
+        return "AdaptedSynchronizer(FDv1)";
+    }
+
     @Override
     public CompletableFuture<FDv2SourceResult> next() {
         synchronized (startLock) {

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSource.java

@@ -2,6 +2,8 @@
 
 import com.google.common.collect.ImmutableList;
 import com.launchdarkly.logging.LDLogger;
+import com.launchdarkly.logging.LogValues;
+import com.launchdarkly.sdk.fdv2.SourceSignal;
 import com.launchdarkly.sdk.server.datasources.FDv2SourceResult;
 import com.launchdarkly.sdk.server.datasources.Initializer;
 import com.launchdarkly.sdk.server.datasources.Synchronizer;
@@ -48,6 +50,13 @@ class FDv2DataSource implements DataSource {
 
     private volatile boolean closed = false;
 
+    /**
+     * Avoid duplicate orchestration logs for the same synchronizer and {@link SourceSignal} (DATASYSTEM 1.6.6).
+     */
+    private String lastLoggedSynchronizerDedupeName;
+
+    private SourceSignal lastLoggedSynchronizerDedupeStatus;
+
     public interface DataSourceFactory<T> {
         T build();
     }
@@ -73,7 +82,6 @@ public FDv2DataSource(
         );
     }
 
-
     public FDv2DataSource(
         ImmutableList<DataSourceFactory<Initializer>> initializers,
         ImmutableList<DataSourceFactory<Synchronizer>> synchronizers,
@@ -117,7 +125,9 @@ private void run() {
             if (!sourceManager.hasAvailableSources()) {
                 // There are not any initializer or synchronizers, so we are at the best state that
                 // can be achieved.
-                logger.info("LaunchDarkly client will not connect to Launchdarkly for feature flag data due to no initializers or synchronizers");
+                logger.warn(
+                    "LaunchDarkly client will not connect to LaunchDarkly for feature flag data due to no initializers or synchronizers configured."
+                );
                 dataSourceUpdates.updateStatus(DataSourceStatusProvider.State.VALID, null);
                 startFuture.complete(true);
                 return;
@@ -157,13 +167,16 @@ private void run() {
     private void runInitializers() {
         boolean anyDataReceived = false;
         Initializer initializer = sourceManager.getNextInitializerAndSetActive();
-        while(initializer != null) {
+        while (initializer != null) {
+            String initializerName = initializer.name();
+            logger.info("Initializer '{}' is starting.", initializerName);
             try {
-                try(FDv2SourceResult result = initializer.run().get()) {
+                try (FDv2SourceResult result = initializer.run().get()) {
                     switch (result.getResultType()) {
                         case CHANGE_SET:
                             dataSourceUpdates.apply(result.getChangeSet());
                             anyDataReceived = true;
+                            logger.info("Initialized via '{}'.", initializerName);
                             if (!result.getChangeSet().getSelector().isEmpty()) {
                                 // We received data with a selector, so we end the initialization process.
                                 dataSourceUpdates.updateStatus(DataSourceStatusProvider.State.VALID, null);
@@ -176,6 +189,9 @@ private void runInitializers() {
                             switch (status.getState()) {
                                 case INTERRUPTED:
                                 case TERMINAL_ERROR:
+                                    logger.warn("Initializer '{}' failed: {}",
+                                        initializerName,
+                                        detailForError(status.getErrorInfo()));
                                     // The data source updates handler will filter the state during initializing, but this
                                     // will make the error information available.
                                     dataSourceUpdates.updateStatus(
@@ -194,8 +210,6 @@ private void runInitializers() {
                     }
                 }
             } catch (ExecutionException | InterruptedException | CancellationException e) {
-                // We don't expect these conditions to happen in practice.
-
                 // The data source updates handler will filter the state during initializing, but this
                 // will make the error information available.
                 dataSourceUpdates.updateStatus(
@@ -204,9 +218,12 @@ private void runInitializers() {
                         0,
                         e.toString(),
                         new Date().toInstant()));
-                logger.warn("Error running initializer: {}", e.toString());
+                Throwable root = e instanceof ExecutionException && e.getCause() != null ? e.getCause() : e;
+                logger.error("Error running initializer '{}': {}",
+                    initializerName,
+                    root.getMessage() != null ? root.getMessage() : LogValues.exceptionSummary(root));
             }
-        initializer = sourceManager.getNextInitializerAndSetActive();
+            initializer = sourceManager.getNextInitializerAndSetActive();
         }
         // We received data without a selector, and we have exhausted initializers, so we are going to
         // consider ourselves initialized.
@@ -249,6 +266,9 @@ private void runSynchronizers() {
 
             // We want to continue running synchronizers for as long as any are available.
             while (synchronizer != null) {
+                String synchronizerName = synchronizer.name();
+                logger.info("Synchronizer '{}' is starting.", synchronizerName);
+                resetSynchronizerStatusDedupe();
                 try {
                     boolean running = true;
 
@@ -264,16 +284,19 @@ private void runSynchronizers() {
                                 Condition c = (Condition) res;
                                 switch (c.getType()) {
                                     case FALLBACK:
-                                        // For fallback, we will move to the next available synchronizer, which may loop.
-                                        // This is the default behavior of exiting the run loop, so we don't need to take
-                                        // any action.
-                                        logger.debug("A synchronizer has experienced an interruption and we are falling back.");
+                                        logger.info(
+                                            "Fallback condition met, falling back from synchronizer '{}'.",
+                                            synchronizer.name()
+                                        );
                                         break;
                                     case RECOVERY:
                                         // For recovery, we will start at the first available synchronizer.
                                         // So we reset the source index, and finding the source will start at the beginning.
                                         sourceManager.resetSourceIndex();
-                                        logger.debug("The data source is attempting to recover to a higher priority synchronizer.");
+                                        logger.info(
+                                            "Recovery condition met, moving from synchronizer '{}' to primary synchronizer.",
+                                            synchronizer.name()
+                                        );
                                         break;
                                 }
                                 // A running synchronizer will only have fallback and recovery conditions that it can act on.
@@ -292,6 +315,8 @@ private void runSynchronizers() {
 
                                 switch (result.getResultType()) {
                                     case CHANGE_SET:
+                                        // A data update breaks the "in a row" streak for status deduplication (DATASYSTEM 1.6.6).
+                                        resetSynchronizerStatusDedupe();
                                         dataSourceUpdates.apply(result.getChangeSet());
                                         dataSourceUpdates.updateStatus(DataSourceStatusProvider.State.VALID, null);
                                         // This could have been completed by any data source. But if it has not been completed before
@@ -302,17 +327,33 @@ private void runSynchronizers() {
                                         FDv2SourceResult.Status status = result.getStatus();
                                         switch (status.getState()) {
                                             case INTERRUPTED:
+                                                maybeLogSynchronizerStatusChange(
+                                                    synchronizer.name(),
+                                                    status.getState()
+                                                );
                                                 // Handled by conditions.
                                                 dataSourceUpdates.updateStatus(
                                                     DataSourceStatusProvider.State.INTERRUPTED,
                                                     status.getErrorInfo());
                                                 break;
                                             case SHUTDOWN:
+                                                maybeLogSynchronizerStatusChange(
+                                                    synchronizer.name(),
+                                                    status.getState()
+                                                );
                                                 // We should be overall shutting down.
                                                 logger.debug("Synchronizer shutdown.");
                                                 return;
                                             case TERMINAL_ERROR:
+                                                maybeLogSynchronizerStatusChange(
+                                                    synchronizer.name(),
+                                                    status.getState()
+                                                );
                                                 sourceManager.blockCurrentSynchronizer();
+                                                logger.warn(
+                                                    "Synchronizer '{}' permanently failed and will not be used again until application restart.",
+                                                    synchronizer.name()
+                                                );
                                                 running = false;
                                                 dataSourceUpdates.updateStatus(
                                                     DataSourceStatusProvider.State.INTERRUPTED,
@@ -335,6 +376,7 @@ private void runSynchronizers() {
                                         !sourceManager.isCurrentSynchronizerFDv1Fallback()
                                 ) {
                                     sourceManager.fdv1Fallback();
+                                    logger.info("Falling back to an FDv1 fallback synchronizer.");
                                     running = false;
                                 }
                             }
@@ -348,20 +390,55 @@ private void runSynchronizers() {
                             e.toString(),
                             new Date().toInstant()
                         ));
-                    logger.warn("Error running synchronizer: {}, will try next synchronizer, or retry.", e.toString());
+                    Throwable root = e instanceof ExecutionException && e.getCause() != null ? e.getCause() : e;
+                    logger.error("Error running synchronizer '{}': {}",
+                        synchronizer.name(),
+                        root.getMessage() != null ? root.getMessage() : LogValues.exceptionSummary(root));
                     // Move to the next synchronizer.
                 }
                 // Get the next available synchronizer and set it active
                 synchronizer = sourceManager.getNextAvailableSynchronizerAndSetActive();
             }
-        } catch(Exception e) {
+            if (!closed) {
+                logger.warn("No more synchronizers available.");
+            }
+        } catch (Exception e) {
             // We are not expecting to encounter this situation, but if we do, then we should log it.
             logger.error("Unexpected error in data source: {}", e.toString());
-        }finally {
+        } finally {
             sourceManager.close();
         }
     }
 
+    private static String detailForError(DataSourceStatusProvider.ErrorInfo errorInfo) {
+        if (errorInfo == null) {
+            return "unknown error";
+        }
+        if (errorInfo.getMessage() != null && !errorInfo.getMessage().isEmpty()) {
+            return errorInfo.getMessage();
+        }
+        return errorInfo.toString();
+    }
+
+    private void resetSynchronizerStatusDedupe() {
+        lastLoggedSynchronizerDedupeName = null;
+        lastLoggedSynchronizerDedupeStatus = null;
+    }
+
+    private void maybeLogSynchronizerStatusChange(String sourceName, SourceSignal state) {
+        if (state == SourceSignal.GOODBYE) {
+            return;
+        }
+        if (sourceName != null
+            && sourceName.equals(lastLoggedSynchronizerDedupeName)
+            && state == lastLoggedSynchronizerDedupeStatus) {
+            return;
+        }
+        lastLoggedSynchronizerDedupeName = sourceName;
+        lastLoggedSynchronizerDedupeStatus = state;
+        logger.info("Synchronizer '{}' reported status: {}.", sourceName, state.name());
+    }
+
     @Override
     public Future<Void> start() {
         if (!started.getAndSet(true)) {

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingInitializerImpl.java

@@ -16,6 +16,11 @@ public PollingInitializerImpl(FDv2Requestor requestor, LDLogger logger, Selector
         this.selectorSource = selectorSource;
     }
 
+    @Override
+    public String name() {
+        return "PollingInitializer(FDv2)";
+    }
+
     @Override
     public CompletableFuture<FDv2SourceResult> run() {
         CompletableFuture<FDv2SourceResult> pollResult = poll(selectorSource.getSelector(), true);

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/PollingSynchronizerImpl.java

@@ -36,6 +36,11 @@ public PollingSynchronizerImpl(
         }
     }
 
+    @Override
+    public String name() {
+        return "PollingSynchronizer(FDv2)";
+    }
+
     private void doPoll() {
         try {
             FDv2SourceResult res = poll(selectorSource.getSelector(), false).get();

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/StreamingSynchronizerImpl.java

@@ -94,6 +94,11 @@ public StreamingSynchronizerImpl(
         // The stream will lazily start when `next` is called.
     }
 
+    @Override
+    public String name() {
+        return "StreamingSynchronizer(FDv2)";
+    }
+
     private void startStream() {
         Headers headers = httpProperties.toHeadersBuilder()
                 .add("Accept", "text/event-stream")

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/datasources/Initializer.java

@@ -51,4 +51,16 @@ public interface Initializer extends Closeable {
      * @return The result of the initializer.
      */
     CompletableFuture<FDv2SourceResult> run();
+
+    /**
+     * Human-readable name for logging and diagnostics. Do not use this for influencing code behavior.
+     * <p>
+     * Implementations may override; the default uses the runtime class simple name.
+     *
+     * @return the name
+     */
+    default String name() {
+        String simple = getClass().getSimpleName();
+        return simple.isEmpty() ? getClass().getName() : simple;
+    }
 }

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/datasources/Synchronizer.java

@@ -57,4 +57,16 @@ public interface Synchronizer extends Closeable {
      * @return a future that will complete when the next result is available
      */
     CompletableFuture<FDv2SourceResult> next();
+
+    /**
+     * Human-readable name for logging and diagnostics. Do not use this for influencing code behavior.
+     * <p>
+     * Implementations may override; the default uses the runtime class simple name.
+     *
+     * @return the name
+     */
+    default String name() {
+        String simple = getClass().getSimpleName();
+        return simple.isEmpty() ? getClass().getName() : simple;
+    }
 }

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/integrations/FileInitializer.java

@@ -33,6 +33,11 @@ final class FileInitializer implements Initializer {
         this.synchronizer = new FileSynchronizer(sources, false, duplicateKeysHandling, logger, persist);
     }
 
+    @Override
+    public String name() {
+        return "FileInitializer(V2)";
+    }
+
     @Override
     public CompletableFuture<FDv2SourceResult> run() {
         return synchronizer.next().thenApply(result -> {

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/integrations/FileSynchronizer.java

@@ -61,6 +61,11 @@ final class FileSynchronizer extends FileDataSourceBase implements Synchronizer
         this.fileWatcher = fw;
     }
 
+    @Override
+    public String name() {
+        return "FileSynchronizer(V2)";
+    }
+
     @Override
     public CompletableFuture<FDv2SourceResult> next() {
         if (!started.getAndSet(true)) {

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/integrations/TestDataV2.java

@@ -295,6 +295,11 @@ private final class TestDataV2SynchronizerImpl implements Synchronizer {
 
     private final AtomicBoolean initialSent = new AtomicBoolean(false);
 
+    @Override
+    public String name() {
+      return "TestDataV2";
+    }
+
     void put(FDv2SourceResult result, CompletableFuture<Void> completion) {
       resultQueue.put(result);
       try {