From ea6d525e94a6c9a98c9a83737fa08476a3db73df Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 15 Oct 2026 17:05:56 -0400 Subject: [PATCH 1/6] Log AAS agent supervisor lifecycle at the default level In Azure App Service the tracer starts and supervises the trace agent and DogStatsD. Overlapping worker processes during platform restarts can leave those agents missing or shared, but starts, exits, and stops were only logged at debug level, so default customer logs could not show what happened. - Log each agent start, exit (with exit code and run time), and stop at info, with the tracer PID and, on Java 9+, the agent PID. - Log when an already-running agent not started by this process is used, and when that agent goes away. - After repeated quick failures, move lifecycle lines to debug until the agent stays up, so a crash loop does not flood the log. - Warn when shutdown leaves an agent running. --- .../trace/util/JDK9ProcessPidFunction.java | 15 +++ .../java/datadog/trace/util/PidHelper.java | 26 +++++ .../datadog/trace/util/ProcessSupervisor.java | 107 +++++++++++++++++- 3 files changed, 144 insertions(+), 4 deletions(-) create mode 100644 internal-api/internal-api-9/src/main/java/datadog/trace/util/JDK9ProcessPidFunction.java diff --git a/internal-api/internal-api-9/src/main/java/datadog/trace/util/JDK9ProcessPidFunction.java b/internal-api/internal-api-9/src/main/java/datadog/trace/util/JDK9ProcessPidFunction.java new file mode 100644 index 00000000000..047174b3018 --- /dev/null +++ b/internal-api/internal-api-9/src/main/java/datadog/trace/util/JDK9ProcessPidFunction.java @@ -0,0 +1,15 @@ +package datadog.trace.util; + +import java.util.function.Function; + +/** Use standard API to retrieve the PID of a child process on Java9+. */ +public final class JDK9ProcessPidFunction implements Function { + @Override + public String apply(Process process) { + try { + return Long.toString(process.pid()); + } catch (Throwable e) { + return ""; + } + } +} diff --git a/internal-api/src/main/java/datadog/trace/util/PidHelper.java b/internal-api/src/main/java/datadog/trace/util/PidHelper.java index 887dd3900e8..a5b4b82b203 100644 --- a/internal-api/src/main/java/datadog/trace/util/PidHelper.java +++ b/internal-api/src/main/java/datadog/trace/util/PidHelper.java @@ -13,6 +13,7 @@ import java.util.Collections; import java.util.Objects; import java.util.Set; +import java.util.function.Function; import java.util.function.Predicate; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -38,6 +39,31 @@ public static long getPidAsLong() { return PID_AS_LONG; } + private static final Function CHILD_PID = findChildPidFunction(); + + /** Returns the PID of a child process, or an empty string when the JVM cannot provide it. */ + public static String getPid(Process process) { + if (null == process || null == CHILD_PID) { + return ""; + } + return CHILD_PID.apply(process); + } + + @SuppressWarnings("unchecked") + private static Function findChildPidFunction() { + if (JavaVirtualMachine.isJavaVersionAtLeast(9)) { + try { + return (Function) + Class.forName("datadog.trace.util.JDK9ProcessPidFunction") + .getDeclaredConstructor() + .newInstance(); + } catch (Throwable e) { + log.debug("JDK9ProcessPidFunction not available", e); + } + } + return null; // Java 8 has no public API for the PID of a child process + } + @SuppressWarnings("unchecked") private static String findPid() { String pid = ""; diff --git a/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java b/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java index b232ddc156c..84a20159169 100644 --- a/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java +++ b/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java @@ -6,6 +6,7 @@ import static datadog.trace.util.ProcessSupervisor.Health.HEALTHY; import static datadog.trace.util.ProcessSupervisor.Health.INTERRUPTED; import static datadog.trace.util.ProcessSupervisor.Health.READY_TO_START; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import datadog.trace.api.internal.VisibleForTesting; import datadog.trace.bootstrap.instrumentation.api.AgentTracer; @@ -38,6 +39,14 @@ public interface HealthCheck { private static final long FAULTED_DELAY_MILLIS = 2_000; private static final int MAX_FAULTS = 5; + // Lifecycle events are logged at info level. A process that keeps failing quickly would flood + // the log, so after this many failures in a row the details go to debug level until the + // process stays up for STABLE_RUN_MILLIS. + private static final int QUIET_AFTER_FAILURES = 5; + private static final long STABLE_RUN_MILLIS = 60_000; + + private static final long STOP_WAIT_MILLIS = 200; + private final String imageName; private final ProcessBuilder processBuilder; private final HealthCheck healthCheck; @@ -45,8 +54,10 @@ public interface HealthCheck { private long nextCheckMillis = 0; private Health currentHealth = Health.NEVER_CHECKED; + private Health lastCheckResult = Health.NEVER_CHECKED; private Process currentProcess; private int faults; + private int failuresInARow; private volatile boolean stopping = false; @@ -80,6 +91,7 @@ private void mainLoop() { Thread.sleep(delayMillis); } currentHealth = healthCheck.run(currentHealth); + logHealthCheckChange(currentHealth); if (currentHealth == READY_TO_START) { startProcessAndWait(); } @@ -96,6 +108,28 @@ private void mainLoop() { } } + private void logHealthCheckChange(Health result) { + Health previous = lastCheckResult; + lastCheckResult = result; + if (result == previous) { + return; + } + if (result == HEALTHY) { + // Health checks only run while this supervisor has no running process of its own, + // so a healthy result means another process (usually another worker) is serving it. + failuresInARow = 0; + log.info( + "Process [{}] is already running and was not started by this process (pid {}); using it", + imageName, + PidHelper.getPid()); + } else if (result == READY_TO_START && previous == HEALTHY) { + logLifecycle( + "Process [{}] that this process (pid {}) was using is no longer available", + imageName, + PidHelper.getPid()); + } + } + private void scheduleNextHealthCheck() { long now = System.currentTimeMillis(); if (currentHealth == HEALTHY) { @@ -115,28 +149,86 @@ private void startProcessAndWait() throws Exception { } currentHealth = HEALTHY; faults = 0; + logLifecycle( + "Started process [{}] with pid {} from this process (pid {})", + imageName, + pidOrUnknown(currentProcess), + PidHelper.getPid()); } + long startedMillis = System.currentTimeMillis(); + String childPid = pidOrUnknown(currentProcess); + // Block until the process exits int code = currentProcess.waitFor(); - log.debug("Process [{}] has exited with code {}", imageName, code); currentHealth = code == 0 ? INTERRUPTED : FAULTED; + long ranMillis = System.currentTimeMillis() - startedMillis; + if (ranMillis >= STABLE_RUN_MILLIS) { + failuresInARow = 0; + } + logLifecycle( + "Process [{}] with pid {} exited with code {} after {} ms", + imageName, + childPid, + code, + ranMillis); + if (code != 0 && ++failuresInARow == QUIET_AFTER_FAILURES) { + log.warn( + "Process [{}] has exited with an error {} times in a row; " + + "further starts and exits are logged at debug level until it stays up for {} seconds", + imageName, + failuresInARow, + MILLISECONDS.toSeconds(STABLE_RUN_MILLIS)); + } + // Process is dead, no longer needs to be tracked currentProcess = null; } private void stopProcess() { if (currentProcess != null) { - log.debug("Stopping process: [{}]", imageName); - currentProcess.destroy(); - if (currentProcess.isAlive()) { + String childPid = pidOrUnknown(currentProcess); + log.info("Stopping process [{}] with pid {}", imageName, childPid); + boolean stopped = false; + try { + currentProcess.destroy(); + stopped = currentProcess.waitFor(STOP_WAIT_MILLIS, MILLISECONDS); + if (!stopped) { + currentProcess.destroyForcibly(); + stopped = currentProcess.waitFor(STOP_WAIT_MILLIS, MILLISECONDS); + } + } catch (InterruptedException e) { currentProcess.destroyForcibly(); + Thread.currentThread().interrupt(); + } + if (stopped) { + log.info("Stopped process [{}] with pid {}", imageName, childPid); + } else { + log.warn( + "Process [{}] with pid {} was still running after the stop request", + imageName, + childPid); } currentProcess = null; + } else { + log.debug("No process [{}] started by this process to stop", imageName); } } + private void logLifecycle(String format, Object... arguments) { + if (failuresInARow < QUIET_AFTER_FAILURES) { + log.info(format, arguments); + } else { + log.debug(format, arguments); + } + } + + private static String pidOrUnknown(Process process) { + String pid = PidHelper.getPid(process); + return pid.isEmpty() ? "unknown" : pid; + } + @Override public void close() { stopping = true; @@ -145,6 +237,13 @@ public void close() { supervisorThread.join(THREAD_JOIN_TIMOUT_MS); } catch (Throwable ignored) { } + if (supervisorThread.isAlive()) { + log.warn( + "Supervisor for process [{}] did not finish stopping within {} ms; " + + "the process may be left running", + imageName, + THREAD_JOIN_TIMOUT_MS); + } } @VisibleForTesting From 29025b98ecfef7ef44baf1d08bddd5227cb13e48 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 15 Oct 2026 17:06:35 -0400 Subject: [PATCH 2/6] Back off AAS agent restarts instead of giving up ProcessSupervisor reset its fault count on every successful start, so an agent that kept exiting was restarted every 2 seconds forever, while a start that kept throwing gave up for good after 5 attempts. - Forget faults only after the process stays up for 60 seconds. - Back off after each fault, doubling from 2 seconds to a cap of 60 seconds, and never stop retrying. - Move repeated start exceptions to debug after the first few, so a permanently failing start does not log a stack trace every minute. --- .../datadog/trace/util/ProcessSupervisor.java | 43 +++++++++++++++---- .../trace/util/ProcessSupervisorTest.java | 31 +++++++++++++ 2 files changed, 65 insertions(+), 9 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java b/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java index 84a20159169..968277782fa 100644 --- a/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java +++ b/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java @@ -37,7 +37,7 @@ public interface HealthCheck { private static final long HEALTHY_DELAY_MILLIS = 10_000; private static final long FAULTED_DELAY_MILLIS = 2_000; - private static final int MAX_FAULTS = 5; + private static final long MAX_FAULTED_DELAY_MILLIS = 60_000; // Lifecycle events are logged at info level. A process that keeps failing quickly would flood // the log, so after this many failures in a row the details go to debug level until the @@ -56,7 +56,7 @@ public interface HealthCheck { private Health currentHealth = Health.NEVER_CHECKED; private Health lastCheckResult = Health.NEVER_CHECKED; private Process currentProcess; - private int faults; + private final FaultBackoff faultBackoff = new FaultBackoff(); private int failuresInARow; private volatile boolean stopping = false; @@ -81,10 +81,6 @@ public ProcessSupervisor( private void mainLoop() { try { while (!stopping) { - if (currentHealth == FAULTED && ++faults >= MAX_FAULTS) { - log.warn("Failed to start process [{}] after {} attempts", imageName, faults); - break; - } try { long delayMillis = nextCheckMillis - System.currentTimeMillis(); if (delayMillis > 0) { @@ -98,7 +94,11 @@ private void mainLoop() { } catch (InterruptedException e) { currentHealth = INTERRUPTED; } catch (Throwable e) { - log.warn("Exception starting process: [{}]", imageName, e); + if (failuresInARow++ < QUIET_AFTER_FAILURES) { + log.warn("Exception starting process: [{}]", imageName, e); + } else { + log.debug("Exception starting process: [{}]", imageName, e); + } currentHealth = FAULTED; } scheduleNextHealthCheck(); @@ -135,7 +135,7 @@ private void scheduleNextHealthCheck() { if (currentHealth == HEALTHY) { nextCheckMillis = now + HEALTHY_DELAY_MILLIS; } else if (currentHealth == FAULTED) { - nextCheckMillis = now + FAULTED_DELAY_MILLIS; + nextCheckMillis = now + faultBackoff.recordFault(); } else { // interrupted nextCheckMillis = Long.max(nextCheckMillis, now + 100); } @@ -148,7 +148,6 @@ private void startProcessAndWait() throws Exception { currentProcess = processBuilder.start(); } currentHealth = HEALTHY; - faults = 0; logLifecycle( "Started process [{}] with pid {} from this process (pid {})", imageName, @@ -164,6 +163,7 @@ private void startProcessAndWait() throws Exception { currentHealth = code == 0 ? INTERRUPTED : FAULTED; long ranMillis = System.currentTimeMillis() - startedMillis; + faultBackoff.recordUptime(ranMillis); if (ranMillis >= STABLE_RUN_MILLIS) { failuresInARow = 0; } @@ -246,6 +246,31 @@ public void close() { } } + /** + * Counts faults in a row and spaces out the retries that follow them. Faults are forgotten only + * once a process has stayed up for {@link #STABLE_RUN_MILLIS}, so an agent that keeps exiting + * soon after it starts backs off instead of restarting every few seconds. Retries never stop. + */ + static final class FaultBackoff { + private static final int MAX_DOUBLINGS = 5; + + private int faults; + + /** Records a fault and returns how long to wait before the next health check. */ + long recordFault() { + faults++; + int doublings = Math.min(faults - 1, MAX_DOUBLINGS); + return Math.min(FAULTED_DELAY_MILLIS << doublings, MAX_FAULTED_DELAY_MILLIS); + } + + /** Forgets earlier faults once a process has been up for a stable period. */ + void recordUptime(long upMillis) { + if (upMillis >= STABLE_RUN_MILLIS) { + faults = 0; + } + } + } + @VisibleForTesting Process getCurrentProcess() { return currentProcess; diff --git a/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java b/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java index d213add1a70..8f080d7d6b1 100644 --- a/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java @@ -1,5 +1,6 @@ package datadog.trace.util; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -52,6 +53,36 @@ void processKilledWhenSupervisorClosed() throws InterruptedException { }); } + @Test + void faultBackoffDoublesUpToSixtySecondsAndNeverStops() { + ProcessSupervisor.FaultBackoff backoff = new ProcessSupervisor.FaultBackoff(); + + assertEquals(2_000, backoff.recordFault()); + assertEquals(4_000, backoff.recordFault()); + assertEquals(8_000, backoff.recordFault()); + assertEquals(16_000, backoff.recordFault()); + assertEquals(32_000, backoff.recordFault()); + assertEquals(60_000, backoff.recordFault()); + for (int i = 0; i < 1_000; i++) { + assertEquals(60_000, backoff.recordFault()); + } + } + + @Test + void faultBackoffResetsOnlyAfterStableUptime() { + ProcessSupervisor.FaultBackoff backoff = new ProcessSupervisor.FaultBackoff(); + backoff.recordFault(); + backoff.recordFault(); + + // a start that exits again quickly keeps the earlier faults + backoff.recordUptime(59_999); + assertEquals(8_000, backoff.recordFault()); + + // staying up for the stable period forgets them + backoff.recordUptime(60_000); + assertEquals(2_000, backoff.recordFault()); + } + @Test void processRespawnsWhenKilled() throws InterruptedException { ProcessBuilder processBuilder = createProcessBuilder(); From ae00f9a4afea9384ec5176ec7216d9aa519d84cb Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 15 Oct 2026 17:06:36 -0400 Subject: [PATCH 3/6] Health check and restart the AAS agent this tracer started Once ProcessSupervisor started an agent it blocked in waitFor() until the agent exited, so an agent that hung but stayed alive was never checked or replaced. - While its own agent runs, wait in keep-alive steps of 10 seconds and run the health check after each one. - After 3 failed health checks in a row, stop that agent and go back through the normal start path, after the fault backoff. Log the first failed check and the restart once. - Keep tracking an agent that is still running after a stop request, so a restart never starts a second copy next to it. - Supervisors without a health check keep only waiting for the exit. --- .../datadog/trace/util/ProcessSupervisor.java | 81 ++++++++++++++++--- 1 file changed, 70 insertions(+), 11 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java b/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java index 968277782fa..5918e1d4958 100644 --- a/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java +++ b/internal-api/src/main/java/datadog/trace/util/ProcessSupervisor.java @@ -47,6 +47,9 @@ public interface HealthCheck { private static final long STOP_WAIT_MILLIS = 200; + // A process this supervisor started is restarted after failing this many health checks in a row + private static final int MAX_FAILED_HEALTH_CHECKS = 3; + private final String imageName; private final ProcessBuilder processBuilder; private final HealthCheck healthCheck; @@ -56,6 +59,8 @@ public interface HealthCheck { private Health currentHealth = Health.NEVER_CHECKED; private Health lastCheckResult = Health.NEVER_CHECKED; private Process currentProcess; + private long currentProcessStartedMillis; + private int failedHealthChecks; private final FaultBackoff faultBackoff = new FaultBackoff(); private int failuresInARow; @@ -86,10 +91,12 @@ private void mainLoop() { if (delayMillis > 0) { Thread.sleep(delayMillis); } - currentHealth = healthCheck.run(currentHealth); - logHealthCheckChange(currentHealth); - if (currentHealth == READY_TO_START) { - startProcessAndWait(); + if (currentProcess == null) { + currentHealth = healthCheck.run(currentHealth); + logHealthCheckChange(currentHealth); + } + if (currentProcess != null || currentHealth == READY_TO_START) { + startProcessAndSupervise(); } } catch (InterruptedException e) { currentHealth = INTERRUPTED; @@ -115,7 +122,7 @@ private void logHealthCheckChange(Health result) { return; } if (result == HEALTHY) { - // Health checks only run while this supervisor has no running process of its own, + // These checks only run while this supervisor has no running process of its own, // so a healthy result means another process (usually another worker) is serving it. failuresInARow = 0; log.info( @@ -141,12 +148,14 @@ private void scheduleNextHealthCheck() { } } - private void startProcessAndWait() throws Exception { + private void startProcessAndSupervise() throws Exception { if (currentProcess == null) { log.debug("Starting process: [{}]", imageName); try (TraceScope ignored = AgentTracer.get().muteTracing()) { currentProcess = processBuilder.start(); } + currentProcessStartedMillis = System.currentTimeMillis(); + failedHealthChecks = 0; currentHealth = HEALTHY; logLifecycle( "Started process [{}] with pid {} from this process (pid {})", @@ -155,14 +164,20 @@ private void startProcessAndWait() throws Exception { PidHelper.getPid()); } - long startedMillis = System.currentTimeMillis(); String childPid = pidOrUnknown(currentProcess); - // Block until the process exits - int code = currentProcess.waitFor(); + // Keep checking on the process until it exits + while (!currentProcess.waitFor(HEALTHY_DELAY_MILLIS, MILLISECONDS)) { + faultBackoff.recordUptime(System.currentTimeMillis() - currentProcessStartedMillis); + if (healthCheck != ALWAYS_READY && !passesHealthCheck(childPid)) { + restartUnhealthyProcess(childPid); + return; + } + } + int code = currentProcess.exitValue(); currentHealth = code == 0 ? INTERRUPTED : FAULTED; - long ranMillis = System.currentTimeMillis() - startedMillis; + long ranMillis = System.currentTimeMillis() - currentProcessStartedMillis; faultBackoff.recordUptime(ranMillis); if (ranMillis >= STABLE_RUN_MILLIS) { failuresInARow = 0; @@ -186,6 +201,49 @@ private void startProcessAndWait() throws Exception { currentProcess = null; } + /** + * Checks on the running process this supervisor started. + * + * @return false once it has failed {@link #MAX_FAILED_HEALTH_CHECKS} health checks in a row + */ + private boolean passesHealthCheck(String childPid) throws InterruptedException { + if (healthCheck.run(HEALTHY) == HEALTHY) { + if (failedHealthChecks > 0) { + logLifecycle( + "Process [{}] with pid {} is passing health checks again", imageName, childPid); + } + failedHealthChecks = 0; + return true; + } + if (++failedHealthChecks == 1) { + logLifecycle( + "Process [{}] with pid {} started by this process (pid {}) failed a health check", + imageName, + childPid, + PidHelper.getPid()); + } + return failedHealthChecks < MAX_FAILED_HEALTH_CHECKS; + } + + private void restartUnhealthyProcess(String childPid) { + if (failuresInARow++ < QUIET_AFTER_FAILURES) { + log.warn( + "Process [{}] with pid {} failed {} health checks in a row; restarting it", + imageName, + childPid, + failedHealthChecks); + } else { + log.debug( + "Process [{}] with pid {} failed {} health checks in a row; restarting it", + imageName, + childPid, + failedHealthChecks); + } + stopProcess(); + // The next health check starts a new process through the normal start path, after a backoff + currentHealth = FAULTED; + } + private void stopProcess() { if (currentProcess != null) { String childPid = pidOrUnknown(currentProcess); @@ -204,13 +262,14 @@ private void stopProcess() { } if (stopped) { log.info("Stopped process [{}] with pid {}", imageName, childPid); + currentProcess = null; } else { + // Keep tracking it, so a restart does not start a second copy while this one runs log.warn( "Process [{}] with pid {} was still running after the stop request", imageName, childPid); } - currentProcess = null; } else { log.debug("No process [{}] started by this process to stop", imageName); } From 9f1ddc1ec4dfacdb78f87fdff4b5d80305623131 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 15 Oct 2026 17:06:37 -0400 Subject: [PATCH 4/6] Treat a busy AAS agent pipe as present On Windows, File.exists() can report false for a named pipe whose instances are all busy. The supervisor then took a busy agent for a missing one and started a competing agent, which lost the pipe race and exited. When File.exists() reports false on Windows, open the pipe once and close it again. Only "cannot find the file" means the pipe is absent; "All pipe instances are busy" and other open errors mean a process is serving it. --- .../ddagent/ExternalAgentLauncher.java | 36 +++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java b/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java index 8eea13de224..7cd35fa4aba 100644 --- a/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java +++ b/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java @@ -10,6 +10,9 @@ 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 org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -91,7 +94,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 } @@ -104,11 +107,40 @@ public ProcessSupervisor.Health run(ProcessSupervisor.Health previousHealth) } // otherwise just check that the pipe is still bound - if (pipe.exists()) { + if (pipeExists()) { return HEALTHY; // keep using external process } else { return READY_TO_START; // start our own process } } + + /** + * {@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"); + } } } From b0c3d30ec1dfa6789eb3d3379546418e9801e905 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 15 Oct 2026 17:06:44 -0400 Subject: [PATCH 5/6] Check that the AAS trace agent answers on its pipe The trace agent health check only tested that its pipe existed, so a hung agent that kept its pipe was trusted forever. - After the pipe check passes, send a minimal GET /info over a pipe handle of the check's own, outside the tracer's HTTP clients, with tracing muted. - Treat any HTTP status line as an answer. A busy pipe counts as present. - Pipe reads cannot time out, so send the request on a daemon thread and give up after 2 seconds. While that request is still blocked, later checks fail without sending another one. - DogStatsD never replies, so its check stays "pipe present". --- .../ddagent/ExternalAgentLauncher.java | 93 +++++++++++++++++-- 1 file changed, 86 insertions(+), 7 deletions(-) diff --git a/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java b/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java index 7cd35fa4aba..e3dfdfff248 100644 --- a/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java +++ b/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java @@ -1,18 +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; @@ -35,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"); } @@ -48,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"); } @@ -66,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 pendingInfoRequest; - NamedPipeHealthCheck(String pipeName) { + NamedPipeHealthCheck(String pipeName, boolean checkResponse) { + this.checkResponse = checkResponse; if (pipeName.startsWith(NAMED_PIPE_PREFIX)) { this.pipe = new File(pipeName); } else { @@ -103,17 +129,70 @@ 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 (pipeExists()) { - return HEALTHY; // keep using external process + 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 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. From d3b4fd3ff9671f6c92e2cd327216b577dfb5a17e Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 16 Oct 2026 12:37:30 -0400 Subject: [PATCH 6/6] Test how ProcessSupervisor supervises the AAS agent The supervisor now health checks its own agent while it runs, restarts it after three failed checks in a row, uses an agent another process started, and never gives up retrying. Only the backoff arithmetic was tested, so coverage of these paths was too low for the build to pass. The new tests start a fake Process from a mocked ProcessBuilder. The fake returns quickly from timed waits, so health checks that normally run ten seconds apart run fast, and each test can choose how the process exits or refuses to stop. They check that the supervisor: - restarts its own process after exactly three failed health checks; - keeps its process when a health check passes before the third failure; - uses a process it did not start, then starts its own once that one goes away (this waits out the real ten second healthy interval); - keeps tracking a process that ignores stop requests instead of starting a second copy; - retries after the process fails to start; - logs the exit code and starts again after the process exits. --- .../trace/util/ProcessSupervisorTest.java | 284 ++++++++++++++++++ 1 file changed, 284 insertions(+) diff --git a/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java b/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java index 8f080d7d6b1..e5877dc0b34 100644 --- a/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java @@ -1,19 +1,61 @@ package datadog.trace.util; +import static datadog.trace.util.ProcessSupervisor.Health.FAULTED; +import static datadog.trace.util.ProcessSupervisor.Health.HEALTHY; +import static datadog.trace.util.ProcessSupervisor.Health.READY_TO_START; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import datadog.trace.test.util.PollingConditions; import datadog.trace.test.util.PortableCommand; +import datadog.trace.util.ProcessSupervisor.HealthCheck; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; // This test looks at the private "currentProcess" variable because the alternative // would be calling "ps -e" repeatedly class ProcessSupervisorTest { + private final Logger supervisorLogger = (Logger) LoggerFactory.getLogger(ProcessSupervisor.class); + private final ListAppender logs = new ListAppender<>(); + + @BeforeEach + void captureLogs() { + logs.start(); + supervisorLogger.addAppender(logs); + } + + @AfterEach + void stopCapturingLogs() { + supervisorLogger.detachAppender(logs); + } + private ProcessBuilder createProcessBuilder() { // Creates a process that never returns on its own return new ProcessBuilder(PortableCommand.runForever()); @@ -127,4 +169,246 @@ void processRespawnsWhenKilled() throws InterruptedException { assertTrue(process == null || !process.isAlive()); }); } + + @Test + void restartsItsOwnProcessAfterThreeFailedHealthChecksInARow() throws IOException { + List started = new CopyOnWriteArrayList<>(); + AtomicInteger checksWhileRunning = new AtomicInteger(); + HealthCheck failsWhileRunning = + previous -> { + if (previous == HEALTHY) { + checksWhileRunning.incrementAndGet(); + return FAULTED; + } + return READY_TO_START; + }; + + try (ProcessSupervisor ignored = + new ProcessSupervisor("test", startsFakeProcesses(started, true), failsWhileRunning)) { + new PollingConditions(5) + .eventually( + () -> { + assertEquals(1, started.size()); + assertFalse(started.get(0).isAlive()); + }); + // the replacement only starts after a fault backoff of a few seconds + assertEquals(3, checksWhileRunning.get()); + assertEquals(1, countLogs(Level.INFO, "failed a health check")); + assertEquals(1, countLogs(Level.WARN, "failed 3 health checks in a row; restarting it")); + + // a replacement is started (it keeps failing its health checks, so it is restarted too) + new PollingConditions(5).eventually(() -> assertTrue(started.size() >= 2)); + } + } + + @Test + void keepsItsOwnProcessWhenHealthChecksRecoverBeforeThreeFailures() throws IOException { + List started = new CopyOnWriteArrayList<>(); + AtomicInteger checksWhileRunning = new AtomicInteger(); + // fails two checks in a row, then passes one, over and over + HealthCheck sometimesFailsWhileRunning = + previous -> { + if (previous == HEALTHY) { + return checksWhileRunning.incrementAndGet() % 3 == 0 ? HEALTHY : FAULTED; + } + return READY_TO_START; + }; + + try (ProcessSupervisor ignored = + new ProcessSupervisor( + "test", startsFakeProcesses(started, true), sometimesFailsWhileRunning)) { + new PollingConditions(5).eventually(() -> assertTrue(checksWhileRunning.get() >= 9)); + + assertEquals(1, started.size()); + assertEquals(0, started.get(0).stopRequests.get()); + assertTrue(countLogs(Level.INFO, "is passing health checks again") >= 2); + } + } + + @Test + void usesAProcessItDidNotStartAndStartsItsOwnWhenThatOneGoesAway() throws IOException { + List started = new CopyOnWriteArrayList<>(); + AtomicBoolean otherProcessRunning = new AtomicBoolean(true); + HealthCheck answersWhileAnyProcessRuns = + previous -> otherProcessRunning.get() || !started.isEmpty() ? HEALTHY : READY_TO_START; + + try (ProcessSupervisor ignored = + new ProcessSupervisor( + "test", startsFakeProcesses(started, true), answersWhileAnyProcessRuns)) { + new PollingConditions(5) + .eventually( + () -> + assertEquals( + 1, + countLogs( + Level.INFO, "is already running and was not started by this process"))); + assertTrue(started.isEmpty()); + + otherProcessRunning.set(false); + + // the next check runs once the ten second healthy interval has passed + new PollingConditions(15) + .eventually(() -> assertEquals(1, countLogs(Level.INFO, "Started process [test]"))); + assertEquals(1, started.size()); + assertEquals(1, countLogs(Level.INFO, "was using is no longer available")); + } + } + + @Test + void keepsTrackingItsProcessWhenItDoesNotStopInsteadOfStartingASecondCopy() throws IOException { + List started = new CopyOnWriteArrayList<>(); + AtomicInteger checksWhileRunning = new AtomicInteger(); + HealthCheck failsWhileRunning = + previous -> { + if (previous == HEALTHY) { + checksWhileRunning.incrementAndGet(); + return FAULTED; + } + return READY_TO_START; + }; + + try (ProcessSupervisor ignored = + new ProcessSupervisor("test", startsFakeProcesses(started, false), failsWhileRunning)) { + // after the failed restart and a fault backoff, it supervises the same process again + new PollingConditions(10).eventually(() -> assertTrue(checksWhileRunning.get() > 3)); + + assertEquals(1, started.size()); + assertTrue(started.get(0).isAlive()); + assertTrue(started.get(0).stopRequests.get() >= 2, "asks to stop, then forcibly"); + assertTrue(countLogs(Level.WARN, "was still running after the stop request") >= 1); + } + } + + @Test + void retriesAfterTheProcessFailsToStart() throws IOException { + List started = new CopyOnWriteArrayList<>(); + ProcessBuilder processBuilder = mock(ProcessBuilder.class); + when(processBuilder.start()) + .thenThrow(new IOException("cannot start")) + .thenAnswer(invocation -> startFakeProcess(started, true)); + + try (ProcessSupervisor ignored = new ProcessSupervisor("test", processBuilder)) { + new PollingConditions(5).eventually(() -> assertEquals(1, started.size())); + + verify(processBuilder, times(2)).start(); + assertEquals(1, countLogs(Level.WARN, "Exception starting process: [test]")); + } + } + + @Test + void logsTheExitCodeAndStartsAgainAfterTheProcessExits() throws IOException { + List started = new CopyOnWriteArrayList<>(); + + try (ProcessSupervisor ignored = + new ProcessSupervisor("test", startsFakeProcesses(started, true))) { + new PollingConditions(5).eventually(() -> assertEquals(1, started.size())); + started.get(0).exit(0); + + new PollingConditions(5).eventually(() -> assertEquals(2, started.size())); + assertEquals(1, countLogs(Level.INFO, "exited with code 0 after")); + } + } + + private static ProcessBuilder startsFakeProcesses( + List started, boolean stopsWhenAsked) throws IOException { + ProcessBuilder processBuilder = mock(ProcessBuilder.class); + when(processBuilder.start()) + .thenAnswer(invocation -> startFakeProcess(started, stopsWhenAsked)); + return processBuilder; + } + + private static FakeProcess startFakeProcess(List started, boolean stopsWhenAsked) { + FakeProcess process = new FakeProcess(stopsWhenAsked); + started.add(process); + return process; + } + + private int countLogs(Level level, String text) { + List events; + synchronized (logs) { + events = new ArrayList<>(logs.list); + } + int count = 0; + for (ILoggingEvent event : events) { + if (event.getLevel() == level && event.getFormattedMessage().contains(text)) { + count++; + } + } + return count; + } + + /** + * Stands in for a child process. Each timed wait returns within {@link #WAIT_MILLIS}, so the + * supervisor's ten second interval between health checks of a running process passes quickly. + */ + private static final class FakeProcess extends Process { + private static final long WAIT_MILLIS = 20; + + private final CountDownLatch exited = new CountDownLatch(1); + private final boolean stopsWhenAsked; + private volatile int exitCode; + final AtomicInteger stopRequests = new AtomicInteger(); + + FakeProcess(boolean stopsWhenAsked) { + this.stopsWhenAsked = stopsWhenAsked; + } + + void exit(int code) { + exitCode = code; + exited.countDown(); + } + + @Override + public boolean waitFor(long timeout, TimeUnit unit) throws InterruptedException { + return exited.await(Math.min(unit.toMillis(timeout), WAIT_MILLIS), MILLISECONDS); + } + + @Override + public int waitFor() throws InterruptedException { + exited.await(); + return exitCode; + } + + @Override + public int exitValue() { + if (isAlive()) { + throw new IllegalThreadStateException("still running"); + } + return exitCode; + } + + @Override + public boolean isAlive() { + return exited.getCount() > 0; + } + + @Override + public void destroy() { + stopRequests.incrementAndGet(); + if (stopsWhenAsked) { + exit(143); + } + } + + @Override + public Process destroyForcibly() { + destroy(); + return this; + } + + @Override + public OutputStream getOutputStream() { + return new ByteArrayOutputStream(); + } + + @Override + public InputStream getInputStream() { + return new ByteArrayInputStream(new byte[0]); + } + + @Override + public InputStream getErrorStream() { + return new ByteArrayInputStream(new byte[0]); + } + } }