diff --git a/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java b/communication/src/main/java/datadog/communication/ddagent/ExternalAgentLauncher.java index 8eea13de224..e3dfdfff248 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, 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 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 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"); + } } } diff --git a/dd-java-agent/agent-logging/build.gradle b/dd-java-agent/agent-logging/build.gradle index 8e8b9d97fc8..ca2da6ec8c0 100644 --- a/dd-java-agent/agent-logging/build.gradle +++ b/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 diff --git a/dd-java-agent/agent-logging/src/main/java/datadog/trace/logging/simplelogger/AasPerProcessLogFile.java b/dd-java-agent/agent-logging/src/main/java/datadog/trace/logging/simplelogger/AasPerProcessLogFile.java new file mode 100644 index 00000000000..c935e3a729f --- /dev/null +++ b/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. + * + *

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: -. and rolled -_. + 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(); + } + } +} diff --git a/dd-java-agent/agent-logging/src/main/java/datadog/trace/logging/simplelogger/SLCompatSettings.java b/dd-java-agent/agent-logging/src/main/java/datadog/trace/logging/simplelogger/SLCompatSettings.java index f3cc1a57a27..f345606ad01 100644 --- a/dd-java-agent/agent-logging/src/main/java/datadog/trace/logging/simplelogger/SLCompatSettings.java +++ b/dd-java-agent/agent-logging/src/main/java/datadog/trace/logging/simplelogger/SLCompatSettings.java @@ -181,14 +181,22 @@ static PrintStream getPrintStream(String logFile) { printStreamWrapper = new PrintStreamWrapper(System.out); break; default: - FileOutputStream outputStream = null; + java.io.OutputStream outputStream = null; try { File outputFile = new File(logFile); File parentFile = outputFile.getParentFile(); if (parentFile != null) { parentFile.mkdirs(); } - outputStream = new FileOutputStream(outputFile); + if (AasPerProcessLogFile.isEnabled()) { + // one file per JVM instead of truncating a shared file + AasPerProcessLogFile.RollingFileOutputStream rolling = + AasPerProcessLogFile.open(outputFile); + outputStream = rolling; + outputFile = rolling.currentFile(); + } else { + outputStream = new FileOutputStream(outputFile); + } PrintStream printStream = new PrintStream(outputStream, true); LogReporter.register(outputFile); return printStream; diff --git a/dd-java-agent/agent-logging/src/test/java/datadog/trace/logging/simplelogger/AasPerProcessLogFileTest.java b/dd-java-agent/agent-logging/src/test/java/datadog/trace/logging/simplelogger/AasPerProcessLogFileTest.java new file mode 100644 index 00000000000..c3e6d0ccad3 --- /dev/null +++ b/dd-java-agent/agent-logging/src/test/java/datadog/trace/logging/simplelogger/AasPerProcessLogFileTest.java @@ -0,0 +1,263 @@ +package datadog.trace.logging.simplelogger; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.concurrent.TimeUnit.DAYS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.environment.EnvironmentVariables; +import datadog.environment.EnvironmentVariables.EnvironmentVariablesProvider; +import datadog.trace.logging.PrintStreamWrapper; +import java.io.File; +import java.io.IOException; +import java.io.PrintStream; +import java.lang.management.ManagementFactory; +import java.nio.file.Files; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.TreeSet; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.tabletest.junit.TableTest; + +class AasPerProcessLogFileTest { + private static final String PID = ManagementFactory.getRuntimeMXBean().getName().split("@", 2)[0]; + + private final Map environment = new HashMap<>(); + private EnvironmentVariablesProvider originalEnvironment; + + @TempDir File logDirectory; + + @BeforeEach + void useTestEnvironment() { + originalEnvironment = EnvironmentVariables.provider; + EnvironmentVariables.provider = + new EnvironmentVariablesProvider() { + @Override + public String get(String name) { + return environment.get(name); + } + + @Override + public Map getAll() { + return environment; + } + }; + } + + @AfterEach + void restoreEnvironment() { + EnvironmentVariables.provider = originalEnvironment; + } + + @TableTest({ + "scenario | azureAppServices | perProcessFile", + "enabled with 1 | 1 | true ", + "enabled with true | true | true ", + "enabled with TRUE | TRUE | true ", + "disabled with false | false | false ", + "not set | | false " + }) + void writesThisJvmsLogToItsOwnFileOnlyInAzureAppService( + String azureAppServices, boolean perProcessFile) throws Exception { + environment.put("DD_AZURE_APP_SERVICES", azureAppServices); + environment.put("DD_TRACE_LOGFILE_RETENTION_DAYS", "0"); + File configuredFile = new File(logDirectory, "dd-java-tracer.log"); + + PrintStream printStream = SLCompatSettings.getPrintStream(configuredFile.getPath()); + printStream.print("a log line"); + printStream.close(); + + File expectedFile = + perProcessFile ? new File(logDirectory, "dd-java-tracer-" + PID + ".log") : configuredFile; + assertEquals(fileNames(expectedFile.getName()), fileNames(logDirectory.list())); + assertEquals("a log line", contents(expectedFile)); + } + + @Test + void fallsBackToStandardErrorWhenThePerProcessLogFileCannotBeOpened() throws Exception { + environment.put("DD_AZURE_APP_SERVICES", "true"); + File notADirectory = new File(logDirectory, "not-a-directory"); + assertTrue(notADirectory.createNewFile()); + + PrintStream printStream = + SLCompatSettings.getPrintStream(new File(notADirectory, "dd-java-tracer.log").getPath()); + waitForCleanup(); + + assertSame(System.err, ((PrintStreamWrapper) printStream).getOriginalPrintStream()); + } + + @TableTest({ + "scenario | configuredName | perProcessName ", + "with extension | dd-java-tracer.log | dd-java-tracer-{pid}.log", + "without extension | dd-java-tracer | dd-java-tracer-{pid} ", + "only an extension | .log | .log-{pid} " + }) + void namesThePerProcessFileAfterTheConfiguredFileAndThePid( + String configuredName, String perProcessName) throws IOException { + environment.put("DD_TRACE_LOGFILE_RETENTION_DAYS", "0"); + + try (AasPerProcessLogFile.RollingFileOutputStream out = + AasPerProcessLogFile.open(new File(logDirectory, configuredName))) { + assertEquals( + new File(logDirectory, perProcessName.replace("{pid}", PID)).getAbsoluteFile(), + out.currentFile().getAbsoluteFile()); + } + } + + @Test + void rollsOverToNumberedFilesAtTheConfiguredSize() throws IOException { + environment.put("DD_MAX_LOGFILE_SIZE", "10"); + environment.put("DD_TRACE_LOGFILE_RETENTION_DAYS", "0"); + + try (AasPerProcessLogFile.RollingFileOutputStream out = + AasPerProcessLogFile.open(new File(logDirectory, "tracer.log"))) { + // a write larger than the limit still goes to the empty file instead of leaving it empty + out.write(bytes("0123456789ABC")); + out.write(bytes("1234")); + out.write(bytes("567890")); + out.write('!'); + out.flush(); + + assertEquals(new File(logDirectory, "tracer-" + PID + "_002.log"), out.currentFile()); + } + assertEquals( + fileNames( + "tracer-" + PID + ".log", "tracer-" + PID + "_001.log", "tracer-" + PID + "_002.log"), + fileNames(logDirectory.list())); + assertEquals("0123456789ABC", contents(new File(logDirectory, "tracer-" + PID + ".log"))); + assertEquals("1234567890", contents(new File(logDirectory, "tracer-" + PID + "_001.log"))); + assertEquals("!", contents(new File(logDirectory, "tracer-" + PID + "_002.log"))); + } + + @TableTest({ + "scenario | maxLogFileSize | files", + "limit reached | 150 | 2 ", + "zero disables | 0 | 1 ", + "invalid: default | not-a-number | 1 ", + "unset: default | | 1 " + }) + void rollsOverOnlyWhenAPositiveSizeIsConfigured(String maxLogFileSize, int files) + throws IOException { + environment.put("DD_MAX_LOGFILE_SIZE", maxLogFileSize); + environment.put("DD_TRACE_LOGFILE_RETENTION_DAYS", "0"); + + try (AasPerProcessLogFile.RollingFileOutputStream out = + AasPerProcessLogFile.open(new File(logDirectory, "tracer.log"))) { + out.write(new byte[100]); + out.write(new byte[100]); + } + assertEquals(files, logDirectory.list().length); + } + + @Test + void appendsToAnExistingFileAndCountsItTowardsTheSize() throws IOException { + File existing = new File(logDirectory, "tracer-" + PID + ".log"); + Files.write(existing.toPath(), bytes("earlier")); + + try (AasPerProcessLogFile.RollingFileOutputStream out = + new AasPerProcessLogFile.RollingFileOutputStream( + logDirectory, "tracer-" + PID, ".log", 10)) { + out.write(bytes("later")); + } + + assertEquals("earlier", contents(existing)); + assertEquals("later", contents(new File(logDirectory, "tracer-" + PID + "_001.log"))); + } + + @TableTest({ + "scenario | retentionDays | ageDays | deleted", + "default keeps 32 days | | 20 | false ", + "default deletes after | | 40 | true ", + "configured keeps | 5 | 1 | false ", + "configured deletes after | 5 | 10 | true ", + "zero disables cleanup | 0 | 400 | false ", + "invalid: default keeps | soon | 20 | false ", + "invalid: default deletes | soon | 40 | true " + }) + void deletesThisTracersPerProcessFilesOlderThanTheRetention( + String retentionDays, int ageDays, boolean deleted) throws Exception { + environment.put("DD_TRACE_LOGFILE_RETENTION_DAYS", retentionDays); + File oldFile = oldFile("dd-java-tracer-1234.log", ageDays); + + AasPerProcessLogFile.open(new File(logDirectory, "dd-java-tracer.log")).close(); + waitForCleanup(); + + assertEquals(deleted, !oldFile.exists()); + } + + @Test + void cleanupDeletesOnlyThisTracersPerProcessFiles() throws Exception { + File ownFile = oldFile("dd-java-tracer-1234.log", 40); + File ownRolledFile = oldFile("dd-java-tracer-1234_001.log", 40); + File configuredFile = oldFile("dd-java-tracer.log", 40); + File otherTracerFile = oldFile("dotnet-tracer-managed-1234.log", 40); + File otherExtension = oldFile("dd-java-tracer-1234.txt", 40); + File notAPid = oldFile("dd-java-tracer-old.log", 40); + + AasPerProcessLogFile.open(new File(logDirectory, "dd-java-tracer.log")).close(); + waitForCleanup(); + + assertFalse(ownFile.exists()); + assertFalse(ownRolledFile.exists()); + assertTrue(configuredFile.exists()); + assertTrue(otherTracerFile.exists()); + assertTrue(otherExtension.exists()); + assertTrue(notAPid.exists()); + } + + @Test + void cleanupStopsQuietlyWhenAFileCannotBeDeleted() throws Exception { + // a non-empty directory with a per-process file name cannot be deleted + File undeletable = new File(logDirectory, "dd-java-tracer-1234.log"); + assertTrue(new File(undeletable, "child").mkdirs()); + assertTrue(undeletable.setLastModified(daysAgo(40))); + + try (AasPerProcessLogFile.RollingFileOutputStream out = + AasPerProcessLogFile.open(new File(logDirectory, "dd-java-tracer.log"))) { + waitForCleanup(); + out.write(bytes("still logging")); + out.flush(); + + assertTrue(undeletable.exists()); + assertEquals("still logging", contents(out.currentFile())); + } + } + + private File oldFile(String name, int ageDays) throws IOException { + File file = new File(logDirectory, name); + assertTrue(file.createNewFile()); + assertTrue(file.setLastModified(daysAgo(ageDays))); + return file; + } + + private static long daysAgo(int days) { + return System.currentTimeMillis() - DAYS.toMillis(days); + } + + /** Waits for the background cleanup that opening a per-process log file starts. */ + private static void waitForCleanup() throws InterruptedException { + for (Thread thread : Thread.getAllStackTraces().keySet()) { + if ("dd-log-file-cleanup".equals(thread.getName())) { + thread.join(10_000); + } + } + } + + private static TreeSet fileNames(String... names) { + return new TreeSet<>(Arrays.asList(names)); + } + + private static byte[] bytes(String text) { + return text.getBytes(UTF_8); + } + + private static String contents(File file) throws IOException { + return new String(Files.readAllBytes(file.toPath()), UTF_8); + } +} 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..5918e1d4958 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 = 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; @@ -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,179 @@ 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()); } - // Block until the process exits - int code = currentProcess.waitFor(); - log.debug("Process [{}] has exited with code {}", imageName, code); + String childPid = pidOrUnknown(currentProcess); + + // 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() - currentProcessStartedMillis; + faultBackoff.recordUptime(ranMillis); + 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; } + /** + * 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); - 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); + 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 +296,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..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,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,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]); + } + } }