From 68020600144577394aed137c358d848a3a51b3fe Mon Sep 17 00:00:00 2001 From: Vikentiy Fesunov Date: Tue, 4 Aug 2026 18:10:48 +0200 Subject: [PATCH 1/2] Pass origin information to the forwarder ForwarderContext holds origin information derived from the environment, and supplies it to the forwarder implementation. CgroupReader is vendored as is from java-dogstatsd-client for now, to keep the prototype isolated from the main library. --- .../dogstatsd/http/CgroupReader.java | 257 ++++++++++++++++++ .../com/datadoghq/dogstatsd/http/EnvMap.java | 19 ++ .../dogstatsd/http/ForwarderContext.java | 151 ++++++++++ .../dogstatsd/http/ForwarderContextTest.java | 243 +++++++++++++++++ dogstatsd-http/forwarder/pom.xml | 5 + .../dogstatsd/http/forwarder/Forwarder.java | 212 +++++++++++---- .../http/forwarder/ForwarderTest.java | 59 +++- 7 files changed, 883 insertions(+), 63 deletions(-) create mode 100644 dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/CgroupReader.java create mode 100644 dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/EnvMap.java create mode 100644 dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/ForwarderContext.java create mode 100644 dogstatsd-http/core/src/test/java/com/datadoghq/dogstatsd/http/ForwarderContextTest.java diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/CgroupReader.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/CgroupReader.java new file mode 100644 index 00000000..550a9aaf --- /dev/null +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/CgroupReader.java @@ -0,0 +1,257 @@ +package com.datadoghq.dogstatsd.http; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * A reader class that retrieves the current container ID or the cgroup controller inode parsed from + * the cgroup file. + */ +class CgroupReader { + private static final Path CGROUP_PATH = Paths.get("/proc/self/cgroup"); + private static final String UUID_SOURCE = "[0-9a-f]{8}(?:[_-][0-9a-f]{4}){3}[_-][0-9a-f]{12}"; + private static final String CONTAINER_SOURCE = "[0-9a-f]{64}"; + private static final String TASK_SOURCE = "[0-9a-f]{32}-\\d+"; + private static final Pattern LINE_RE = + Pattern.compile("^\\d+:[^:]*:(.+)$", Pattern.MULTILINE | Pattern.UNIX_LINES); + private static final Pattern CONTAINER_RE = + Pattern.compile( + "(" + + UUID_SOURCE + + "|" + + CONTAINER_SOURCE + + "|" + + TASK_SOURCE + + ")(?:.scope)?$"); + + /** DEFAULT_CGROUP_MOUNT_PATH is the default cgroup mount path. */ + private static final Path DEFAULT_CGROUP_MOUNT_PATH = Paths.get("/sys/fs/cgroup"); + + /** CGROUP_NS_PATH is the path to the cgroup namespace file. */ + private static final Path CGROUP_NS_PATH = Paths.get("/proc/self/ns/cgroup"); + + /** + * CGROUPV1_BASE_CONTROLLER is the controller used to identify the container-id in cgroup v1 + * (memory). + */ + private static final String CGROUPV1_BASE_CONTROLLER = "memory"; + + /** + * CGROUPV2_BASE_CONTROLLER is the controller used to identify the container-id in cgroup v2. + */ + private static final String CGROUPV2_BASE_CONTROLLER = ""; + + /** HOST_CGROUP_NAMESPACE_INODE is the inode of the host cgroup namespace. */ + private static final long HOST_CGROUP_NAMESPACE_INODE = 0xEFFFFFFBL; + + private final Path MOUNTINFO_PATH = Paths.get("/proc/self/mountinfo"); + + private final Pattern MOUNTINFO_RE = + Pattern.compile( + ".*/([^\\s/]+)/(([0-9a-f]{64})|([0-9a-f]{32}-\\d+)|([0-9a-f]{8}(-[0-9a-f]{4}){4})$)/[\\S]*hostname"); + + interface Fs { + String getContents(Path path) throws IOException; + + long getInode(Path path) throws IOException; + } + + static class FilesFs implements Fs { + @Override + public String getContents(Path path) throws IOException { + return new String(Files.readAllBytes(path)); + } + + @Override + public long getInode(Path path) throws IOException { + return (long) Files.getAttribute(path, "unix:ino"); + } + } + + private final Fs fs; + + CgroupReader() { + this(new FilesFs()); + } + + CgroupReader(Fs fs) { + super(); + this.fs = fs; + } + + /** + * Returns the container ID if available or the cgroup controller inode. + * + * @throws IOException if /proc/self/cgroup is readable and still an I/O error occurs reading + * from the stream. + */ + public String getContainerID() { + String containerID = null; + + String cgroupContent = null; + try { + cgroupContent = fs.getContents(CGROUP_PATH); + } catch (IOException ex) { + // ignored + } + + if (!isEmpty(cgroupContent)) { + containerID = parseSelfCgroup(cgroupContent); + } + + if (!isEmpty(containerID)) { + return containerID; + } + + containerID = trySelfMountInfo(); + if (!isEmpty(containerID)) { + return containerID; + } + + /* + * If the container ID is not available it means that the application is either + * not running in a container or running is private cgroup namespace, we + * fallback to the cgroup controller inode. The agent (7.51+) will use it to get + * the container ID. + * In Host cgroup namespace, the container ID should be found. If it is not + * found, it means that the application is running on a host/vm. + * + */ + if (!isEmpty(cgroupContent) && !isHostCgroupNamespace(CGROUP_NS_PATH)) { + containerID = getCgroupInode(DEFAULT_CGROUP_MOUNT_PATH, cgroupContent); + } + return containerID; + } + + /** + * Parses a Cgroup file (=/proc/self/cgroup) content and returns the corresponding container ID. + * It can be found only if the container is running in host cgroup namespace. + * + * @param cgroupsContent Cgroup file content + */ + public static String parseSelfCgroup(final String cgroupsContent) { + final Matcher lines = LINE_RE.matcher(cgroupsContent); + while (lines.find()) { + final String path = lines.group(1); + final Matcher matcher = CONTAINER_RE.matcher(path); + if (matcher.find()) { + return matcher.group(1); + } + } + + return null; + } + + /** + * Returns true if the host cgroup namespace is used. It looks at the inode of + * `/proc/self/ns/cgroup` and compares it to HOST_CGROUP_NAMESPACE_INODE. + * + * @param path Path to the cgroup namespace file. + */ + private boolean isHostCgroupNamespace(final Path path) { + long hostCgroupInode = inodeForPath(path); + return hostCgroupInode == HOST_CGROUP_NAMESPACE_INODE; + } + + /** + * Returns the inode for the given path. + * + * @param path Path to the cgroup namespace file. + */ + private long inodeForPath(final Path path) { + try { + long inode = (long) fs.getInode(path); + return inode; + } catch (Exception e) { + return 0; + } + } + + /** + * Returns the cgroup controller inode for the given cgroup mount path and procSelfCgroupPath. + * + * @param cgroupMountPath Path to the cgroup mount point. + * @param cgroupContent String content of the cgroup file. + */ + public String getCgroupInode(final Path cgroupMountPath, final String cgroupContent) { + Map cgroupControllersPaths = parseCgroupNodePath(cgroupContent); + if (cgroupControllersPaths == null) { + return null; + } + + // Retrieve the cgroup inode from /sys/fs/cgroup+controller+cgroupNodePath + List controllers = + Arrays.asList(CGROUPV1_BASE_CONTROLLER, CGROUPV2_BASE_CONTROLLER); + for (String controller : controllers) { + String cgroupNodePath = cgroupControllersPaths.get(controller); + if (cgroupNodePath == null) { + continue; + } + Path path = Paths.get(cgroupMountPath.toString(), controller, cgroupNodePath); + long inode = inodeForPath(path); + /* + * Inode 0 is not a valid inode. Inode 1 is a bad block inode and inode 2 is the + * root of a filesystem. We can safely ignore them. + */ + if (inode > 2) { + return "in-" + inode; + } + } + + return null; + } + + /** + * Returns a map of cgroup controllers and their corresponding cgroup path. + * + * @param cgroupContent Cgroup file content. + */ + public Map parseCgroupNodePath(final String cgroupContent) { + Map res = new HashMap<>(); + + for (String line : cgroupContent.split("\n")) { + String[] tokens = line.split(":"); + if (tokens.length != 3) { + continue; + } + if (CGROUPV1_BASE_CONTROLLER.equals(tokens[1]) + || CGROUPV2_BASE_CONTROLLER.equals(tokens[1])) { + res.put(tokens[1], tokens[2]); + } + } + + return res; + } + + private static boolean isEmpty(String str) { + return str == null || str.isEmpty(); + } + + String trySelfMountInfo() { + String mountInfo; + try { + mountInfo = fs.getContents(MOUNTINFO_PATH); + } catch (IOException ex) { + return null; + } + + for (String line : mountInfo.split("\n")) { + Matcher matcher = MOUNTINFO_RE.matcher(line); + if (matcher.find()) { + if (!"sandboxes".equals(matcher.group(1))) { + return matcher.group(2); + } + } + } + + return null; + } +} diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/EnvMap.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/EnvMap.java new file mode 100644 index 00000000..10cbfcfe --- /dev/null +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/EnvMap.java @@ -0,0 +1,19 @@ +package com.datadoghq.dogstatsd.http; + +import java.util.Map; + +class EnvMap { + private final Map env; + + EnvMap() { + env = null; + } + + EnvMap(Map provided) { + env = provided; + } + + String get(String name) { + return env != null ? env.get(name) : System.getenv(name); + } +} diff --git a/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/ForwarderContext.java b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/ForwarderContext.java new file mode 100644 index 00000000..c3527997 --- /dev/null +++ b/dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/ForwarderContext.java @@ -0,0 +1,151 @@ +/* Unless explicitly stated otherwise all files in this repository are + * licensed under the Apache 2.0 License. + * + * This product includes software developed at Datadog + * (https://www.datadoghq.com/) Copyright 2026 Datadog, Inc. + */ + +package com.datadoghq.dogstatsd.http; + +import java.util.Map; + +/** Provides common parameters to the forwarder implementations. */ +public class ForwarderContext { + private final String localData; + private final String externalData; + + private ForwarderContext(final String localData, final String externalData) { + this.localData = localData; + this.externalData = externalData; + } + + /** + * Creates a builder for a context with explicit values or detection disabled. + * + * @return a new builder. + */ + public static Builder builder() { + return new Builder(); + } + + /** + * Returns the local-data value: a container ID, an {@code in-} cgroup fallback, or null + * when neither could be determined. + * + * @return the local-data value, or null. + */ + public String localData() { + return localData; + } + + /** + * Returns the external-data value from {@code DD_EXTERNAL_ENV}, or null when it is unset. + * + * @return the external-data value, or null. + */ + public String externalData() { + return externalData; + } + + /** + * Returns new instance with default settings. + * + * @return new default instance. + */ + public static ForwarderContext defaults() { + return builder().build(); + } + + /** Builds a {@link ForwarderContext}. Obtained via {@link ForwarderContext#builder}. */ + public static final class Builder { + private Boolean originDetectionEnabled = null; + private EnvMap env = new EnvMap(); + private CgroupReader cgroupReader = new CgroupReader(); + private String localData; + private String externalData; + + private Builder() {} + + /** + * Sets the local-data value explicitly, skipping detection for it. + * + * @param val the local-data value, or null to detect it. + * @return this builder. + */ + public Builder localData(final String val) { + localData = val; + return this; + } + + /** + * Sets the external-data value explicitly, skipping detection for it. + * + * @param val the external-data value, or null to detect it. + * @return this builder. + */ + public Builder externalData(final String val) { + externalData = val; + return this; + } + + /** + * Enables or disables detection of values that were not set explicitly. Defaults to the + * {@code DD_ORIGIN_DETECTION_ENABLED} environment variable, and to true when that is unset. + * + * @param val whether to detect local and external data. + * @return this builder. + */ + public Builder originDetectionEnabled(final boolean val) { + originDetectionEnabled = val; + return this; + } + + Builder environment(final Map val) { + env = new EnvMap(val); + return this; + } + + Builder cgroupReader(final CgroupReader val) { + cgroupReader = val; + return this; + } + + /** + * Builds the context, running detection for any value not set explicitly. + * + * @return a new context. + */ + public ForwarderContext build() { + String local = localData; + String external = externalData; + + if (resolveOriginDetectionEnabled()) { + if (local == null) { + local = cgroupReader.getContainerID(); + } + if (external == null) { + external = env.get("DD_EXTERNAL_ENV"); + } + } + + return new ForwarderContext(local, external); + } + + boolean resolveOriginDetectionEnabled() { + if (originDetectionEnabled != null) { + return originDetectionEnabled; + } + + final String value = env.get("DD_ORIGIN_DETECTION_ENABLED"); + if (value == null) { + return true; + } + final String normalized = value.trim().toLowerCase(); + return !("no".equals(normalized) + || "false".equals(normalized) + || "0".equals(normalized) + || "n".equals(normalized) + || "off".equals(normalized)); + } + } +} diff --git a/dogstatsd-http/core/src/test/java/com/datadoghq/dogstatsd/http/ForwarderContextTest.java b/dogstatsd-http/core/src/test/java/com/datadoghq/dogstatsd/http/ForwarderContextTest.java new file mode 100644 index 00000000..2f11591f --- /dev/null +++ b/dogstatsd-http/core/src/test/java/com/datadoghq/dogstatsd/http/ForwarderContextTest.java @@ -0,0 +1,243 @@ +/* Unless explicitly stated otherwise all files in this repository are + * licensed under the Apache 2.0 License. + * + * This product includes software developed at Datadog + * (https://www.datadoghq.com/) Copyright 2026 Datadog, Inc. + */ + +package com.datadoghq.dogstatsd.http; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.junit.Test; + +public class ForwarderContextTest { + /** A reader that reports a fixed container ID instead of reading /proc. */ + private static class StubCgroupReader extends CgroupReader { + private final String containerID; + + StubCgroupReader(String containerID) { + this.containerID = containerID; + } + + @Override + public String getContainerID() { + return containerID; + } + } + + private static ForwarderContext.Builder builder(Map env, String containerID) { + return ForwarderContext.builder() + .environment(env) + .cgroupReader(new StubCgroupReader(containerID)); + } + + @Test + public void detectionFillsBothValues() { + Map env = new HashMap(); + env.put("DD_EXTERNAL_ENV", "en-xyz"); + + ForwarderContext ctx = builder(env, "container-id").build(); + assertEquals("container-id", ctx.localData()); + assertEquals("en-xyz", ctx.externalData()); + } + + /** The cgroup reader supplies local data; nothing else does. */ + @Test + public void cgroupReaderSuppliesLocalData() { + ForwarderContext ctx = builder(new HashMap(), "in-1234567").build(); + assertEquals("in-1234567", ctx.localData()); + assertNull(ctx.externalData()); + } + + @Test + public void nullContainerIdLeavesLocalDataUnset() { + assertNull(builder(new HashMap(), null).build().localData()); + } + + @Test + public void unsetExternalEnvLeavesExternalDataUnset() { + assertNull(builder(new HashMap(), "container-id").build().externalData()); + } + + /** Explicit values take precedence over anything detection would find. */ + @Test + public void explicitValuesSkipDetection() { + Map env = new HashMap(); + env.put("DD_EXTERNAL_ENV", "detected-en"); + + ForwarderContext ctx = + builder(env, "detected-ci") + .localData("explicit-ci") + .externalData("explicit-en") + .build(); + assertEquals("explicit-ci", ctx.localData()); + assertEquals("explicit-en", ctx.externalData()); + } + + /** Detection still runs for the value that was not set explicitly. */ + @Test + public void explicitLocalDataStillDetectsExternalData() { + Map env = new HashMap(); + env.put("DD_EXTERNAL_ENV", "en-xyz"); + + ForwarderContext ctx = builder(env, "detected-ci").localData("explicit-ci").build(); + assertEquals("explicit-ci", ctx.localData()); + assertEquals("en-xyz", ctx.externalData()); + } + + /** + * One {@code DD_ORIGIN_DETECTION_ENABLED} value, and whether it leaves origin detection + * enabled. A null value means the variable is not set at all. + */ + private static class Case { + final String value; + final boolean detects; + + Case(String value, boolean detects) { + this.value = value; + this.detects = detects; + } + + @Override + public String toString() { + return "DD_ORIGIN_DETECTION_ENABLED=" + + (value == null ? "" : "[" + value + "]") + + " should " + + (detects ? "detect" : "not detect"); + } + } + + private static Case detects(String value) { + return new Case(value, true); + } + + private static Case ignores(String value) { + return new Case(value, false); + } + + /** + * The accepted values and their meanings match {@code + * NonBlockingStatsDClient.isOriginDetectionEnabled}: only an explicitly falsy value disables + * detection, and everything else leaves it enabled. + */ + private static final Case[] ORIGIN_DETECTION_CASES = { + detects(null), + detects(""), + detects(" "), + detects(" "), + detects("\t"), + detects("\n"), + ignores("no"), + ignores("false"), + ignores("0"), + ignores("n"), + ignores("off"), + ignores("NO"), + ignores("False"), + ignores("N"), + ignores("OFF"), + ignores("oFf"), + ignores(" no"), + ignores("false "), + ignores(" 0 "), + ignores("\toff\t"), + ignores("\n false \n"), + detects("yes"), + detects("true"), + detects("1"), + detects("y"), + detects("on"), + detects("YES"), + detects("True"), + detects("unknown"), + }; + + @Test + public void originDetectionEnabledEnvVar() { + for (Case c : ORIGIN_DETECTION_CASES) { + Map env = new HashMap(); + if (c.value != null) { + env.put("DD_ORIGIN_DETECTION_ENABLED", c.value); + } + + boolean detects = + ForwarderContext.builder().environment(env).resolveOriginDetectionEnabled(); + assertEquals(c.value, c.detects, detects); + } + } + + /** Disabling detection skips the cgroup read and the DD_EXTERNAL_ENV lookup alike. */ + @Test + public void disabledOriginDetectionSkipsBothLookups() { + Map env = new HashMap(); + env.put("DD_ORIGIN_DETECTION_ENABLED", "false"); + env.put("DD_EXTERNAL_ENV", "en-xyz"); + + ForwarderContext ctx = builder(env, "container-id").build(); + assertNull(ctx.localData()); + assertNull(ctx.externalData()); + } + + /** Values set explicitly on the builder are kept even with detection turned off. */ + @Test + public void disabledOriginDetectionKeepsExplicitValues() { + Map env = new HashMap(); + env.put("DD_ORIGIN_DETECTION_ENABLED", "false"); + + ForwarderContext ctx = + builder(env, "detected-ci") + .localData("explicit-ci") + .externalData("explicit-en") + .build(); + assertEquals("explicit-ci", ctx.localData()); + assertEquals("explicit-en", ctx.externalData()); + } + + @Test + public void builderOverridesEnvironmentWhenDisabling() { + Map env = new HashMap(); + env.put("DD_ORIGIN_DETECTION_ENABLED", "true"); + env.put("DD_EXTERNAL_ENV", "en-xyz"); + + ForwarderContext ctx = builder(env, "container-id").originDetectionEnabled(false).build(); + assertNull(ctx.localData()); + assertNull(ctx.externalData()); + } + + @Test + public void builderOverridesEnvironmentWhenEnabling() { + Map env = new HashMap(); + env.put("DD_ORIGIN_DETECTION_ENABLED", "false"); + env.put("DD_EXTERNAL_ENV", "en-xyz"); + + ForwarderContext ctx = builder(env, "container-id").originDetectionEnabled(true).build(); + assertEquals("container-id", ctx.localData()); + assertEquals("en-xyz", ctx.externalData()); + } + + /** An empty DD_EXTERNAL_ENV is passed through as-is rather than treated as unset. */ + @Test + public void emptyExternalEnvIsPassedThrough() { + Map env = new HashMap(); + env.put("DD_EXTERNAL_ENV", ""); + + ForwarderContext ctx = builder(env, "container-id").build(); + assertEquals("", ctx.externalData()); + } + + @Test + public void emptyEnvironmentDetectsLocalDataOnly() { + ForwarderContext ctx = + ForwarderContext.builder() + .environment(Collections.emptyMap()) + .cgroupReader(new StubCgroupReader("container-id")) + .build(); + assertEquals("container-id", ctx.localData()); + assertNull(ctx.externalData()); + } +} diff --git a/dogstatsd-http/forwarder/pom.xml b/dogstatsd-http/forwarder/pom.xml index 40bed66e..d8c34732 100644 --- a/dogstatsd-http/forwarder/pom.xml +++ b/dogstatsd-http/forwarder/pom.xml @@ -15,6 +15,11 @@ HTTP forwarder for DogStatsD metrics. + + com.datadoghq + dogstatsd-http-core + ${project.version} + junit junit diff --git a/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java b/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java index 0cfadf9b..f29ad640 100644 --- a/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java +++ b/dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Forwarder.java @@ -10,6 +10,7 @@ import static java.net.http.HttpRequest.BodyPublishers; import static java.net.http.HttpResponse.BodyHandlers; +import com.datadoghq.dogstatsd.http.ForwarderContext; import java.io.IOException; import java.net.URI; import java.net.http.HttpClient; @@ -37,34 +38,36 @@ public class Forwarder extends Thread { final Duration requestTimeout; final Random rng = new Random(); - String localData; - String externalData; + final String localData; + final String externalData; final Telemetry telemetry; /** - * Creates a new forwarder. + * Creates a builder for a forwarder. * - * @param maxRequestsBytes maximum total size of buffered payloads, in bytes - * @param maxTries maximum number of delivery attempts per payload - * @param whenFull action to take when the queue is at capacity - * @param connectTimeout timeout for establishing the TCP connection - * @param requestTimeout timeout from sending the request until response headers are received; - * {@code null} disables the request timeout + * @return a new builder. */ - public Forwarder( - long maxRequestsBytes, - long maxTries, - WhenFull whenFull, - Duration connectTimeout, - Duration requestTimeout) { + public static Builder builder() { + return new Builder(); + } + + Forwarder(final Builder builder) { this.telemetry = new Telemetry(); - this.queue = new BoundedQueue(maxRequestsBytes, maxTries, whenFull, this.telemetry); - this.requestTimeout = requestTimeout; + this.queue = + new BoundedQueue( + builder.maxRequestsBytes, + builder.maxTries, + builder.whenFull, + this.telemetry); + this.requestTimeout = builder.requestTimeout; + this.localData = builder.localData; + this.externalData = builder.externalData; + this.client = HttpClient.newBuilder() .version(HttpClient.Version.HTTP_2) - .connectTimeout(connectTimeout) + .connectTimeout(builder.connectTimeout) .build(); } @@ -97,8 +100,8 @@ public void run() { /** * Enqueues a payload for delivery to the given endpoint. * - *

If the queue is full, behaviour is determined by the {@link WhenFull} policy supplied at - * construction time. + *

If the queue is full, behaviour is determined by the {@link WhenFull} policy set with + * {@link Builder#whenFull}. * * @param url the remote HTTP endpoint to POST the payload to * @param payload the raw bytes to deliver @@ -199,46 +202,6 @@ void backoff() throws InterruptedException { } } - /** - * Sets the local-data value sent as the {@code x-dsd-ld} header with each request. - * - *

Local data carries the container ID or cgroup node inode used by the Datadog Agent for - * origin detection (DogStatsD protocol v1.4). - * - * @param data the local-data string, or {@code null} to omit the header - */ - public void setLocalData(String data) { - validateHeaderValue(data); - logger.log(Level.INFO, "using local data: {0}", data); - localData = data; - } - - /** - * Sets the external-data value sent as the {@code x-dsd-ed} header with each request. - * - *

External data is supplied by the Datadog Agent Admission Controller and is used by the - * Agent to enrich metrics with container tags when a container ID is unavailable (DogStatsD - * protocol v1.5, Agent ≥ v7.57.0). - * - * @param data the external-data string, or {@code null} to omit the header - */ - public void setExternalData(String data) { - validateHeaderValue(data); - logger.log(Level.INFO, "using external data: {0}", data); - externalData = data; - } - - private static final Pattern validHeaderValue = Pattern.compile("[\\t\\x20-\\x7E\\x80-\\xFF]*"); - - private static void validateHeaderValue(String value) { - if (value == null) { - return; - } - if (!validHeaderValue.matcher(value).matches()) { - throw new IllegalArgumentException("invalid character"); - } - } - /** * Closes the forwarder: stops accepting new payloads and drains the remaining backlog. * @@ -268,4 +231,133 @@ public boolean close(Duration timeout) throws InterruptedException { } return queue.empty(); } + + /** Builds a {@link Forwarder}. Obtained via {@link Forwarder#builder}. */ + public static final class Builder { + private long maxRequestsBytes = 8L * 1024 * 1024; + private long maxTries = 20; + private WhenFull whenFull = WhenFull.DROP; + private Duration connectTimeout = Duration.ofSeconds(1); + private Duration requestTimeout = Duration.ofSeconds(1); + private String localData; + private String externalData; + private boolean contextSet; + + private Builder() {} + + /** + * Sets the maximum total size of buffered payloads, in bytes. Defaults to 8 MiB. + * + *

Payloads larger than this are rejected by {@link Forwarder#send}. + * + * @param val the maximum number of buffered bytes; must be positive. + * @return this builder. + */ + public Builder maxRequestsBytes(final long val) { + if (val <= 0) { + throw new IllegalArgumentException("maxRequestsBytes must be positive"); + } + maxRequestsBytes = val; + return this; + } + + /** + * Sets the maximum number of delivery attempts per payload. Defaults to 20. + * + * @param val the maximum number of attempts; must be at least 1. + * @return this builder. + */ + public Builder maxTries(final long val) { + if (val < 1) { + throw new IllegalArgumentException("maxTries must be at least 1"); + } + maxTries = val; + return this; + } + + /** + * Sets the action to take when the queue is at capacity. Defaults to {@link WhenFull#DROP}. + * + * @param val the action to take. + * @return this builder. + */ + public Builder whenFull(final WhenFull val) { + whenFull = Objects.requireNonNull(val, "whenFull"); + return this; + } + + /** + * Sets the timeout for establishing the TCP connection. Defaults to one second. + * + * @param val the connect timeout; must be positive. + * @return this builder. + */ + public Builder connectTimeout(final Duration val) { + Objects.requireNonNull(val, "connectTimeout"); + if (val.isNegative() || val.isZero()) { + throw new IllegalArgumentException("connectTimeout must be positive"); + } + connectTimeout = val; + return this; + } + + /** + * Sets the timeout from sending the request until response headers are received. Defaults + * to one second. + * + * @param val the request timeout, or {@code null} to disable it; must be positive when + * non-null. + * @return this builder. + */ + public Builder requestTimeout(final Duration val) { + if (val != null && (val.isNegative() || val.isZero())) { + throw new IllegalArgumentException("requestTimeout must be positive"); + } + requestTimeout = val; + return this; + } + + /** + * Sets the shared context for this forwarder. + * + *

Defaults to {@code ForwarderContext.defaults()}. + * + * @param context the context to take the values from, or {@code null}. + * @return this builder. + */ + public Builder context(final ForwarderContext context) { + contextSet = true; + if (context == null) { + localData = null; + externalData = null; + } else { + localData = validateHeaderValue(context.localData()); + externalData = validateHeaderValue(context.externalData()); + } + return this; + } + + /** + * Builds the forwarder. The returned forwarder is a {@link Thread} that has not been + * started yet. + * + * @return a new forwarder. + */ + public Forwarder build() { + if (!contextSet) { + context(ForwarderContext.defaults()); + } + return new Forwarder(this); + } + + private static final Pattern validHeaderValue = + Pattern.compile("[\\t\\x20-\\x7E\\x80-\\xFF]*"); + + private static String validateHeaderValue(final String value) { + if (value != null && !validHeaderValue.matcher(value).matches()) { + throw new IllegalArgumentException("invalid character"); + } + return value; + } + } } diff --git a/dogstatsd-http/forwarder/src/test/java/com/datadoghq/dogstatsd/http/forwarder/ForwarderTest.java b/dogstatsd-http/forwarder/src/test/java/com/datadoghq/dogstatsd/http/forwarder/ForwarderTest.java index 40ad7a38..7a65d92d 100644 --- a/dogstatsd-http/forwarder/src/test/java/com/datadoghq/dogstatsd/http/forwarder/ForwarderTest.java +++ b/dogstatsd-http/forwarder/src/test/java/com/datadoghq/dogstatsd/http/forwarder/ForwarderTest.java @@ -10,9 +10,11 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import com.datadoghq.dogstatsd.http.ForwarderContext; import java.net.URI; import java.time.Duration; import java.util.Map; @@ -24,7 +26,58 @@ public class ForwarderTest { private static final URI URL = URI.create("http://localhost:0/"); private static Forwarder newForwarder(long maxBytes, WhenFull whenFull) { - return new Forwarder(maxBytes, 1, whenFull, Duration.ofSeconds(1), Duration.ofSeconds(1)); + return Forwarder.builder() + .maxRequestsBytes(maxBytes) + .maxTries(1) + .whenFull(whenFull) + .build(); + } + + @Test + public void builderRejectsInvalidValues() { + Forwarder.Builder b = Forwarder.builder(); + assertThrows(IllegalArgumentException.class, () -> b.maxRequestsBytes(0)); + assertThrows(IllegalArgumentException.class, () -> b.maxTries(0)); + assertThrows(NullPointerException.class, () -> b.whenFull(null)); + assertThrows(NullPointerException.class, () -> b.connectTimeout(null)); + assertThrows(IllegalArgumentException.class, () -> b.connectTimeout(Duration.ZERO)); + assertThrows(IllegalArgumentException.class, () -> b.requestTimeout(Duration.ZERO)); + } + + /** A null request timeout is legal and means requests have no timeout at all. */ + @Test + public void nullRequestTimeoutIsAllowed() { + Forwarder f = Forwarder.builder().requestTimeout(null).build(); + assertNull(f.requestTimeout); + } + + @Test + public void contextSuppliesOriginDetectionHeaders() { + Forwarder f = + Forwarder.builder() + .context( + ForwarderContext.builder() + .localData("ci-abc") + .externalData("en-xyz") + .build()) + .build(); + assertEquals("ci-abc", f.localData); + assertEquals("en-xyz", f.externalData); + } + + @Test + public void nullContextOmitsOriginDetectionHeaders() { + Forwarder f = Forwarder.builder().context(null).build(); + assertNull(f.localData); + assertNull(f.externalData); + } + + /** Values that can't be sent as a header value are rejected where they're supplied. */ + @Test + public void contextRejectsUnsendableHeaderValue() { + ForwarderContext ctx = ForwarderContext.builder().localData("bad\nvalue").build(); + Forwarder.Builder b = Forwarder.builder(); + assertThrows(IllegalArgumentException.class, () -> b.context(ctx)); } @Test @@ -128,7 +181,7 @@ public void closeDrainsEmptyQueueReturnsTrue() throws InterruptedException { public void closeDrainsPendingItemsReturnsTrue() throws InterruptedException { AtomicInteger processed = new AtomicInteger(); Forwarder f = - new Forwarder(100, 1, WhenFull.DROP, Duration.ofSeconds(1), Duration.ofSeconds(1)) { + new Forwarder(Forwarder.builder()) { @Override void runOnce(Map.Entry item) { processed.incrementAndGet(); @@ -156,7 +209,7 @@ public void sendAfterCloseThrows() throws InterruptedException { public void closeTimesOutReturnsFalse() throws InterruptedException { CountDownLatch entered = new CountDownLatch(1); Forwarder f = - new Forwarder(100, 1, WhenFull.DROP, Duration.ofSeconds(1), Duration.ofSeconds(1)) { + new Forwarder(Forwarder.builder()) { @Override void runOnce(Map.Entry item) throws InterruptedException { From 8d2428137cb3f4cf845ea82c0b676169b38a9fa6 Mon Sep 17 00:00:00 2001 From: Vikentiy Fesunov Date: Wed, 5 Aug 2026 13:29:02 +0200 Subject: [PATCH 2/2] Check if vendored code is in sync --- .github/workflows/ci.yml | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ed6e089e..0c76de67 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,3 +61,14 @@ jobs: run: mvn clean install - name: Test with latest jnr dependencies run: mvn test -P jnr-latest + + check-vendored: + name: Check vendored code + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + - name: CgroupReader is in sync + run: > + diff + <(grep -v ^package src/main/java/com/timgroup/statsd/CgroupReader.java) + <(grep -v ^package dogstatsd-http/core/src/main/java/com/datadoghq/dogstatsd/http/CgroupReader.java)