Support multiline statsd datagrams and single datagrams ending with a newline. (#64)

Author: vjkoskela

pom.xml

@@ -356,6 +356,7 @@
                 <ports>
                   <port>${debugJavaPort}:${debugJavaPort}</port>
                   <port>7090:7090</port>
+                  <port>8125:8125/udp</port>
                 </ports>
                 <volumes>
                   <bind>

src/main/java/com/arpnetworking/metrics/common/parsers/exceptions/ParsingException.java

@@ -51,7 +51,7 @@ public byte[] getOffendingData() {
         return _offendingData;
     }
 
-    // TODO(barp): change this into a List or similar struture to ensure no modifications
+    // TODO(barp): change this into a List or similar structure to ensure no modifications
     private final byte[] _offendingData;
 
     private static final long serialVersionUID = 1L;

src/main/java/com/arpnetworking/metrics/common/sources/StatsdSource.java

@@ -64,7 +64,7 @@ private StatsdSource(final Builder builder) {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(StatsdSource.class);
     private static final Parser<List<Record>, ByteBuffer> PARSER = new StatsdToRecordParser();
-    
+
     /**
      * Name of the actor created to receive the Statsd datagrams.
      */
@@ -90,18 +90,20 @@ public void onReceive(final Object message) throws Exception {
                 getSender().tell(_isReady, getSelf());
             } else if (message instanceof Udp.Bound) {
                 final Udp.Bound updBound = (Udp.Bound) message;
-                LOGGER.debug()
+                _socket = getSender();
+                _isReady = true;
+                LOGGER.info()
                         .setMessage("Statsd server binding complete")
                         .addData("address", updBound.localAddress().getAddress().getHostAddress())
                         .addData("port", updBound.localAddress().getPort())
+                        .addData("socket", _socket)
                         .log();
-                _socket = getSender();
-                _isReady = true;
             } else if (message instanceof Udp.Received) {
                 final Udp.Received updReceived = (Udp.Received) message;
                 LOGGER.trace()
                         .setMessage("Statsd received datagram")
                         .addData("bytes", updReceived.data().size())
+                        .addData("socket", _socket)
                         .log();
 
                 try {
@@ -112,14 +114,23 @@ public void onReceive(final Object message) throws Exception {
                 } catch (final ParserException e) {
                     BAD_REQUEST_LOGGER.warn()
                             .setMessage("Error handling statsd datagram")
+                            .addData("socket", _socket)
                             .setThrowable(e)
                             .log();
                 }
 
             } else if (Objects.equals(message, UdpMessage.unbind())) {
+                LOGGER.debug()
+                        .setMessage("Statsd unbind")
+                        .addData("socket", _socket)
+                        .log();
                 _socket.tell(message, getSelf());
 
             } else if (message instanceof Udp.Unbound) {
+                LOGGER.debug()
+                        .setMessage("Statsd unbound")
+                        .addData("socket", _socket)
+                        .log();
                 getContext().stop(getSelf());
 
             } else {

src/main/java/com/arpnetworking/metrics/mad/parsers/StatsdToRecordParser.java

@@ -24,7 +24,9 @@
 import com.arpnetworking.tsdcore.model.Quantity;
 import com.arpnetworking.tsdcore.model.Unit;
 import com.google.common.base.Charsets;
+import com.google.common.base.Splitter;
 import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
@@ -77,38 +79,43 @@ public final class StatsdToRecordParser implements Parser<List<Record>, ByteBuff
     public List<Record> parse(final ByteBuffer datagram) throws ParsingException {
         // CHECKSTYLE.OFF: IllegalInstantiation - This is the recommended way
         final String datagramAsString = new String(datagram.array(), Charsets.UTF_8);
-        // CHECKSTYLE.ON: IllegalInstantiation
-        final Matcher matcher = STATSD_PATTERN.matcher(datagramAsString);
-        if (!matcher.matches()) {
-            throw new ParsingException("Invalid statsd datagram", datagram.array());
-        }
+        final ImmutableList.Builder<Record> recordListBuilder = ImmutableList.builder();
+        for (final String line : LINE_SPLITTER.split(datagramAsString)) {
+            // CHECKSTYLE.ON: IllegalInstantiation
+            final Matcher matcher = STATSD_PATTERN.matcher(line);
+            if (!matcher.matches()) {
+                throw new ParsingException("Invalid statsd line", line.getBytes(Charsets.UTF_8));
+            }
 
-        // Parse the name
-        final String name = parseName(datagram, matcher.group("NAME"));
+            // Parse the name
+            final String name = parseName(datagram, matcher.group("NAME"));
 
-        // Parse the _metricType
-        final StatsdType type = parseStatsdType(datagram, matcher.group("TYPE"));
+            // Parse the _metricType
+            final StatsdType type = parseStatsdType(datagram, matcher.group("TYPE"));
 
-        // Parse the value
-        final Number value = parseValue(datagram, matcher.group("VALUE"), type);
+            // Parse the value
+            final Number value = parseValue(datagram, matcher.group("VALUE"), type);
 
-        // Parse the value
-        final Optional<Double> sampleRate = parseSampleRate(datagram, matcher.group("SAMPLERATE"), type);
+            // Parse the value
+            final Optional<Double> sampleRate = parseSampleRate(datagram, matcher.group("SAMPLERATE"), type);
 
-        // Parse the tags
-        final ImmutableMap<String, String> annotations = parseTags(matcher.group("TAGS"));
+            // Parse the tags
+            final ImmutableMap<String, String> annotations = parseTags(matcher.group("TAGS"));
 
-        // Enforce sampling
-        if (sampleRate.isPresent() && sampleRate.get().compareTo(1.0) != 0) {
-            if (sampleRate.get().compareTo(0.0) == 0) {
-                return Collections.emptyList();
-            }
-            if (Double.compare(_randomSupplier.get().nextDouble(), sampleRate.get()) > 0) {
-                return Collections.emptyList();
+            // Enforce sampling
+            if (sampleRate.isPresent() && sampleRate.get().compareTo(1.0) != 0) {
+                if (sampleRate.get().compareTo(0.0) == 0) {
+                    return Collections.emptyList();
+                }
+                if (Double.compare(_randomSupplier.get().nextDouble(), sampleRate.get()) > 0) {
+                    return Collections.emptyList();
+                }
             }
+
+            recordListBuilder.add(createRecord(name, value, type, annotations));
         }
 
-        return Collections.singletonList(createRecord(name, value, type, annotations));
+        return recordListBuilder.build();
     }
 
     private StatsdType parseStatsdType(final ByteBuffer datagram, final @Nullable String statsdTypeAsString) throws ParsingException {
@@ -225,6 +232,7 @@ public StatsdToRecordParser() {
             StatsdType.COUNTER,
             StatsdType.HISTOGRAM,
             StatsdType.TIMER);
+    private static final Splitter LINE_SPLITTER = Splitter.on('\n').omitEmptyStrings();
     private static final ThreadLocal<NumberFormat> NUMBER_FORMAT = ThreadLocal.withInitial(NumberFormat::getInstance);
     private static final Pattern STATSD_PATTERN = Pattern.compile(
             "^(?<NAME>[^:@|]+):(?<VALUE>[^|]+)\\|(?<TYPE>[^|]+)(\\|@(?<SAMPLERATE>[^|]+))?(\\|#(?<TAGS>.+))?$");