Add capability to have configurable aggregation temporality for OTLP Registry

implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/OtlpConfig.java

@@ -17,15 +17,15 @@
 
 import io.micrometer.core.instrument.config.validate.Validated;
 import io.micrometer.core.instrument.push.PushRegistryConfig;
+import io.opentelemetry.proto.metrics.v1.AggregationTemporality;
 
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
 import static io.micrometer.core.instrument.config.MeterRegistryConfigValidator.*;
-import static io.micrometer.core.instrument.config.validate.PropertyValidator.getString;
-import static io.micrometer.core.instrument.config.validate.PropertyValidator.getUrlString;
+import static io.micrometer.core.instrument.config.validate.PropertyValidator.*;
 
 /**
  * Config for {@link OtlpMeterRegistry}.
@@ -87,6 +87,19 @@ default Map<String, String> resourceAttributes() {
         return resourceAttributes;
     }
 
+    /**
+     * {@link AggregationTemporality} of the OtlpMeterRegistry. This determines whether
+     * the meters should be cumulative(AGGREGATION_TEMPORALITY_CUMULATIVE) or
+     * step(AGGREGATION_TEMPORALITY_DELTA).
+     * @return the aggregationTemporality for OtlpRegistry
+     * @see <a href=
+     * "https://opentelemetry.io/docs/reference/specification/metrics/data-model/#temporality">Temporality</a>
+     */
+    default int getAggregationTemporality() {
+        return getInteger(this, "aggregationTemporality")
+                .orElse(AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE_VALUE);
+    }
+
     /**
      * Additional headers to send with exported metrics. This may be needed for
      * authorization headers, for example.

implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/OtlpDistributionSummary.java → implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/OtlpCumulativeDistributionSummary.java

@@ -23,7 +23,7 @@
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
 
-class OtlpDistributionSummary extends CumulativeDistributionSummary implements StartTimeAwareMeter {
+class OtlpCumulativeDistributionSummary extends CumulativeDistributionSummary implements StartTimeAwareMeter {
 
     private static final CountAtBucket[] EMPTY_HISTOGRAM = new CountAtBucket[0];
 
@@ -32,8 +32,8 @@ class OtlpDistributionSummary extends CumulativeDistributionSummary implements S
     @Nullable
     private final Histogram monotonicBucketCountHistogram;
 
-    OtlpDistributionSummary(Id id, Clock clock, DistributionStatisticConfig distributionStatisticConfig, double scale,
-            boolean supportsAggregablePercentiles) {
+    OtlpCumulativeDistributionSummary(Id id, Clock clock, DistributionStatisticConfig distributionStatisticConfig,
+            double scale, boolean supportsAggregablePercentiles) {
         super(id, clock, DistributionStatisticConfig.builder().percentilesHistogram(false) // avoid
                                                                                            // a
                                                                                            // histogram
@@ -79,13 +79,9 @@ public HistogramSnapshot takeSnapshot() {
             return snapshot;
         }
 
+        CountAtBucket[] histogramCounts = this.monotonicBucketCountHistogram.takeSnapshot(0, 0, 0).histogramCounts();
         return new HistogramSnapshot(snapshot.count(), snapshot.total(), snapshot.max(), snapshot.percentileValues(),
-                histogramCounts(), snapshot::outputSummary);
-    }
-
-    private CountAtBucket[] histogramCounts() {
-        return this.monotonicBucketCountHistogram == null ? EMPTY_HISTOGRAM
-                : this.monotonicBucketCountHistogram.takeSnapshot(0, 0, 0).histogramCounts();
+                histogramCounts, snapshot::outputSummary);
     }
 
     @Override

implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/OtlpTimer.java → implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/OtlpCumulativeTimer.java

@@ -25,15 +25,15 @@
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
 
-class OtlpTimer extends CumulativeTimer implements StartTimeAwareMeter {
+class OtlpCumulativeTimer extends CumulativeTimer implements StartTimeAwareMeter {
 
     private final long startTimeNanos;
 
     @Nullable
     private final Histogram monotonicCountBucketHistogram;
 
-    OtlpTimer(Id id, Clock clock, DistributionStatisticConfig distributionStatisticConfig, PauseDetector pauseDetector,
-            TimeUnit baseTimeUnit) {
+    OtlpCumulativeTimer(Id id, Clock clock, DistributionStatisticConfig distributionStatisticConfig,
+            PauseDetector pauseDetector, TimeUnit baseTimeUnit) {
         super(id, clock, DistributionStatisticConfig.builder().percentilesHistogram(false) // avoid
                                                                                            // a
                                                                                            // histogram

implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/OtlpMeterRegistry.java

@@ -24,8 +24,10 @@
 import io.micrometer.core.instrument.distribution.*;
 import io.micrometer.core.instrument.distribution.pause.PauseDetector;
 import io.micrometer.core.instrument.internal.DefaultGauge;
+import io.micrometer.core.instrument.internal.DefaultLongTaskTimer;
 import io.micrometer.core.instrument.internal.DefaultMeter;
 import io.micrometer.core.instrument.push.PushMeterRegistry;
+import io.micrometer.core.instrument.step.*;
 import io.micrometer.core.instrument.util.MeterPartition;
 import io.micrometer.core.instrument.util.NamedThreadFactory;
 import io.micrometer.core.instrument.util.TimeUtils;
@@ -123,19 +125,27 @@ protected <T> Gauge newGauge(Meter.Id id, @Nullable T obj, ToDoubleFunction<T> v
 
     @Override
     protected Counter newCounter(Meter.Id id) {
-        return new OtlpCounter(id, this.clock);
+        return isDeltaAggregationTemporality() ? new StepCounter(id, this.clock, config.step().toMillis())
+                : new OtlpCounter(id, this.clock);
     }
 
     @Override
     protected Timer newTimer(Meter.Id id, DistributionStatisticConfig distributionStatisticConfig,
             PauseDetector pauseDetector) {
-        return new OtlpTimer(id, this.clock, distributionStatisticConfig, pauseDetector, getBaseTimeUnit());
+        return isDeltaAggregationTemporality()
+                ? new OtlpStepTimer(id, clock, distributionStatisticConfig, pauseDetector, getBaseTimeUnit(),
+                        config.step().toMillis())
+                : new OtlpCumulativeTimer(id, this.clock, distributionStatisticConfig, pauseDetector,
+                        getBaseTimeUnit());
     }
 
     @Override
     protected DistributionSummary newDistributionSummary(Meter.Id id,
             DistributionStatisticConfig distributionStatisticConfig, double scale) {
-        return new OtlpDistributionSummary(id, this.clock, distributionStatisticConfig, scale, true);
+        return isDeltaAggregationTemporality()
+                ? new OtlpStepDistributionSummary(id, clock, distributionStatisticConfig, scale,
+                        config.step().toMillis())
+                : new OtlpCumulativeDistributionSummary(id, this.clock, distributionStatisticConfig, scale, true);
     }
 
     @Override
@@ -146,18 +156,25 @@ protected Meter newMeter(Meter.Id id, Meter.Type type, Iterable<Measurement> mea
     @Override
     protected <T> FunctionTimer newFunctionTimer(Meter.Id id, T obj, ToLongFunction<T> countFunction,
             ToDoubleFunction<T> totalTimeFunction, TimeUnit totalTimeFunctionUnit) {
-        return new OtlpFunctionTimer<>(id, obj, countFunction, totalTimeFunction, totalTimeFunctionUnit,
-                getBaseTimeUnit(), this.clock);
+        return isDeltaAggregationTemporality()
+                ? new StepFunctionTimer<>(id, clock, config.step().toMillis(), obj, countFunction, totalTimeFunction,
+                        totalTimeFunctionUnit, getBaseTimeUnit())
+                : new OtlpFunctionTimer<>(id, obj, countFunction, totalTimeFunction, totalTimeFunctionUnit,
+                        getBaseTimeUnit(), this.clock);
     }
 
     @Override
     protected <T> FunctionCounter newFunctionCounter(Meter.Id id, T obj, ToDoubleFunction<T> countFunction) {
-        return new OtlpFunctionCounter<>(id, obj, countFunction, this.clock);
+        return isDeltaAggregationTemporality()
+                ? new StepFunctionCounter<>(id, clock, config.step().toMillis(), obj, countFunction)
+                : new OtlpFunctionCounter<>(id, obj, countFunction, this.clock);
     }
 
     @Override
     protected LongTaskTimer newLongTaskTimer(Meter.Id id, DistributionStatisticConfig distributionStatisticConfig) {
-        return new OtlpLongTaskTimer(id, this.clock, getBaseTimeUnit(), distributionStatisticConfig);
+        return isDeltaAggregationTemporality()
+                ? new DefaultLongTaskTimer(id, clock, getBaseTimeUnit(), distributionStatisticConfig, false)
+                : new OtlpLongTaskTimer(id, this.clock, getBaseTimeUnit(), distributionStatisticConfig);
     }
 
     @Override
@@ -181,31 +198,29 @@ private Metric writeMeter(Meter meter) {
     Metric writeGauge(Gauge gauge) {
         return getMetricBuilder(gauge.getId())
                 .setGauge(io.opentelemetry.proto.metrics.v1.Gauge.newBuilder()
-                        .addDataPoints(NumberDataPoint.newBuilder()
-                                .setTimeUnixNano(TimeUnit.MILLISECONDS.toNanos(this.clock.wallTime()))
+                        .addDataPoints(NumberDataPoint.newBuilder().setTimeUnixNano(getEndTimeNanos())
                                 .setAsDouble(gauge.value()).addAllAttributes(getTagsForId(gauge.getId())).build()))
                 .build();
     }
 
     // VisibleForTesting
     Metric writeCounter(Counter counter) {
-        return writeSum((StartTimeAwareMeter) counter, counter::count);
+        return writeSum(counter, counter::count);
     }
 
     // VisibleForTesting
     Metric writeFunctionCounter(FunctionCounter functionCounter) {
-        return writeSum((StartTimeAwareMeter) functionCounter, functionCounter::count);
+        return writeSum(functionCounter, functionCounter::count);
     }
 
-    private Metric writeSum(StartTimeAwareMeter meter, DoubleSupplier count) {
-        return getMetricBuilder(meter.getId())
-                .setSum(Sum.newBuilder()
-                        .addDataPoints(NumberDataPoint.newBuilder().setStartTimeUnixNano(meter.getStartTimeNanos())
-                                .setTimeUnixNano(TimeUnit.MILLISECONDS.toNanos(this.clock.wallTime()))
-                                .setAsDouble(count.getAsDouble()).addAllAttributes(getTagsForId(meter.getId())).build())
-                        .setIsMonotonic(true)
-                        .setAggregationTemporality(AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE).build())
-                .build();
+    private Metric writeSum(Meter meter, DoubleSupplier count) {
+        return getMetricBuilder(meter.getId()).setSum(Sum.newBuilder()
+                .addDataPoints(NumberDataPoint.newBuilder().setStartTimeUnixNano(getStartTimeNanos(meter))
+                        .setTimeUnixNano(getEndTimeNanos()).setAsDouble(count.getAsDouble())
+                        .addAllAttributes(getTagsForId(meter.getId())).build())
+                .setIsMonotonic(true)
+                .setAggregationTemporality(AggregationTemporality.forNumber(config.getAggregationTemporality()))
+                .build()).build();
     }
 
     // VisibleForTesting
@@ -215,8 +230,8 @@ Metric writeHistogramSupport(HistogramSupport histogramSupport) {
         HistogramSnapshot histogramSnapshot = histogramSupport.takeSnapshot();
 
         Iterable<? extends KeyValue> tags = getTagsForId(histogramSupport.getId());
-        long startTimeNanos = ((StartTimeAwareMeter) histogramSupport).getStartTimeNanos();
-        long wallTimeNanos = TimeUnit.MILLISECONDS.toNanos(this.clock.wallTime());
+        long startTimeNanos = getStartTimeNanos(histogramSupport);
+        long wallTimeNanos = getEndTimeNanos();
         double total = isTimeBased ? histogramSnapshot.total(getBaseTimeUnit()) : histogramSnapshot.total();
         long count = histogramSnapshot.count();
 
@@ -236,6 +251,9 @@ Metric writeHistogramSupport(HistogramSupport histogramSupport) {
         HistogramDataPoint.Builder histogramDataPoint = HistogramDataPoint.newBuilder().addAllAttributes(tags)
                 .setStartTimeUnixNano(startTimeNanos).setTimeUnixNano(wallTimeNanos).setSum(total).setCount(count);
 
+        if (isDeltaAggregationTemporality()) {
+            histogramDataPoint.setMax(histogramSnapshot.max(getBaseTimeUnit()));
+        }
         // if histogram enabled, add histogram buckets
         if (histogramSnapshot.histogramCounts().length != 0) {
             for (CountAtBucket countAtBucket : histogramSnapshot.histogramCounts()) {
@@ -244,26 +262,39 @@ Metric writeHistogramSupport(HistogramSupport histogramSupport) {
                 histogramDataPoint.addBucketCounts((long) countAtBucket.count());
             }
             metricBuilder.setHistogram(Histogram.newBuilder()
-                    .setAggregationTemporality(AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE)
+                    .setAggregationTemporality(AggregationTemporality.forNumber(config.getAggregationTemporality()))
                     .addDataPoints(histogramDataPoint));
             return metricBuilder.build();
         }
 
         return metricBuilder.setHistogram(Histogram.newBuilder()
-                .setAggregationTemporality(AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE)
+                .setAggregationTemporality(AggregationTemporality.forNumber(config.getAggregationTemporality()))
                 .addDataPoints(histogramDataPoint)).build();
     }
 
     // VisibleForTesting
     Metric writeFunctionTimer(FunctionTimer functionTimer) {
         return getMetricBuilder(functionTimer.getId()).setHistogram(Histogram.newBuilder()
                 .addDataPoints(HistogramDataPoint.newBuilder().addAllAttributes(getTagsForId(functionTimer.getId()))
-                        .setStartTimeUnixNano(((StartTimeAwareMeter) functionTimer).getStartTimeNanos())
-                        .setTimeUnixNano(TimeUnit.MILLISECONDS.toNanos(this.clock.wallTime()))
+                        .setStartTimeUnixNano(getStartTimeNanos((functionTimer))).setTimeUnixNano(getEndTimeNanos())
                         .setSum(functionTimer.totalTime(getBaseTimeUnit())).setCount((long) functionTimer.count())))
                 .build();
     }
 
+    private boolean isDeltaAggregationTemporality() {
+        return config.getAggregationTemporality() == AggregationTemporality.AGGREGATION_TEMPORALITY_DELTA_VALUE;
+    }
+
+    private long getStartTimeNanos(Meter meter) {
+        return isDeltaAggregationTemporality() ? getEndTimeNanos() - config.step().toNanos()
+                : ((StartTimeAwareMeter) meter).getStartTimeNanos();
+    }
+
+    private long getEndTimeNanos() {
+        return isDeltaAggregationTemporality() ? (clock.wallTime() / config.step().toMillis()) * config.step().toNanos()
+                : TimeUnit.MILLISECONDS.toNanos(clock.wallTime());
+    }
+
     private Metric.Builder getMetricBuilder(Meter.Id id) {
         Metric.Builder builder = Metric.newBuilder().setName(getConventionName(id));
         if (id.getBaseUnit() != null) {

implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/OtlpStepDistributionSummary.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.registry.otlp;
+
+import io.micrometer.common.lang.Nullable;
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.distribution.*;
+import io.micrometer.core.instrument.step.StepDistributionSummary;
+
+class OtlpStepDistributionSummary extends StepDistributionSummary {
+
+    @Nullable
+    private final Histogram countBucketHistogram;
+
+    /**
+     * Create a new {@code StepDistributionSummary}.
+     * @param id ID
+     * @param clock clock
+     * @param distributionStatisticConfig distribution static configuration
+     * @param scale scale
+     * @param stepMillis step in milliseconds
+     */
+    public OtlpStepDistributionSummary(Id id, Clock clock, DistributionStatisticConfig distributionStatisticConfig,
+            double scale, long stepMillis) {
+        super(id, clock, DistributionStatisticConfig.builder().percentilesHistogram(false).serviceLevelObjectives()
+                .build().merge(distributionStatisticConfig), scale, stepMillis, false);
+        if (distributionStatisticConfig.isPublishingHistogram()) {
+            this.countBucketHistogram = new TimeWindowFixedBoundaryHistogram(clock, distributionStatisticConfig, true,
+                    false);
+        }
+        else {
+            this.countBucketHistogram = null;
+        }
+    }
+
+    @Override
+    protected void recordNonNegative(double amount) {
+        super.recordNonNegative(amount);
+        if (this.countBucketHistogram != null) {
+            this.countBucketHistogram.recordDouble(amount);
+        }
+    }
+
+    @Override
+    public HistogramSnapshot takeSnapshot() {
+        HistogramSnapshot snapshot = super.takeSnapshot();
+        if (countBucketHistogram == null) {
+            return snapshot;
+        }
+
+        CountAtBucket[] histogramCounts = this.countBucketHistogram.takeSnapshot(0, 0, 0).histogramCounts();
+        return new HistogramSnapshot(snapshot.count(), snapshot.total(), snapshot.max(), snapshot.percentileValues(),
+                histogramCounts, snapshot::outputSummary);
+    }
+
+}