Fix trace and StatsD connections over Windows named pipes

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));

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 : "<auto-detect>") + (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<DDAgentStatsDConnection> {
+    public static final RecreateTask INSTANCE = new RecreateTask();
+
+    @Override
+    public void run(final DDAgentStatsDConnection target) {
+      target.recreateClient();
+    }
+  }
+
   private static final class ConnectTask
       implements AgentTaskScheduler.Task<DDAgentStatsDConnection> {
     public static final ConnectTask INSTANCE = new ConnectTask();

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

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<File, NamedPipeSocket> 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

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