Telemetry: fix polling gauges always reporting NaN on Prometheus and timers

igor-core/igor-core.gradle

@@ -7,4 +7,6 @@ dependencies {
 
   // TODO(rz): Get rid of this dependency!
   implementation "com.squareup.retrofit:retrofit"
+
+  testImplementation "com.netflix.spectator:spectator-reg-micrometer"
 }

igor-core/src/main/java/com/netflix/spinnaker/igor/polling/CommonPollingMonitor.java

@@ -24,13 +24,17 @@
 import com.netflix.spinnaker.kork.discovery.RemoteStatusChangedEvent;
 import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService;
 import java.time.Duration;
+import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import javax.annotation.Nullable;
 import javax.annotation.PreDestroy;
 import net.logstash.logback.argument.StructuredArguments;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.scheduling.TaskScheduler;
@@ -44,16 +48,16 @@ public abstract class CommonPollingMonitor<I extends DeltaItem, T extends Pollin
   protected final Id missedNotificationId;
   private final DiscoveryStatusListener discoveryStatusListener;
   private final AtomicLong lastPoll = new AtomicLong();
-  private final Id itemsCachedId;
-  private final Id itemsOverThresholdId;
-  private final Id pollCycleFailedId;
-  private final Id pollCycleTimingId;
   private final Optional<LockService> lockService;
   private ScheduledFuture<?> monitor;
+  private final CommonPollingMonitorInstrumentation instrumentation;
   protected Logger log = LoggerFactory.getLogger(getClass());
   protected TaskScheduler scheduler;
   protected final DynamicConfigService dynamicConfigService;
 
+  private Map<String, AtomicInteger> itemsOverThresholdMap = new ConcurrentHashMap<>();
+  private Map<String, AtomicInteger> itemsCachedMap = new ConcurrentHashMap<>();
+
   public CommonPollingMonitor(
       IgorConfigurationProperties igorProperties,
       Registry registry,
@@ -67,12 +71,9 @@ public CommonPollingMonitor(
     this.discoveryStatusListener = discoveryStatusListener;
     this.lockService = lockService;
     this.scheduler = scheduler;
+    this.instrumentation = new CommonPollingMonitorInstrumentation(registry);
 
-    itemsCachedId = registry.createId("pollingMonitor.newItems");
-    itemsOverThresholdId = registry.createId("pollingMonitor.itemsOverThreshold");
-    pollCycleFailedId = registry.createId("pollingMonitor.failed");
     missedNotificationId = registry.createId("pollingMonitor.missedEchoNotification");
-    pollCycleTimingId = registry.createId("pollingMonitor.pollTiming");
   }
 
   @Override
@@ -86,20 +87,19 @@ public void onApplicationEvent(RemoteStatusChangedEvent event) {
     this.monitor =
         scheduler.schedule(
             () ->
-                registry
-                    .timer(pollCycleTimingId.withTag("monitor", getClass().getSimpleName()))
-                    .record(
-                        () -> {
-                          if (isInService()) {
-                            poll(true);
-                            lastPoll.set(System.currentTimeMillis());
-                          } else {
-                            log.info(
-                                "not in service (lastPoll: {})",
-                                (lastPoll.get() == 0) ? "n/a" : lastPoll.toString());
-                            lastPoll.set(0);
-                          }
-                        }),
+                instrumentation.trackPollCycleTime(
+                    this.getName(),
+                    () -> {
+                      if (isInService()) {
+                        poll(true);
+                        lastPoll.set(System.currentTimeMillis());
+                      } else {
+                        log.info(
+                            "not in service (lastPoll: {})",
+                            (lastPoll.get() == 0) ? "n/a" : lastPoll.toString());
+                        lastPoll.set(0);
+                      }
+                    }),
             new PeriodicTrigger(getPollInterval(), TimeUnit.SECONDS));
   }
 
@@ -163,7 +163,17 @@ protected String getLockName(String name, String partition) {
   }
 
   protected void internalPollSingle(PollContext ctx) {
-    String monitorName = getClass().getSimpleName();
+    String monitorName =
+        !StringUtils.isBlank(this.getName()) ? this.getName() : getClass().getSimpleName();
+
+    itemsCachedMap.putIfAbsent(ctx.partitionName, new AtomicInteger(0));
+    itemsOverThresholdMap.putIfAbsent(ctx.partitionName, new AtomicInteger(0));
+
+    instrumentation.trackItemsCached(
+        itemsCachedMap.get(ctx.partitionName), monitorName, ctx.partitionName);
+
+    instrumentation.trackItemsOverThreshold(
+        itemsOverThresholdMap.get(ctx.partitionName), monitorName, ctx.partitionName);
 
     try {
       T delta = generateDelta(ctx);
@@ -175,58 +185,40 @@ protected void internalPollSingle(PollContext ctx) {
       boolean sendEvents = !ctx.fastForward;
       int deltaSize = delta.getItems().size();
       if (deltaSize > upperThreshold) {
-        registry
-            .gauge(
-                itemsOverThresholdId.withTags(
-                    "monitor", monitorName, "partition", ctx.partitionName))
-            .set(deltaSize);
+        itemsOverThresholdMap.get(ctx.partitionName).set(deltaSize);
         if (ctx.fastForward) {
           log.warn(
               "Fast forwarding items ({}) in {} {}",
-              deltaSize,
+              itemsOverThresholdMap.get(ctx.partitionName).get(),
               StructuredArguments.kv("monitor", monitorName),
               StructuredArguments.kv("partition", ctx.partitionName));
           sendEvents = false;
         } else {
           log.error(
               "Number of items ({}) to cache exceeds upper threshold ({}) in {} {}",
-              deltaSize,
+              itemsOverThresholdMap.get(ctx.partitionName).get(),
               upperThreshold,
               StructuredArguments.kv("monitor", monitorName),
               StructuredArguments.kv("partition", ctx.partitionName));
           return;
         }
       } else {
-        registry
-            .gauge(
-                itemsOverThresholdId.withTags(
-                    "monitor", monitorName, "partition", ctx.partitionName))
-            .set(0);
+        itemsOverThresholdMap.get(ctx.partitionName).set(0);
       }
 
       sendEvents = sendEvents && isSendEventsEnabled();
 
       commitDelta(delta, sendEvents);
-      registry
-          .gauge(itemsCachedId.withTags("monitor", monitorName, "partition", ctx.partitionName))
-          .set(deltaSize);
+      itemsCachedMap.get(ctx.partitionName).set(deltaSize);
     } catch (Exception e) {
       log.error(
           "Failed to update monitor items for {}:{}",
           StructuredArguments.kv("monitor", monitorName),
           StructuredArguments.kv("partition", ctx.partitionName),
           e);
-      registry
-          .counter(
-              pollCycleFailedId.withTags("monitor", monitorName, "partition", ctx.partitionName))
-          .increment();
-      registry
-          .gauge(itemsCachedId.withTags("monitor", monitorName, "partition", ctx.partitionName))
-          .set(0);
-      registry
-          .gauge(
-              itemsOverThresholdId.withTags("monitor", monitorName, "partition", ctx.partitionName))
-          .set(0);
+      instrumentation.trackPollCycleFailed(monitorName, ctx.partitionName);
+      itemsCachedMap.get(ctx.partitionName).set(0);
+      itemsOverThresholdMap.get(ctx.partitionName).set(0);
     }
   }
 

igor-core/src/main/java/com/netflix/spinnaker/igor/polling/CommonPollingMonitorInstrumentation.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2024 Wise, PLC.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netflix.spinnaker.igor.polling;
+
+import com.netflix.spectator.api.Gauge;
+import com.netflix.spectator.api.Id;
+import com.netflix.spectator.api.Registry;
+import com.netflix.spectator.api.patterns.PolledMeter;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class CommonPollingMonitorInstrumentation {
+
+  private final Registry registry;
+  private final Id itemsCachedId;
+  private final Id itemsOverThresholdId;
+  private final Id pollCycleFailedId;
+  private final Id pollCycleTimingId;
+
+  public CommonPollingMonitorInstrumentation(Registry registry) {
+    this.registry = registry;
+    itemsCachedId = registry.createId("pollingMonitor.newItems");
+    itemsOverThresholdId = registry.createId("pollingMonitor.itemsOverThreshold");
+    pollCycleFailedId = registry.createId("pollingMonitor.failed");
+    pollCycleTimingId = registry.createId("pollingMonitor.pollTiming");
+  }
+
+  public void trackItemsCached(AtomicInteger numberOfItems, String monitor, String partition) {
+    Gauge gauge =
+        (Gauge) registry.get(itemsCachedId.withTags("monitor", monitor, "partition", partition));
+
+    /*
+    Spectator gauges are slightly different from Micrometer ones: the polling gauge
+    has been deprecated in favour of PolledMeter.
+    Previous implementation resulted in NaN most of the times, while we want to store observations
+    on Prometheus. We don't need a DistributionSummary, but just a gauge which doesn't get garbage
+    collected. For a unique combination of (metricName, tags), PolledMeter will sum up the observed
+    values, so to avoid this, we set the gauge ONCE and we pass a strong reference holding the
+    latest value to be observed.
+    */
+
+    if (gauge == null) {
+      PolledMeter.using(registry)
+          .withId(itemsCachedId.withTags("monitor", monitor, "partition", partition))
+          .monitorValue(numberOfItems);
+    }
+  }
+
+  public void trackItemsOverThreshold(
+      AtomicInteger numberOfItems, String monitor, String partition) {
+    Gauge gauge =
+        (Gauge)
+            registry.get(itemsOverThresholdId.withTags("monitor", monitor, "partition", partition));
+    if (gauge == null) {
+      PolledMeter.using(registry)
+          .withId(itemsOverThresholdId.withTags("monitor", monitor, "partition", partition))
+          .monitorValue(numberOfItems);
+    }
+  }
+
+  public void trackPollCycleTime(String monitor, Runnable lambda) {
+    registry.timer(pollCycleTimingId.withTags("monitor", monitor)).record(lambda);
+  }
+
+  public void trackPollCycleFailed(String monitor, String partition) {
+    registry
+        .counter(getPollCycleFailedId().withTags("monitor", monitor, "partition", partition))
+        .increment();
+  }
+
+  public Id getItemsCachedId() {
+    return itemsCachedId;
+  }
+
+  public Id getItemsOverThresholdId() {
+    return itemsOverThresholdId;
+  }
+
+  public Id getPollCycleFailedId() {
+    return pollCycleFailedId;
+  }
+
+  public Id getPollCycleTimingId() {
+    return pollCycleTimingId;
+  }
+}