diff --git a/communication/src/main/java/datadog/communication/http/OkHttpUtils.java b/communication/src/main/java/datadog/communication/http/OkHttpUtils.java index 44105ae772d..9dc5cbd4c41 100644 --- a/communication/src/main/java/datadog/communication/http/OkHttpUtils.java +++ b/communication/src/main/java/datadog/communication/http/OkHttpUtils.java @@ -177,12 +177,13 @@ private static OkHttpClient buildHttpClient( .dispatcher( dispatcher != null ? dispatcher : new Dispatcher(RejectingExecutorService.INSTANCE)); + final boolean useNamedPipe = unixDomainSocketPath == null && namedPipe != null; if (unixDomainSocketPath != null) { builder.socketFactory( new UnixDomainSocketFactory( new File(unixDomainSocketPath), useJdkUnixDomainSocket, agentConfiguredUsingDefault)); log.debug("Using UnixDomainSocket as http transport"); - } else if (namedPipe != null) { + } else if (useNamedPipe) { builder.socketFactory(new NamedPipeSocketFactory(namedPipe)); log.debug("Using NamedPipe as http transport"); } @@ -204,7 +205,11 @@ private static OkHttpClient buildHttpClient( builder.retryOnConnectionFailure(retryOnConnectionFailure); } - if (maxRunningRequests != null) { + if (useNamedPipe) { + // NamedPipeSocket can't time out a read, so OkHttp's health check on a pooled connection can + // block forever. Don't keep idle pipe connections; each request opens its own. + builder.connectionPool(new ConnectionPool(0, 1, SECONDS)); + } else if (maxRunningRequests != null) { // Reusing connections causes non daemon threads to be created which causes agent to prevent // app from exiting. See https://github.com/square/okhttp/issues/4029 for some details. builder.connectionPool(new ConnectionPool(maxRunningRequests, 1, SECONDS)); diff --git a/products/metrics/metrics-lib/src/main/java/datadog/metrics/impl/statsd/DDAgentStatsDConnection.java b/products/metrics/metrics-lib/src/main/java/datadog/metrics/impl/statsd/DDAgentStatsDConnection.java index f64fdbd285c..efb0c51f6cf 100644 --- a/products/metrics/metrics-lib/src/main/java/datadog/metrics/impl/statsd/DDAgentStatsDConnection.java +++ b/products/metrics/metrics-lib/src/main/java/datadog/metrics/impl/statsd/DDAgentStatsDConnection.java @@ -3,6 +3,8 @@ import static datadog.trace.api.ConfigDefaults.DEFAULT_DOGSTATSD_SOCKET_PATH; import static datadog.trace.util.AgentThreadFactory.AgentThread.STATSD_CLIENT; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.MINUTES; +import static java.util.concurrent.TimeUnit.NANOSECONDS; import static java.util.concurrent.TimeUnit.SECONDS; import com.timgroup.statsd.NoOpDirectStatsDClient; @@ -11,6 +13,7 @@ import datadog.common.filesystem.Files; import datadog.environment.OperatingSystem; import datadog.logging.IOLogger; +import datadog.logging.RatelimitedLogger; import datadog.trace.api.Config; import datadog.trace.util.AgentTaskScheduler; import datadog.trace.util.AgentThreadFactory; @@ -23,6 +26,7 @@ final class DDAgentStatsDConnection implements StatsDClientErrorHandler { private static final Logger log = LoggerFactory.getLogger(DDAgentStatsDConnection.class); private static final IOLogger ioLogger = new IOLogger(log); + private static final RatelimitedLogger recreateLogger = new RatelimitedLogger(log, 5, MINUTES); private static final com.timgroup.statsd.StatsDClient NO_OP = new NoOpDirectStatsDClient(); @@ -31,18 +35,19 @@ final class DDAgentStatsDConnection implements StatsDClientErrorHandler { private static final AgentThreadFactory STATSD_CLIENT_THREAD_FACTORY = new AgentThreadFactory(STATSD_CLIENT); - private static final int RETRY_DELAY = 10; - private static final int MAX_RETRIES = 20; + private static final long RETRY_DELAY = 10; + private static final long MAX_RETRY_DELAY = 60; private boolean usingDefaultPort; private volatile String host; - private volatile Integer port; + volatile Integer port; private final String namedPipe; private final boolean useAggregation; private final AtomicInteger clientCount = new AtomicInteger(0); private final AtomicInteger errorCount = new AtomicInteger(0); private final AtomicInteger retries = new AtomicInteger(0); + private final WriteFailureStreak writeFailureStreak = new WriteFailureStreak(); volatile com.timgroup.statsd.StatsDClient statsd = NO_OP; @@ -59,6 +64,10 @@ public void handle(final Exception e) { errorCount.incrementAndGet(); String message = e.getClass().getSimpleName() + " in StatsD client - " + statsDAddress(); ioLogger.error(message, e); + if (writeFailureStreak.recordError(NANOSECONDS.toMillis(System.nanoTime()))) { + // handle() runs on a client thread, so close the client from another thread + AgentTaskScheduler.get().schedule(RecreateTask.INSTANCE, this, 0, MILLISECONDS); + } } public void acquire() { @@ -159,20 +168,28 @@ private void doConnect() { try { statsd = clientBuilder.build(); - if (log.isDebugEnabled()) { + int failedAttempts = retries.getAndSet(0); + if (failedAttempts > 0) { + log.info( + "StatsD connected to {} after {} failed attempts", statsDAddress(), failedAttempts); + } else if (log.isDebugEnabled()) { log.debug("StatsD connected to {}", statsDAddress()); } } catch (final Exception e) { - if (retries.getAndIncrement() < MAX_RETRIES) { - if (log.isDebugEnabled()) { - log.debug( - "Scheduling StatsD connection in {} seconds - {}", RETRY_DELAY, statsDAddress()); - } - AgentTaskScheduler.get() - .scheduleWithJitter(ConnectTask.INSTANCE, this, RETRY_DELAY, SECONDS); - } else { - log.debug("Max retries have been reached. Will not attempt again."); + int failedAttempts = retries.incrementAndGet(); + long delaySeconds = connectRetryDelaySeconds(failedAttempts); + if (failedAttempts == 1) { + log.warn( + "Unable to create StatsD client - {} - Will keep retrying, at least every {} seconds: {}", + statsDAddress(), + MAX_RETRY_DELAY, + e.getMessage()); + } else if (log.isDebugEnabled()) { + log.debug( + "Scheduling StatsD connection in {} seconds - {}", delaySeconds, statsDAddress()); } + // no jitter, so the delay stays within MAX_RETRY_DELAY + AgentTaskScheduler.get().schedule(ConnectTask.INSTANCE, this, delaySeconds, SECONDS); } catch (Throwable t) { if (log.isDebugEnabled()) { // Display full stack traces on debug logs @@ -195,9 +212,26 @@ private void doConnect() { } } + /** + * Delay before the next connection attempt: doubles from {@link #RETRY_DELAY} up to {@link + * #MAX_RETRY_DELAY} seconds. Attempts never stop, so a DogStatsD that comes up late is still + * picked up. + */ + static long connectRetryDelaySeconds(int failedAttempts) { + long delaySeconds = RETRY_DELAY; + for (int i = 1; i < failedAttempts && delaySeconds < MAX_RETRY_DELAY; i++) { + delaySeconds *= 2; + } + return Math.min(delaySeconds, MAX_RETRY_DELAY); + } + @SuppressFBWarnings("DMI_HARDCODED_ABSOLUTE_FILENAME") - private void discoverConnectionSettings() { + void discoverConnectionSettings() { if (namedPipe != null) { + if (null == port) { + // the client ignores the port for a named pipe, but the builder needs one + port = 0; + } return; } @@ -238,6 +272,29 @@ void handleDefaultPortChange(final int newPort) { } } + /** + * The client never reopens its named pipe or socket, so after the DogStatsD it was writing to + * goes away, every write fails even once a new DogStatsD serves the same name. + */ + private void recreateClient() { + synchronized (this) { + if (NO_OP != statsd && clientCount.get() > 0) { + recreateLogger.warn( + "StatsD client writes to {} have kept failing for at least {} seconds; recreating the client", + statsDAddress(), + MILLISECONDS.toSeconds(WriteFailureStreak.RECREATE_AFTER_MILLIS)); + try { + statsd.close(); + } catch (final Exception e) { + log.debug("Problem closing StatsD client - {}", statsDAddress(), e); + } finally { + statsd = NO_OP; + } + doConnect(); + } + } + } + private void doClose() { synchronized (this) { if (NO_OP != statsd && 0 == clientCount.get()) { @@ -263,6 +320,49 @@ private String statsDAddress() { return (null != host ? host : "") + (null != port && port > 0 ? ":" + port : ""); } + /** + * Tracks how long StatsD writes have kept failing. Errors more than {@link #MAX_ERROR_GAP_MILLIS} + * apart start a new streak, and a streak needs {@link #MIN_ERRORS} errors spanning {@link + * #RECREATE_AFTER_MILLIS}, so brief or isolated timeouts under load do not recreate the client. + */ + static final class WriteFailureStreak { + static final long RECREATE_AFTER_MILLIS = 30_000; + // longer than the 30 second tracer health metrics flush, so a dead pipe keeps one streak going + static final long MAX_ERROR_GAP_MILLIS = 60_000; + static final int MIN_ERRORS = 3; + + private int errors; + private long firstErrorMillis; + private long lastErrorMillis; + + /** + * @return true when the client should be recreated; the streak then starts over + */ + synchronized boolean recordError(long nowMillis) { + if (errors == 0 || nowMillis - lastErrorMillis > MAX_ERROR_GAP_MILLIS) { + errors = 0; + firstErrorMillis = nowMillis; + } + errors++; + lastErrorMillis = nowMillis; + if (errors >= MIN_ERRORS && nowMillis - firstErrorMillis >= RECREATE_AFTER_MILLIS) { + errors = 0; + return true; + } + return false; + } + } + + private static final class RecreateTask + implements AgentTaskScheduler.Task { + public static final RecreateTask INSTANCE = new RecreateTask(); + + @Override + public void run(final DDAgentStatsDConnection target) { + target.recreateClient(); + } + } + private static final class ConnectTask implements AgentTaskScheduler.Task { public static final ConnectTask INSTANCE = new ConnectTask(); diff --git a/products/metrics/metrics-lib/src/test/java/datadog/metrics/impl/statsd/DDAgentStatsDConnectionTest.java b/products/metrics/metrics-lib/src/test/java/datadog/metrics/impl/statsd/DDAgentStatsDConnectionTest.java new file mode 100644 index 00000000000..4103f8f9542 --- /dev/null +++ b/products/metrics/metrics-lib/src/test/java/datadog/metrics/impl/statsd/DDAgentStatsDConnectionTest.java @@ -0,0 +1,76 @@ +package datadog.metrics.impl.statsd; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +class DDAgentStatsDConnectionTest { + + @Test + void connectRetryDelayDoublesUpToSixtySecondsAndNeverStops() { + assertEquals(10, DDAgentStatsDConnection.connectRetryDelaySeconds(1)); + assertEquals(20, DDAgentStatsDConnection.connectRetryDelaySeconds(2)); + assertEquals(40, DDAgentStatsDConnection.connectRetryDelaySeconds(3)); + assertEquals(60, DDAgentStatsDConnection.connectRetryDelaySeconds(4)); + assertEquals(60, DDAgentStatsDConnection.connectRetryDelaySeconds(1_000)); + assertEquals(60, DDAgentStatsDConnection.connectRetryDelaySeconds(Integer.MAX_VALUE)); + } + + @Test + void namedPipeConnectionUsesPortZero() { + DDAgentStatsDConnection connection = + new DDAgentStatsDConnection(null, null, "dogstatsd", false); + + connection.discoverConnectionSettings(); + + assertEquals(Integer.valueOf(0), connection.port); + } + + @Test + void writeErrorsAboutFifteenSecondsApartRecreateAfterThirtySeconds() { + DDAgentStatsDConnection.WriteFailureStreak streak = + new DDAgentStatsDConnection.WriteFailureStreak(); + + assertFalse(streak.recordError(0)); + assertFalse(streak.recordError(15_000)); + assertTrue(streak.recordError(30_000)); + + // the streak starts over after a recreate + assertFalse(streak.recordError(45_000)); + assertFalse(streak.recordError(60_000)); + assertTrue(streak.recordError(75_000)); + } + + @Test + void isolatedWriteErrorsDoNotRecreate() { + DDAgentStatsDConnection.WriteFailureStreak streak = + new DDAgentStatsDConnection.WriteFailureStreak(); + + assertFalse(streak.recordError(0)); + assertFalse(streak.recordError(30_000)); + } + + @Test + void briefBurstOfWriteErrorsDoesNotRecreate() { + DDAgentStatsDConnection.WriteFailureStreak streak = + new DDAgentStatsDConnection.WriteFailureStreak(); + + for (long now = 0; now < 29_000; now += 100) { + assertFalse(streak.recordError(now)); + } + } + + @Test + void gapOfMoreThanSixtySecondsStartsANewStreak() { + DDAgentStatsDConnection.WriteFailureStreak streak = + new DDAgentStatsDConnection.WriteFailureStreak(); + + assertFalse(streak.recordError(0)); + assertFalse(streak.recordError(15_000)); + assertFalse(streak.recordError(75_001)); + assertFalse(streak.recordError(90_000)); + assertTrue(streak.recordError(105_001)); + } +} diff --git a/utils/socket-utils/src/main/java/datadog/common/socket/NamedPipeSocketFactory.java b/utils/socket-utils/src/main/java/datadog/common/socket/NamedPipeSocketFactory.java index 02fe777d130..48da4d21003 100644 --- a/utils/socket-utils/src/main/java/datadog/common/socket/NamedPipeSocketFactory.java +++ b/utils/socket-utils/src/main/java/datadog/common/socket/NamedPipeSocketFactory.java @@ -4,13 +4,10 @@ import java.io.IOException; import java.net.InetAddress; import java.net.Socket; -import java.util.HashMap; -import java.util.Map; import javax.net.SocketFactory; public class NamedPipeSocketFactory extends SocketFactory { private static final String NAMED_PIPE_PREFIX = "\\\\.\\pipe\\"; - private static final Map fileToSocket = new HashMap<>(); private final File pipe; public NamedPipeSocketFactory(String pipeName) { @@ -19,19 +16,15 @@ public NamedPipeSocketFactory(String pipeName) { this.pipe = new File(pipeNameWithPrefix); } + NamedPipeSocketFactory(File pipe) { + this.pipe = pipe; + } + @Override public Socket createSocket() throws IOException { - // Getting a new socket is rare enough that simple synchronization is sufficient - synchronized (fileToSocket) { - NamedPipeSocket socket = fileToSocket.get(pipe); - - if (socket == null || socket.isClosed()) { - socket = new NamedPipeSocket(pipe); - fileToSocket.put(pipe, socket); - } - - return socket; - } + // Each connection gets its own pipe handle. A shared handle lets a blocked read on one + // connection, such as a pooled-connection health check, block every other request. + return new NamedPipeSocket(pipe); } @Override diff --git a/utils/socket-utils/src/test/java/datadog/common/socket/NamedPipeSocketFactoryTest.java b/utils/socket-utils/src/test/java/datadog/common/socket/NamedPipeSocketFactoryTest.java new file mode 100644 index 00000000000..d3251e822b0 --- /dev/null +++ b/utils/socket-utils/src/test/java/datadog/common/socket/NamedPipeSocketFactoryTest.java @@ -0,0 +1,28 @@ +package datadog.common.socket; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotSame; + +import java.io.File; +import java.net.Socket; +import java.nio.file.Path; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class NamedPipeSocketFactoryTest { + + @Test + void createsANewSocketForEachConnection(@TempDir Path tempDir) throws Exception { + // a regular file stands in for the pipe, since NamedPipeSocket only opens it as a file + File pipe = tempDir.resolve("pipe").toFile(); + NamedPipeSocketFactory factory = new NamedPipeSocketFactory(pipe); + + try (Socket first = factory.createSocket(); + Socket second = factory.createSocket()) { + assertNotSame(first, second); + + first.close(); + assertFalse(second.isClosed()); + } + } +}