Define OTLP Trace Metrics Histogram and Writers

dd-trace-core/build.gradle

@@ -65,6 +65,7 @@ excludedClassesCoverage += [
   'datadog.trace.core.otlp.common.OtlpGrpcSender',
   // covered by OTLP system-tests
   'datadog.trace.core.otlp.logs.OtlpLogsService',
+  'datadog.trace.core.otlp.metrics.OtlpMetricsSenderFactory',
   'datadog.trace.core.otlp.metrics.OtlpMetricsService'
 ]
 

dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java

@@ -21,7 +21,7 @@
  * static); tests using {@link PeerTagSchema} must also call {@code PeerTagSchema#resetHandlers} on
  * the schema instance.
  */
-final class AggregateEntry extends Hashtable.Entry {
+public final class AggregateEntry extends Hashtable.Entry {
 
   static final long ERROR_TAG = 0x8000000000000000L;
   static final long TOP_LEVEL_TAG = 0x4000000000000000L;
@@ -165,31 +165,31 @@ AggregateEntry recordOneDuration(long tagAndDuration) {
     return this;
   }
 
-  int getErrorCount() {
+  public int getErrorCount() {
     return errorCount;
   }
 
-  int getHitCount() {
+  public int getHitCount() {
     return hitCount;
   }
 
-  int getTopLevelCount() {
+  public int getTopLevelCount() {
     return topLevelCount;
   }
 
-  long getDuration() {
+  public long getDuration() {
     return okDuration + errorDuration;
   }
 
-  long getOkDuration() {
+  public long getOkDuration() {
     return okDuration;
   }
 
-  long getErrorDuration() {
+  public long getErrorDuration() {
     return errorDuration;
   }
 
-  Histogram getOkLatencies() {
+  public Histogram getOkLatencies() {
     return okLatencies;
   }
 
@@ -199,7 +199,7 @@ Histogram getOkLatencies() {
    * {@link SerializingMetricWriter}.
    */
   @Nullable
-  Histogram getErrorLatencies() {
+  public Histogram getErrorLatencies() {
     return errorLatencies;
   }
 
@@ -286,19 +286,19 @@ static long hashOf(
   }
 
   // Accessors for SerializingMetricWriter.
-  UTF8BytesString getResource() {
+  public UTF8BytesString getResource() {
     return resource;
   }
 
-  UTF8BytesString getService() {
+  public UTF8BytesString getService() {
     return service;
   }
 
-  UTF8BytesString getOperationName() {
+  public UTF8BytesString getOperationName() {
     return operationName;
   }
 
-  UTF8BytesString getServiceSource() {
+  public UTF8BytesString getServiceSource() {
     return serviceSource;
   }
 
@@ -308,65 +308,65 @@ UTF8BytesString getServiceSource() {
    * captured" (see field comment) -- callers that need a presence check should go through this
    * predicate rather than comparing against {@code EMPTY} directly.
    */
-  boolean hasServiceSource() {
+  public boolean hasServiceSource() {
     return serviceSource.length() > 0;
   }
 
-  UTF8BytesString getType() {
+  public UTF8BytesString getType() {
     return type;
   }
 
-  UTF8BytesString getSpanKind() {
+  public UTF8BytesString getSpanKind() {
     return spanKind;
   }
 
-  UTF8BytesString getHttpMethod() {
+  public UTF8BytesString getHttpMethod() {
     return httpMethod;
   }
 
   /**
    * Whether the snapshot carried an HTTP method. See {@link #hasServiceSource} for the contract.
    */
-  boolean hasHttpMethod() {
+  public boolean hasHttpMethod() {
     return httpMethod.length() > 0;
   }
 
-  UTF8BytesString getHttpEndpoint() {
+  public UTF8BytesString getHttpEndpoint() {
     return httpEndpoint;
   }
 
   /**
    * Whether the snapshot carried an HTTP endpoint. See {@link #hasServiceSource} for the contract.
    */
-  boolean hasHttpEndpoint() {
+  public boolean hasHttpEndpoint() {
     return httpEndpoint.length() > 0;
   }
 
-  UTF8BytesString getGrpcStatusCode() {
+  public UTF8BytesString getGrpcStatusCode() {
     return grpcStatusCode;
   }
 
   /**
    * Whether the snapshot carried a gRPC status code. See {@link #hasServiceSource} for the
    * contract.
    */
-  boolean hasGrpcStatusCode() {
+  public boolean hasGrpcStatusCode() {
     return grpcStatusCode.length() > 0;
   }
 
-  int getHttpStatusCode() {
+  public int getHttpStatusCode() {
     return httpStatusCode;
   }
 
-  boolean isSynthetics() {
+  public boolean isSynthetics() {
     return synthetic;
   }
 
-  boolean isTraceRoot() {
+  public boolean isTraceRoot() {
     return traceRoot;
   }
 
-  List<UTF8BytesString> getPeerTags() {
+  public List<UTF8BytesString> getPeerTags() {
     return peerTags;
   }
 

dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProto.java

@@ -60,6 +60,20 @@ public static int recordMetricMessage(
       OtelInstrumentDescriptor descriptor,
       int nestedDataPointBytes,
       OtlpProtoBuffer protobuf) {
+    return recordMetricMessage(buf, descriptor, nestedDataPointBytes, protobuf, false);
+  }
+
+  /**
+   * Records a metric message after its nested data point messages have been recorded. When {@code
+   * forceDelta} is {@code true}, a histogram is encoded as DELTA regardless of the configured
+   * temporality preference (for sources whose data points are inherently per-interval deltas).
+   */
+  public static int recordMetricMessage(
+      GrowableBuffer buf,
+      OtelInstrumentDescriptor descriptor,
+      int nestedDataPointBytes,
+      OtlpProtoBuffer protobuf,
+      boolean forceDelta) {
 
     writeTag(buf, 1, LEN_WIRE_TYPE);
     writeString(buf, descriptor.getName().getUtf8Bytes());
@@ -107,7 +121,7 @@ public static int recordMetricMessage(
         writeTag(buf, 9, LEN_WIRE_TYPE);
         writeVarInt(buf, nestedDataPointBytes + 2);
         writeTag(buf, 2, VARINT_WIRE_TYPE);
-        writeVarInt(buf, HISTOGRAM_TEMPORALITY);
+        writeVarInt(buf, forceDelta ? TEMPORALITY_DELTA : HISTOGRAM_TEMPORALITY);
         break;
       default:
         throw new IllegalArgumentException("Unknown instrument type: " + descriptor.getType());

dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProtoCollector.java

@@ -14,7 +14,6 @@
 import static datadog.trace.core.otlp.metrics.OtlpMetricsProto.recordScopedMetricsMessage;
 
 import datadog.communication.serialization.GrowableBuffer;
-import datadog.trace.api.time.SystemTimeSource;
 import datadog.trace.api.time.TimeSource;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
 import datadog.trace.bootstrap.otel.metrics.OtelInstrumentDescriptor;
@@ -46,14 +45,13 @@
 public final class OtlpMetricsProtoCollector extends OtlpMetricsCollector
     implements OtlpMetricsVisitor, OtlpScopedMetricsVisitor, OtlpMetricVisitor {
 
-  public static final OtlpMetricsProtoCollector INSTANCE =
-      new OtlpMetricsProtoCollector(SystemTimeSource.INSTANCE);
-
   private final GrowableBuffer buf = new GrowableBuffer(512);
   private final OtlpProtoBuffer protobuf = new OtlpProtoBuffer(8192);
 
   private final TimeSource timeSource;
 
+  private final boolean forceHistogramDelta;
+
   private long startNanos;
   private long endNanos;
 
@@ -66,8 +64,13 @@ public final class OtlpMetricsProtoCollector extends OtlpMetricsCollector
   private OtelInstrumentDescriptor currentMetric;
 
   public OtlpMetricsProtoCollector(TimeSource timeSource) {
+    this(timeSource, false);
+  }
+
+  OtlpMetricsProtoCollector(TimeSource timeSource, boolean forceHistogramDelta) {
     this.timeSource = timeSource;
     this.endNanos = timeSource.getCurrentTimeNanos();
+    this.forceHistogramDelta = forceHistogramDelta;
   }
 
   /**
@@ -82,6 +85,16 @@ public OtlpPayload collectMetrics() {
 
   OtlpPayload collectMetrics(Consumer<OtlpMetricsVisitor> registry) {
     start();
+    return run(registry);
+  }
+
+  OtlpPayload collectMetrics(
+      Consumer<OtlpMetricsVisitor> registry, long startNanos, long endNanos) {
+    startWithWindow(startNanos, endNanos);
+    return run(registry);
+  }
+
+  private OtlpPayload run(Consumer<OtlpMetricsVisitor> registry) {
     try {
       registry.accept(this);
       return completePayload();
@@ -96,6 +109,18 @@ private void start() {
     startNanos = endNanos;
     endNanos = timeSource.getCurrentTimeNanos();
 
+    recalibrate();
+  }
+
+  /** Prepare temporary elements to collect metrics over an explicit window. */
+  private void startWithWindow(long startNanos, long endNanos) {
+    this.startNanos = startNanos;
+    this.endNanos = endNanos;
+
+    recalibrate();
+  }
+
+  private void recalibrate() {
     // remove stale entries from caches
     OtlpCommonProto.recalibrateCaches();
   }
@@ -193,7 +218,8 @@ private void completeMetric() {
 
     // add metric message prefix to its nested chunks and promote to scoped
     if (metricBytes > 0) {
-      scopedBytes += recordMetricMessage(buf, currentMetric, metricBytes, protobuf);
+      scopedBytes +=
+          recordMetricMessage(buf, currentMetric, metricBytes, protobuf, forceHistogramDelta);
     }
 
     // reset temporary elements for next metric

dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsSenderFactory.java

@@ -0,0 +1,43 @@
+package datadog.trace.core.otlp.metrics;
+
+import datadog.trace.api.Config;
+import datadog.trace.core.otlp.common.OtlpGrpcSender;
+import datadog.trace.core.otlp.common.OtlpHttpSender;
+import datadog.trace.core.otlp.common.OtlpSender;
+import javax.annotation.Nullable;
+
+/**
+ * Selects the {@link OtlpSender} for the configured OTLP metrics protocol. Shared by every OTLP
+ * metrics export path ({@code OtlpMetricsService} for OpenTelemetry-API metrics, {@code
+ * OtlpStatsMetricWriter} for trace stats) so they all pick their transport identically and stay in
+ * sync as protocols/endpoints evolve.
+ */
+final class OtlpMetricsSenderFactory {
+  private OtlpMetricsSenderFactory() {}
+
+  /**
+   * Builds the sender for {@code config}'s OTLP metrics protocol, or {@code null} if the protocol
+   * has no protobuf encoder yet (e.g. HTTP_JSON).
+   */
+  @Nullable
+  static OtlpSender create(Config config) {
+    switch (config.getOtlpMetricsProtocol()) {
+      case GRPC:
+        return new OtlpGrpcSender(
+            config.getOtlpMetricsEndpoint(),
+            "/opentelemetry.proto.collector.metrics.v1.MetricsService/Export",
+            config.getOtlpMetricsHeaders(),
+            config.getOtlpMetricsTimeout(),
+            config.getOtlpMetricsCompression());
+      case HTTP_PROTOBUF:
+        return new OtlpHttpSender(
+            config.getOtlpMetricsEndpoint(),
+            "/v1/metrics",
+            config.getOtlpMetricsHeaders(),
+            config.getOtlpMetricsTimeout(),
+            config.getOtlpMetricsCompression());
+      default:
+        return null;
+    }
+  }
+}

dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java

@@ -3,8 +3,7 @@
 import static datadog.trace.util.AgentThreadFactory.AgentThread.OTLP_METRICS_EXPORTER;
 
 import datadog.trace.api.Config;
-import datadog.trace.core.otlp.common.OtlpGrpcSender;
-import datadog.trace.core.otlp.common.OtlpHttpSender;
+import datadog.trace.api.time.SystemTimeSource;
 import datadog.trace.core.otlp.common.OtlpPayload;
 import datadog.trace.core.otlp.common.OtlpSender;
 import datadog.trace.util.AgentTaskScheduler;
@@ -30,31 +29,12 @@ public final class OtlpMetricsService {
   private OtlpMetricsService(Config config) {
     this.scheduler = new AgentTaskScheduler(OTLP_METRICS_EXPORTER);
 
-    switch (config.getOtlpMetricsProtocol()) {
-      case GRPC:
-        this.collector = OtlpMetricsProtoCollector.INSTANCE;
-        this.sender =
-            new OtlpGrpcSender(
-                config.getOtlpMetricsEndpoint(),
-                "/opentelemetry.proto.collector.metrics.v1.MetricsService/Export",
-                config.getOtlpMetricsHeaders(),
-                config.getOtlpMetricsTimeout(),
-                config.getOtlpMetricsCompression());
-        break;
-      case HTTP_PROTOBUF:
-        this.collector = OtlpMetricsProtoCollector.INSTANCE;
-        this.sender =
-            new OtlpHttpSender(
-                config.getOtlpMetricsEndpoint(),
-                "/v1/metrics",
-                config.getOtlpMetricsHeaders(),
-                config.getOtlpMetricsTimeout(),
-                config.getOtlpMetricsCompression());
-        break;
-      default:
-        LOGGER.debug("Unsupported OTLP metrics protocol: {}", config.getOtlpMetricsProtocol());
-        this.collector = null;
-        this.sender = null;
+    this.sender = OtlpMetricsSenderFactory.create(config);
+    if (this.sender == null) {
+      LOGGER.debug("Unsupported OTLP metrics protocol: {}", config.getOtlpMetricsProtocol());
+      this.collector = null;
+    } else {
+      this.collector = new OtlpMetricsProtoCollector(SystemTimeSource.INSTANCE);
     }
 
     this.intervalMillis = config.getMetricsOtelInterval();