Recover Azure App Service agents after overlapping restarts

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

@@ -1,15 +1,27 @@
 package datadog.communication.ddagent;
 
+import static datadog.trace.util.AgentThreadFactory.AgentThread.PROCESS_SUPERVISOR;
 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 static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 import datadog.environment.OperatingSystem;
 import datadog.trace.api.Config;
+import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
+import datadog.trace.context.TraceScope;
+import datadog.trace.util.AgentThreadFactory;
 import datadog.trace.util.ProcessSupervisor;
 import java.io.Closeable;
 import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeoutException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -32,7 +44,7 @@ public ExternalAgentLauncher(Config config) {
 
         traceProcessSupervisor =
             new ProcessSupervisor(
-                "trace-agent", traceProcessBuilder, healthCheck(config.getAgentNamedPipe()));
+                "trace-agent", traceProcessBuilder, healthCheck(config.getAgentNamedPipe(), true));
       } else {
         log.warn("Trace agent path not set. Will not start trace agent process");
       }
@@ -45,7 +57,9 @@ public ExternalAgentLauncher(Config config) {
 
         dogStatsDProcessSupervisor =
             new ProcessSupervisor(
-                "dogstatsd", dogStatsDProcessBuilder, healthCheck(config.getDogStatsDNamedPipe()));
+                "dogstatsd",
+                dogStatsDProcessBuilder,
+                healthCheck(config.getDogStatsDNamedPipe(), false));
       } else {
         log.warn("DogStatsD path not set. Will not start DogStatsD process");
       }
@@ -63,18 +77,33 @@ public void close() {
     }
   }
 
-  private static ProcessSupervisor.HealthCheck healthCheck(String pipeName) {
+  /**
+   * @param checkResponse whether the agent answers requests on its pipe. DogStatsD never replies,
+   *     so for it the pipe being present is the whole check.
+   */
+  private static ProcessSupervisor.HealthCheck healthCheck(String pipeName, boolean checkResponse) {
     return null != pipeName && !pipeName.trim().isEmpty()
-        ? new NamedPipeHealthCheck(pipeName)
+        ? new NamedPipeHealthCheck(pipeName, checkResponse)
         : ALWAYS_READY;
   }
 
   static final class NamedPipeHealthCheck implements ProcessSupervisor.HealthCheck {
     private static final String NAMED_PIPE_PREFIX = "\\\\.\\pipe\\";
 
+    private static final byte[] INFO_REQUEST =
+        "GET /info HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"
+            .getBytes(StandardCharsets.US_ASCII);
+    private static final int MAX_STATUS_LINE_LENGTH = 64;
+    private static final long RESPONSE_TIMEOUT_MILLIS = 2_000;
+
     private final File pipe;
+    private final boolean checkResponse;
+
+    // Only the supervisor thread runs the health check, so this needs no synchronization
+    private FutureTask<Boolean> pendingInfoRequest;
 
-    NamedPipeHealthCheck(String pipeName) {
+    NamedPipeHealthCheck(String pipeName, boolean checkResponse) {
+      this.checkResponse = checkResponse;
       if (pipeName.startsWith(NAMED_PIPE_PREFIX)) {
         this.pipe = new File(pipeName);
       } else {
@@ -91,7 +120,7 @@ public ProcessSupervisor.Health run(ProcessSupervisor.Health previousHealth)
 
         double delayMillis = 50;
         for (int retries = 0; retries < 7; retries++) {
-          if (!pipe.exists()) {
+          if (!pipeExists()) {
             return READY_TO_START; // no longer bound, start our own external process
           }
 
@@ -100,15 +129,97 @@ public ProcessSupervisor.Health run(ProcessSupervisor.Health previousHealth)
           delayMillis = delayMillis * 1.75;
         }
 
-        return HEALTHY; // use existing external process
+        return respondsOrReadyToStart(); // use existing external process if it responds
       }
 
       // otherwise just check that the pipe is still bound
-      if (pipe.exists()) {
-        return HEALTHY; // keep using external process
+      if (pipeExists()) {
+        return respondsOrReadyToStart(); // keep using external process if it responds
       } else {
         return READY_TO_START; // start our own process
       }
     }
+
+    private ProcessSupervisor.Health respondsOrReadyToStart() throws InterruptedException {
+      return !checkResponse || agentResponds() ? HEALTHY : READY_TO_START;
+    }
+
+    /**
+     * Sends a request on another thread and waits up to {@link #RESPONSE_TIMEOUT_MILLIS} for the
+     * answer. Reads from a pipe cannot time out, so a request to a hung agent stays blocked until
+     * the agent exits. Until then later checks fail without sending another request.
+     */
+    private boolean agentResponds() throws InterruptedException {
+      if (pendingInfoRequest != null && !pendingInfoRequest.isDone()) {
+        return false; // the last request is still waiting for an answer
+      }
+      FutureTask<Boolean> infoRequest = new FutureTask<>(this::sendInfoRequest);
+      pendingInfoRequest = infoRequest;
+      AgentThreadFactory.newAgentThread(PROCESS_SUPERVISOR, "-health-check", infoRequest, true)
+          .start();
+      try {
+        return infoRequest.get(RESPONSE_TIMEOUT_MILLIS, MILLISECONDS);
+      } catch (ExecutionException | TimeoutException e) {
+        return false;
+      }
+    }
+
+    /**
+     * Sends a minimal request over a pipe handle of its own, outside the tracer's HTTP clients.
+     *
+     * @return true if the agent answers with an HTTP status line, whatever the status
+     */
+    private boolean sendInfoRequest() throws IOException {
+      try (TraceScope ignored = AgentTracer.get().muteTracing()) {
+        RandomAccessFile file;
+        try {
+          file = new RandomAccessFile(pipe, "rw");
+        } catch (FileNotFoundException e) {
+          // a pipe with all instances busy cannot be asked, but an agent is serving it
+          return !isPipeNotFound(e);
+        }
+        try {
+          file.write(INFO_REQUEST);
+          StringBuilder statusLine = new StringBuilder();
+          for (int c = file.read();
+              c != -1 && c != '\r' && statusLine.length() < MAX_STATUS_LINE_LENGTH;
+              c = file.read()) {
+            statusLine.append((char) c);
+          }
+          return statusLine.toString().startsWith("HTTP/1.");
+        } finally {
+          file.close();
+        }
+      }
+    }
+
+    /**
+     * {@link File#exists()} can report false for a pipe whose instances are all busy, for example
+     * under load. In that case open the pipe once to tell busy from absent.
+     */
+    private boolean pipeExists() {
+      if (pipe.exists()) {
+        return true;
+      }
+      if (!OperatingSystem.isWindows()) {
+        return false;
+      }
+      try (RandomAccessFile ignored = new RandomAccessFile(pipe, "rw")) {
+        return true;
+      } catch (FileNotFoundException e) {
+        return !isPipeNotFound(e);
+      } catch (IOException e) {
+        return true; // opened, then failed to close
+      }
+    }
+
+    /**
+     * Only "cannot find the file" (ERROR_FILE_NOT_FOUND) means no process is serving the pipe.
+     * Errors such as "All pipe instances are busy" (ERROR_PIPE_BUSY) mean that one is.
+     */
+    private static boolean isPipeNotFound(FileNotFoundException e) {
+      String message = e.getMessage();
+      return message != null && message.contains("cannot find the file");
+    }
   }
 }

communication/src/main/java/datadog/communication/http/OkHttpUtils.java

@@ -177,12 +177,13 @@ private static OkHttpClient buildHttpClient(
         .dispatcher(
             dispatcher != null ? dispatcher : new Dispatcher(RejectingExecutorService.INSTANCE));
 
+    final boolean useNamedPipe = unixDomainSocketPath == null && namedPipe != null;
     if (unixDomainSocketPath != null) {
       builder.socketFactory(
           new UnixDomainSocketFactory(
               new File(unixDomainSocketPath), useJdkUnixDomainSocket, agentConfiguredUsingDefault));
       log.debug("Using UnixDomainSocket as http transport");
-    } else if (namedPipe != null) {
+    } else if (useNamedPipe) {
       builder.socketFactory(new NamedPipeSocketFactory(namedPipe));
       log.debug("Using NamedPipe as http transport");
     }
@@ -204,7 +205,11 @@ private static OkHttpClient buildHttpClient(
       builder.retryOnConnectionFailure(retryOnConnectionFailure);
     }
 
-    if (maxRunningRequests != null) {
+    if (useNamedPipe) {
+      // NamedPipeSocket can't time out a read, so OkHttp's health check on a pooled connection can
+      // block forever. Don't keep idle pipe connections; each request opens its own.
+      builder.connectionPool(new ConnectionPool(0, 1, SECONDS));
+    } else if (maxRunningRequests != null) {
       // Reusing connections causes non daemon threads to be created which causes agent to prevent
       // app from exiting. See https://github.com/square/okhttp/issues/4029 for some details.
       builder.connectionPool(new ConnectionPool(maxRunningRequests, 1, SECONDS));

dd-java-agent/agent-logging/build.gradle

@@ -21,6 +21,11 @@ excludedClassesCoverage += [
   'datadog.trace.logging.LogReporter',
 ]
 
+excludedClassesInstructionCoverage += [
+  // The PID lookup has separate Java 8 and Java 9+ paths, and no single JVM runs both
+  'datadog.trace.logging.simplelogger.AasPerProcessLogFile',
+]
+
 dependencies {
   // This is fine since this project is shadowed into the agent-jar by dd-java-agent:agent-bootstrap
   api libs.slf4j