feat: add support for Otlp exponential histograms

kotlin/goodmetrics/src/main/kotlin/goodmetrics/MetricsSetups.kt

@@ -13,6 +13,7 @@ import goodmetrics.pipeline.AggregatedBatch
 import goodmetrics.pipeline.Aggregator
 import goodmetrics.pipeline.BatchSender.Companion.launchSender
 import goodmetrics.pipeline.Batcher
+import goodmetrics.pipeline.DistributionMode
 import goodmetrics.pipeline.MetricsSink
 import goodmetrics.pipeline.SynchronizingBuffer
 import io.grpc.Metadata
@@ -31,7 +32,7 @@ data class ConfiguredMetrics(
 
 class MetricsSetups private constructor() {
     companion object {
-        fun CoroutineScope.goodMetrics(goodmetricsHost: String = "localhost", port: Int = 9573, aggregationWidth: Duration = 10.seconds): ConfiguredMetrics {
+        fun CoroutineScope.goodMetrics(goodmetricsHost: String = "localhost", port: Int = 9573, aggregationWidth: Duration = 10.seconds, distributionMode: DistributionMode = DistributionMode.Histogram): ConfiguredMetrics {
             val unaryIncomingBuffer = SynchronizingBuffer()
             val unaryFactory = MetricsFactory(unaryIncomingBuffer, timeSource = NanoTimeSource.preciseNanoTime, totaltimeType = MetricsFactory.TotaltimeType.DistributionMicroseconds)
 
@@ -40,7 +41,7 @@ class MetricsSetups private constructor() {
                 sendMetricsBatch(batch)
             }
 
-            val preaggregatedIncomingBuffer = Aggregator(aggregationWidth)
+            val preaggregatedIncomingBuffer = Aggregator(aggregationWidth, distributionMode = distributionMode)
             val preaggregatedFactory = MetricsFactory(preaggregatedIncomingBuffer, timeSource = NanoTimeSource.fastNanoTime, totaltimeType = MetricsFactory.TotaltimeType.DistributionMicroseconds)
 
             val preaggregatedBatcher = Batcher(preaggregatedIncomingBuffer)
@@ -362,8 +363,10 @@ class MetricsSetups private constructor() {
             client: OpentelemetryClient,
             logError: (message: String, exception: Exception) -> Unit,
             onSendPreaggregated: (List<AggregatedBatch>) -> Unit,
+            // By default, use Histogram distribution mode since ExponentialHistograms are not frequently used
+            distributionMode: DistributionMode = DistributionMode.Histogram,
         ): Aggregator {
-            val sink = Aggregator(aggregationWidth = aggregationWidth)
+            val sink = Aggregator(aggregationWidth = aggregationWidth, distributionMode = distributionMode)
             val batcher = Batcher(
                 sink,
                 batchSize = batchSize,

kotlin/goodmetrics/src/main/kotlin/goodmetrics/downstream/GoodmetricsClient.kt

@@ -17,6 +17,7 @@ import io.goodmetrics.statisticSet
 import io.grpc.netty.GrpcSslContexts
 import io.grpc.netty.NettyChannelBuilder
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory
+import kotlin.math.roundToLong
 
 class GoodmetricsClient private constructor(
     private val stub: MetricsGrpcKt.MetricsCoroutineStub,
@@ -156,6 +157,14 @@ fun Aggregation.toProto(): Measurement = measurement {
                 }
             }
         }
+        is Aggregation.ExponentialHistogram -> {
+            val bucketCounts = this@toProto.valueCounts().map { (value, count) -> Pair(value.roundToLong(), count)}.toMap()
+            histogram = histogram {
+                for ((bucket, count) in bucketCounts) {
+                    buckets[bucket] = count
+                }
+            }
+        }
         is Aggregation.StatisticSet -> {
             statisticSet = statisticSet {
                 minimum = min.get()

kotlin/goodmetrics/src/main/kotlin/goodmetrics/downstream/OpentelemetryClient.kt

@@ -8,9 +8,12 @@ import goodmetrics.io.opentelemetry.proto.common.v1.anyValue
 import goodmetrics.io.opentelemetry.proto.common.v1.instrumentationScope
 import goodmetrics.io.opentelemetry.proto.common.v1.keyValue
 import goodmetrics.io.opentelemetry.proto.metrics.v1.AggregationTemporality
+import goodmetrics.io.opentelemetry.proto.metrics.v1.ExponentialHistogramDataPointKt.buckets
 import goodmetrics.io.opentelemetry.proto.metrics.v1.Metric
 import goodmetrics.io.opentelemetry.proto.metrics.v1.ResourceMetrics
 import goodmetrics.io.opentelemetry.proto.metrics.v1.ScopeMetrics
+import goodmetrics.io.opentelemetry.proto.metrics.v1.exponentialHistogram
+import goodmetrics.io.opentelemetry.proto.metrics.v1.exponentialHistogramDataPoint
 import goodmetrics.io.opentelemetry.proto.metrics.v1.gauge
 import goodmetrics.io.opentelemetry.proto.metrics.v1.histogram
 import goodmetrics.io.opentelemetry.proto.metrics.v1.histogramDataPoint
@@ -184,6 +187,15 @@ class OpentelemetryClient(
                             }
                         )
                     }
+                    is Aggregation.ExponentialHistogram -> {
+                        yield(
+                            metric {
+                                name = "${this@asGoofyOtlpMetricSequence.metric}_$measurementName"
+                                unit = "1"
+                                exponentialHistogram = aggregation.asOtlpExponentialHistogram(otlpDimensions, this@asGoofyOtlpMetricSequence.timestampNanos, aggregationWidth)
+                            }
+                        )
+                    }
                     is Aggregation.StatisticSet -> {
                         yieldAll(aggregation.statisticSetToOtlp(this@asGoofyOtlpMetricSequence.metric, measurementName, timestampNanos, aggregationWidth, otlpDimensions))
                     }
@@ -355,6 +367,39 @@ class OpentelemetryClient(
         )
     }
 
+    private fun Aggregation.ExponentialHistogram.asOtlpExponentialHistogram(
+        otlpDimensions: Iterable<KeyValue>,
+        timestampNanos: Long,
+        aggregationWidth: Duration,
+    ) = exponentialHistogram {
+        aggregationTemporality = AggregationTemporality.AGGREGATION_TEMPORALITY_DELTA
+        dataPoints.add(
+            exponentialHistogramDataPoint {
+                attributes.addAll(otlpDimensions)
+                startTimeUnixNano = timestampNanos - aggregationWidth.inWholeNanoseconds
+                timeUnixNano = timestampNanos
+                count = this@asOtlpExponentialHistogram.count()
+                sum = if(this@asOtlpExponentialHistogram.hasNegatives()) 0.0 else (this@asOtlpExponentialHistogram.sum())
+                min = this@asOtlpExponentialHistogram.min()
+                max = this@asOtlpExponentialHistogram.max()
+                scale = this@asOtlpExponentialHistogram.scale()
+                zeroCount = 0
+                positive = buckets {
+                   offset = this@asOtlpExponentialHistogram.bucketStartOffset()
+                   bucketCounts.addAll(
+                       this@asOtlpExponentialHistogram.takePositives()
+                   )
+                }
+                negative = buckets {
+                    offset = this@asOtlpExponentialHistogram.bucketStartOffset()
+                    bucketCounts.addAll(
+                        this@asOtlpExponentialHistogram.takeNegatives()
+                    )
+                }
+            }
+        )
+    }
+
     private val library = instrumentationScope {
         name = "goodmetrics_kotlin"
         version = OpentelemetryClient::class.java.`package`.implementationVersion ?: "development"