Merge pull request #42 from DataDog/xvello/uds

Add UDS support

Author: xvello

README.md

@@ -60,3 +60,17 @@ public class Foo {
   }
 }
 ```
+
+Unix Domain Socket support
+---------------------------
+
+As an alternative to UDP, Agent6 can receive metrics via a UNIX Socket (on Linux only). This library supports
+transmission via this protocol. To use it, simply pass the socket path as a hostname, and 0 as port.
+
+By default, all exceptions are ignored, mimicking UDP behaviour. When using Unix Sockets, transmission errors will
+trigger exceptions you can choose to handle by passing a `StatsDClientErrorHandler`:
+
+- Connection error because of an invalid/missing socket will trigger a `java.io.IOException: No such file or directory`
+- If dogstatsd's reception buffer were to fill up, the send will timeout after 100ms and throw either a
+`java.io.IOException: No buffer space available` or a `java.io.IOException: Resource temporarily unavailable`
+

pom.xml

@@ -66,6 +66,11 @@
       <scope>test</scope>
       <version>4.12</version>
     </dependency>
+    <dependency>
+      <groupId>com.github.jnr</groupId>
+      <artifactId>jnr-unixsocket</artifactId>
+      <version>0.18</version>
+    </dependency>
   </dependencies>
 
   <distributionManagement>

src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java

@@ -2,6 +2,7 @@
 
 import java.io.IOException;
 import java.net.InetAddress;
+import java.net.SocketAddress;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
@@ -19,7 +20,9 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
-
+import jnr.unixsocket.UnixSocketAddress;
+import jnr.unixsocket.UnixDatagramChannel;
+import jnr.unixsocket.UnixSocketOptions;
 
 
 /**
@@ -305,7 +308,7 @@ public NonBlockingStatsDClient(final String prefix, final String hostname, final
      *     if the client could not be started
      */
     public NonBlockingStatsDClient(final String prefix,  final int queueSize, String[] constantTags, final StatsDClientErrorHandler errorHandler,
-                                   final Callable<InetSocketAddress> addressLookup) throws StatsDClientException {
+                                   final Callable<SocketAddress> addressLookup) throws StatsDClientException {
         if((prefix != null) && (!prefix.isEmpty())) {
             this.prefix = new StringBuilder(prefix).append(".").toString();
         } else {
@@ -330,7 +333,15 @@ public NonBlockingStatsDClient(final String prefix,  final int queueSize, String
         }
 
         try {
-            clientChannel = DatagramChannel.open();
+            final SocketAddress address = addressLookup.call();
+            if (address instanceof UnixSocketAddress) {
+                clientChannel = UnixDatagramChannel.open();
+                // Set send timeout to 100ms, to handle the case where the transmission buffer is full
+                // If no timeout is set, the send becomes blocking
+                clientChannel.setOption(UnixSocketOptions.SO_SNDTIMEO, Integer.valueOf(100));
+            } else{
+                clientChannel = DatagramChannel.open();
+            }
         } catch (final Exception e) {
             throw new StatsDClientException("Failed to start StatsD client", e);
         }
@@ -995,10 +1006,11 @@ private boolean isInvalidSample(double sampleRate) {
 
     private class QueueConsumer implements Runnable {
         private final ByteBuffer sendBuffer = ByteBuffer.allocate(PACKET_SIZE_BYTES);
+        private final Callable<SocketAddress> addressLookup;
 
-        private final Callable<InetSocketAddress> addressLookup;
 
-        QueueConsumer(final Callable<InetSocketAddress> addressLookup) {
+
+        QueueConsumer(final Callable<SocketAddress> addressLookup) {
             this.addressLookup = addressLookup;
         }
 
@@ -1007,7 +1019,7 @@ private class QueueConsumer implements Runnable {
                 try {
                     final String message = queue.poll(1, TimeUnit.SECONDS);
                     if(null != message) {
-                        final InetSocketAddress address = addressLookup.call();
+                        final SocketAddress address = addressLookup.call();
                         final byte[] data = message.getBytes(MESSAGE_CHARSET);
                         if(sendBuffer.remaining() < (data.length + 1)) {
                             blockingSend(address);
@@ -1026,7 +1038,7 @@ private class QueueConsumer implements Runnable {
             }
         }
 
-        private void blockingSend(final InetSocketAddress address) throws IOException {
+        private void blockingSend(final SocketAddress address) throws IOException {
             final int sizeOfBuffer = sendBuffer.position();
             sendBuffer.flip();
 
@@ -1038,10 +1050,9 @@ private void blockingSend(final InetSocketAddress address) throws IOException {
                 handler.handle(
                         new IOException(
                             String.format(
-                                "Could not send entirely stat %s to host %s:%d. Only sent %d bytes out of %d bytes",
+                                "Could not send entirely stat %s to %s. Only sent %d bytes out of %d bytes",
                                 sendBuffer.toString(),
-                                address.getHostName(),
-                                address.getPort(),
+                                address.toString(),
                                 sentBytes,
                                 sizeOfBuffer)));
             }
@@ -1055,10 +1066,14 @@ private void blockingSend(final InetSocketAddress address) throws IOException {
      * @param port     the port of the targeted StatsD server
      * @return a function to perform the lookup
      */
-    public static Callable<InetSocketAddress> volatileAddressResolution(final String hostname, final int port) {
-        return new Callable<InetSocketAddress>() {
-            @Override public InetSocketAddress call() throws UnknownHostException {
-                return new InetSocketAddress(InetAddress.getByName(hostname), port);
+    public static Callable<SocketAddress> volatileAddressResolution(final String hostname, final int port) {
+        return new Callable<SocketAddress>() {
+            @Override public SocketAddress call() throws UnknownHostException {
+                if (port == 0) { // Hostname is a file path to the socket
+                    return new UnixSocketAddress(hostname);
+                } else {
+                    return new InetSocketAddress(InetAddress.getByName(hostname), port);
+                }
             }
         };
     }
@@ -1071,16 +1086,16 @@ public static Callable<InetSocketAddress> volatileAddressResolution(final String
    * @return a function that cached the result of the lookup
    * @throws Exception if the lookup fails, i.e. {@link UnknownHostException}
    */
-    public static Callable<InetSocketAddress> staticAddressResolution(final String hostname, final int port) throws Exception {
-        final InetSocketAddress address = volatileAddressResolution(hostname, port).call();
-        return new Callable<InetSocketAddress>() {
-            @Override public InetSocketAddress call() {
+    public static Callable<SocketAddress> staticAddressResolution(final String hostname, final int port) throws Exception {
+        final SocketAddress address = volatileAddressResolution(hostname, port).call();
+        return new Callable<SocketAddress>() {
+            @Override public SocketAddress call() {
                 return address;
             }
         };
     }
 
-    private static Callable<InetSocketAddress> staticStatsDAddressResolution(final String hostname, final int port) throws StatsDClientException {
+    private static Callable<SocketAddress> staticStatsDAddressResolution(final String hostname, final int port) throws StatsDClientException {
         try {
             return staticAddressResolution(hostname, port);
         } catch (final Exception e) {

src/test/java/com/timgroup/statsd/DummyStatsDServer.java

@@ -2,30 +2,54 @@
 package com.timgroup.statsd;
 
 import java.io.IOException;
-import java.net.DatagramPacket;
-import java.net.DatagramSocket;
-import java.net.SocketException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
 import java.util.ArrayList;
 import java.util.List;
+import jnr.unixsocket.UnixDatagramChannel;
+import jnr.unixsocket.UnixSocketAddress;
+import java.nio.charset.StandardCharsets;
 
 
 final class DummyStatsDServer {
     private final List<String> messagesReceived = new ArrayList<String>();
-    private final DatagramSocket server;
+    private final DatagramChannel server;
+    private volatile Boolean freeze = false;
 
-    public DummyStatsDServer(int port) throws SocketException {
-        server = new DatagramSocket(port);
+    public DummyStatsDServer(int port) throws IOException {
+        server = DatagramChannel.open();
+        server.bind(new InetSocketAddress(port));
+        this.listen();
+    }
+
+    public DummyStatsDServer(String socketPath) throws IOException {
+        server = UnixDatagramChannel.open();
+        server.bind(new UnixSocketAddress(socketPath));
+        this.listen();
+    }
+
+    private void listen() {
         Thread thread = new Thread(new Runnable() {
             @Override
             public void run() {
-                while(!server.isClosed()) {
-                    try {
-                        final DatagramPacket packet = new DatagramPacket(new byte[1500], 1500);
-                        server.receive(packet);
-                        for(String msg : new String(packet.getData(), NonBlockingStatsDClient.MESSAGE_CHARSET).split("\n")) {
-                            messagesReceived.add(msg.trim());
+                final ByteBuffer packet = ByteBuffer.allocate(1500);
+                while(server.isOpen()) {
+                    if (freeze) {
+                        try {
+                            Thread.sleep(10);
+                        } catch (InterruptedException e) {
+                        }
+                    } else {
+                        try {
+                            packet.clear();
+                            server.receive(packet);
+                            packet.flip();
+                            for (String msg : StandardCharsets.UTF_8.decode(packet).toString().split("\n")) {
+                                messagesReceived.add(msg.trim());
+                            }
+                        } catch (IOException e) {
                         }
-                    } catch (IOException e) {
                     }
                 }
             }
@@ -47,7 +71,15 @@ public List<String> messagesReceived() {
         return new ArrayList<String>(messagesReceived);
     }
 
-    public void close() {
+    public void freeze() {
+        freeze = true;
+    }
+
+    public void unfreeze() {
+        freeze = false;
+    }
+
+    public void close() throws IOException {
         server.close();
     }
 

src/test/java/com/timgroup/statsd/NonBlockingStatsDClientPerfTest.java

@@ -1,6 +1,7 @@
 package com.timgroup.statsd;
 
 
+import java.io.IOException;
 import java.net.SocketException;
 import java.util.Random;
 import java.util.concurrent.ExecutorService;
@@ -18,11 +19,11 @@ public final class NonBlockingStatsDClientPerfTest {
     private static final int STATSD_SERVER_PORT = 17255;
     private static final Random RAND = new Random();
     private static final NonBlockingStatsDClient client = new NonBlockingStatsDClient("my.prefix", "localhost", STATSD_SERVER_PORT);
-    private final ExecutorService executor = Executors.newFixedThreadPool(20);
+    private final ExecutorService executor = Executors.newFixedThreadPool(10);
     private static DummyStatsDServer server;
 
     @BeforeClass
-    public static void start() throws SocketException {
+    public static void start() throws IOException {
         server = new DummyStatsDServer(STATSD_SERVER_PORT);
     }
 
@@ -39,7 +40,7 @@ public void perf_test() throws Exception {
         for(int i = 0; i < testSize; ++i) {
             executor.submit(new Runnable() {
                 public void run() {
-                    client.count("mycount", RAND.nextInt());
+                    client.count("mycount", 1);
                 }
             });
 
@@ -48,7 +49,7 @@ public void run() {
         executor.shutdown();
         executor.awaitTermination(20, TimeUnit.SECONDS);
 
-        for(int i = 0; i < 20000 && server.messagesReceived().size() < testSize; i += 50) {
+        while(server.messagesReceived().size() < testSize) {
             try {
                 Thread.sleep(50);
             } catch (InterruptedException ex) {}

src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java

@@ -5,6 +5,7 @@
 import org.junit.After;
 import org.junit.Test;
 
+import java.io.IOException;
 import java.net.SocketException;
 import java.util.Locale;
 
@@ -19,7 +20,7 @@ public class NonBlockingStatsDClientTest {
     private static DummyStatsDServer server;
 
     @BeforeClass
-    public static void start() throws SocketException {
+    public static void start() throws IOException {
         server = new DummyStatsDServer(STATSD_SERVER_PORT);
     }