diff --git a/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java b/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java index 8eea13de224..08592f83935 100644 --- a/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java +++ b/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, answersOnPipe(config.getAgentNamedPipe())); } else { log.warn("Trace agent path not set. Will not start trace agent process"); } @@ -45,7 +57,7 @@ public ExternalAgentLauncher(Config config) { dogStatsDProcessSupervisor = new ProcessSupervisor( - "dogstatsd", dogStatsDProcessBuilder, healthCheck(config.getDogStatsDNamedPipe())); + "dogstatsd", dogStatsDProcessBuilder, pipePresent(config.getDogStatsDNamedPipe())); } else { log.warn("DogStatsD path not set. Will not start DogStatsD process"); } @@ -63,18 +75,40 @@ public void close() { } } - private static ProcessSupervisor.HealthCheck healthCheck(String pipeName) { + /** Healthy while the agent's pipe is present and the agent answers a request on it. */ + private static ProcessSupervisor.HealthCheck answersOnPipe(String pipeName) { + return healthCheck(pipeName, true); + } + + /** Healthy while the agent's pipe is present. DogStatsD never replies on its pipe. */ + private static ProcessSupervisor.HealthCheck pipePresent(String pipeName) { + return healthCheck(pipeName, false); + } + + 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; - NamedPipeHealthCheck(String pipeName) { + // A request that got no answer in time. Only the supervisor thread runs the health check, so + // this needs no synchronization. + private FutureTask timedOutInfoRequest; + + NamedPipeHealthCheck(String pipeName, boolean checkResponse) { + this.checkResponse = checkResponse; if (pipeName.startsWith(NAMED_PIPE_PREFIX)) { this.pipe = new File(pipeName); } else { @@ -91,7 +125,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 +134,117 @@ 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. An answer that + * arrives after the timeout still counts at the next check, so a slow agent is not restarted. + */ + private boolean agentResponds() throws InterruptedException { + FutureTask lateRequest = timedOutInfoRequest; + if (lateRequest != null) { + if (!lateRequest.isDone()) { + return false; // still waiting for an answer + } + timedOutInfoRequest = null; + if (answered(lateRequest)) { + return true; + } + } + FutureTask infoRequest = new FutureTask<>(this::sendInfoRequest); + AgentThreadFactory.newAgentThread(PROCESS_SUPERVISOR, "-health-check", infoRequest, true) + .start(); + try { + return infoRequest.get(RESPONSE_TIMEOUT_MILLIS, MILLISECONDS); + } catch (TimeoutException e) { + timedOutInfoRequest = infoRequest; + return false; + } catch (ExecutionException e) { + return false; + } + } + + /** Returns the result of a finished request, without waiting. */ + private static boolean answered(FutureTask finishedRequest) + throws InterruptedException { + try { + return finishedRequest.get(); + } catch (ExecutionException 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"); + } } } 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..54c3a925424 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; @@ -36,7 +37,18 @@ 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 + // 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 = 400; + + // 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; @@ -45,8 +57,12 @@ 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 long currentProcessStartedMillis; + private int failedHealthChecks; + private final FaultBackoff faultBackoff = new FaultBackoff(); + private int failuresInARow; private volatile boolean stopping = false; @@ -70,23 +86,26 @@ 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) { Thread.sleep(delayMillis); } - currentHealth = healthCheck.run(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; } 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(); @@ -96,47 +115,184 @@ private void mainLoop() { } } + private void logHealthCheckChange(Health result) { + Health previous = lastCheckResult; + lastCheckResult = result; + if (result == previous) { + return; + } + if (result == HEALTHY) { + // 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( + "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) { 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); } } - 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; - faults = 0; + logLifecycle( + "Started process [{}] with pid {} from this process (pid {})", + imageName, + pidOrUnknown(currentProcess), + PidHelper.getPid()); + } + + String childPid = pidOrUnknown(currentProcess); + + // Keep checking on the process until it exits + while (!currentProcess.waitFor(HEALTHY_DELAY_MILLIS, MILLISECONDS)) { + long upMillis = System.currentTimeMillis() - currentProcessStartedMillis; + faultBackoff.recordUptime(upMillis); + if (upMillis >= STABLE_RUN_MILLIS) { + failuresInARow = 0; // it stayed up, so log its lifecycle at info level again + } + if (healthCheck != ALWAYS_READY && !passesHealthCheck(childPid)) { + restartUnhealthyProcess(childPid); + return; + } } + int code = currentProcess.exitValue(); + long ranMillis = System.currentTimeMillis() - currentProcessStartedMillis; + boolean stayedUp = ranMillis >= STABLE_RUN_MILLIS; + // An exit before the process stayed up is a fault whatever its code. The trace agent exits + // with 0 when APM is disabled, for example, and restarting that every 100 ms never ends. + boolean failed = code != 0 || !stayedUp; + currentHealth = failed ? FAULTED : INTERRUPTED; - // Block until the process exits - int code = currentProcess.waitFor(); - log.debug("Process [{}] has exited with code {}", imageName, code); - currentHealth = code == 0 ? INTERRUPTED : FAULTED; + faultBackoff.recordUptime(ranMillis); + if (stayedUp) { + failuresInARow = 0; + } + logLifecycle( + "Process [{}] with pid {} exited with code {} after {} ms", + imageName, + childPid, + code, + ranMillis); + if (failed && ++failuresInARow == QUIET_AFTER_FAILURES) { + log.warn( + "Process [{}] has failed {} 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; } + /** + * 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) { - log.debug("Stopping process: [{}]", imageName); + String childPid = pidOrUnknown(currentProcess); + log.info("Stopping process [{}] with pid {}", imageName, childPid); + // The supervised agents run only on Windows, where destroy() and destroyForcibly() both call + // TerminateProcess: the process stops at once, without a chance to flush. currentProcess.destroy(); - if (currentProcess.isAlive()) { - currentProcess.destroyForcibly(); + boolean stopped = false; + try { + stopped = currentProcess.waitFor(STOP_WAIT_MILLIS, MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + 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); } } + 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 +301,38 @@ 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); + } + } + + /** + * 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 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..19fcfc89e5a 100644 --- a/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java +++ b/internal-api/src/test/java/datadog/trace/util/ProcessSupervisorTest.java @@ -1,18 +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()); @@ -52,6 +95,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(); @@ -96,4 +169,261 @@ 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() >= 1, "asks it to stop"); + 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")); + } + } + + @Test + void backsOffWhenTheProcessKeepsExitingWithCodeZero() throws Exception { + // the trace agent exits with 0 at once when APM is disabled + List started = new CopyOnWriteArrayList<>(); + ProcessBuilder processBuilder = mock(ProcessBuilder.class); + when(processBuilder.start()) + .thenAnswer( + invocation -> { + FakeProcess process = startFakeProcess(started, true); + process.exit(0); + return process; + }); + + try (ProcessSupervisor ignored = new ProcessSupervisor("test", processBuilder)) { + new PollingConditions(5).eventually(() -> assertEquals(1, started.size())); + // the first fault waits 2 s; restarting after a clean exit would take about 100 ms + Thread.sleep(1_000); + assertEquals(1, started.size()); + } + } + + 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 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]); + } + } }