tim-group · prodjito · Jul 23, 2020
diff --git a/src/main/java/com/timgroup/statsd/NoOpStatsDClient.java b/src/main/java/com/timgroup/statsd/NoOpStatsDClient.java
@@ -16,4 +16,12 @@ public final class NoOpStatsDClient extends ConvenienceMethodProvidingStatsDClie
     @Override public void recordGaugeDelta(String aspect, double delta) { }
     @Override public void recordSetEvent(String aspect, String value) { }
     @Override public void recordExecutionTime(String aspect, long timeInMs, double sampleRate) { }
+    @Override public void count(String aspect, long delta, double sampleRate, String[] tags) { }
+    @Override public void recordGaugeValue(String aspect, long value, String[] tags) { }
+    @Override public void recordGaugeValue(String aspect, double value, String[] tags) { }
+    @Override public void recordGaugeDelta(String aspect, long delta, String[] tags) { }
+    @Override public void recordGaugeDelta(String aspect, double delta, String[] tags) { }
+    @Override public void recordSetEvent(String aspect, String value, String[] tags) { }
+    @Override public void recordExecutionTime(String aspect, long timeInMs, double sampleRate, String[] tags) { }
+    @Override public void setClientTags(String[] tags) { }
 }
diff --git a/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java b/src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java
@@ -2,6 +2,8 @@
 
 import java.nio.charset.Charset;
 import java.text.NumberFormat;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Locale;
 
 /**
@@ -33,13 +35,16 @@
 public final class NonBlockingStatsDClient extends ConvenienceMethodProvidingStatsDClient {
 
     private static final Charset STATS_D_ENCODING = Charset.forName("UTF-8");
+    private static final String STATS_D_TAG_PREFIX = "|#";
 
     private static final StatsDClientErrorHandler NO_OP_HANDLER = new StatsDClientErrorHandler() {
         @Override public void handle(Exception e) { /* No-op */ }
     };
 
     private final String prefix;
     private final NonBlockingUdpSender sender;
+
+    private String clientTags = "";
 
     /**
      * Create a new StatsD client communicating with a StatsD instance on the
@@ -122,6 +127,25 @@ public void stop() {
     public void count(String aspect, long delta, double sampleRate) {
         send(messageFor(aspect, Long.toString(delta), "c", sampleRate));
     }
+
+    /**
+     * Adjusts the specified counter by a given delta.
+     * 
+     * <p>This method is non-blocking and is guaranteed not to throw an exception.</p>
+     * 
+     * @param aspect
+     *     the name of the counter to adjust
+     * @param delta
+     *     the amount to adjust the counter by
+     * @param sampleRate
+     *     the sampling rate being employed. For example, a rate of 0.1 would tell StatsD that this counter is being sent
+     *     sampled every 1/10th of the time.
+     * @param tags
+     *     A string array containing one or more tags. Each tag can be in the format of key:value, e.g. key1:value1. Or it can be just a key, e.g. key3.
+     */
+    public void count(String aspect, long delta, double sampleRate, String[] tags) {
+        send(messageFor(aspect, Long.toString(delta), "c", sampleRate, tags));
+    }
 
     /**
      * Records the latest fixed value for the specified named gauge.
@@ -152,6 +176,22 @@ public void recordGaugeDelta(String aspect, long value) {
     public void recordGaugeDelta(String aspect, double value) {
         recordGaugeCommon(aspect, stringValueOf(value), value < 0, true);
     }
+
+    public void recordGaugeValue(String aspect, long value, String[] tags) {
+        recordGaugeCommon(aspect, Long.toString(value), value < 0, false, tags);
+    }
+
+    public void recordGaugeValue(String aspect, double value, String[] tags) {
+        recordGaugeCommon(aspect, stringValueOf(value), value < 0, false, tags);
+    }
+
+    public void recordGaugeDelta(String aspect, long value, String[] tags) {
+        recordGaugeCommon(aspect, Long.toString(value), value < 0, true, tags);
+    }
+
+    public void recordGaugeDelta(String aspect, double value, String[] tags) {
+        recordGaugeCommon(aspect, stringValueOf(value), value < 0, true, tags);
+    }
 
     private void recordGaugeCommon(String aspect, String value, boolean negative, boolean delta) {
         final StringBuilder message = new StringBuilder();
@@ -161,6 +201,15 @@ private void recordGaugeCommon(String aspect, String value, boolean negative, bo
         message.append(messageFor(aspect, (delta && !negative) ? ("+" + value) : value, "g"));
         send(message.toString());
     }
+
+    private void recordGaugeCommon(String aspect, String value, boolean negative, boolean delta, String[] tags) {
+        final StringBuilder message = new StringBuilder();
+        if (!delta && negative) {
+            message.append(messageFor(aspect, "0", "g", 1.0, tags)).append('\n');
+        }
+        message.append(messageFor(aspect, (delta && !negative) ? ("+" + value) : value, "g", 1.0, tags));
+        send(message.toString());
+    }
 
     /**
      * StatsD supports counting unique occurrences of events between flushes, Call this method to records an occurrence
@@ -177,6 +226,10 @@ private void recordGaugeCommon(String aspect, String value, boolean negative, bo
     public void recordSetEvent(String aspect, String eventName) {
         send(messageFor(aspect, eventName, "s"));
     }
+
+    public void recordSetEvent(String aspect, String eventName, String[] tags) {
+        send(messageFor(aspect, eventName, "s", 1.0, tags));
+    }
 
     /**
      * Records an execution time in milliseconds for the specified named operation.
@@ -192,18 +245,37 @@ public void recordSetEvent(String aspect, String eventName) {
     public void recordExecutionTime(String aspect, long timeInMs, double sampleRate) {
         send(messageFor(aspect, Long.toString(timeInMs), "ms", sampleRate));
     }
+
+    public void recordExecutionTime(String aspect, long timeInMs, double sampleRate, String[] tags) {
+        send(messageFor(aspect, Long.toString(timeInMs), "ms", sampleRate, tags));
+    }
+
+    /**
+     * Set tags at the client level. These tags will be added to all metrics.
+     * 
+     * @param tags
+     *     A string array containing one or more tags. Each tag can be in the format of key:value, e.g. key1:value1. Or it can be just a key, e.g. key3.
+     */
+    public void setClientTags(String[] tags) {
+    	List<String> tagsList = Arrays.asList(tags);
+    	clientTags += String.join(",", tagsList);
+    }
 
     private String messageFor(String aspect, String value, String type) {
         return messageFor(aspect, value, type, 1.0);
     }
-
+    
     private String messageFor(String aspect, String value, String type, double sampleRate) {
+    	return messageFor(aspect, value, type, sampleRate, null);
+    }
+
+    private String messageFor(String aspect, String value, String type, double sampleRate, String[] tags) {
         final String message = prefix + aspect + ':' + value + '|' + type;
-        return (sampleRate == 1.0)
+        return addTags(tags, (sampleRate == 1.0)
                 ? message
-                : (message + "|@" + stringValueOf(sampleRate));
+                : (message + "|@" + stringValueOf(sampleRate)));
     }
-
+    
     private void send(final String message) {
         sender.send(message);
     }
@@ -214,4 +286,18 @@ private String stringValueOf(double value) {
         formatter.setMaximumFractionDigits(19);
         return formatter.format(value);
     }
+
+    private String addTags(final String[] tags, String message) {
+    	if(tags == null && clientTags.isEmpty()) return message;
+    	StringBuilder sb = new StringBuilder(message);
+    	sb.append(STATS_D_TAG_PREFIX);
+    	if(!clientTags.isEmpty()) {
+    		sb.append(clientTags);
+    		if(tags != null) {
+    			sb.append(",");
+    		}
+    	}
+    	sb.append(String.join(",",Arrays.asList(tags)));
+        return sb.toString();
+    }
 }
diff --git a/src/main/java/com/timgroup/statsd/StatsDClient.java b/src/main/java/com/timgroup/statsd/StatsDClient.java
@@ -50,6 +50,23 @@ public interface StatsDClient {
      *     sampled every 1/10th of the time.
      */
     void count(String aspect, long delta, double sampleRate);
+
+    /**
+     * Adjusts the specified counter by a given delta.
+     *
+     * <p>This method is non-blocking and is guaranteed not to throw an exception.</p>
+     *
+     * @param aspect
+     *     the name of the counter to adjust
+     * @param delta
+     *     the amount to adjust the counter by
+     * @param sampleRate
+     *     the sampling rate being employed. For example, a rate of 0.1 would tell StatsD that this counter is being sent
+     *     sampled every 1/10th of the time.
+     * @param tags
+     *     A string array containing one or more tags. Each tag can be in the format of key:value, e.g. key1:value1. Or it can be just a key, e.g. key3.
+     */
+    void count(String aspect, long delta, double sampleRate, String[] tags);
 
     /**
      * Increments the specified counter by one.
@@ -116,7 +133,45 @@ public interface StatsDClient {
     void recordGaugeDelta(String aspect, double delta);
 
     /**
-     * Convenience method equivalent to {@link #recordGaugeValue(String, long)}.
+     * Records the latest fixed value for the specified named gauge.
+     * 
+     * <p>This method is non-blocking and is guaranteed not to throw an exception.</p>
+     * 
+     * @param aspect
+     *     the name of the gauge
+     * @param value
+     *     the new reading of the gauge
+     * @param tags
+     *     A string array containing one or more tags. Each tag can be in the format of key:value, e.g. key1:value1. Or it can be just a key, e.g. key3.
+     */
+    void recordGaugeValue(String aspect, long value, String[] tags);
+
+    /**
+     * Convenience method equivalent to {@link #recordGaugeValue(String, long, String[])} but for double values.
+     */
+    void recordGaugeValue(String aspect, double value, String[] tags);
+
+    /**
+     * Records a change in the value of the specified named gauge.
+     *
+     * <p>This method is non-blocking and is guaranteed not to throw an exception.</p>
+     *
+     * @param aspect
+     *     the name of the gauge
+     * @param delta
+     *     the +/- delta to apply to the gauge
+     * @param tags
+     *     A string array containing one or more tags. Each tag can be in the format of key:value, e.g. key1:value1. Or it can be just a key, e.g. key3.
+     */
+    void recordGaugeDelta(String aspect, long delta, String[] tags);
+
+    /**
+     * Convenience method equivalent to {@link #recordGaugeDelta(String, long, String[])} but for double deltas.
+     */
+    void recordGaugeDelta(String aspect, double delta, String[] tags);
+
+    /**
+     * Convenience method equivalent to {@link #recordGaugeValue(String, long, String[])}.
      */
     void gauge(String aspect, long value);
 
@@ -137,6 +192,21 @@ public interface StatsDClient {
      *     the value to be added to the set
      */
     void recordSetEvent(String aspect, String eventName);
+
+    /**
+     * StatsD supports counting unique occurrences of events between flushes, Call this method to records an occurrence
+     * of the specified named event.
+     * 
+     * <p>This method is non-blocking and is guaranteed not to throw an exception.</p>
+     * 
+     * @param aspect
+     *     the name of the set
+     * @param eventName
+     *     the value to be added to the set
+     * @param tags
+     *     A string array containing one or more tags. Each tag can be in the format of key:value, e.g. key1:value1. Or it can be just a key, e.g. key3.
+     */
+    void recordSetEvent(String aspect, String eventName, String[] tags);
 
     /**
      * Convenience method equivalent to {@link #recordSetEvent(String, String)}.
@@ -169,6 +239,23 @@ public interface StatsDClient {
      *     sampled every 1/10th of the time, so that it updates its timer_counters appropriately.
      */
     void recordExecutionTime(String aspect, long timeInMs, double sampleRate);
+
+    /**
+     * Adjusts the specified counter by a given delta.
+     *
+     * <p>This method is non-blocking and is guaranteed not to throw an exception.</p>
+     *
+     * @param aspect
+     *     the name of the counter to adjust
+     * @param delta
+     *     the amount to adjust the counter by
+     * @param sampleRate
+     *     the sampling rate being employed. For example, a rate of 0.1 would tell StatsD that this timer is being sent
+     *     sampled every 1/10th of the time, so that it updates its timer_counters appropriately.
+     * @param tags
+     *     A string array containing one or more tags. Each tag can be in the format of key:value, e.g. key1:value1. Or it can be just a key, e.g. key3.
+     */
+    void recordExecutionTime(String aspect, long timeInMs, double sampleRate, String[] tags);
 
     /**
      * Records an execution time in milliseconds for the specified named operation. The execution
@@ -188,5 +275,13 @@ public interface StatsDClient {
      * Convenience method equivalent to {@link #recordExecutionTime(String, long)}.
      */
     void time(String aspect, long value);
+
+    /**
+     * Set tags at the client level. These tags will be added to all metrics.
+     * 
+     * @param tags
+     *     A string array containing one or more tags. Each tag can be in the format of key:value, e.g. key1:value1. Or it can be just a key, e.g. key3.
+     */
+    void setClientTags(String[] tags);
 
 }
diff --git a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java
@@ -97,7 +97,7 @@ public void stop() throws Exception {
     }
 
     @Test(timeout=5000L) public void
-    sends_negagive_gauge_to_statsd_by_resetting_to_zero_first() throws Exception {
+    sends_negative_gauge_to_statsd_by_resetting_to_zero_first() throws Exception {
         client.recordGaugeValue("mygauge", -423L);
         server.waitForMessage();
 
@@ -178,6 +178,39 @@ public void stop() throws Exception {
 
         assertThat(server.messagesReceived(), contains("my.prefix.mytime:0|ms"));
     }
+
+    @Test(timeout=5000L) public void
+    sends_tagged_counter_value_with_rate_to_statsd() throws Exception {
+    	client.setClientTags(new String[]{"key1:value1", "key2:value2", "key3"});
+        client.count("mycount", Long.MAX_VALUE, 0.00024, new String[]{"key4:value4", "key5:value5", "key6"});
+        server.waitForMessage();
+
+        assertThat(server.messagesReceived(), contains("my.prefix.mycount:9223372036854775807|c|@0.00024|#key1:value1,key2:value2,key3,key4:value4,key5:value5,key6"));
+    }
+
+    @Test(timeout=5000L) public void
+    sends_tagged_negative_gauge_to_statsd_by_resetting_to_zero_first() throws Exception {
+    	client.recordGaugeValue("mygauge", -423L, new String[]{"key1:value1", "key2:value2", "key3"});
+        server.waitForMessage();
+
+        assertThat(server.messagesReceived(), contains("my.prefix.mygauge:0|g|#key1:value1,key2:value2,key3\nmy.prefix.mygauge:-423|g|#key1:value1,key2:value2,key3"));
+    }
+
+    @Test(timeout=5000L) public void
+    sends_tagged_set_to_statsd() throws Exception {
+        client.recordSetEvent("myset", "test", new String[]{"key1:value1", "key2:value2", "key3"});
+        server.waitForMessage();
+
+        assertThat(server.messagesReceived(), contains("my.prefix.myset:test|s|#key1:value1,key2:value2,key3"));
+    }
+
+    @Test(timeout=5000L) public void
+    sends_tagged_timer_with_rate_to_statsd() throws Exception {
+        client.recordExecutionTime("mytime", 123L, 0.000123, new String[]{"key1:value1", "key2:value2", "key3"});
+        server.waitForMessage();
+
+        assertThat(server.messagesReceived(), contains("my.prefix.mytime:123|ms|@0.000123|#key1:value1,key2:value2,key3"));
+    }
 
     @Test(timeout=5000L) public void
     allows_empty_prefix() {