diff --git a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java index 2cb8a004..8a2a3a8e 100644 --- a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java +++ b/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 diff --git a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java index cfef457d..be0a9a26 100644 --- a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java +++ b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java @@ -431,6 +431,15 @@ private Callable 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 { + private final NamedPipeSocketAddress pipe; + final Callable fallbackLookup; + + SystemPropertyNamedPipeLookup( + NamedPipeSocketAddress pipe, Callable 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 namedPipeFallbackAddressLookup() { + return new Callable() { + @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 staticAddress(final String hostname, final int port) { try { return staticAddressResolution(hostname, port); diff --git a/src/test/java/com/timgroup/statsd/BuilderAddressTest.java b/src/test/java/com/timgroup/statsd/BuilderAddressTest.java index ea344be5..5d799794 100644 --- a/src/test/java/com/timgroup/statsd/BuilderAddressTest.java +++ b/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() { + @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()); + } } diff --git a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java index bc349081..6145130b 100644 --- a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java +++ b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java @@ -1,5 +1,7 @@ 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; @@ -7,11 +9,76 @@ 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 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 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 env = new HashMap<>();