Add ability to pause remote components until SharedCommunicationObjec…

…ts is ready
DataDog · Dec 28, 2024 · 84d2b6d · 84d2b6d
1 parent 40c7583
commit 84d2b6d
Showing 1 changed file with 48 additions and 4 deletions.
diff --git a/communication/src/main/java/datadog/communication/ddagent/SharedCommunicationObjects.java b/communication/src/main/java/datadog/communication/ddagent/SharedCommunicationObjects.java
@@ -11,6 +11,8 @@
 import datadog.remoteconfig.DefaultConfigurationPoller;
 import datadog.trace.api.Config;
 import datadog.trace.util.AgentTaskScheduler;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import okhttp3.HttpUrl;
@@ -21,12 +23,23 @@
 public class SharedCommunicationObjects {
   private static final Logger log = LoggerFactory.getLogger(SharedCommunicationObjects.class);
 
+  private final List<Runnable> pausedComponents = new ArrayList<>();
+  private volatile boolean paused;
+
   public OkHttpClient okHttpClient;
   public HttpUrl agentUrl;
   public Monitoring monitoring;
   private DDAgentFeaturesDiscovery featuresDiscovery;
   private ConfigurationPoller configurationPoller;
 
+  public SharedCommunicationObjects() {
+    this(false);
+  }
+
+  public SharedCommunicationObjects(boolean paused) {
+    this.paused = paused;
+  }
+
   public void createRemaining(Config config) {
     if (monitoring == null) {
       monitoring = Monitoring.DISABLED;
@@ -46,6 +59,32 @@ public void createRemaining(Config config) {
     }
   }
 
+  public void whenReady(Runnable callback) {
+    if (paused) {
+      synchronized (pausedComponents) {
+        if (paused) {
+          pausedComponents.add(callback);
+          return;
+        }
+      }
+    }
+    callback.run(); // not paused, run immediately
+  }
+
+  public void resume() {
+    paused = false;
+    synchronized (pausedComponents) {
+      for (Runnable callback : pausedComponents) {
+        try {
+          callback.run();
+        } catch (Throwable e) {
+          log.warn("Problem resuming remote component {}", callback, e);
+        }
+      }
+      pausedComponents.clear();
+    }
+  }
+
   private static HttpUrl parseAgentUrl(Config config) {
     String agentUrl = config.getAgentUrl();
     if (agentUrl.startsWith("unix:")) {
@@ -100,11 +139,16 @@ public DDAgentFeaturesDiscovery featuresDiscovery(Config config) {
               agentUrl,
               config.isTraceAgentV05Enabled(),
               config.isTracerMetricsEnabled());
-      if (AGENT_THREAD_GROUP.equals(Thread.currentThread().getThreadGroup())) {
-        featuresDiscovery.discover(); // safe to run on same thread
+
+      if (paused) {
+        // defer remote discovery until remote I/O is allowed
       } else {
-        // avoid performing blocking I/O operation on application thread
-        AgentTaskScheduler.INSTANCE.execute(featuresDiscovery::discover);
+        if (AGENT_THREAD_GROUP.equals(Thread.currentThread().getThreadGroup())) {
+          featuresDiscovery.discover(); // safe to run on same thread
+        } else {
+          // avoid performing blocking I/O operation on application thread
+          AgentTaskScheduler.INSTANCE.execute(featuresDiscovery::discover);
+        }
       }
     }
     return featuresDiscovery;