Add process health-check for Azure so we can re-use named-pipes bound to existing healthy external processes (#5541)

Author: mcculls

communication/build.gradle

@@ -26,6 +26,7 @@ ext {
   minimumInstructionCoverage = 0.8
   excludedClassesCoverage = [
     'datadog.communication.ddagent.ExternalAgentLauncher',
+    'datadog.communication.ddagent.ExternalAgentLauncher.NamedPipeHealthCheck',
     'datadog.communication.ddagent.SharedCommunicationObjects.FixedConfigUrlSupplier',
     'datadog.communication.ddagent.SharedCommunicationObjects.RetryConfigUrlSupplier',
     'datadog.communication.http.OkHttpUtils',

communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java

@@ -1,5 +1,10 @@
 package datadog.communication.ddagent;
 
+import static datadog.trace.util.ProcessSupervisor.ALWAYS_READY;
+import static datadog.trace.util.ProcessSupervisor.Health.HEALTHY;
+import static datadog.trace.util.ProcessSupervisor.Health.NEVER_CHECKED;
+import static datadog.trace.util.ProcessSupervisor.Health.READY_TO_START;
+
 import datadog.trace.api.Config;
 import datadog.trace.api.Platform;
 import datadog.trace.util.ProcessSupervisor;
@@ -25,7 +30,9 @@ public ExternalAgentLauncher(Config config) {
         traceProcessBuilder.redirectError(DISCARD);
         traceProcessBuilder.command().addAll(config.getTraceAgentArgs());
 
-        traceProcessSupervisor = new ProcessSupervisor("datadog-trace-agent", traceProcessBuilder);
+        traceProcessSupervisor =
+            new ProcessSupervisor(
+                "trace-agent", traceProcessBuilder, healthCheck(config.getAgentNamedPipe()));
       } else {
         log.warn("Trace agent path not set. Will not start trace agent process");
       }
@@ -36,7 +43,9 @@ public ExternalAgentLauncher(Config config) {
         dogStatsDProcessBuilder.redirectError(DISCARD);
         dogStatsDProcessBuilder.command().addAll(config.getDogStatsDArgs());
 
-        dogStatsDProcessSupervisor = new ProcessSupervisor("dogstatsd", dogStatsDProcessBuilder);
+        dogStatsDProcessSupervisor =
+            new ProcessSupervisor(
+                "dogstatsd", dogStatsDProcessBuilder, healthCheck(config.getDogStatsDNamedPipe()));
       } else {
         log.warn("DogStatsD path not set. Will not start DogStatsD process");
       }
@@ -53,4 +62,53 @@ public void close() {
       dogStatsDProcessSupervisor.close();
     }
   }
+
+  private static ProcessSupervisor.HealthCheck healthCheck(String pipeName) {
+    return null != pipeName && !pipeName.trim().isEmpty()
+        ? new NamedPipeHealthCheck(pipeName)
+        : ALWAYS_READY;
+  }
+
+  static final class NamedPipeHealthCheck implements ProcessSupervisor.HealthCheck {
+    private static final String NAMED_PIPE_PREFIX = "\\\\.\\pipe\\";
+
+    private final File pipe;
+
+    NamedPipeHealthCheck(String pipeName) {
+      if (pipeName.startsWith(NAMED_PIPE_PREFIX)) {
+        this.pipe = new File(pipeName);
+      } else {
+        this.pipe = new File(NAMED_PIPE_PREFIX + pipeName);
+      }
+    }
+
+    @Override
+    public ProcessSupervisor.Health run(ProcessSupervisor.Health previousHealth)
+        throws InterruptedException {
+
+      // first-time round do a more detailed check for existing bound named-pipe
+      if (previousHealth == NEVER_CHECKED) {
+
+        double delayMillis = 50;
+        for (int retries = 0; retries < 7; retries++) {
+          if (!pipe.exists()) {
+            return READY_TO_START; // no longer bound, start our own external process
+          }
+
+          // check at increasing intervals to make sure it's bound to a healthy process
+          Thread.sleep((long) delayMillis);
+          delayMillis = delayMillis * 1.75;
+        }
+
+        return HEALTHY; // use existing external process
+      }
+
+      // otherwise just check that the pipe is still bound
+      if (pipe.exists()) {
+        return HEALTHY; // keep using external process
+      } else {
+        return READY_TO_START; // start our own process
+      }
+    }
+  }
 }

internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java

@@ -27,6 +27,8 @@ public interface HealthCheck {
     Health run(Health previousHealth) throws InterruptedException;
   }
 
+  public static final HealthCheck ALWAYS_READY = health -> READY_TO_START;
+
   private static final Logger log = LoggerFactory.getLogger(ProcessSupervisor.class);
 
   private static final long HEALTHY_DELAY_MILLIS = 10_000;
@@ -50,7 +52,7 @@ public interface HealthCheck {
    * @param processBuilder Builder to create the process
    */
   public ProcessSupervisor(String imageName, ProcessBuilder processBuilder) {
-    this(imageName, processBuilder, health -> READY_TO_START);
+    this(imageName, processBuilder, ALWAYS_READY);
   }
 
   public ProcessSupervisor(