Statsd reporter to require host name and optional port to support UDS

pom.xml

@@ -37,7 +37,7 @@
 
         <commons-io.version>2.4</commons-io.version>
         <apache-commons-lang3.version>3.5</apache-commons-lang3.version>
-        <java-dogstatsd-client.version>2.9.0</java-dogstatsd-client.version>
+        <java-dogstatsd-client.version>2.10.3</java-dogstatsd-client.version>
         <jcommander.version>1.35</jcommander.version>
         <!-- log4j 2.13+ drops support for Java 7, so stick to 2.12 -->
         <log4j.version>2.12.1</log4j.version>

src/main/java/org/datadog/jmxfetch/AppConfig.java

@@ -106,7 +106,8 @@ public class AppConfig {
     @Parameter(
             names = {"--reporter", "-r"},
             description =
-                    "Reporter to use: should be either \"statsd:[STATSD_PORT]\", "
+                    "Reporter to use: should be either \"statsd:[STATSD_HOST][STATSD_PORT]\", "
+                     + "\"statsd:unix://[STATSD_UNIX_SOCKET_PATH]\", "
                      + "\"console\" or \"json\"",
             validateWith = ReporterValidator.class,
             converter = ReporterConverter.class,

src/main/java/org/datadog/jmxfetch/reporter/ReporterFactory.java

@@ -3,10 +3,12 @@
 import org.datadog.jmxfetch.util.StringUtils;
 
 import java.util.Arrays;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 public class ReporterFactory {
 
-    /** Gets the reporter for the correspndonding type string (console, statsd). */
+    /** Gets the reporter for the corresponding type string (console, statsd). */
     public static Reporter getReporter(String type) {
         if (type == null || type.length() <= 0) {
             throw new IllegalArgumentException("Null or empty reporter type");
@@ -16,17 +18,21 @@ public static Reporter getReporter(String type) {
         } else if ("json".equals(type)) {
             return new JsonReporter();
         } else if (type.startsWith("statsd:")) {
-            String[] typeElements = type.split(":");
-            String host = "localhost";
-            Integer port = Integer.valueOf(typeElements[typeElements.length - 1]);
-            if (typeElements.length > 2) {
-                host = StringUtils.join(":",
-                        Arrays.copyOfRange(typeElements, 1, typeElements.length - 1));
+
+            Matcher matcher = Pattern.compile("^statsd:(.*):(\\d+)$").matcher(type);
+            if (matcher.find() && matcher.groupCount() == 2) {
+                String host = matcher.group(1);
+                Integer port = Integer.valueOf(matcher.group(2));
+                return new StatsdReporter(host, port);
+            }
+
+            matcher = Pattern.compile("^statsd:unix://(.*)$").matcher(type);
+            if (matcher.find() && matcher.groupCount() == 1) {
+                String socketPath = matcher.group(1);
+                return new StatsdReporter(socketPath, 0);
             }
-            return new StatsdReporter(host, port);
-        } else {
-            throw new IllegalArgumentException("Invalid reporter type: " + type);
         }
+        throw new IllegalArgumentException("Invalid reporter type: " + type);
     }
 
     private ReporterFactory() {}

src/main/java/org/datadog/jmxfetch/reporter/StatsdReporter.java

@@ -1,6 +1,6 @@
 package org.datadog.jmxfetch.reporter;
 
-import com.timgroup.statsd.NonBlockingStatsDClient;
+import com.timgroup.statsd.NonBlockingStatsDClientBuilder;
 import com.timgroup.statsd.ServiceCheck;
 import com.timgroup.statsd.StatsDClient;
 import com.timgroup.statsd.StatsDClientErrorHandler;
@@ -41,15 +41,18 @@ private void init() {
         /* Create the StatsDClient with "entity-id" set to "none" to avoid
            having dogstatsd server adding origin tags, when the connection is
            done with UDS. */
-        statsDClient =
-                new NonBlockingStatsDClient(
-                        null,
-                        this.statsdHost,
-                        this.statsdPort,
-                        Integer.MAX_VALUE,
-                        new String[] {},
-                        new LoggingErrorHandler(),
-                        entityId);
+        NonBlockingStatsDClientBuilder builder = new NonBlockingStatsDClientBuilder()
+                .enableTelemetry(false)
+                .hostname(this.statsdHost)
+                .port(this.statsdPort)
+                .errorHandler(new LoggingErrorHandler())
+                .entityID(entityId);
+
+        // When using UDS set the datagram size to 8k
+        if (this.statsdPort == 0) {
+            builder.maxPacketSizeBytes(8192);
+        }
+        statsDClient = builder.build();
     }
 
     protected void sendMetricPoint(

src/main/java/org/datadog/jmxfetch/validator/ReporterValidator.java

@@ -5,29 +5,15 @@
 
 public class ReporterValidator implements IParameterValidator {
 
-    private static final String STATSD_PREFIX = "statsd:";
-    private final PositiveIntegerValidator positiveIntegerValidator =
-            new PositiveIntegerValidator();
-
     /** Validates a reporter configurations (console, statsd). */
     public void validate(String name, String value) throws ParameterException {
-        if (value.startsWith(STATSD_PREFIX) && value.length() > STATSD_PREFIX.length()) {
-            String[] splitValue = value.split(":");
-            String port = splitValue[splitValue.length - 1];
-            try {
-                positiveIntegerValidator.validate(name, port);
-            } catch (ParameterException pe) {
-                throw new ParameterException(
-                        "Statsd Port should be a positive integer (found " + port + ")");
-            }
-            return;
-        }
-        if (!value.equals("console") && !value.equals("json")) {
+        if (!value.matches("^statsd:.+$") && !value.equals("console") && !value.equals("json")) {
             throw new ParameterException(
                     "Parameter "
                             + name
-                            + " should be either 'console', 'json', 'statsd:[STATSD_PORT]' "
-                            + "or 'statsd:[STATSD_HOST]:[STATSD_PORT]'");
+                            + " should be either 'console', 'json',"
+                            + " 'statsd:[STATSD_HOST]:[STATSD_PORT]'"
+                            + " or 'statsd:unix://[STATSD_UNIX_SOCKET_PATH]'");
         }
     }
 }

src/test/java/org/datadog/jmxfetch/TestParsingJCommander.java

@@ -149,7 +149,7 @@ public void testParsingReporter() {
         params =
                 new String[] {
                     "--reporter",
-                    "statsd:10",
+                    "statsd:localhost:10",
                     "--check",
                     SINGLE_CHECK,
                     "--conf_directory",
@@ -212,27 +212,26 @@ public void testParsingReporter() {
             fail("Should have failed because reporter is invalid");
         } catch (ParameterException pe) {
             assertEquals(
-                    "Parameter --reporter should be either 'console', 'json', 'statsd:[STATSD_PORT]' or 'statsd:[STATSD_HOST]:[STATSD_PORT]'",
+                    "Parameter --reporter should be either 'console', 'json', 'statsd:[STATSD_HOST]:[STATSD_PORT]' or 'statsd:unix://[STATSD_UNIX_SOCKET_PATH]'",
                     pe.getMessage());
         }
 
-        // invalid port
+        // UDS
         params =
                 new String[] {
                     "-r",
-                    "statsd:-1",
+                    "statsd:unix:///path/to/dsd.socket",
                     "--check",
                     SINGLE_CHECK,
                     "--conf_directory",
                     CONF_DIR,
                     AppConfig.ACTION_COLLECT
                 };
-        try {
-            testCommand(params);
-            fail("Should have failed because statsd reporter port is invalid");
-        } catch (ParameterException pe) {
-            assertEquals("Statsd Port should be a positive integer (found -1)", pe.getMessage());
-        }
+        appConfig = testCommand(params);
+        assertNotNull(appConfig.getReporter());
+        assertTrue(appConfig.getReporter() instanceof StatsdReporter);
+        assertEquals("/path/to/dsd.socket", ((StatsdReporter) appConfig.getReporter()).getStatsdHost());
+        assertEquals(0, ((StatsdReporter) appConfig.getReporter()).getStatsdPort());
     }
 
     @Test