Merge pull request #95 from cisco-open/feature/master/histogram-metrics-types-support

preet-dev · web-flow · commit 408314bc093a · 2023-06-26T00:03:57.000+05:30
Add support for histogram metrics
diff --git a/example-definitions/json/metric-definition.json b/example-definitions/json/metric-definition.json
@@ -91,14 +91,16 @@
     {
       "name": "container.cpu.used.utilization",
       "unit": "%",
-      "otelType": "summary",
-      "valueFunction": "arithmeticSequenceSummary(1, 2, \"\", 7)",
+      "otelType": "histogram",
+      "aggregationTemporality": "delta",
+      "valueFunction": "arithmeticSequenceSummary(5, 2, \"\", 7)",
       "isDouble": true,
-      "quantiles": [
-        0,
-        0.5,
-        0.75,
-        1
+      "bounds": [
+        10,
+        30,
+        50,
+        70,
+        90
       ],
       "reportingResources": [
         "container"
diff --git a/example-definitions/qa/metric-definition.yaml b/example-definitions/qa/metric-definition.yaml
@@ -63,10 +63,11 @@ metrics:
     reportingResources: [resource_quota]
   - name: container.cpu.used.utilization
     unit: "%"
-    otelType: summary
-    valueFunction: 'arithmeticSequenceSummary(1, 2, "", 7)'
+    otelType: histogram
+    aggregationTemporality: delta
+    valueFunction: 'arithmeticSequenceSummary(5, 2, "", 7)'
     isDouble: true
-    quantiles: [ 0, 0.5, 0.75, 1 ]
+    bounds: [10, 30, 50, 70, 90]
     reportingResources: [ container ]
   - name: cpu.usage
     unit: "{cores}"
diff --git a/src/main/java/io/opentelemetry/contrib/generator/telemetry/metrics/HistogramGenerator.java b/src/main/java/io/opentelemetry/contrib/generator/telemetry/metrics/HistogramGenerator.java
@@ -0,0 +1,98 @@
+package io.opentelemetry.contrib.generator.telemetry.metrics;
+
+import io.opentelemetry.contrib.generator.core.jel.ExpressionProcessor;
+import io.opentelemetry.contrib.generator.telemetry.GeneratorsStateProvider;
+import io.opentelemetry.contrib.generator.telemetry.metrics.dto.MetricDefinition;
+import io.opentelemetry.contrib.generator.telemetry.misc.GeneratorUtils;
+import io.opentelemetry.proto.metrics.v1.*;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Class to generate metric values of Histogram type.
+ */
+@Slf4j
+public class HistogramGenerator {
+
+    private final String requestID;
+    private final ExpressionProcessor jelProcessor;
+
+    public HistogramGenerator(String requestID, ExpressionProcessor jelProcessor) {
+        this.requestID = requestID;
+        this.jelProcessor = jelProcessor;
+    }
+
+    public Metric.Builder getOTelMetric(MetricDefinition metricDefinition) {
+        return Metric.newBuilder()
+                .setName(metricDefinition.getName())
+                .setUnit(metricDefinition.getUnit())
+                .setHistogram(getHistogramDataPoint(metricDefinition));
+
+    }
+
+    private Histogram getHistogramDataPoint(MetricDefinition metricDefinition) {
+        long[] times = getTimes(metricDefinition);
+        List<Double> values = getCountValues(metricDefinition);
+        double sum = values.stream().mapToDouble(Double::doubleValue).sum();
+        int count = (values.size());
+        return Histogram.newBuilder()
+                .setAggregationTemporality(metricDefinition.getAggregationTemporality())
+                .addDataPoints(HistogramDataPoint.newBuilder()
+                        .setCount(count)
+                        .setSum(sum)
+                        .setStartTimeUnixNano(times[0])
+                        .setTimeUnixNano(times[1])
+                        .addAllExplicitBounds(metricDefinition.getBounds())
+                        .addAllBucketCounts(getBucketCounts(metricDefinition, values))
+                        .addAllAttributes(GeneratorUtils.getEvaluatedAttributes(jelProcessor, metricDefinition.getAttributes()))
+                        .build())
+                .build();
+    }
+
+    private long[] getTimes(MetricDefinition metricDefinition) {
+        long[] times = GeneratorUtils.normalizeTimestamp(TimeUnit.MILLISECONDS.toNanos(System.currentTimeMillis()),
+                metricDefinition.getPayloadFrequencySeconds());
+        if (metricDefinition.getAggregationTemporality() == AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE) {
+            if (GeneratorsStateProvider.getMetricGeneratorState(requestID).getFixedStartTime() == 0) {
+                GeneratorsStateProvider.getMetricGeneratorState(requestID).setFixedStartTime(times[0]);
+            }
+            times[0] = GeneratorsStateProvider.getMetricGeneratorState(requestID).getFixedStartTime();
+        }
+        return times;
+    }
+
+    private List<Double> getCountValues(MetricDefinition metricDefinition) {
+        Object value = jelProcessor.eval(metricDefinition.getValueFunction());
+        List<Object> rawValues = value instanceof List<?> ? (List<Object>) value : Collections.nCopies(5, value);
+        return rawValues.stream().map(val -> Double.parseDouble(val.toString())).toList();
+    }
+
+    private List<Long> getBucketCounts(MetricDefinition metricDefinition, List<Double> values) {
+        List<Long> bucketCounts = new ArrayList<>();
+        for (int i = 0; i < metricDefinition.getBounds().size(); i++) {
+            double lowerBound = i == 0 ? Double.MIN_VALUE : metricDefinition.getBounds().get(i - 1) + 1;
+            double upperBound = i == metricDefinition.getBounds().size() - 1 ? Double.MAX_VALUE :
+                    metricDefinition.getBounds().get(i);
+            long count = countValuesInRange(values, lowerBound, upperBound);
+            bucketCounts.add(count);
+        }
+
+        return bucketCounts;
+    }
+
+    private long countValuesInRange(List<Double> values, double lowerBound, double upperBound) {
+        long count = 0;
+        for (Double eachValue : values) {
+            if (eachValue >= lowerBound && eachValue <= upperBound) {
+                count++;
+            }
+        }
+        return count;
+    }
+
+
+}
diff --git a/src/main/java/io/opentelemetry/contrib/generator/telemetry/metrics/MetricGeneratorThread.java b/src/main/java/io/opentelemetry/contrib/generator/telemetry/metrics/MetricGeneratorThread.java
@@ -52,6 +52,7 @@ public class MetricGeneratorThread implements Runnable {
     private final GeneratorState<MetricGeneratorThread> metricGeneratorState;
     private final GaugeGenerator gaugeGenerator;
     private final SumGenerator sumGenerator;
+    private final HistogramGenerator histogramGenerator;
     private final SummaryGenerator summaryGenerator;
     private int currentCount;
 
@@ -65,6 +66,7 @@ public MetricGeneratorThread(String groupKey, List<MetricDefinition> metrics, Pa
         ExpressionProcessor jelProcessor = JELProvider.getJelProcessor();
         gaugeGenerator = new GaugeGenerator(jelProcessor);
         sumGenerator = new SumGenerator(requestID, jelProcessor);
+        histogramGenerator = new HistogramGenerator(requestID, jelProcessor);
         summaryGenerator = new SummaryGenerator(jelProcessor);
         currentCount = 0;
     }
@@ -116,7 +118,7 @@ public void run() {
                         .addAllResourceMetrics(resourceMetricsList)
                         .build();
                 log.info(requestID + ": Sending payload for: " + groupKey);
-                //log.debug(requestID + ": Complete payload for " + groupKey + ": " + resourceMetrics);
+                log.debug(requestID + ": Complete payload for " + groupKey + ": " + resourceMetrics);
                 boolean responseStatus = payloadHandler.postPayload(resourceMetrics);
                 if (metricGeneratorState.getTransportStorage() != null) {
                     metricGeneratorState.getTransportStorage().store(groupKey, resourceMetrics, responseStatus);
@@ -131,6 +133,7 @@ private Metric.Builder getMetric(MetricDefinition metricDefinition) {
         return switch (metricDefinition.getOtelType()) {
             case Constants.GAUGE -> gaugeGenerator.getOTelMetric(metricDefinition);
             case Constants.SUM -> sumGenerator.getOTelMetric(metricDefinition);
+            case Constants.HISTOGRAM -> histogramGenerator.getOTelMetric(metricDefinition);
             default -> summaryGenerator.getOTelMetric(metricDefinition);
         };
     }
@@ -195,18 +198,29 @@ private Metric getMetricWithResourceAttributes(Metric.Builder partialMetric, Lis
                     .map(NumberDataPoint::toBuilder)
                     .map(bdp -> bdp.addAllAttributes(resourceAttributes).build())
                     .toList();
-            partialMetric.getSum().toBuilder()
+            Sum newSum = partialMetric.getSum().toBuilder()
                     .clearDataPoints()
                     .addAllDataPoints(dataPointsWAttrs)
                     .build();
-            Sum newSum = partialMetric.getSum().toBuilder()
+            return Metric.newBuilder()
+                    .setName(partialMetric.getName())
+                    .setUnit(partialMetric.getUnit())
+                    .setSum(newSum)
+                    .build();
+        } else if (metricType.equals(Metric.DataCase.HISTOGRAM)) {
+            List<HistogramDataPoint> dataPoints = partialMetric.getHistogram().getDataPointsList();
+            List<HistogramDataPoint> dataPointsWAttrs = dataPoints.stream()
+                    .map(HistogramDataPoint::toBuilder)
+                    .map(bdp -> bdp.addAllAttributes(resourceAttributes).build())
+                    .toList();
+            Histogram newHistogram = partialMetric.getHistogram().toBuilder()
                     .clearDataPoints()
                     .addAllDataPoints(dataPointsWAttrs)
                     .build();
             return Metric.newBuilder()
                     .setName(partialMetric.getName())
                     .setUnit(partialMetric.getUnit())
-                    .setSum(newSum)
+                    .setHistogram(newHistogram)
                     .build();
         } else {
             List<SummaryDataPoint> dataPoints = partialMetric.getSummary().getDataPointsList();
diff --git a/src/main/java/io/opentelemetry/contrib/generator/telemetry/metrics/dto/MetricDefinition.java b/src/main/java/io/opentelemetry/contrib/generator/telemetry/metrics/dto/MetricDefinition.java
@@ -28,6 +28,7 @@
 import org.apache.commons.lang3.StringUtils;
 
 import java.util.*;
+import java.util.stream.IntStream;
 
 @Data
 @Slf4j
@@ -40,6 +41,7 @@ public class MetricDefinition implements Cloneable {
     private Boolean isMonotonic;
     private Boolean isDouble;
     private List<Double> quantiles;
+    private List<Double> bounds;
     private String valueFunction;
     private Integer payloadFrequencySeconds;
     private Integer payloadCount;
@@ -93,14 +95,15 @@ private void validateOTelType() {
             throw new GeneratorException("Invalid OTeltype '" + otelType + "' found for metric " + name  + " ." +
                     "Valid types are " + StringUtils.join(Constants.validMetricTypes));
         }
-        validateAggregationTemporalityForSum();
+        validateAggregationTemporality();
     }
 
-    private void validateAggregationTemporalityForSum() {
-        if (otelType.equalsIgnoreCase(Constants.SUM)) {
+    private void validateAggregationTemporality() {
+        if (otelType.equals(Constants.SUM) || otelType.equals(Constants.HISTOGRAM) ||
+                otelType.equals(Constants.EXP_HISTOGRAM)) {
             if (aggregationTemporality==null || aggregationTemporality.isBlank()) {
-                throw new GeneratorException("OTel type for metric " + name + " is of 'sum' type but Aggregation " +
-                        "temporality not provided");
+                throw new GeneratorException("OTel type for metric " + name + " is of '" + otelType + "' type but " +
+                        "Aggregation temporality not provided");
             }
             //Check aggregation temporality is valid
             if (!(aggregationTemporality.equalsIgnoreCase(Constants.CUMULATIVE) ||
@@ -109,11 +112,48 @@ private void validateAggregationTemporalityForSum() {
                         " specified for metric " + name + ". Valid types are (Cumulative, Delta)");
             }
         }
+        validateQuantiles();
+    }
+
+    private void validateQuantiles() {
+        if (otelType.equals(Constants.SUMMARY)) {
+            if (quantiles == null || quantiles.isEmpty()) {
+                throw new GeneratorException("OTel metric " + name + " of type summary does not have any quantiles " +
+                        "specified.");
+            }
+        }
+        validateBounds();
+    }
+
+    private void validateBounds() {
+        if (otelType.equals(Constants.HISTOGRAM)) {
+            if (bounds == null || bounds.isEmpty()) {
+                throw new GeneratorException("OTel metric " + name + " of type histogram does not have any bounds " +
+                        "specified.");
+            }
+            if (bounds.size() > 1 && IntStream.range(0, bounds.size()-1)
+                    .anyMatch(idx -> bounds.get(idx) >= bounds.get(idx + 1))) {
+                throw new GeneratorException("Invalid bounds provided for metric " + name + ". Bound values must be " +
+                        "in strictly ascending order");
+            }
+        }
         validateAttributes();
     }
 
     private void validateAttributes() {
         attributes = GeneratorUtils.validateAttributes(attributes);
+        checkValueFunction();
+    }
+
+    private void checkValueFunction() {
+        if ((otelType.equals(Constants.SUMMARY) || otelType.equals(Constants.HISTOGRAM) ||
+                otelType.equals(Constants.EXP_HISTOGRAM)) && (!valueFunction.contains("Summary"))) {
+            log.warn("Metric " + name + " should be using a summary variant of the value expression");
+        }
+        if ((otelType.equals(Constants.SUM) || otelType.equals(Constants.GAUGE)) &&
+                (valueFunction.contains("Summary"))) {
+            log.warn("Metric " + name + " should not be using the summary variant of the value expression");
+        }
     }
 
     private void validateResourceTypes(Set<String> allResourceTypes) {
diff --git a/src/main/java/io/opentelemetry/contrib/generator/telemetry/misc/Constants.java b/src/main/java/io/opentelemetry/contrib/generator/telemetry/misc/Constants.java
@@ -26,7 +26,9 @@ private Constants() {}
     public static final String SUM = "sum";
     public static final String GAUGE = "gauge";
     public static final String SUMMARY = "summary";
-    public static final List<String> validMetricTypes = Arrays.asList(SUM, GAUGE, SUMMARY);
+    public static final String HISTOGRAM = "histogram";
+    public static final String EXP_HISTOGRAM = "exponential-histogram";
+    public static final List<String> validMetricTypes = Arrays.asList(SUM, GAUGE, SUMMARY, HISTOGRAM, EXP_HISTOGRAM);
     public static final String CUMULATIVE = "cumulative";
     public static final String DELTA = "delta";
 
diff --git a/src/test/resources/test-definitions/metrics-test.yaml b/src/test/resources/test-definitions/metrics-test.yaml
@@ -19,10 +19,11 @@ payloadCount: 10
 metrics:
   - name: system.network.in.kb.sec
     unit: kBy/s
-    otelType: summary
+    otelType: histogram
     valueFunction: 'absoluteCosineSequenceSummary("*7000", 5)'
     isDouble: true
-    quantiles: [0, 50, 100]
+    aggregationTemporality: cumulative
+    bounds: [10, 100, 1000, 10000, 100000]
     reportingResources: [network_interface, container, machine]
     attributes:
       system.internal.ip: 'IPv4Sequence("10.134.1.34")'
