Fix prometheus reporting issue (#507)

MetricsType.ACTIVITY_SCHEDULED_TO_START_LATENCY is reported twice with different set of tags.
It causes prometheus to fail with an IllegalArgumentException: "Prometheus requires that all meters with the same name have the same set of tag keys."
Made all metrics to be reported with the same set of tags.

Improved reporter to track gauges as well.

Author: AleksandrKolosov

src/main/java/com/uber/cadence/reporter/CadenceClientStatsReporter.java

@@ -17,6 +17,11 @@
 
 package com.uber.cadence.reporter;
 
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.AtomicDouble;
+import com.uber.cadence.internal.metrics.MetricsTag;
 import com.uber.m3.tally.Buckets;
 import com.uber.m3.tally.Capabilities;
 import com.uber.m3.tally.CapableOf;
@@ -25,11 +30,13 @@
 import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.Tag;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
 
 public class CadenceClientStatsReporter implements StatsReporter {
 
+  private final Map<String, AtomicDouble> gauges = new ConcurrentHashMap<>();
+
   @Override
   public Capabilities capabilities() {
     return CapableOf.REPORTING;
@@ -52,7 +59,12 @@ public void reportCounter(String name, Map<String, String> tags, long value) {
 
   @Override
   public void reportGauge(String name, Map<String, String> tags, double value) {
-    // NOOP
+    AtomicDouble gauge = gauges.computeIfAbsent(name, metricName -> {
+      AtomicDouble result = Metrics.gauge(name, getTags(tags), new AtomicDouble());
+      Preconditions.checkNotNull(result, "Metrics.gauge should not return null ever");
+      return result;
+    });
+    gauge.set(value);
   }
 
   @Override
@@ -83,9 +95,11 @@ public void reportHistogramDurationSamples(
   }
 
   private Iterable<Tag> getTags(Map<String, String> tags) {
-    return tags.entrySet()
-        .stream()
-        .map(entry -> Tag.of(entry.getKey(), entry.getValue()))
-        .collect(Collectors.toList());
+    return ImmutableList.of(
+        Tag.of(MetricsTag.ACTIVITY_TYPE, Strings.nullToEmpty(tags.get(MetricsTag.ACTIVITY_TYPE))),
+        Tag.of(MetricsTag.DOMAIN, Strings.nullToEmpty(tags.get(MetricsTag.DOMAIN))),
+        Tag.of(MetricsTag.TASK_LIST, Strings.nullToEmpty(tags.get(MetricsTag.TASK_LIST))),
+        Tag.of(MetricsTag.WORKFLOW_TYPE, Strings.nullToEmpty(tags.get(MetricsTag.WORKFLOW_TYPE)))
+    );
   }
 }

src/test/java/com/uber/cadence/reporter/CadenceClientStatsReporterTest.java

@@ -19,13 +19,15 @@
 
 import static org.junit.Assert.assertEquals;
 
+import com.uber.cadence.internal.metrics.MetricsTag;
 import com.uber.m3.tally.CapableOf;
 import com.uber.m3.util.Duration;
 import com.uber.m3.util.ImmutableMap;
 import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import org.junit.After;
@@ -36,9 +38,15 @@ public class CadenceClientStatsReporterTest {
 
   private static final String DEFAULT_REPORT_NAME = "cadence_workflow_start";
   private static final Map<String, String> DEFAULT_REPORT_TAGS =
-      ImmutableMap.of("Domain", "domain_name", "TaskList", "task_list");
+      ImmutableMap.of(MetricsTag.DOMAIN, "domain_name", MetricsTag.TASK_LIST, "task_list");
   private static final long DEFAULT_COUNT = 10;
+  private static final double DEFAULT_VALUE = 1.0;
   private static final Duration DEFAULT_DURATION = Duration.ofSeconds(10);
+  private static final List<Tag> EXPECTED_REPORT_TAGS = Arrays.asList(
+      Tag.of(MetricsTag.ACTIVITY_TYPE, ""),
+      Tag.of(MetricsTag.DOMAIN, "domain_name"),
+      Tag.of(MetricsTag.TASK_LIST, "task_list"),
+      Tag.of(MetricsTag.WORKFLOW_TYPE, ""));
 
   private CadenceClientStatsReporter cadenceClientStatsReporter = new CadenceClientStatsReporter();
 
@@ -61,8 +69,7 @@ public void testReporterCapabilitiesShouldReturnReporting() {
   public void testCounterShouldCallMetricRegistryForMonitoredCounterCadenceAction() {
     callDefaultCounter();
 
-    assertEquals(
-        Arrays.asList(Tag.of("Domain", "domain_name"), Tag.of("TaskList", "task_list")),
+    assertEquals(EXPECTED_REPORT_TAGS,
         Metrics.globalRegistry.get(DEFAULT_REPORT_NAME).counter().getId().getTags());
     assertEquals(10, Metrics.globalRegistry.get(DEFAULT_REPORT_NAME).counter().count(), 0);
   }
@@ -71,13 +78,21 @@ public void testCounterShouldCallMetricRegistryForMonitoredCounterCadenceAction(
   public void testTimerShouldCallMetricRegistryForMonitoredCounterCadenceAction() {
     callDefaultTimer();
 
-    assertEquals(
-        Arrays.asList(Tag.of("Domain", "domain_name"), Tag.of("TaskList", "task_list")),
+    assertEquals(EXPECTED_REPORT_TAGS,
         Metrics.globalRegistry.get(DEFAULT_REPORT_NAME).timer().getId().getTags());
     assertEquals(
         10, Metrics.globalRegistry.get(DEFAULT_REPORT_NAME).timer().totalTime(TimeUnit.SECONDS), 0);
   }
 
+  @Test
+  public void testGaugeShouldCallMetricRegistryForMonitoredGaugeCadenceAction() {
+    callDefaultGauge();
+
+    assertEquals(EXPECTED_REPORT_TAGS,
+        Metrics.globalRegistry.get(DEFAULT_REPORT_NAME).gauge().getId().getTags());
+    assertEquals(1.0, Metrics.globalRegistry.get(DEFAULT_REPORT_NAME).gauge().value(), 0);
+  }
+
   private void callDefaultCounter() {
     cadenceClientStatsReporter.reportCounter(
         DEFAULT_REPORT_NAME, DEFAULT_REPORT_TAGS, DEFAULT_COUNT);
@@ -87,4 +102,8 @@ private void callDefaultTimer() {
     cadenceClientStatsReporter.reportTimer(
         DEFAULT_REPORT_NAME, DEFAULT_REPORT_TAGS, DEFAULT_DURATION);
   }
+
+  private void callDefaultGauge() {
+    cadenceClientStatsReporter.reportGauge(DEFAULT_REPORT_NAME, DEFAULT_REPORT_TAGS, DEFAULT_VALUE);
+  }
 }