From bcaa39f5f3b0a13b80156bc13488215c234832fa Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 9 Oct 2026 17:27:20 -0400 Subject: [PATCH 1/2] Read dd.dogstatsd.pipe.name for the default named pipe A JVM agent can set system properties after startup but not environment variables. serverless-compat-java will set dd.dogstatsd.pipe.name to a per-function pipe on Azure Functions shared plans, so a client built with default settings now uses it. Precedence: explicit builder settings, then the system property, then the existing DD_DOGSTATSD_URL, DD_DOGSTATSD_PIPE_NAME and DD_AGENT_HOST environment variables. If the system-property pipe can't be opened at build time, the client falls back to UDP (DD_AGENT_HOST and DD_DOGSTATSD_PORT, else localhost:8125) instead of throwing. The serverless compat binary also listens on UDP, so this covers a cold start where the client starts before the binary creates its pipe. Explicit and environment-variable pipes still throw as before. SVLS-8246 --- .../statsd/NonBlockingStatsDClient.java | 32 ++++++++-- .../NonBlockingStatsDClientBuilder.java | 46 ++++++++++++++ .../timgroup/statsd/BuilderAddressTest.java | 55 ++++++++++++++++ .../NonBlockingStatsDClientBuilderTest.java | 62 +++++++++++++++++++ 4 files changed, 189 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java index 2cb8a004..183a6528 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,30 @@ 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. + 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..df74077a 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,71 @@ 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())); + NonBlockingStatsDClient client = null; + try { + client = + new NonBlockingStatsDClientBuilder() + .withEnvironmentVariables(env) + .prefix("my.prefix") + .enableAggregation(false) + .build(); + + client.gauge("value", 42); + server.waitForMessage("my.prefix"); + assertThat(server.messagesReceived(), hasItem("my.prefix.value:42|g")); + } 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<>(); From c548a8ce2bd7442acf3cc1d20d241c1153ac53df Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 9 Oct 2026 17:38:17 -0400 Subject: [PATCH 2/2] Report the named pipe UDP fallback to the error handler The fallback from dd.dogstatsd.pipe.name to UDP was silent. The client now passes one StatsDClientException to the error handler when it falls back. It happens once, while the client is built, so it can't repeat per message. SVLS-8246 --- .../java/com/timgroup/statsd/NonBlockingStatsDClient.java | 7 +++++++ .../statsd/NonBlockingStatsDClientBuilderTest.java | 7 ++++++- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java index 183a6528..8a2a3a8e 100644 --- a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java +++ b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java @@ -257,6 +257,13 @@ public NonBlockingStatsDClient(final NonBlockingStatsDClientBuilder builder) 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) diff --git a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java index df74077a..6145130b 100644 --- a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java +++ b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java @@ -36,6 +36,7 @@ public void system_property_named_pipe_falls_back_to_udp() throws Exception { env.put( NonBlockingStatsDClient.DD_DOGSTATSD_PORT_ENV_VAR, Integer.toString(server.getPort())); + final RecordingErrorHandler errorHandler = new RecordingErrorHandler(); NonBlockingStatsDClient client = null; try { client = @@ -43,11 +44,15 @@ public void system_property_named_pipe_falls_back_to_udp() throws Exception { .withEnvironmentVariables(env) .prefix("my.prefix") .enableAggregation(false) + .errorHandler(errorHandler) .build(); client.gauge("value", 42); - server.waitForMessage("my.prefix"); + 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();