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");
+    }
   }
 }

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

dd-java-agent/agent-logging/src/main/java/datadog/trace/logging/simplelogger/AasPerProcessLogFile.java

@@ -0,0 +1,187 @@
+package datadog.trace.logging.simplelogger;
+
+import datadog.environment.EnvironmentVariables;
+import de.thetaphi.forbiddenapis.SuppressForbidden;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.management.ManagementFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+
+/**
+ * In Azure App Service, overlapping worker processes run several JVMs that share one configured log
+ * file, and each new JVM truncates it. This gives each JVM its own file, the same way the .NET
+ * tracer does: the PID goes in the name, the file rolls over at a size limit, and old per-process
+ * files are deleted at startup.
+ *
+ * <p>Settings mirror the .NET tracer: DD_TRACE_LOGFILE_RETENTION_DAYS (default 32, 0 disables
+ * cleanup) and DD_MAX_LOGFILE_SIZE in bytes (default 10 MB).
+ */
+@SuppressForbidden
+final class AasPerProcessLogFile {
+  private static final int DEFAULT_RETENTION_DAYS = 32;
+  private static final long DEFAULT_MAX_SIZE_BYTES = 10L * 1024 * 1024;
+
+  private AasPerProcessLogFile() {}
+
+  static boolean isEnabled() {
+    String aas = EnvironmentVariables.get("DD_AZURE_APP_SERVICES");
+    return "1".equals(aas) || "true".equalsIgnoreCase(aas);
+  }
+
+  /** Opens the per-process log file and starts the cleanup of old per-process files. */
+  static RollingFileOutputStream open(File configuredFile) throws IOException {
+    String pid = currentPid();
+    String name = configuredFile.getName();
+    int dot = name.lastIndexOf('.');
+    String stem = dot > 0 ? name.substring(0, dot) : name;
+    String extension = dot > 0 ? name.substring(dot) : "";
+    File directory = configuredFile.getAbsoluteFile().getParentFile();
+
+    startCleanup(directory, stem, extension);
+    return new RollingFileOutputStream(directory, stem + "-" + pid, extension, maxSizeBytes());
+  }
+
+  static String currentPid() {
+    try {
+      // Java 9+; reflection keeps this class loadable on Java 8.
+      Class<?> handle = Class.forName("java.lang.ProcessHandle");
+      Object current = handle.getMethod("current").invoke(null);
+      return String.valueOf(handle.getMethod("pid").invoke(current));
+    } catch (Throwable ignored) {
+      // Java 8
+    }
+    try {
+      String vmName = ManagementFactory.getRuntimeMXBean().getName();
+      int at = vmName.indexOf('@');
+      if (at > 0) {
+        return vmName.substring(0, at);
+      }
+    } catch (Throwable ignored) {
+      // fall through
+    }
+    return "unknown";
+  }
+
+  private static long maxSizeBytes() {
+    return parseLong(EnvironmentVariables.get("DD_MAX_LOGFILE_SIZE"), DEFAULT_MAX_SIZE_BYTES);
+  }
+
+  private static int retentionDays() {
+    return (int)
+        parseLong(
+            EnvironmentVariables.get("DD_TRACE_LOGFILE_RETENTION_DAYS"), DEFAULT_RETENTION_DAYS);
+  }
+
+  private static long parseLong(String value, long defaultValue) {
+    if (value == null) {
+      return defaultValue;
+    }
+    try {
+      return Long.parseLong(value.trim());
+    } catch (NumberFormatException e) {
+      return defaultValue;
+    }
+  }
+
+  private static void startCleanup(final File directory, String stem, String extension) {
+    final int days = retentionDays();
+    if (days <= 0 || directory == null) {
+      return;
+    }
+    // Only this tracer's per-process files: <stem>-<pid>.<ext> and rolled <stem>-<pid>_<n>.<ext>
+    final Pattern ownFiles =
+        Pattern.compile(Pattern.quote(stem) + "-\\d+(_\\d+)?" + Pattern.quote(extension));
+    Thread cleanup =
+        new Thread(
+            () -> {
+              long cutoff = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(days);
+              try {
+                File[] files = directory.listFiles();
+                if (files == null) {
+                  return;
+                }
+                for (File file : files) {
+                  if (ownFiles.matcher(file.getName()).matches()
+                      && file.lastModified() < cutoff
+                      && !file.delete()) {
+                    return; // stop at the first failure, like the .NET tracer
+                  }
+                }
+              } catch (Throwable ignored) {
+                // stop quietly at the first IO problem
+              }
+            },
+            "dd-log-file-cleanup");
+    cleanup.setDaemon(true);
+    cleanup.start();
+  }
+
+  /** Appends to base.ext, then base_001.ext, base_002.ext, ... once each reaches maxSizeBytes. */
+  static final class RollingFileOutputStream extends OutputStream {
+    private final File directory;
+    private final String base;
+    private final String extension;
+    private final long maxSizeBytes;
+    private FileOutputStream current;
+    private long currentSize;
+    private int sequence;
+
+    RollingFileOutputStream(File directory, String base, String extension, long maxSizeBytes)
+        throws IOException {
+      this.directory = directory;
+      this.base = base;
+      this.extension = extension;
+      this.maxSizeBytes = maxSizeBytes;
+      if (directory != null) {
+        directory.mkdirs();
+      }
+      openNext();
+    }
+
+    File currentFile() {
+      String suffix = sequence == 0 ? "" : String.format("_%03d", sequence);
+      return new File(directory, base + suffix + extension);
+    }
+
+    private void openNext() throws IOException {
+      File file = currentFile();
+      current = new FileOutputStream(file, true);
+      currentSize = file.length();
+    }
+
+    private void rollIfNeeded(int incoming) throws IOException {
+      if (maxSizeBytes > 0 && currentSize > 0 && currentSize + incoming > maxSizeBytes) {
+        current.close();
+        sequence++;
+        openNext();
+      }
+    }
+
+    @Override
+    public synchronized void write(int b) throws IOException {
+      rollIfNeeded(1);
+      current.write(b);
+      currentSize++;
+    }
+
+    @Override
+    public synchronized void write(byte[] b, int off, int len) throws IOException {
+      rollIfNeeded(len);
+      current.write(b, off, len);
+      currentSize += len;
+    }
+
+    @Override
+    public synchronized void flush() throws IOException {
+      current.flush();
+    }
+
+    @Override
+    public synchronized void close() throws IOException {
+      current.close();
+    }
+  }
+}