map custom counters to monotonic counter (#449)

Before these were incorrectly getting treated as regular
gauges. This happens at registration time.

servo-core/src/main/java/com/netflix/servo/SpectatorContext.java

@@ -37,7 +37,9 @@
 
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -107,9 +109,13 @@ public static LazyDistributionSummary distributionSummary(MonitorConfig config)
 
   /** Convert servo config to spectator id. */
   public static Id createId(MonitorConfig config) {
+    // Need to ensure that Servo type tag is removed to avoid incorrectly reprocessing the
+    // data in later transforms
+    Map<String, String> tags = new HashMap<>(config.getTags().asMap());
+    tags.remove("type");
     return registry
         .createId(config.getName())
-        .withTags(config.getTags().asMap());
+        .withTags(tags);
   }
 
   /** Dedicated thread pool for polling user defined functions registered as gauges. */
@@ -134,6 +140,25 @@ public static void register(Monitor<?> monitor) {
       ((SpectatorMonitor) monitor).initializeSpectator(BasicTagList.EMPTY);
     } else {
       PolledMeter.monitorMeter(registry, new ServoMeter(monitor));
+      monitorMonitonicValues(monitor);
+    }
+  }
+
+  private static boolean isCounter(MonitorConfig config) {
+    return "COUNTER".equals(config.getTags().getValue("type"));
+  }
+
+  private static void monitorMonitonicValues(Monitor<?> monitor) {
+    if (!(monitor instanceof SpectatorMonitor)) {
+      if (monitor instanceof CompositeMonitor<?>) {
+        CompositeMonitor<?> cm = (CompositeMonitor<?>) monitor;
+        for (Monitor<?> m : cm.getMonitors()) {
+          monitorMonitonicValues(m);
+        }
+      } else if (isCounter(monitor.getConfig())) {
+        polledGauge(monitor.getConfig())
+            .monitorMonotonicCounter(monitor, m -> ((Number) m.getValue()).longValue());
+      }
     }
   }
 
@@ -164,7 +189,7 @@ private void addMeasurements(Monitor<?> m, List<Measurement> measurements) {
           for (Monitor<?> v : cm.getMonitors()) {
             addMeasurements(v, measurements);
           }
-        } else {
+        } else if (!isCounter(m.getConfig())) {
           try {
             Object obj = m.getValue();
             if (obj instanceof Number) {

servo-core/src/test/java/com/netflix/servo/monitor/SpectatorIntegrationTest.java

@@ -158,9 +158,8 @@ public void testAnnotatedCounter() {
     Id id = registry.createId("counter")
         .withTag("class", "AnnotateExample")
         .withTag("level", "INFO")
-        .withTag("id", "foo")
-        .withTag("type", "COUNTER");
-    assertEquals(1, registry.counter(id.withTag(COUNTER)).count());
+        .withTag("id", "foo");
+    assertEquals(1, registry.counter(id).count());
   }
 
   @Test
@@ -207,7 +206,7 @@ public void testPeakRateCounter() {
     DefaultMonitorRegistry.getInstance().register(c);
     c.increment();
     PolledMeter.update(registry);
-    assertEquals(1.0, registry.gauge(ID.withTag("type", "GAUGE")).value());
+    assertEquals(1.0, registry.gauge(ID).value());
   }
 
   @Test
@@ -220,6 +219,15 @@ public void testPeakRateCounterRemove() {
     assertEquals(0, registry.stream().count());
   }
 
+  @Test
+  public void testCustomCounter() {
+    CustomCounter c = new CustomCounter();
+    register(c);
+    c.increment();
+    PolledMeter.update(registry);
+    assertEquals(1, registry.counter(ID).count());
+  }
+
   @Test
   public void testDoubleGauge() {
     DoubleGauge c = new DoubleGauge(CONFIG);
@@ -252,8 +260,7 @@ public void testAnnotatedGauge() {
     Id id = registry.createId("gauge")
         .withTag("class", "AnnotateExample")
         .withTag("level", "INFO")
-        .withTag("id", "foo")
-        .withTag("type", "GAUGE");
+        .withTag("id", "foo");
     assertEquals(42.0, registry.gauge(id).value(), 1e-12);
   }
 
@@ -279,7 +286,7 @@ public void testMinGauge() {
     g.update(42);
     clock.set(60000);
     PolledMeter.update(registry);
-    assertEquals(42.0, registry.gauge(ID.withTag("type", "GAUGE")).value());
+    assertEquals(42.0, registry.gauge(ID).value());
   }
 
   @Test
@@ -475,4 +482,26 @@ public Long getValue(int pollerIndex) {
       return 0L;
     }
   }
+
+  public static class CustomCounter implements com.netflix.servo.monitor.Monitor<Long> {
+
+    private final MonitorConfig config = CONFIG.withAdditionalTag(DataSourceType.COUNTER);
+    private long value = 0L;
+
+    public void increment() {
+      ++value;
+    }
+
+    @Override public Long getValue() {
+      return value;
+    }
+
+    @Override public Long getValue(int pollerIndex) {
+      return value;
+    }
+
+    @Override public MonitorConfig getConfig() {
+      return config;
+    }
+  }
 }