Substitute null for tag values that are null as per Mika.

ultrabrew · Dec 11, 2020 · d758ec1 · d758ec1
1 parent 9045e9f
commit d758ec1
Show file tree

Hide file tree

Showing 5 changed files with 65 additions and 14 deletions.
diff --git a/reporter-influxdb/src/main/java/io/ultrabrew/metrics/reporters/influxdb/InfluxDBClient.java b/reporter-influxdb/src/main/java/io/ultrabrew/metrics/reporters/influxdb/InfluxDBClient.java
@@ -33,6 +33,7 @@ public class InfluxDBClient {
   private static final byte COMMA = ',';
   private static final byte EQUALS = '=';
   private static final byte NEWLINE = '\n';
+  private static final byte[] NULL_STRING = new byte[] { 'N', 'U', 'L', 'L' };
 
   private final ByteBuffer byteBuffer;
   private final URI dbUri;
@@ -76,15 +77,15 @@ private void doWrite(final String measurement, final String[] tags, final String
         return;
       }
       if (Strings.isNullOrEmpty(tags[i + 1])) {
-        LOGGER.warn("Null or empty tag value in tags array: {} for measurement {}", 
-          Arrays.toString(tags), measurement);
-        byteBuffer.position(rollback);
-        return;
+        // TODO - Some users want this, some don't. Set a flag in the builder.
+        //LOGGER.warn("Null or empty tag value in tags array: {} for measurement {}", 
+        //  Arrays.toString(tags), measurement);
       }
       byteBuffer.put(COMMA)
           .put(tags[i].getBytes(UTF_8))
           .put(EQUALS)
-          .put(tags[i + 1].getBytes(UTF_8));
+          .put(Strings.isNullOrEmpty(tags[i + 1]) ? NULL_STRING : 
+            tags[i + 1].getBytes(UTF_8));
     }
     byteBuffer.put(WHITESPACE);
 

diff --git a/...er-influxdb/src/test/java/io/ultrabrew/metrics/reporters/influxdb/InfluxDBClientTest.java b/...er-influxdb/src/test/java/io/ultrabrew/metrics/reporters/influxdb/InfluxDBClientTest.java
@@ -163,13 +163,18 @@ public void testNullStrings() throws IOException {
     client.write("cpu_load_short", new String[] { null, "web01" }, new String[] { "temp", "80" }, 1534055562000000003L);
     // one good one in the middle
     client.write("cpu_load_short", new String[] { "host", "web01" }, new String[] { "temp", "80" }, 1534055562000000003L);
+    // this is ok too
     client.write("cpu_load_short", new String[] { "host", null }, new String[] { "temp", "80" }, 1534055562000000003L);
     client.write("cpu_load_short", new String[] { "host", "web01" }, new String[] { null, "80" }, 1534055562000000003L);
     client.write("cpu_load_short", new String[] { "host", "web01" }, new String[] { "temp", null }, 1534055562000000003L);
     client.flush();
     new Verifications() {{
-      httpClient.execute((HttpUriRequest) any);
+      HttpPost request;
+      httpClient.execute(request = withCapture());
       times = 1;
+      assertEquals("cpu_load_short,host=web01 temp=80 1534055562000000003\n" +
+                   "cpu_load_short,host=NULL temp=80 1534055562000000003\n", 
+                     EntityUtils.toString(request.getEntity()));
     }};
   }
 

diff --git a/...-influxdb/src/test/java/io/ultrabrew/metrics/reporters/influxdb/InfluxDBReporterTest.java b/...-influxdb/src/test/java/io/ultrabrew/metrics/reporters/influxdb/InfluxDBReporterTest.java
@@ -19,6 +19,7 @@
 import mockit.Verifications;
 import org.apache.http.StatusLine;
 import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpUriRequest;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.junit.jupiter.api.Test;
@@ -66,17 +67,39 @@ public void testSetBufferSize() {
   }
 
   @Test
-  public void testSeEndpoint() {
+  public void testSeEndpoint(@Mocked CloseableHttpClient httpClient,
+      @Mocked CloseableHttpResponse closeableHttpResponse, @Mocked StatusLine statusLine)
+      throws InterruptedException, IOException {
+    new Expectations() {{
+      httpClient.execute((HttpUriRequest) any);
+      result = closeableHttpResponse;
+      closeableHttpResponse.getStatusLine();
+      result = statusLine;
+      statusLine.getStatusCode();
+      result = 200;
+    }};
+
+    MetricRegistry registry = new MetricRegistry();
     InfluxDBReporter r = InfluxDBReporter.builder()
         .withBaseUri(TEST_URI)
         .withDatabase("test") // ignored
         .withEndpoint("/my/change?db=foo")
         .withBufferSize(12765)
         .build();
 
-    InfluxDBClient c = Deencapsulation.getField(r, "dbClient");
-    URI uri = Deencapsulation.getField(c, "dbUri");
-    assertEquals(TEST_URI + "/my/change?db=foo", uri.toString());
+    registry.addReporter(r);
+
+    Counter counter = registry.counter("counter");
+    counter.inc("tag", "value");
+
+    Thread.sleep(3000);
+
+    new Verifications() {{
+      HttpPost request;
+      httpClient.execute(request = withCapture());
+      times = 1;
+      assertEquals(TEST_URI + "/my/change?db=foo", request.getURI().toString());
+    }};
   }
 
   @Test

diff --git a/...er-opentsdb/src/main/java/io/ultrabrew/metrics/reporters/opentsdb/OpenTSDBHttpClient.java b/...er-opentsdb/src/main/java/io/ultrabrew/metrics/reporters/opentsdb/OpenTSDBHttpClient.java
@@ -4,7 +4,6 @@
 
 package io.ultrabrew.metrics.reporters.opentsdb;
 
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.ultrabrew.metrics.util.Strings;
 import java.util.Arrays;
 import java.io.ByteArrayOutputStream;
@@ -98,8 +97,8 @@ void write(final String metricName,
       return;
     }
     for (int i = 0; i < tags.length; i++) {
-      if (Strings.isNullOrEmpty(tags[i])) {
-        LOG.warn("Null tag key or value in: {} for metric {}", Arrays.toString(tags), metricName);
+      if (i % 2 == 0 && Strings.isNullOrEmpty(tags[i])) {
+        LOG.warn("Null tag key in: {} for metric {}", Arrays.toString(tags), metricName);
         return;
       }
     }
@@ -118,7 +117,11 @@ void write(final String metricName,
     writer.write(TAGS);
     boolean toggle = true;
     for (int i = 0; i < tags.length; i++) {
-      writeEscapedString(tags[i]);
+      if (i % 2 != 0 && tags[i] == null) {
+        writeEscapedString("NULL");
+      } else {
+        writeEscapedString(tags[i]);
+      }
       if (toggle) {
         writer.write(':');
       } else if (i + 1 < tags.length) {

diff --git a/...pentsdb/src/test/java/io/ultrabrew/metrics/reporters/opentsdb/OpenTSDBHttpClientTest.java b/...pentsdb/src/test/java/io/ultrabrew/metrics/reporters/opentsdb/OpenTSDBHttpClientTest.java
@@ -9,14 +9,17 @@
 
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import mockit.Capturing;
 import mockit.Expectations;
 import mockit.Mocked;
 import mockit.Verifications;
 import org.apache.http.StatusLine;
 import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpUriRequest;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.junit.jupiter.api.Test;
+import org.apache.http.util.EntityUtils;
 
 public class OpenTSDBHttpClientTest {
 
@@ -89,13 +92,29 @@ public void testWriteFailsManualFlush() throws Exception {
 
   @Test
   public void testNullStringsAndTagsValidation() throws Exception {
+    new Expectations() {{
+      httpClient.execute((HttpUriRequest) any);
+      result = closeableHttpResponse;
+      closeableHttpResponse.getStatusLine();
+      result = statusLine;
+      statusLine.getStatusCode();
+      result = 200;
+    }};
     OpenTSDBHttpClient client = new OpenTSDBHttpClient(DUMMY_DB_URI, 64, true);
     client.write(null, new String[] { "host", "web01" }, 1534055562000000003L, "80");
     client.write("cpu_load_short.temp", new String[] { null, "web01" }, 1534055562000000003L, "80");
     client.write("cpu_load_short.temp", new String[] { "host", null }, 1534055562000000003L, "80");
     client.write("cpu_load_short.temp", new String[] { "host", "web01" }, 1534055562000000003L, null);
     client.write("cpu_load_short.temp", null, 1534055562000000003L, "80");
     client.write("cpu_load_short.temp", new String[] { "host" }, 1534055562000000003L, "80");
+    client.flush();
+    new Verifications() {{
+      HttpPost request;
+      httpClient.execute(request = withCapture());
+      times = 1;
+      assertEquals("[{\"metric\":\"cpu_load_short.temp\",\"timestamp\":1534055562000000003,\"tags\":{\"host\":\"NULL\"},\"value\":80}]", 
+                     EntityUtils.toString(request.getEntity()));
+    }};
   }
 
   @Test