Add capability to correctly export fixed buckets histogram to SignalFx (#2977)

* Add capability to correctly export fixed buckets histogram to SignalFx

* Fix to always include the "+Inf" bucket.
* Fix to never expire per bucket counters.
* The metrics are still reported as Gauges by default, to fix this user has to enable a property.

Signed-off-by: Bogdan Drutu <[email protected]>

* Address feedback

Signed-off-by: Bogdan Drutu <[email protected]>

* Make SignalFx histograms cumulative without breaking max()

* Hide DistributionStatisticsConfig changes behind a config flag

* Polishing pass

Co-authored-by: Mateusz Rzeszutek <[email protected]>
Co-authored-by: Jonatan Ivanov <[email protected]>

Author: 3 people

implementations/micrometer-registry-signalfx/src/main/java/io/micrometer/signalfx/CumulativeHistogramConfigUtil.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2022 VMware, Inc.
+ *
+ * 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
+ *
+ * https://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 io.micrometer.signalfx;
+
+import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
+
+import java.time.Duration;
+import java.util.Arrays;
+
+/**
+ * Adds cumulative histogram capabilities to the {@link DistributionStatisticConfig}.
+ *
+ * @author Bogdan Drutu
+ * @author Mateusz Rzeszutek
+ */
+final class CumulativeHistogramConfigUtil {
+
+    static DistributionStatisticConfig updateConfig(DistributionStatisticConfig distributionStatisticConfig) {
+        double[] sloBoundaries = distributionStatisticConfig.getServiceLevelObjectiveBoundaries();
+        if (sloBoundaries == null || sloBoundaries.length == 0) {
+            return distributionStatisticConfig;
+        }
+        double[] newSloBoundaries = sloBoundaries;
+        // Add the +Inf bucket since the "count" resets every export.
+        if (!isPositiveInf(sloBoundaries[sloBoundaries.length - 1])) {
+            newSloBoundaries = Arrays.copyOf(sloBoundaries, sloBoundaries.length + 1);
+            newSloBoundaries[newSloBoundaries.length - 1] = Double.MAX_VALUE;
+        }
+
+        return DistributionStatisticConfig.builder()
+                // Set the expiration duration for the histogram counts to be effectively infinite.
+                // Without this, the counts are reset every expiry duration.
+                .expiry(Duration.ofNanos(Long.MAX_VALUE)) // effectively infinite
+                .bufferLength(1)
+                .serviceLevelObjectives(newSloBoundaries)
+                .build()
+                .merge(distributionStatisticConfig);
+    }
+
+    private static boolean isPositiveInf(double bucket) {
+        return bucket == Double.POSITIVE_INFINITY || bucket == Double.MAX_VALUE || (long) bucket == Long.MAX_VALUE;
+    }
+
+    private CumulativeHistogramConfigUtil() {
+    }
+}

implementations/micrometer-registry-signalfx/src/main/java/io/micrometer/signalfx/SignalFxConfig.java

@@ -42,6 +42,13 @@ default String accessToken() {
         return getSecret(this, "accessToken").required().get();
     }
 
+    /**
+     * @return {@code true} if the SignalFx registry should emit cumulative histogram buckets.
+     */
+    default boolean publishCumulativeHistogram() {
+        return getBoolean(this, "publishCumulativeHistogram").orElse(false);
+    }
+
     /**
      * @return The URI to ship metrics to. If you need to publish metrics to an internal proxy en route to
      * SignalFx, you can define the location of the proxy with this.

implementations/micrometer-registry-signalfx/src/main/java/io/micrometer/signalfx/SignalFxMeterRegistry.java

@@ -23,8 +23,21 @@
 import com.signalfx.metrics.errorhandler.OnSendErrorHandler;
 import com.signalfx.metrics.flush.AggregateMetricSender;
 import com.signalfx.metrics.protobuf.SignalFxProtocolBuffers;
-import io.micrometer.core.instrument.*;
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.FunctionCounter;
+import io.micrometer.core.instrument.FunctionTimer;
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.LongTaskTimer;
+import io.micrometer.core.instrument.Meter;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.TimeGauge;
+import io.micrometer.core.instrument.Timer;
 import io.micrometer.core.instrument.config.NamingConvention;
+import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
+import io.micrometer.core.instrument.distribution.HistogramGauges;
+import io.micrometer.core.instrument.distribution.pause.PauseDetector;
 import io.micrometer.core.instrument.step.StepMeterRegistry;
 import io.micrometer.core.instrument.util.MeterPartition;
 import io.micrometer.core.instrument.util.NamedThreadFactory;
@@ -41,6 +54,7 @@
 import java.util.stream.Stream;
 
 import static com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.MetricType.COUNTER;
+import static com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.MetricType.CUMULATIVE_COUNTER;
 import static com.signalfx.metrics.protobuf.SignalFxProtocolBuffers.MetricType.GAUGE;
 import static java.util.stream.StreamSupport.stream;
 
@@ -59,6 +73,7 @@ public class SignalFxMeterRegistry extends StepMeterRegistry {
     private final HttpEventProtobufReceiverFactory eventReceiverFactory;
     private final Set<OnSendErrorHandler> onSendErrorHandlerCollection = Collections.singleton(
             metricError -> this.logger.warn("failed to send metrics: {}", metricError.getMessage()));
+    private final boolean publishCumulativeHistograms;
 
     public SignalFxMeterRegistry(SignalFxConfig config, Clock clock) {
         this(config, clock, DEFAULT_THREAD_FACTORY);
@@ -71,16 +86,17 @@ public SignalFxMeterRegistry(SignalFxConfig config, Clock clock, ThreadFactory t
         URI apiUri = URI.create(config.uri());
         int port = apiUri.getPort();
         if (port == -1) {
-            if ("http" .equals(apiUri.getScheme())) {
+            if ("http".equals(apiUri.getScheme())) {
                 port = 80;
-            } else if ("https" .equals(apiUri.getScheme())) {
+            } else if ("https".equals(apiUri.getScheme())) {
                 port = 443;
             }
         }
 
         SignalFxReceiverEndpoint signalFxEndpoint = new SignalFxEndpoint(apiUri.getScheme(), apiUri.getHost(), port);
         this.dataPointReceiverFactory = new HttpDataPointProtobufReceiverFactory(signalFxEndpoint);
         this.eventReceiverFactory = new HttpEventProtobufReceiverFactory(signalFxEndpoint);
+        this.publishCumulativeHistograms = config.publishCumulativeHistogram();
 
         config().namingConvention(new SignalFxNamingConvention());
 
@@ -118,7 +134,27 @@ protected void publish() {
         }
     }
 
-    private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addMeter(Meter meter) {
+    @Override
+    protected Timer newTimer(Meter.Id id, DistributionStatisticConfig distributionStatisticConfig, PauseDetector pauseDetector) {
+        if (!publishCumulativeHistograms) {
+            return super.newTimer(id, distributionStatisticConfig, pauseDetector);
+        }
+        Timer timer = new SignalfxTimer(id, clock, distributionStatisticConfig, pauseDetector, getBaseTimeUnit(), config.step().toMillis());
+        HistogramGauges.registerWithCommonFormat(timer, this);
+        return timer;
+    }
+
+    @Override
+    protected DistributionSummary newDistributionSummary(Meter.Id id, DistributionStatisticConfig distributionStatisticConfig, double scale) {
+        if (!publishCumulativeHistograms) {
+            return super.newDistributionSummary(id, distributionStatisticConfig, scale);
+        }
+        DistributionSummary summary = new SignalfxDistributionSummary(id, clock, distributionStatisticConfig, scale, config.step().toMillis());
+        HistogramGauges.registerWithCommonFormat(summary, this);
+        return summary;
+    }
+
+    Stream<SignalFxProtocolBuffers.DataPoint.Builder> addMeter(Meter meter) {
         return stream(meter.measure().spliterator(), false).flatMap(measurement -> {
             String statSuffix = NamingConvention.camelCase.tagKey(measurement.getStatistic().toString());
             switch (measurement.getStatistic()) {
@@ -170,23 +206,28 @@ Stream<SignalFxProtocolBuffers.DataPoint.Builder> addLongTaskTimer(LongTaskTimer
         );
     }
 
-    private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addTimeGauge(TimeGauge timeGauge) {
+    Stream<SignalFxProtocolBuffers.DataPoint.Builder> addTimeGauge(TimeGauge timeGauge) {
         return Stream.of(addDatapoint(timeGauge, GAUGE, null, timeGauge.value(getBaseTimeUnit())));
     }
 
-    private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addGauge(Gauge gauge) {
+    Stream<SignalFxProtocolBuffers.DataPoint.Builder> addGauge(Gauge gauge) {
+        if (publishCumulativeHistograms
+                && gauge.getId().syntheticAssociation() != null
+                && gauge.getId().getName().endsWith(".histogram")) {
+            return Stream.of(addDatapoint(gauge, CUMULATIVE_COUNTER, null, gauge.value()));
+        }
         return Stream.of(addDatapoint(gauge, GAUGE, null, gauge.value()));
     }
 
-    private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addCounter(Counter counter) {
+    Stream<SignalFxProtocolBuffers.DataPoint.Builder> addCounter(Counter counter) {
         return Stream.of(addDatapoint(counter, COUNTER, null, counter.count()));
     }
 
-    private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addFunctionCounter(FunctionCounter counter) {
+    Stream<SignalFxProtocolBuffers.DataPoint.Builder> addFunctionCounter(FunctionCounter counter) {
         return Stream.of(addDatapoint(counter, COUNTER, null, counter.count()));
     }
 
-    private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addTimer(Timer timer) {
+    Stream<SignalFxProtocolBuffers.DataPoint.Builder> addTimer(Timer timer) {
         return Stream.of(
                 addDatapoint(timer, COUNTER, "count", timer.count()),
                 addDatapoint(timer, COUNTER, "totalTime", timer.totalTime(getBaseTimeUnit())),
@@ -195,15 +236,15 @@ private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addTimer(Timer timer)
         );
     }
 
-    private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addFunctionTimer(FunctionTimer timer) {
+    Stream<SignalFxProtocolBuffers.DataPoint.Builder> addFunctionTimer(FunctionTimer timer) {
         return Stream.of(
                 addDatapoint(timer, COUNTER, "count", timer.count()),
                 addDatapoint(timer, COUNTER, "totalTime", timer.totalTime(getBaseTimeUnit())),
                 addDatapoint(timer, GAUGE, "avg", timer.mean(getBaseTimeUnit()))
         );
     }
 
-    private Stream<SignalFxProtocolBuffers.DataPoint.Builder> addDistributionSummary(DistributionSummary summary) {
+    Stream<SignalFxProtocolBuffers.DataPoint.Builder> addDistributionSummary(DistributionSummary summary) {
         return Stream.of(
                 addDatapoint(summary, COUNTER, "count", summary.count()),
                 addDatapoint(summary, COUNTER, "totalTime", summary.totalAmount()),

implementations/micrometer-registry-signalfx/src/main/java/io/micrometer/signalfx/SignalfxDistributionSummary.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2022 VMware, Inc.
+ *
+ * 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
+ *
+ * https://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 io.micrometer.signalfx;
+
+import io.micrometer.core.instrument.AbstractDistributionSummary;
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
+import io.micrometer.core.instrument.distribution.TimeWindowMax;
+import io.micrometer.core.instrument.step.StepTuple2;
+
+import java.util.concurrent.atomic.DoubleAdder;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * This class is mostly the same as {@link io.micrometer.core.instrument.step.StepDistributionSummary}, with one notable
+ * difference: the {@link DistributionStatisticConfig} is modified before being passed to the super class constructor -
+ * that forces the histogram generated by this meter to be cumulative.
+ *
+ * @author Bogdan Drutu
+ * @author Mateusz Rzeszutek
+ */
+final class SignalfxDistributionSummary extends AbstractDistributionSummary {
+
+    private final LongAdder count = new LongAdder();
+    private final DoubleAdder total = new DoubleAdder();
+    private final StepTuple2<Long, Double> countTotal;
+    private final TimeWindowMax max;
+
+    SignalfxDistributionSummary(Id id, Clock clock, DistributionStatisticConfig distributionStatisticConfig, double scale, long stepMillis) {
+        super(id, clock, CumulativeHistogramConfigUtil.updateConfig(distributionStatisticConfig), scale, false);
+        this.countTotal = new StepTuple2<>(clock, stepMillis, 0L, 0.0, count::sumThenReset, total::sumThenReset);
+        max = new TimeWindowMax(clock, distributionStatisticConfig);
+    }
+
+    @Override
+    protected void recordNonNegative(double amount) {
+        count.increment();
+        total.add(amount);
+        max.record(amount);
+    }
+
+    @Override
+    public long count() {
+        return countTotal.poll1();
+    }
+
+    @Override
+    public double totalAmount() {
+        return countTotal.poll2();
+    }
+
+    @Override
+    public double max() {
+        return max.poll();
+    }
+}

implementations/micrometer-registry-signalfx/src/main/java/io/micrometer/signalfx/SignalfxTimer.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2022 VMware, Inc.
+ *
+ * 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
+ *
+ * https://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 io.micrometer.signalfx;
+
+import io.micrometer.core.instrument.AbstractTimer;
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
+import io.micrometer.core.instrument.distribution.TimeWindowMax;
+import io.micrometer.core.instrument.distribution.pause.PauseDetector;
+import io.micrometer.core.instrument.step.StepTuple2;
+import io.micrometer.core.instrument.util.TimeUtils;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+
+/**
+ * This class is mostly the same as {@link io.micrometer.core.instrument.step.StepTimer}, with one notable difference:
+ * the {@link DistributionStatisticConfig} is modified before being passed to the super class constructor -
+ * that forces the histogram generated by this meter to be cumulative.
+ *
+ * @author Bogdan Drutu
+ * @author Mateusz Rzeszutek
+ */
+final class SignalfxTimer extends AbstractTimer {
+
+    private final LongAdder count = new LongAdder();
+    private final LongAdder total = new LongAdder();
+    private final StepTuple2<Long, Long> countTotal;
+    private final TimeWindowMax max;
+
+    SignalfxTimer(Id id, Clock clock, DistributionStatisticConfig distributionStatisticConfig, PauseDetector pauseDetector, TimeUnit baseTimeUnit, long stepMillis) {
+        super(id, clock, CumulativeHistogramConfigUtil.updateConfig(distributionStatisticConfig), pauseDetector, baseTimeUnit, false);
+        countTotal = new StepTuple2<>(clock, stepMillis, 0L, 0L, count::sumThenReset, total::sumThenReset);
+        max = new TimeWindowMax(clock, distributionStatisticConfig);
+    }
+
+    @Override
+    protected void recordNonNegative(long amount, TimeUnit unit) {
+        final long nanoAmount = (long) TimeUtils.convert(amount, unit, TimeUnit.NANOSECONDS);
+        count.increment();
+        total.add(nanoAmount);
+        max.record(amount, unit);
+    }
+
+    @Override
+    public long count() {
+        return countTotal.poll1();
+    }
+
+    @Override
+    public double totalTime(TimeUnit unit) {
+        return TimeUtils.nanosToUnit(countTotal.poll2(), unit);
+    }
+
+    @Override
+    public double max(TimeUnit unit) {
+        return max.poll(unit);
+    }
+}