Read dd.dogstatsd.pipe.name for the default named pipe

src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java

@@ -1,5 +1,6 @@
 package com.timgroup.statsd;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
@@ -59,6 +60,7 @@ public class NonBlockingStatsDClient implements StatsDClient {
     public static final String DD_DOGSTATSD_PORT_ENV_VAR = "DD_DOGSTATSD_PORT";
     public static final String DD_AGENT_HOST_ENV_VAR = "DD_AGENT_HOST";
     public static final String DD_NAMED_PIPE_ENV_VAR = "DD_DOGSTATSD_PIPE_NAME";
+    static final String DD_NAMED_PIPE_SYSTEM_PROPERTY = "dd.dogstatsd.pipe.name";
     public static final String DD_ENTITY_ID_ENV_VAR = "DD_ENTITY_ID";
     private static final String ENTITY_ID_TAG_NAME = "dd.internal.entity_id";
     public static final String ORIGIN_DETECTION_ENABLED_ENV_VAR = "DD_ORIGIN_DETECTION_ENABLED";
@@ -241,12 +243,37 @@ public NonBlockingStatsDClient(final NonBlockingStatsDClientBuilder builder)
         externalEnv = originDetectionEnabled ? Utf8.sanitize(env.get("DD_EXTERNAL_ENV")) : "";
 
         try {
-            clientChannel =
-                    createByteChannel(
-                            builder.addressLookup,
-                            builder.timeout,
-                            builder.connectionTimeout,
-                            builder.socketBufferSize);
+            ClientChannel channel;
+            try {
+                channel =
+                        createByteChannel(
+                                builder.addressLookup,
+                                builder.timeout,
+                                builder.connectionTimeout,
+                                builder.socketBufferSize);
+            } catch (FileNotFoundException e) {
+                if (!(builder.addressLookup
+                        instanceof NonBlockingStatsDClientBuilder.SystemPropertyNamedPipeLookup)) {
+                    throw e;
+                }
+                // The system property pipe may not exist yet; its server also listens on UDP.
+                // Report the fallback once, at build time, so it doesn't repeat per message.
+                handler.handle(
+                        new StatsDClientException(
+                                "Could not open named pipe from system property "
+                                        + DD_NAMED_PIPE_SYSTEM_PROPERTY
+                                        + "; falling back to UDP",
+                                e));
+                channel =
+                        createByteChannel(
+                                ((NonBlockingStatsDClientBuilder.SystemPropertyNamedPipeLookup)
+                                                builder.addressLookup)
+                                        .fallbackLookup,
+                                builder.timeout,
+                                builder.connectionTimeout,
+                                builder.socketBufferSize);
+            }
+            clientChannel = channel;
 
             ThreadFactory threadFactory =
                     builder.threadFactory != null

src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java

@@ -431,6 +431,15 @@ private Callable<SocketAddress> getAddressLookup() {
             return staticAddress(hostname, port);
         }
 
+        // Next, try the system property, which can be set after the JVM has started.
+        String namedPipeFromProperty =
+                System.getProperty(NonBlockingStatsDClient.DD_NAMED_PIPE_SYSTEM_PROPERTY);
+        if (namedPipeFromProperty != null) {
+            return new SystemPropertyNamedPipeLookup(
+                    new NamedPipeSocketAddress(namedPipeFromProperty),
+                    namedPipeFallbackAddressLookup());
+        }
+
         // Next, try various environment variables.
         String url = env.get(NonBlockingStatsDClient.DD_DOGSTATSD_URL_ENV_VAR);
         if (url != null) {
@@ -547,6 +556,43 @@ public SocketAddress call() {
         };
     }
 
+    /**
+     * Named pipe lookup from the system property, with the address the client uses instead when the
+     * pipe cannot be opened at build time.
+     */
+    static final class SystemPropertyNamedPipeLookup implements Callable<SocketAddress> {
+        private final NamedPipeSocketAddress pipe;
+        final Callable<SocketAddress> fallbackLookup;
+
+        SystemPropertyNamedPipeLookup(
+                NamedPipeSocketAddress pipe, Callable<SocketAddress> fallbackLookup) {
+            this.pipe = pipe;
+            this.fallbackLookup = fallbackLookup;
+        }
+
+        @Override
+        public SocketAddress call() {
+            return pipe;
+        }
+    }
+
+    /**
+     * Resolves the address from "DD_AGENT_HOST" (default "localhost") and "DD_DOGSTATSD_PORT" when
+     * called, so that a pipe which opens never depends on these values.
+     */
+    private Callable<SocketAddress> namedPipeFallbackAddressLookup() {
+        return new Callable<SocketAddress>() {
+            @Override
+            public SocketAddress call() throws Exception {
+                String hostFromEnv = env.get(NonBlockingStatsDClient.DD_AGENT_HOST_ENV_VAR);
+                if (hostFromEnv == null) {
+                    hostFromEnv = "localhost";
+                }
+                return volatileAddressResolution(hostFromEnv, getPortFromEnvVar(port)).call();
+            }
+        };
+    }
+
     private static Callable<SocketAddress> staticAddress(final String hostname, final int port) {
         try {
             return staticAddressResolution(hostname, port);

src/test/java/com/timgroup/statsd/BuilderAddressTest.java

@@ -9,7 +9,9 @@
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.Callable;
 import jnr.unixsocket.UnixSocketAddress;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -196,6 +198,11 @@ public void set() {
         set(NonBlockingStatsDClient.DD_NAMED_PIPE_ENV_VAR, pipe);
     }
 
+    @After
+    public void clearNamedPipeSystemProperty() {
+        System.clearProperty(NonBlockingStatsDClient.DD_NAMED_PIPE_SYSTEM_PROPERTY);
+    }
+
     void set(String name, String val) {
         if (val != null) {
             env.put(name, val);
@@ -245,4 +252,52 @@ public void address_resolution() throws Exception {
                         .resolve();
         assertEquals(new NamedPipeSocketAddress("ook"), b.addressLookup.call());
     }
+
+    @Test(timeout = 5000L)
+    public void system_property_named_pipe_resolution() throws Exception {
+        System.setProperty(NonBlockingStatsDClient.DD_NAMED_PIPE_SYSTEM_PROPERTY, "sysprop");
+        NonBlockingStatsDClientBuilder b;
+
+        // Default configuration uses the system property regardless of environment variables.
+        b = new NonBlockingStatsDClientBuilder().withEnvironmentVariables(env).resolve();
+        assertEquals(new NamedPipeSocketAddress("sysprop"), b.addressLookup.call());
+        assertEquals(b.addressLookup, b.telemetryAddressLookup);
+
+        // Explicit configuration is used regardless of the system property.
+        b =
+                new NonBlockingStatsDClientBuilder()
+                        .withEnvironmentVariables(env)
+                        .hostname("2.2.2.2")
+                        .port(2222)
+                        .resolve();
+        assertEquals(new InetSocketAddress("2.2.2.2", 2222), b.addressLookup.call());
+
+        b =
+                new NonBlockingStatsDClientBuilder()
+                        .withEnvironmentVariables(env)
+                        .namedPipe("ook")
+                        .resolve();
+        assertEquals(new NamedPipeSocketAddress("ook"), b.addressLookup.call());
+
+        final SocketAddress explicit = new InetSocketAddress("3.3.3.3", 3333);
+        b =
+                new NonBlockingStatsDClientBuilder()
+                        .withEnvironmentVariables(env)
+                        .addressLookup(
+                                new Callable<SocketAddress>() {
+                                    @Override
+                                    public SocketAddress call() {
+                                        return explicit;
+                                    }
+                                })
+                        .resolve();
+        assertEquals(explicit, b.addressLookup.call());
+
+        b =
+                new NonBlockingStatsDClientBuilder()
+                        .withEnvironmentVariables(env)
+                        .address("udp://4.4.4.4:4444")
+                        .resolve();
+        assertEquals(new InetSocketAddress("4.4.4.4", 4444), b.addressLookup.call());
+    }
 }

src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java

@@ -1,17 +1,84 @@
 package com.timgroup.statsd;
 
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Random;
+import org.junit.After;
 import org.junit.Test;
 import org.junit.function.ThrowingRunnable;
 
 public class NonBlockingStatsDClientBuilderTest {
 
+    @After
+    public void clearNamedPipeSystemProperty() {
+        System.clearProperty(NonBlockingStatsDClient.DD_NAMED_PIPE_SYSTEM_PROPERTY);
+    }
+
+    // A pipe inside a missing directory fails to open on every platform without creating a file.
+    private static String unopenablePipeName() {
+        return "missing-dir-" + new Random().nextInt(1000000) + "/pipe";
+    }
+
+    @Test(timeout = 5000L)
+    public void system_property_named_pipe_falls_back_to_udp() throws Exception {
+        System.setProperty(
+                NonBlockingStatsDClient.DD_NAMED_PIPE_SYSTEM_PROPERTY, unopenablePipeName());
+        final UDPDummyStatsDServer server = new UDPDummyStatsDServer(0);
+        final Map<String, String> env = new HashMap<>();
+        env.put(NonBlockingStatsDClient.DD_AGENT_HOST_ENV_VAR, "localhost");
+        env.put(
+                NonBlockingStatsDClient.DD_DOGSTATSD_PORT_ENV_VAR,
+                Integer.toString(server.getPort()));
+        final RecordingErrorHandler errorHandler = new RecordingErrorHandler();
+        NonBlockingStatsDClient client = null;
+        try {
+            client =
+                    new NonBlockingStatsDClientBuilder()
+                            .withEnvironmentVariables(env)
+                            .prefix("my.prefix")
+                            .enableAggregation(false)
+                            .errorHandler(errorHandler)
+                            .build();
+
+            client.gauge("value", 42);
+            client.gauge("value", 43);
+            server.waitForMessage("my.prefix.value:43");
+            assertThat(server.messagesReceived(), hasItem("my.prefix.value:42|g"));
+            // The fallback is reported once at build, not for each message sent.
+            assertEquals(1, errorHandler.getExceptions().size());
+        } finally {
+            if (client != null) {
+                client.stop();
+            }
+            server.close();
+        }
+    }
+
+    @Test(timeout = 5000L)
+    public void environment_named_pipe_does_not_fall_back() throws Exception {
+        final Map<String, String> env = new HashMap<>();
+        env.put(NonBlockingStatsDClient.DD_NAMED_PIPE_ENV_VAR, unopenablePipeName());
+        env.put(NonBlockingStatsDClient.DD_AGENT_HOST_ENV_VAR, "localhost");
+        assertThrows(
+                StatsDClientException.class,
+                new ThrowingRunnable() {
+                    @Override
+                    public void run() {
+                        new NonBlockingStatsDClientBuilder()
+                                .withEnvironmentVariables(env)
+                                .enableAggregation(false)
+                                .build();
+                    }
+                });
+    }
+
     @Test(timeout = 5000L)
     public void origin_detection_env_false() throws Exception {
         final Map<String, String> env = new HashMap<>();