[SVLS-8245] Give each Windows Azure Function its own named pipes

src/main/java/com/datadog/NamedPipeNames.java

@@ -0,0 +1,87 @@
+package com.datadog;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+
+/**
+ * Chooses the trace and DogStatsD pipe names for one function, following the .NET
+ * compatibility layer: the binary's *_WINDOWS_PIPE_NAME setting, then a name the customer
+ * already set for the tracer, then a generated per-function name.
+ */
+final class NamedPipeNames {
+    static final int MAX_PIPE_NAME_LENGTH = 256;
+
+    static final String TRACE_WINDOWS_PIPE_NAME_ENV = "DD_APM_WINDOWS_PIPE_NAME";
+    static final String TRACE_PIPE_NAME_ENV = "DD_TRACE_PIPE_NAME";
+    static final String TRACE_PIPE_NAME_PROPERTY = "dd.trace.pipe.name";
+
+    static final String DOGSTATSD_WINDOWS_PIPE_NAME_ENV = "DD_DOGSTATSD_WINDOWS_PIPE_NAME";
+    static final String DOGSTATSD_PIPE_NAME_ENV = "DD_DOGSTATSD_PIPE_NAME";
+    static final String DOGSTATSD_PIPE_NAME_PROPERTY = "dd.dogstatsd.pipe.name";
+
+    final String tracePipeName;
+    final String dogstatsdPipeName;
+    final List<String> warnings;
+
+    private NamedPipeNames(String tracePipeName, String dogstatsdPipeName, List<String> warnings) {
+        this.tracePipeName = tracePipeName;
+        this.dogstatsdPipeName = dogstatsdPipeName;
+        this.warnings = Collections.unmodifiableList(warnings);
+    }
+
+    static NamedPipeNames choose(Map<String, String> env, Properties systemProperties) {
+        List<String> warnings = new ArrayList<>();
+        String tracePipeName = choosePipeName(env, systemProperties, TRACE_WINDOWS_PIPE_NAME_ENV,
+                TRACE_PIPE_NAME_PROPERTY, TRACE_PIPE_NAME_ENV, "dd_trace_", warnings);
+        String dogstatsdPipeName = choosePipeName(env, systemProperties, DOGSTATSD_WINDOWS_PIPE_NAME_ENV,
+                DOGSTATSD_PIPE_NAME_PROPERTY, DOGSTATSD_PIPE_NAME_ENV, "dd_dogstatsd_", warnings);
+        return new NamedPipeNames(tracePipeName, dogstatsdPipeName, warnings);
+    }
+
+    private static String choosePipeName(Map<String, String> env, Properties systemProperties,
+            String windowsPipeNameEnv, String pipeNameProperty, String pipeNameEnv,
+            String generatedPrefix, List<String> warnings) {
+        // The tracer reads system properties before environment variables, so a JVM option
+        // the customer passed is the customer-set name when both are present.
+        String customerSetName = nonEmpty(systemProperties.getProperty(pipeNameProperty));
+        String customerSetSource = pipeNameProperty;
+        if (customerSetName == null) {
+            customerSetName = nonEmpty(env.get(pipeNameEnv));
+            customerSetSource = pipeNameEnv;
+        }
+
+        String windowsPipeName = nonEmpty(env.get(windowsPipeNameEnv));
+        if (windowsPipeName != null) {
+            if (customerSetName != null && !customerSetName.equals(windowsPipeName)) {
+                warnings.add(String.format("%s ('%s') differs from %s ('%s'). Using %s.",
+                        customerSetSource, customerSetName, windowsPipeNameEnv, windowsPipeName,
+                        windowsPipeNameEnv));
+            }
+            warnIfTooLong(windowsPipeNameEnv, windowsPipeName, warnings);
+            return windowsPipeName;
+        }
+
+        if (customerSetName != null) {
+            warnIfTooLong(customerSetSource, customerSetName, warnings);
+            return customerSetName;
+        }
+
+        return generatedPrefix + UUID.randomUUID().toString().replace("-", "");
+    }
+
+    private static void warnIfTooLong(String source, String pipeName, List<String> warnings) {
+        if (pipeName.length() > MAX_PIPE_NAME_LENGTH) {
+            warnings.add(String.format(
+                    "%s is %d characters long. Windows pipe names can be at most %d characters.",
+                    source, pipeName.length(), MAX_PIPE_NAME_LENGTH));
+        }
+    }
+
+    private static String nonEmpty(String value) {
+        return value == null || value.isEmpty() ? null : value;
+    }
+}

src/main/java/com/datadog/ServerlessCompatAgent.java

@@ -10,6 +10,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.helpers.NOPLogger;
@@ -43,6 +44,8 @@ private static String mapDdLogLevelToSlf4jLogLevel(String ddLogLevel) {
 
     private static final String os = System.getProperty("os.name").toLowerCase();
     private static final String binaryPath = System.getenv("DD_SERVERLESS_COMPAT_PATH");
+    static final String DOGSTATSD_PORT_ENV = "DD_DOGSTATSD_PORT";
+    static final String DEFAULT_DOGSTATSD_PORT = "8125";
 
     public static boolean isWindows() {
         return os.contains("win");
@@ -106,6 +109,61 @@ public static boolean isAzureFlexWithoutDDAzureResourceGroup() {
                 && System.getenv("DD_AZURE_RESOURCE_GROUP") == null;
     }
 
+    /**
+     * Chooses this function's named pipes. The binary always gets them, so it warns instead of
+     * panicking when another binary on the instance owns a port.
+     *
+     * <p>dd-java-agent gets them, as system properties, only when the gate says it can use pipes;
+     * the properties must be set before its premain runs. Otherwise the tracer keeps what it
+     * inherited. It learns a pipe name only from its own configuration, so the binary's pipes
+     * can't reach an older tracer.
+     *
+     * @return the chosen pipe names for the binary
+     */
+    static NamedPipeNames configureNamedPipes() {
+        return configureNamedPipes(TracerVersionGate.check(), System.getenv(), System.getProperties());
+    }
+
+    static NamedPipeNames configureNamedPipes(TracerVersionGate gate, Map<String, String> env,
+            Properties systemProperties) {
+        NamedPipeNames pipeNames = NamedPipeNames.choose(env, systemProperties);
+        for (String warning : pipeNames.warnings) {
+            log.warn(warning);
+        }
+
+        if (!gate.passed) {
+            log.info("Tracer named pipes are off, so dd-java-agent keeps its inherited trace and DogStatsD "
+                    + "settings: {}. The binary still listens on pipes {} and {}, TCP, and UDP.",
+                    gate.reason, pipeNames.tracePipeName, pipeNames.dogstatsdPipeName);
+            return pipeNames;
+        }
+        log.debug("Tracer named pipes are on: {}", gate.reason);
+        if (gate.tracerListedBeforeCompat) {
+            log.warn("dd-java-agent is listed before the Datadog Serverless Compatibility Layer in the JVM's "
+                    + "-javaagent options, so it may have read its settings before the named pipes were set and "
+                    + "keep sending traces and its own metrics over TCP and UDP. List the Compatibility Layer's "
+                    + "-javaagent option first.");
+        }
+
+        systemProperties.setProperty(NamedPipeNames.TRACE_PIPE_NAME_PROPERTY, pipeNames.tracePipeName);
+        systemProperties.setProperty(NamedPipeNames.DOGSTATSD_PIPE_NAME_PROPERTY, pipeNames.dogstatsdPipeName);
+        log.info("Configured named pipes - Trace: {}, DogStatsD: {}",
+                pipeNames.tracePipeName, pipeNames.dogstatsdPipeName);
+        return pipeNames;
+    }
+
+    /**
+     * Passes the pipe names to the binary and asks it to keep UDP too. With a pipe, the binary
+     * listens on UDP only when {@code DD_DOGSTATSD_PORT} is set. A JVM can't change its own
+     * environment, so DogStatsD clients that don't read {@code dd.dogstatsd.pipe.name}, or can't
+     * open the pipe, still send to UDP 8125. A port the customer set wins.
+     */
+    static void addPipeSettings(Map<String, String> binaryEnvironment, NamedPipeNames pipeNames) {
+        binaryEnvironment.put(NamedPipeNames.TRACE_WINDOWS_PIPE_NAME_ENV, pipeNames.tracePipeName);
+        binaryEnvironment.put(NamedPipeNames.DOGSTATSD_WINDOWS_PIPE_NAME_ENV, pipeNames.dogstatsdPipeName);
+        binaryEnvironment.putIfAbsent(DOGSTATSD_PORT_ENV, DEFAULT_DOGSTATSD_PORT);
+    }
+
     public static void premain(String agentArgs, Instrumentation instrumentation) {
         CloudEnvironment environment = getEnvironment();
         log.debug("Environment detected: {}", environment);
@@ -169,8 +227,16 @@ public static void premain(String agentArgs, Instrumentation instrumentation) {
             String packageVersion = getPackageVersion();
             log.debug("Found package version {}", packageVersion);
 
+            NamedPipeNames pipeNames = null;
+            if (environment == CloudEnvironment.AZURE_FUNCTION && isWindows()) {
+                pipeNames = configureNamedPipes();
+            }
+
             ProcessBuilder processBuilder = new ProcessBuilder(executableFile.getAbsolutePath());
             processBuilder.environment().put("DD_SERVERLESS_COMPAT_VERSION", packageVersion);
+            if (pipeNames != null) {
+                addPipeSettings(processBuilder.environment(), pipeNames);
+            }
             processBuilder.inheritIO();
             processBuilder.start();
         } catch (Exception e) {

src/main/java/com/datadog/TracerVersionGate.java

@@ -0,0 +1,147 @@
+package com.datadog;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.util.List;
+import java.util.jar.Attributes;
+import java.util.jar.JarFile;
+import java.util.jar.Manifest;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Decides whether the dd-java-agent in this JVM supports named pipes. Older tracers hang
+ * uploading traces over a pipe and never send their own metrics over one, so pipe names are
+ * only handed out when the tracer is at least {@link #MINIMUM_TRACER_VERSION}.
+ */
+final class TracerVersionGate {
+    static final String TRACER_PREMAIN_CLASS = "datadog.trace.bootstrap.AgentPreCheck";
+    static final String COMPAT_PREMAIN_CLASS = ServerlessCompatAgent.class.getName();
+
+    // Placeholder: the next unreleased dd-trace-java version on master. Replace it with the
+    // first dd-trace-java release that includes the named-pipe transport and StatsD fixes.
+    static final String MINIMUM_TRACER_VERSION = "1.68.0";
+
+    private static final String JAVAAGENT_PREFIX = "-javaagent:";
+    private static final Pattern VERSION_PATTERN = Pattern.compile("^(\\d+)\\.(\\d+)\\.(\\d+)(\\D.*)?$");
+
+    final boolean passed;
+    final String reason;
+    // The JVM runs -javaagent premains in command-line order, so a tracer listed first reads its
+    // settings before this agent sets the pipe system properties.
+    final boolean tracerListedBeforeCompat;
+
+    private TracerVersionGate(boolean passed, String reason, boolean tracerListedBeforeCompat) {
+        this.passed = passed;
+        this.reason = reason;
+        this.tracerListedBeforeCompat = tracerListedBeforeCompat;
+    }
+
+    static TracerVersionGate check() {
+        List<String> jvmArguments;
+        try {
+            jvmArguments = ManagementFactory.getRuntimeMXBean().getInputArguments();
+        } catch (Exception e) {
+            return new TracerVersionGate(false, "unable to read JVM arguments to find dd-java-agent: " + e, false);
+        }
+        return check(jvmArguments);
+    }
+
+    static TracerVersionGate check(List<String> jvmArguments) {
+        return check(jvmArguments, MINIMUM_TRACER_VERSION);
+    }
+
+    static TracerVersionGate check(List<String> jvmArguments, String minimumVersion) {
+        int[] minimum = parseVersion(minimumVersion);
+        if (minimum == null) {
+            return new TracerVersionGate(false, "invalid minimum dd-java-agent version " + minimumVersion, false);
+        }
+
+        String jarPath = null;
+        Manifest tracerManifest = null;
+        boolean tracerListedBeforeCompat = false;
+        for (String argument : jvmArguments) {
+            if (!argument.startsWith(JAVAAGENT_PREFIX)) {
+                continue;
+            }
+            String agentJarPath = javaagentJarPath(argument);
+            Manifest manifest = readManifest(agentJarPath);
+            if (manifest == null) {
+                continue;
+            }
+            String premainClass = manifest.getMainAttributes().getValue("Premain-Class");
+            if (COMPAT_PREMAIN_CLASS.equals(premainClass) && tracerManifest != null) {
+                tracerListedBeforeCompat = true;
+            } else if (TRACER_PREMAIN_CLASS.equals(premainClass) && tracerManifest == null) {
+                jarPath = agentJarPath;
+                tracerManifest = manifest;
+            }
+        }
+
+        if (tracerManifest == null) {
+            return new TracerVersionGate(false,
+                    "no readable dd-java-agent found among -javaagent options (Premain-Class " + TRACER_PREMAIN_CLASS + ")",
+                    false);
+        }
+
+        String version = tracerManifest.getMainAttributes().getValue(Attributes.Name.IMPLEMENTATION_VERSION);
+        if (version == null || version.isEmpty()) {
+            return new TracerVersionGate(false,
+                    "dd-java-agent at " + jarPath + " has no Implementation-Version in its manifest",
+                    tracerListedBeforeCompat);
+        }
+        int[] tracer = parseVersion(version);
+        if (tracer == null) {
+            return new TracerVersionGate(false,
+                    "unable to read dd-java-agent version '" + version + "' at " + jarPath, tracerListedBeforeCompat);
+        }
+        if (compare(tracer, minimum) < 0) {
+            return new TracerVersionGate(false, "dd-java-agent " + version + " at " + jarPath
+                    + " is older than " + minimumVersion + ", the first version that supports named pipes",
+                    tracerListedBeforeCompat);
+        }
+        return new TracerVersionGate(true, "dd-java-agent " + version + " at " + jarPath
+                + " supports named pipes (minimum " + minimumVersion + ")", tracerListedBeforeCompat);
+    }
+
+    // The JVM treats everything after the first '=' as agent options.
+    private static String javaagentJarPath(String argument) {
+        String value = argument.substring(JAVAAGENT_PREFIX.length());
+        int optionsStart = value.indexOf('=');
+        return optionsStart >= 0 ? value.substring(0, optionsStart) : value;
+    }
+
+    private static Manifest readManifest(String jarPath) {
+        try (JarFile jarFile = new JarFile(jarPath, false)) {
+            return jarFile.getManifest();
+        } catch (IOException | RuntimeException e) {
+            return null;
+        }
+    }
+
+    // Reads major.minor.patch and ignores any suffix, so 1.68.0-SNAPSHOT counts as 1.68.0.
+    private static int[] parseVersion(String version) {
+        Matcher matcher = VERSION_PATTERN.matcher(version);
+        if (!matcher.matches()) {
+            return null;
+        }
+        try {
+            return new int[] {
+                    Integer.parseInt(matcher.group(1)),
+                    Integer.parseInt(matcher.group(2)),
+                    Integer.parseInt(matcher.group(3))
+            };
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+
+    private static int compare(int[] left, int[] right) {
+        for (int i = 0; i < left.length; i++) {
+            if (left[i] != right[i]) {
+                return Integer.compare(left[i], right[i]);
+            }
+        }
+        return 0;
+    }
+}