Skip to content

Commit a996fee

Browse files
mhlidddevflow.devflow-routing-intake
mhlidd
and
devflow.devflow-routing-intake
authored
BETA: Define OTLP Trace Metrics Histogram and Writers (#11656)
init init adding exact sums for ok/error remove host.name changes init tests adding datadog-specific datapoint attribtues more specific logging refactor origin usage in test helpers and add origin specific testS remove irrelevant test following rebase rename histogram class and refactor tests force temporality delta on tracemetrics histogram gating error latencies on empty as well update metricintegrationtest cleanup make AggregateEntry public and move trace metrics files to datadog.trace.core.otlp.metrics wire otlp trace metrics thru OtlpMetricsProtoCollector Merge branch 'master' into mhlidd/otlp_trace_metrics_setup jacoco Add Resource Attributes for OTLP Trace Metrics (#11667) * init * wiring datadog resource attributes into metric writer * cleanup * spotless rename write... w/ emit... Wire OTLP Trace Metrics Writers into Metrics Aggregators (#11680) * init * cleanup * init * adding bugfixes for system-tests * remove unnecessary code * cleanup * cleanup pt 2 * cleanup pt 2 * rebasing updates * renaming config to otel.traces.span.metrics.enabled * updating comments * updating buildResourceMessage to take a map instead of multiple booleans * Gate Additional OTLP Trace Metric Features (#11718) add shutdown for OKhttp client in OtlpStatsMetricWriter Co-authored-by: devflow.devflow-routing-intake <devflow.devflow-routing-intake@kubernetes.us1.ddbuild.io>
1 parent ba289f3 commit a996fee

31 files changed

Lines changed: 1794 additions & 136 deletions

File tree

‎communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -185,10 +185,11 @@ private void doDiscovery(State newState) {
185185

186186
if (log.isDebugEnabled()) {
187187
log.debug(
188-
"discovered traceEndpoint={}, metricsEndpoint={}, supportsDropping={}, supportsLongRunning={}, dataStreamsEndpoint={}, configEndpoint={}, logEndpoint={}, snapshotEndpoint={}, diagnosticsEndpoint={}, evpProxyEndpoint={}, telemetryProxyEndpoint={}",
188+
"discovered traceEndpoint={}, metricsEndpoint={}, supportsDropping={}, supportsClientSideStats={}, supportsLongRunning={}, dataStreamsEndpoint={}, configEndpoint={}, logEndpoint={}, snapshotEndpoint={}, diagnosticsEndpoint={}, evpProxyEndpoint={}, telemetryProxyEndpoint={}",
189189
newState.traceEndpoint,
190190
newState.metricsEndpoint,
191191
newState.supportsDropping,
192+
newState.supportsClientSideStats,
192193
newState.supportsLongRunning,
193194
newState.dataStreamsEndpoint,
194195
newState.configEndpoint,

‎dd-trace-api/src/main/java/datadog/trace/api/config/OtlpConfig.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ public final class OtlpConfig {
3131
public static final String OTLP_METRICS_TEMPORALITY_PREFERENCE =
3232
"otlp.metrics.temporality.preference";
3333

34-
public static final String TRACES_SPAN_METRICS_ENABLED = "traces.span.metrics.enabled";
34+
public static final String OTEL_TRACES_SPAN_METRICS_ENABLED = "otel.traces.span.metrics.enabled";
3535

3636
public static final String TRACE_OTEL_ENABLED = "trace.otel.enabled";
3737
public static final String TRACE_OTEL_EXPORTER = "trace.otel.exporter";

‎dd-trace-core/build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,7 @@ excludedClassesCoverage += [
6565
'datadog.trace.core.otlp.common.OtlpGrpcSender',
6666
// covered by OTLP system-tests
6767
'datadog.trace.core.otlp.logs.OtlpLogsService',
68+
'datadog.trace.core.otlp.metrics.OtlpMetricsSenderFactory',
6869
'datadog.trace.core.otlp.metrics.OtlpMetricsService'
6970
]
7071

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

Lines changed: 26 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
* static); tests using {@link PeerTagSchema} must also call {@code PeerTagSchema#resetHandlers} on
2323
* the schema instance.
2424
*/
25-
final class AggregateEntry extends Hashtable.Entry {
25+
public final class AggregateEntry extends Hashtable.Entry {
2626

2727
static final long ERROR_TAG = 0x8000000000000000L;
2828
static final long TOP_LEVEL_TAG = 0x4000000000000000L;
@@ -163,31 +163,31 @@ AggregateEntry recordOneDuration(long tagAndDuration) {
163163
return this;
164164
}
165165

166-
int getErrorCount() {
166+
public int getErrorCount() {
167167
return errorCount;
168168
}
169169

170-
int getHitCount() {
170+
public int getHitCount() {
171171
return hitCount;
172172
}
173173

174-
int getTopLevelCount() {
174+
public int getTopLevelCount() {
175175
return topLevelCount;
176176
}
177177

178-
long getDuration() {
178+
public long getDuration() {
179179
return okDuration + errorDuration;
180180
}
181181

182-
long getOkDuration() {
182+
public long getOkDuration() {
183183
return okDuration;
184184
}
185185

186-
long getErrorDuration() {
186+
public long getErrorDuration() {
187187
return errorDuration;
188188
}
189189

190-
Histogram getOkLatencies() {
190+
public Histogram getOkLatencies() {
191191
return okLatencies;
192192
}
193193

@@ -197,7 +197,7 @@ Histogram getOkLatencies() {
197197
* {@link SerializingMetricWriter}.
198198
*/
199199
@Nullable
200-
Histogram getErrorLatencies() {
200+
public Histogram getErrorLatencies() {
201201
return errorLatencies;
202202
}
203203

@@ -284,19 +284,19 @@ static long hashOf(
284284
}
285285

286286
// Accessors for SerializingMetricWriter.
287-
UTF8BytesString getResource() {
287+
public UTF8BytesString getResource() {
288288
return resource;
289289
}
290290

291-
UTF8BytesString getService() {
291+
public UTF8BytesString getService() {
292292
return service;
293293
}
294294

295-
UTF8BytesString getOperationName() {
295+
public UTF8BytesString getOperationName() {
296296
return operationName;
297297
}
298298

299-
UTF8BytesString getServiceSource() {
299+
public UTF8BytesString getServiceSource() {
300300
return serviceSource;
301301
}
302302

@@ -306,65 +306,65 @@ UTF8BytesString getServiceSource() {
306306
* captured" (see field comment) -- callers that need a presence check should go through this
307307
* predicate rather than comparing against {@code EMPTY} directly.
308308
*/
309-
boolean hasServiceSource() {
309+
public boolean hasServiceSource() {
310310
return serviceSource.length() > 0;
311311
}
312312

313-
UTF8BytesString getType() {
313+
public UTF8BytesString getType() {
314314
return type;
315315
}
316316

317-
UTF8BytesString getSpanKind() {
317+
public UTF8BytesString getSpanKind() {
318318
return spanKind;
319319
}
320320

321-
UTF8BytesString getHttpMethod() {
321+
public UTF8BytesString getHttpMethod() {
322322
return httpMethod;
323323
}
324324

325325
/**
326326
* Whether the snapshot carried an HTTP method. See {@link #hasServiceSource} for the contract.
327327
*/
328-
boolean hasHttpMethod() {
328+
public boolean hasHttpMethod() {
329329
return httpMethod.length() > 0;
330330
}
331331

332-
UTF8BytesString getHttpEndpoint() {
332+
public UTF8BytesString getHttpEndpoint() {
333333
return httpEndpoint;
334334
}
335335

336336
/**
337337
* Whether the snapshot carried an HTTP endpoint. See {@link #hasServiceSource} for the contract.
338338
*/
339-
boolean hasHttpEndpoint() {
339+
public boolean hasHttpEndpoint() {
340340
return httpEndpoint.length() > 0;
341341
}
342342

343-
UTF8BytesString getGrpcStatusCode() {
343+
public UTF8BytesString getGrpcStatusCode() {
344344
return grpcStatusCode;
345345
}
346346

347347
/**
348348
* Whether the snapshot carried a gRPC status code. See {@link #hasServiceSource} for the
349349
* contract.
350350
*/
351-
boolean hasGrpcStatusCode() {
351+
public boolean hasGrpcStatusCode() {
352352
return grpcStatusCode.length() > 0;
353353
}
354354

355-
int getHttpStatusCode() {
355+
public int getHttpStatusCode() {
356356
return httpStatusCode;
357357
}
358358

359-
boolean isSynthetics() {
359+
public boolean isSynthetics() {
360360
return synthetic;
361361
}
362362

363-
boolean isTraceRoot() {
363+
public boolean isTraceRoot() {
364364
return traceRoot;
365365
}
366366

367-
List<UTF8BytesString> getPeerTags() {
367+
public List<UTF8BytesString> getPeerTags() {
368368
return peerTags;
369369
}
370370

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

Lines changed: 90 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import static datadog.trace.api.DDSpanTypes.RPC;
55
import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_ENDPOINT;
66
import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_METHOD;
7+
import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_ROUTE;
78
import static datadog.trace.common.metrics.AggregateEntry.ERROR_TAG;
89
import static datadog.trace.common.metrics.AggregateEntry.TOP_LEVEL_TAG;
910
import static datadog.trace.common.metrics.SignalItem.ClearSignal.CLEAR;
@@ -12,6 +13,7 @@
1213
import static datadog.trace.util.AgentThreadFactory.AgentThread.METRICS_AGGREGATOR;
1314
import static datadog.trace.util.AgentThreadFactory.THREAD_JOIN_TIMOUT_MS;
1415
import static datadog.trace.util.AgentThreadFactory.newAgentThread;
16+
import static java.util.concurrent.TimeUnit.MILLISECONDS;
1517
import static java.util.concurrent.TimeUnit.SECONDS;
1618

1719
import datadog.common.queue.Queues;
@@ -26,6 +28,7 @@
2628
import datadog.trace.core.DDTraceCoreInfo;
2729
import datadog.trace.core.SpanKindFilter;
2830
import datadog.trace.core.monitor.HealthMetrics;
31+
import datadog.trace.core.otlp.metrics.OtlpStatsMetricWriter;
2932
import datadog.trace.util.AgentTaskScheduler;
3033
import java.util.Collections;
3134
import java.util.List;
@@ -62,16 +65,27 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList
6265
private static final SpanKindFilter INTERNAL_KIND =
6366
SpanKindFilter.builder().includeInternal().build();
6467

68+
// gRPC status-code source tags, probed in priority order on the OTLP export path
69+
private static final String[] GRPC_STATUS_CODE_KEYS = {
70+
InstrumentationTags.GRPC_STATUS_CODE, // "rpc.grpc.status_code"
71+
"grpc.code",
72+
"rpc.grpc.status.code",
73+
"grpc.status.code",
74+
"rpc.response.status_code",
75+
};
76+
6577
private final Set<String> ignoredResources;
6678
private final Thread thread;
6779
private final MessagePassingQueue<InboxItem> inbox;
6880
private final Sink sink;
81+
private final MetricWriter metricWriter;
6982
private final Aggregator aggregator;
7083
private final long reportingInterval;
7184
private final TimeUnit reportingIntervalTimeUnit;
7285
private final DDAgentFeaturesDiscovery features;
7386
private final HealthMetrics healthMetrics;
7487
private final boolean includeEndpointInMetrics;
88+
private final boolean otlpStatsExportEnabled;
7589

7690
/**
7791
* Cached peer-aggregation schema read by producer threads.
@@ -114,6 +128,30 @@ public ClientStatsAggregator(
114128
config.isTraceResourceRenamingEnabled());
115129
}
116130

131+
/**
132+
* OTLP span-metrics export variant. Reuses the same span selection + DDSketch aggregation as the
133+
* native path, but emits via the injected {@link OtlpStatsMetricWriter} instead of msgpack. No
134+
* agent {@link Sink} is needed, so a {@link NoOpSink} satisfies the register()/backpressure
135+
* contract, and the reporting interval comes from {@code trace.stats.interval} (milliseconds).
136+
*/
137+
public ClientStatsAggregator(
138+
Config config,
139+
SharedCommunicationObjects sharedCommunicationObjects,
140+
HealthMetrics healthMetrics,
141+
OtlpStatsMetricWriter metricWriter) {
142+
this(
143+
config.getMetricsIgnoredResources(),
144+
sharedCommunicationObjects.featuresDiscovery(config),
145+
healthMetrics,
146+
NoOpSink.INSTANCE,
147+
metricWriter,
148+
config.getTracerMetricsMaxAggregates(),
149+
config.getTracerMetricsMaxPending(),
150+
config.getTraceStatsInterval(),
151+
MILLISECONDS,
152+
true);
153+
}
154+
117155
ClientStatsAggregator(
118156
WellKnownTags wellKnownTags,
119157
Set<String> ignoredResources,
@@ -173,10 +211,12 @@ public ClientStatsAggregator(
173211
boolean includeEndpointInMetrics) {
174212
this.ignoredResources = ignoredResources;
175213
this.includeEndpointInMetrics = includeEndpointInMetrics;
214+
this.otlpStatsExportEnabled = metricWriter instanceof OtlpStatsMetricWriter;
176215
this.inbox = Queues.mpscArrayQueue(queueSize);
177216
this.features = features;
178217
this.healthMetrics = healthMetric;
179218
this.sink = sink;
219+
this.metricWriter = metricWriter;
180220
this.aggregator =
181221
new Aggregator(
182222
metricWriter,
@@ -191,6 +231,22 @@ public ClientStatsAggregator(
191231
this.reportingIntervalTimeUnit = timeUnit;
192232
}
193233

234+
// ── visible for testing ─────────────────────────────────────────────────────
235+
// Expose the writer-selection outcome and reporting cadence so tests can assert
236+
// the native-vs-OTLP XOR choice without reflecting into private fields.
237+
238+
boolean isOtlpStatsExportEnabled() {
239+
return otlpStatsExportEnabled;
240+
}
241+
242+
long reportingInterval() {
243+
return reportingInterval;
244+
}
245+
246+
TimeUnit reportingIntervalTimeUnit() {
247+
return reportingIntervalTimeUnit;
248+
}
249+
194250
@Override
195251
public void start() {
196252
sink.register(this);
@@ -206,11 +262,16 @@ public void start() {
206262
log.debug("started metrics aggregator");
207263
}
208264

265+
private boolean statsExportEnabled() {
266+
return otlpStatsExportEnabled || features.supportsMetrics();
267+
}
268+
209269
private boolean isMetricsEnabled() {
210-
if (features.getMetricsEndpoint() == null) {
270+
// The discovery refresh only helps the native path.
271+
if (!otlpStatsExportEnabled && features.getMetricsEndpoint() == null) {
211272
features.discoverIfOutdated();
212273
}
213-
return features.supportsMetrics();
274+
return statsExportEnabled();
214275
}
215276

216277
@Override
@@ -266,7 +327,7 @@ public Future<Boolean> forceReport() {
266327
public boolean publish(List<? extends CoreSpan<?>> trace) {
267328
boolean forceKeep = false;
268329
int counted = 0;
269-
if (features.supportsMetrics()) {
330+
if (statsExportEnabled()) {
270331
// Producer-side fast path: one volatile read and use whatever schema is currently cached.
271332
// The aggregator thread keeps this schema in sync with feature discovery in
272333
// resetCardinalityHandlers(). The only producer-side rebuild is the one-time bootstrap on
@@ -315,12 +376,21 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel, PeerTagSchema peer
315376
Object httpMethodObj = span.unsafeGetTag(HTTP_METHOD);
316377
httpMethod = httpMethodObj != null ? httpMethodObj.toString() : null;
317378
Object httpEndpointObj = span.unsafeGetTag(HTTP_ENDPOINT);
379+
// OTLP path falls back to http.route (mirrors libdatadog). The native v0.6 path keeps its
380+
// http.endpoint-only lookup so this doesn't change its aggregation key / wire output.
381+
if (otlpStatsExportEnabled && httpEndpointObj == null) {
382+
httpEndpointObj = span.unsafeGetTag(HTTP_ROUTE);
383+
}
318384
httpEndpoint = httpEndpointObj != null ? httpEndpointObj.toString() : null;
319385
}
320386

321387
CharSequence spanType = span.getType();
322388
String grpcStatusCode = null;
323-
if (spanType != null && RPC.contentEquals(spanType)) {
389+
if (otlpStatsExportEnabled) {
390+
// OTLP path: probe every known gRPC status-code convention, no span-type gate, so a span
391+
// typed "grpc" (or carrying an OTel-style key) still surfaces rpc.response.status_code.
392+
grpcStatusCode = firstTag(span, GRPC_STATUS_CODE_KEYS);
393+
} else if (spanType != null && RPC.contentEquals(spanType)) {
324394
Object grpcStatusObj = span.unsafeGetTag(InstrumentationTags.GRPC_STATUS_CODE);
325395
grpcStatusCode = grpcStatusObj != null ? grpcStatusObj.toString() : null;
326396
}
@@ -367,6 +437,17 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel, PeerTagSchema peer
367437
return error;
368438
}
369439

440+
/** Returns the first non-null span tag among {@code keys}, in order, or {@code null} if none. */
441+
private static String firstTag(CoreSpan<?> span, String[] keys) {
442+
for (String key : keys) {
443+
Object value = span.unsafeGetTag(key);
444+
if (value != null) {
445+
return value.toString();
446+
}
447+
}
448+
return null;
449+
}
450+
370451
/**
371452
* One-time producer-side bootstrap of {@link #cachedPeerTagSchema}. Synchronized double-check
372453
* guards against two producers racing on the very first publish; after this returns, {@code
@@ -526,6 +607,11 @@ public void close() {
526607
thread.join(THREAD_JOIN_TIMOUT_MS);
527608
} catch (InterruptedException ignored) {
528609
}
610+
// The OTLP stats writer owns a dedicated OTLP sender. Evict it now that the aggregator thread
611+
// has stopped, so no in-flight finishBucket() flush is still using the sender.
612+
if (metricWriter instanceof OtlpStatsMetricWriter) {
613+
((OtlpStatsMetricWriter) metricWriter).shutdown();
614+
}
529615
}
530616

531617
@Override

0 commit comments

Comments
 (0)