From 203ce5cd0c183c8891333909583fb11a1def34ff Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Wed, 26 Aug 2026 19:55:54 -0400 Subject: [PATCH 1/4] test: cover OTel metrics lifecycle operations --- manifests/dotnet.yml | 1 + manifests/golang.yml | 1 + manifests/java.yml | 1 + manifests/nodejs.yml | 1 + manifests/ruby.yml | 1 + manifests/rust.yml | 1 + tests/parametric/test_otel_metrics.py | 77 +++++++++++++++++++ .../OpenTelemetryMetricsController.java | 37 +++++++++ .../trace/opentelemetry/dto/FlushArgs.java | 4 +- .../build/docker/nodejs/parametric/server.js | 44 +++++++++-- .../parametric/apm_test_client/server.py | 32 +++++++- .../_test_clients/_test_client_parametric.py | 24 +++++- 12 files changed, 211 insertions(+), 13 deletions(-) diff --git a/manifests/dotnet.yml b/manifests/dotnet.yml index 774815390e7..d1c1a60fb35 100644 --- a/manifests/dotnet.yml +++ b/manifests/dotnet.yml @@ -1006,6 +1006,7 @@ manifest: tests/parametric/test_otel_metrics.py: v3.29.0 tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Api_MeterProvider::test_otel_get_meter_by_distinct_schema_url: missing_feature (Not supported by .NET's System.Diagnostics.Metrics API) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: missing_feature (DD_HOSTNAME to host.name resource attribute mapping not yet implemented) + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (OTel metrics lifecycle API not implemented) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_exporter_configurations: # Modified by easy win activation script - declaration: missing_feature (OTel metrics telemetry metrics (otel.metrics_export_attempts) not yet fully flushed in time) component_version: <3.36.0 diff --git a/manifests/golang.yml b/manifests/golang.yml index 1041600d993..1a9fe4c51f4 100644 --- a/manifests/golang.yml +++ b/manifests/golang.yml @@ -1492,6 +1492,7 @@ manifest: tests/parametric/test_otel_logs.py::Test_FR11_Telemetry: missing_feature # Created by easy win activation script tests/parametric/test_otel_logs.py::Test_FR12_Log_Levels: missing_feature # Created by easy win activation script tests/parametric/test_otel_logs.py::Test_FR13_Scope_Fields: missing_feature # Created by easy win activation script + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (OTel metrics lifecycle API not implemented) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_default_configurations: missing_feature tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_exporter_configurations: missing_feature tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_exporter_metrics_configurations: missing_feature diff --git a/manifests/java.yml b/manifests/java.yml index f56d208458c..a97b55ac272 100644 --- a/manifests/java.yml +++ b/manifests/java.yml @@ -3989,6 +3989,7 @@ manifest: : incomplete_test_app (GPRC fails for system-test but works with real collector) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_OTLP_Exporter_Metrics_Protocol::test_otlp_protocol_grpc: incomplete_test_app (GPRC fails for system-test but works with real collector) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: irrelevant (DD_HOSTNAME is only supported in Python) + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (Public OTel metrics lifecycle API not released) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry: missing_feature (not yet implemented) tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_add_event_meta_serialization: - declaration: missing_feature (Not implemented) diff --git a/manifests/nodejs.yml b/manifests/nodejs.yml index 0a412f144a5..bdff5244c0e 100644 --- a/manifests/nodejs.yml +++ b/manifests/nodejs.yml @@ -2417,6 +2417,7 @@ manifest: ? tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_OTLP_Exporter_Metrics_Endpoint::test_otlp_metrics_custom_endpoint_grpc : missing_feature (Does not support grpc) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: missing_feature (DD_HOSTNAME to host.name resource attribute mapping not yet implemented) + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (Public OTel metrics lifecycle API not released) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_metrics_grpc: missing_feature (Does not support grpc) tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_add_event_meta_serialization: - declaration: missing_feature (Implemented in v5.17.0 & v4.41.0) diff --git a/manifests/ruby.yml b/manifests/ruby.yml index e36a3206854..e81abbc9383 100644 --- a/manifests/ruby.yml +++ b/manifests/ruby.yml @@ -2500,6 +2500,7 @@ manifest: ? tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_OTLP_Exporter_Metrics_Endpoint::test_otlp_metrics_custom_endpoint_grpc : missing_feature (Does not support grpc) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_Temporality_Preference::test_otel_aggregation_temporality: missing_feature (default Histogram bucket ranges is not up to spec) + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (OTel metrics lifecycle API not implemented) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_metrics_grpc: missing_feature (Does not support grpc) tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods: v1.17.0 tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_add_event_meta_serialization: # TODO: a lower version might be supported diff --git a/manifests/rust.yml b/manifests/rust.yml index 8743419d232..40af19fed22 100644 --- a/manifests/rust.yml +++ b/manifests/rust.yml @@ -255,6 +255,7 @@ manifest: : missing_feature (OpenTelemetry Rust SDK does not filter negative values for Histogram.record(). This is a bug in the upstream SDK) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Api_MeterProvider::test_otel_get_meter_by_distinct_scope_attributes: missing_feature (Rust OpenTelemetry SDK 0.31.0's meter() method does not support scope attributes parameter) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: missing_feature + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (OTel metrics lifecycle API not implemented) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_metrics_grpc: '>=0.2.1' # Modified by easy win activation script tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_metrics_http_protobuf: '>=0.2.1' # Modified by easy win activation script tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods: v0.0.1 diff --git a/tests/parametric/test_otel_metrics.py b/tests/parametric/test_otel_metrics.py index b4a075e9436..60917e0ed38 100644 --- a/tests/parametric/test_otel_metrics.py +++ b/tests/parametric/test_otel_metrics.py @@ -228,6 +228,17 @@ def find_metric_by_name(scope_metric: dict, name: str) -> dict: raise ValueError(f"Metric with name {name} not found") +def find_metrics_by_name(metric_requests: list[dict], name: str) -> list[dict]: + return [ + metric + for metric_request in metric_requests + for resource_metrics in metric_request["resource_metrics"] + for scope_metrics in resource_metrics["scope_metrics"] + for metric in scope_metrics["metrics"] + if metric["name"] == name + ] + + def get_expected_bucket_counts(entries: list[int], bucket_boundaries: list[float]) -> list[int]: bucket_counts = [0] * (len(bucket_boundaries) + 1) for entry in entries: @@ -240,6 +251,72 @@ def get_expected_bucket_counts(entries: list[int], bucket_boundaries: list[float return bucket_counts +@scenarios.parametric +@features.otel_metrics_api +class Test_Otel_Metrics_Lifecycle: + @staticmethod + def generate_pending_counter(test_library: APMLibrary, metric_name: str) -> None: + test_library.otel_get_meter(DEFAULT_METER_NAME, DEFAULT_METER_VERSION, DEFAULT_SCHEMA_URL, {}) + test_library.otel_create_counter( + DEFAULT_METER_NAME, + metric_name, + DEFAULT_INSTRUMENT_UNIT, + DEFAULT_INSTRUMENT_DESCRIPTION, + ) + test_library.otel_counter_add( + DEFAULT_METER_NAME, + metric_name, + DEFAULT_INSTRUMENT_UNIT, + DEFAULT_INSTRUMENT_DESCRIPTION, + 42, + DEFAULT_MEASUREMENT_ATTRIBUTES, + ) + + @staticmethod + def assert_exported_once(test_agent: TestAgentAPI, metric_name: str) -> None: + matching_metrics = find_metrics_by_name(test_agent.metrics(), metric_name) + assert len(matching_metrics) == 1 + assert_sum_aggregation( + matching_metrics[0]["sum"], + "AGGREGATION_TEMPORALITY_DELTA", + is_monotonic=True, + value=42, + attributes=DEFAULT_MEASUREMENT_ATTRIBUTES, + ) + + @pytest.mark.parametrize("library_env", [{**DEFAULT_ENVVARS}]) + def test_force_flush_exports_pending_metric_before_return( + self, test_agent: TestAgentAPI, test_library: APMLibrary, test_id: str + ) -> None: + metric_name = f"lifecycle-force-flush-{test_id}" + self.generate_pending_counter(test_library, metric_name) + assert find_metrics_by_name(test_agent.metrics(), metric_name) == [] + + try: + success = test_library.otel_metrics_force_flush(10, public_only=True) + finally: + test_library.terminate() + + assert success + self.assert_exported_once(test_agent, metric_name) + + @pytest.mark.parametrize("library_env", [{**DEFAULT_ENVVARS}]) + def test_shutdown_exports_pending_metric_before_return( + self, test_agent: TestAgentAPI, test_library: APMLibrary, test_id: str + ) -> None: + metric_name = f"lifecycle-shutdown-{test_id}" + self.generate_pending_counter(test_library, metric_name) + assert find_metrics_by_name(test_agent.metrics(), metric_name) == [] + + try: + success = test_library.otel_metrics_shutdown(10) + finally: + test_library.terminate() + + assert success + self.assert_exported_once(test_agent, metric_name) + + @scenarios.parametric @features.otel_metrics_api class Test_Otel_Metrics_Configuration_Enabled: diff --git a/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/controller/OpenTelemetryMetricsController.java b/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/controller/OpenTelemetryMetricsController.java index 00113d45a3a..32786c635b5 100644 --- a/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/controller/OpenTelemetryMetricsController.java +++ b/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/controller/OpenTelemetryMetricsController.java @@ -10,9 +10,12 @@ import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.common.AttributesBuilder; import io.opentelemetry.api.metrics.*; +import java.lang.reflect.Method; import java.util.Locale; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; @@ -214,6 +217,13 @@ public void histogramRecord(@RequestBody HistogramRecordArgs args) { public FlushResult forceFlush(@RequestBody FlushArgs args) { LOGGER.info("Flushing OTel metrics: {}", args); try { + if (invokeLifecycle("forceFlush", args.seconds())) { + MetricsController.skipNextFlush(); + return new FlushResult(true); + } + if (args.publicOnly()) { + return new FlushResult(false); + } if (GlobalTracer.get() instanceof InternalTracer internalTracer) { internalTracer.flushMetrics(); // skip the next general metrics flush, as it's covered by the same call @@ -229,6 +239,33 @@ public FlushResult forceFlush(@RequestBody FlushArgs args) { } } + @PostMapping("shutdown") + public FlushResult shutdown(@RequestBody FlushArgs args) { + LOGGER.info("Shutting down OTel metrics: {}", args); + try { + return new FlushResult(invokeLifecycle("shutdown", args.seconds())); + } catch (Exception e) { + LOGGER.warn("Failed to shut down OTel metrics", e); + return new FlushResult(false); + } + } + + private static boolean invokeLifecycle(String methodName, long seconds) throws Exception { + Class lifecycleClass; + Method method; + try { + lifecycleClass = Class.forName("datadog.trace.api.metrics.OpenTelemetryMetrics"); + method = lifecycleClass.getMethod(methodName); + } catch (ClassNotFoundException | NoSuchMethodException ignored) { + return false; + } + Object result = method.invoke(null); + if (!(result instanceof Future future)) { + return false; + } + return Boolean.TRUE.equals(future.get(seconds, TimeUnit.SECONDS)); + } + /** Builds {@link Attributes} from a map of strings. */ private static Attributes fromMap(Map map) { AttributesBuilder builder = Attributes.builder(); diff --git a/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/dto/FlushArgs.java b/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/dto/FlushArgs.java index 89554a6099d..c346d8f1cb4 100644 --- a/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/dto/FlushArgs.java +++ b/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/dto/FlushArgs.java @@ -1,4 +1,6 @@ package com.datadoghq.trace.opentelemetry.dto; -public record FlushArgs(long seconds) { +import com.fasterxml.jackson.annotation.JsonProperty; + +public record FlushArgs(long seconds, @JsonProperty("public_only") boolean publicOnly) { } diff --git a/utils/build/docker/nodejs/parametric/server.js b/utils/build/docker/nodejs/parametric/server.js index 10fb172b8b5..b43b4716934 100644 --- a/utils/build/docker/nodejs/parametric/server.js +++ b/utils/build/docker/nodejs/parametric/server.js @@ -72,6 +72,20 @@ function createInstrumentKey(meterName, name, kind, unit, description) { return `${meterName}:${name}:${kind}:${unit}:${description}`; } +async function waitForMetricsLifecycle (operation, seconds) { + let timeout + try { + await Promise.race([ + operation, + new Promise((resolve, reject) => { + timeout = setTimeout(() => reject(new Error('Metrics lifecycle operation timed out')), seconds * 1000) + }) + ]) + } finally { + clearTimeout(timeout) + } +} + app.post('/trace/span/inject_headers', (req, res) => { const request = req.body; const span = spans[request.span_id] @@ -807,13 +821,33 @@ app.post('/metrics/otel/create_asynchronous_gauge', (req, res) => { res.json({}); }); -app.post('/metrics/otel/force_flush', (req, res) => { +app.post('/metrics/otel/force_flush', async (req, res) => { const meterProvider = metrics.getMeterProvider(); - if (meterProvider.reader) { - meterProvider.reader.forceFlush() + try { + if (typeof meterProvider.forceFlush === 'function') { + await waitForMetricsLifecycle(meterProvider.forceFlush(), req.body.seconds || 10) + return res.json({ success: true }); + } + if (!req.body.public_only && meterProvider.reader) { + await waitForMetricsLifecycle(meterProvider.reader.forceFlush(), req.body.seconds || 10) + return res.json({ success: true }); + } + } catch (error) { + return res.json({ success: false, message: error.message }); + } + res.json({ success: false, message: 'Force flush not supported' }); +}); + +app.post('/metrics/otel/shutdown', async (req, res) => { + const meterProvider = metrics.getMeterProvider(); + if (typeof meterProvider.shutdown !== 'function') { + return res.json({ success: false, message: 'Shutdown not supported' }); + } + try { + await waitForMetricsLifecycle(meterProvider.shutdown(), req.body.seconds || 10) res.json({ success: true }); - } else { - res.json({ success: false, message: 'Force flush not supported' }); + } catch (error) { + res.json({ success: false, message: error.message }); } }); diff --git a/utils/build/docker/python/parametric/apm_test_client/server.py b/utils/build/docker/python/parametric/apm_test_client/server.py index 0ff77f4e9c0..5a907079aa2 100644 --- a/utils/build/docker/python/parametric/apm_test_client/server.py +++ b/utils/build/docker/python/parametric/apm_test_client/server.py @@ -1292,7 +1292,8 @@ def otel_create_asynchronous_gauge(args: OtelCreateAsynchronousGaugeArgs): class OtelMetricsForceFlushArgs(BaseModel): - pass + seconds: int = 10 + public_only: bool = False class OtelMetricsForceFlushReturn(BaseModel): @@ -1309,9 +1310,34 @@ def otel_metrics_force_flush(args: OtelMetricsForceFlushArgs): # a default _ProxyMeterProvider provided by the API which does not # have the method. if hasattr(meter_provider, "force_flush"): - meter_provider.force_flush() + try: + result = meter_provider.force_flush(timeout_millis=args.seconds * 1000) + return OtelMetricsForceFlushReturn(success=result is not False) + except Exception: + return OtelMetricsForceFlushReturn(success=False) + + return OtelMetricsForceFlushReturn(success=not args.public_only) + + +class OtelMetricsShutdownArgs(BaseModel): + seconds: int = 10 + - return OtelMetricsForceFlushReturn(success=True) +class OtelMetricsShutdownReturn(BaseModel): + success: bool + + +@app.post("/metrics/otel/shutdown") +def otel_metrics_shutdown(args: OtelMetricsShutdownArgs): + meter_provider = get_meter_provider() + if not hasattr(meter_provider, "shutdown"): + return OtelMetricsShutdownReturn(success=False) + + try: + meter_provider.shutdown(timeout_millis=args.seconds * 1000) + return OtelMetricsShutdownReturn(success=True) + except Exception: + return OtelMetricsShutdownReturn(success=False) class LogCreateLoggerArgs(BaseModel): diff --git a/utils/docker_fixtures/_test_clients/_test_client_parametric.py b/utils/docker_fixtures/_test_clients/_test_client_parametric.py index 703b808ec75..db7131f61a2 100644 --- a/utils/docker_fixtures/_test_clients/_test_client_parametric.py +++ b/utils/docker_fixtures/_test_clients/_test_client_parametric.py @@ -289,6 +289,10 @@ def crash(self) -> None: except RequestException as e: logger.info(f"Expected exception when calling /trace/crash: {e}") + def terminate(self) -> None: + self.container.kill(signal="SIGKILL") + self.container.wait(timeout=10) + def get_logs(self) -> str: return self.container.logs().decode("utf-8") @@ -967,8 +971,14 @@ def otel_create_asynchronous_gauge( }, ) - def otel_metrics_force_flush(self) -> bool: - resp = self._session.post(self._url("/metrics/otel/force_flush"), json={}).json() + def otel_metrics_force_flush(self, seconds: int = 10, *, public_only: bool = False) -> bool: + resp = self._session.post( + self._url("/metrics/otel/force_flush"), json={"seconds": seconds, "public_only": public_only} + ).json() + return resp["success"] + + def otel_metrics_shutdown(self, seconds: int = 10) -> bool: + resp = self._session.post(self._url("/metrics/otel/shutdown"), json={"seconds": seconds}).json() return resp["success"] def llmobs_trace( @@ -1213,8 +1223,14 @@ def otel_create_asynchronous_gauge( ) -> None: self._client.otel_create_asynchronous_gauge(meter_name, name, unit, description, value, attributes) - def otel_metrics_force_flush(self) -> bool: - return self._client.otel_metrics_force_flush() + def otel_metrics_force_flush(self, seconds: int = 10, *, public_only: bool = False) -> bool: + return self._client.otel_metrics_force_flush(seconds, public_only=public_only) + + def otel_metrics_shutdown(self, seconds: int = 10) -> bool: + return self._client.otel_metrics_shutdown(seconds) + + def terminate(self) -> None: + self._client.terminate() def is_alive(self) -> bool: return self._client.is_alive() From 42826253c288dbea02464c4d3d461bc73b2f1c8e Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Tue, 1 Sep 2026 16:04:29 -0400 Subject: [PATCH 2/4] test: align OTel metrics lifecycle coverage --- manifests/dotnet.yml | 2 +- manifests/golang.yml | 2 +- manifests/java.yml | 4 ++- manifests/nodejs.yml | 4 ++- manifests/ruby.yml | 1 - manifests/rust.yml | 1 - tests/parametric/test_otel_metrics.py | 14 +++++---- .../build/docker/golang/parametric/helpers.go | 1 + utils/build/docker/golang/parametric/main.go | 1 + .../docker/golang/parametric/otel_metrics.go | 27 ++++++++++++++--- .../build/docker/nodejs/parametric/server.js | 16 ++++++---- utils/build/docker/ruby/parametric/server.rb | 29 +++++++++++++++++-- .../rust/parametric/src/opentelemetry/mod.rs | 14 +++++++++ .../_test_clients/_test_client_parametric.py | 8 +++-- 14 files changed, 98 insertions(+), 26 deletions(-) diff --git a/manifests/dotnet.yml b/manifests/dotnet.yml index d1c1a60fb35..48e6e5ac079 100644 --- a/manifests/dotnet.yml +++ b/manifests/dotnet.yml @@ -1006,7 +1006,7 @@ manifest: tests/parametric/test_otel_metrics.py: v3.29.0 tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Api_MeterProvider::test_otel_get_meter_by_distinct_schema_url: missing_feature (Not supported by .NET's System.Diagnostics.Metrics API) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: missing_feature (DD_HOSTNAME to host.name resource attribute mapping not yet implemented) - tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (OTel metrics lifecycle API not implemented) + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (No public Datadog meter-provider shutdown operation) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_exporter_configurations: # Modified by easy win activation script - declaration: missing_feature (OTel metrics telemetry metrics (otel.metrics_export_attempts) not yet fully flushed in time) component_version: <3.36.0 diff --git a/manifests/golang.yml b/manifests/golang.yml index 1a9fe4c51f4..2faaabc4748 100644 --- a/manifests/golang.yml +++ b/manifests/golang.yml @@ -1492,7 +1492,7 @@ manifest: tests/parametric/test_otel_logs.py::Test_FR11_Telemetry: missing_feature # Created by easy win activation script tests/parametric/test_otel_logs.py::Test_FR12_Log_Levels: missing_feature # Created by easy win activation script tests/parametric/test_otel_logs.py::Test_FR13_Scope_Fields: missing_feature # Created by easy win activation script - tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (OTel metrics lifecycle API not implemented) + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: v2.6.0 tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_default_configurations: missing_feature tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_exporter_configurations: missing_feature tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_exporter_metrics_configurations: missing_feature diff --git a/manifests/java.yml b/manifests/java.yml index a97b55ac272..cf1a8996ba9 100644 --- a/manifests/java.yml +++ b/manifests/java.yml @@ -3989,7 +3989,9 @@ manifest: : incomplete_test_app (GPRC fails for system-test but works with real collector) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_OTLP_Exporter_Metrics_Protocol::test_otlp_protocol_grpc: incomplete_test_app (GPRC fails for system-test but works with real collector) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: irrelevant (DD_HOSTNAME is only supported in Python) - tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (Public OTel metrics lifecycle API not released) + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: + - declaration: missing_feature (Implemented in 1.66.0) + component_version: '<=1.66.0-SNAPSHOT' tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry: missing_feature (not yet implemented) tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_add_event_meta_serialization: - declaration: missing_feature (Not implemented) diff --git a/manifests/nodejs.yml b/manifests/nodejs.yml index bdff5244c0e..347471e5585 100644 --- a/manifests/nodejs.yml +++ b/manifests/nodejs.yml @@ -2417,7 +2417,9 @@ manifest: ? tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_OTLP_Exporter_Metrics_Endpoint::test_otlp_metrics_custom_endpoint_grpc : missing_feature (Does not support grpc) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: missing_feature (DD_HOSTNAME to host.name resource attribute mapping not yet implemented) - tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (Public OTel metrics lifecycle API not released) + tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: + - declaration: missing_feature (Implemented in 7.0.0) + component_version: '<=7.0.0-pre' tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_metrics_grpc: missing_feature (Does not support grpc) tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_add_event_meta_serialization: - declaration: missing_feature (Implemented in v5.17.0 & v4.41.0) diff --git a/manifests/ruby.yml b/manifests/ruby.yml index e81abbc9383..e36a3206854 100644 --- a/manifests/ruby.yml +++ b/manifests/ruby.yml @@ -2500,7 +2500,6 @@ manifest: ? tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_OTLP_Exporter_Metrics_Endpoint::test_otlp_metrics_custom_endpoint_grpc : missing_feature (Does not support grpc) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_Temporality_Preference::test_otel_aggregation_temporality: missing_feature (default Histogram bucket ranges is not up to spec) - tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (OTel metrics lifecycle API not implemented) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_metrics_grpc: missing_feature (Does not support grpc) tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods: v1.17.0 tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_add_event_meta_serialization: # TODO: a lower version might be supported diff --git a/manifests/rust.yml b/manifests/rust.yml index 40af19fed22..8743419d232 100644 --- a/manifests/rust.yml +++ b/manifests/rust.yml @@ -255,7 +255,6 @@ manifest: : missing_feature (OpenTelemetry Rust SDK does not filter negative values for Histogram.record(). This is a bug in the upstream SDK) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Api_MeterProvider::test_otel_get_meter_by_distinct_scope_attributes: missing_feature (Rust OpenTelemetry SDK 0.31.0's meter() method does not support scope attributes parameter) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: missing_feature - tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: missing_feature (OTel metrics lifecycle API not implemented) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_metrics_grpc: '>=0.2.1' # Modified by easy win activation script tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry::test_telemetry_metrics_http_protobuf: '>=0.2.1' # Modified by easy win activation script tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods: v0.0.1 diff --git a/tests/parametric/test_otel_metrics.py b/tests/parametric/test_otel_metrics.py index 60917e0ed38..038dfbc0394 100644 --- a/tests/parametric/test_otel_metrics.py +++ b/tests/parametric/test_otel_metrics.py @@ -48,6 +48,8 @@ "CORECLR_ENABLE_PROFILING": "1", } +LIFECYCLE_ENVVARS = {**DEFAULT_ENVVARS, "OTEL_METRIC_EXPORT_INTERVAL": "3600000"} + @pytest.fixture def otlp_metrics_endpoint_library_env( @@ -273,8 +275,8 @@ def generate_pending_counter(test_library: APMLibrary, metric_name: str) -> None ) @staticmethod - def assert_exported_once(test_agent: TestAgentAPI, metric_name: str) -> None: - matching_metrics = find_metrics_by_name(test_agent.metrics(), metric_name) + def assert_exported_once(metric_requests: list[dict], metric_name: str) -> None: + matching_metrics = find_metrics_by_name(metric_requests, metric_name) assert len(matching_metrics) == 1 assert_sum_aggregation( matching_metrics[0]["sum"], @@ -284,7 +286,7 @@ def assert_exported_once(test_agent: TestAgentAPI, metric_name: str) -> None: attributes=DEFAULT_MEASUREMENT_ATTRIBUTES, ) - @pytest.mark.parametrize("library_env", [{**DEFAULT_ENVVARS}]) + @pytest.mark.parametrize("library_env", [{**LIFECYCLE_ENVVARS}]) def test_force_flush_exports_pending_metric_before_return( self, test_agent: TestAgentAPI, test_library: APMLibrary, test_id: str ) -> None: @@ -298,9 +300,9 @@ def test_force_flush_exports_pending_metric_before_return( test_library.terminate() assert success - self.assert_exported_once(test_agent, metric_name) + self.assert_exported_once(test_agent.metrics(), metric_name) - @pytest.mark.parametrize("library_env", [{**DEFAULT_ENVVARS}]) + @pytest.mark.parametrize("library_env", [{**LIFECYCLE_ENVVARS}]) def test_shutdown_exports_pending_metric_before_return( self, test_agent: TestAgentAPI, test_library: APMLibrary, test_id: str ) -> None: @@ -314,7 +316,7 @@ def test_shutdown_exports_pending_metric_before_return( test_library.terminate() assert success - self.assert_exported_once(test_agent, metric_name) + self.assert_exported_once(test_agent.metrics(), metric_name) @scenarios.parametric diff --git a/utils/build/docker/golang/parametric/helpers.go b/utils/build/docker/golang/parametric/helpers.go index b4958d24186..92e786ce3d7 100644 --- a/utils/build/docker/golang/parametric/helpers.go +++ b/utils/build/docker/golang/parametric/helpers.go @@ -284,6 +284,7 @@ type OtelCreateAsynchronousGaugeArgs struct { } type OtelMetricsForceFlushArgs struct { + Seconds int `json:"seconds"` } type OtelMetricsForceFlushReturn struct { diff --git a/utils/build/docker/golang/parametric/main.go b/utils/build/docker/golang/parametric/main.go index 509c3fd172c..926c5fdf6ee 100644 --- a/utils/build/docker/golang/parametric/main.go +++ b/utils/build/docker/golang/parametric/main.go @@ -129,6 +129,7 @@ func main() { http.HandleFunc("/metrics/otel/create_asynchronous_updowncounter", s.otelCreateAsynchronousUpDownCounterHandler) http.HandleFunc("/metrics/otel/create_asynchronous_gauge", s.otelCreateAsynchronousGaugeHandler) http.HandleFunc("/metrics/otel/force_flush", s.otelMetricsForceFlushHandler) + http.HandleFunc("/metrics/otel/shutdown", s.otelMetricsShutdownHandler) err = http.ListenAndServe(fmt.Sprintf(":%d", port), nil) if err != nil { diff --git a/utils/build/docker/golang/parametric/otel_metrics.go b/utils/build/docker/golang/parametric/otel_metrics.go index a6b79ce9a00..662ca712a4e 100644 --- a/utils/build/docker/golang/parametric/otel_metrics.go +++ b/utils/build/docker/golang/parametric/otel_metrics.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "strings" + "time" ddmetric "github.com/DataDog/dd-trace-go/v2/ddtrace/opentelemetry/metric" "go.opentelemetry.io/otel" @@ -569,7 +570,7 @@ func (s *apmClientServer) otelMetricsForceFlushHandler(w http.ResponseWriter, r return } - success := s.OtelMetricsForceFlush() + success := s.OtelMetricsForceFlush(args.Seconds) w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(&OtelMetricsForceFlushReturn{Success: success}); err != nil { @@ -577,10 +578,10 @@ func (s *apmClientServer) otelMetricsForceFlushHandler(w http.ResponseWriter, r } } -func (s *apmClientServer) OtelMetricsForceFlush() bool { - // Use the dd-trace-go helper to flush metrics +func (s *apmClientServer) OtelMetricsForceFlush(seconds int) bool { mp := otel.GetMeterProvider() - ctx := context.Background() + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(seconds)*time.Second) + defer cancel() if err := ddmetric.ForceFlush(ctx, mp); err != nil { fmt.Printf("Error flushing metrics: %v\n", err) return false @@ -588,6 +589,24 @@ func (s *apmClientServer) OtelMetricsForceFlush() bool { return true } +func (s *apmClientServer) otelMetricsShutdownHandler(w http.ResponseWriter, r *http.Request) { + var args OtelMetricsForceFlushArgs + if err := json.NewDecoder(r.Body).Decode(&args); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + mp := otel.GetMeterProvider() + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(args.Seconds)*time.Second) + defer cancel() + success := ddmetric.Shutdown(ctx, mp) == nil + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(&OtelMetricsForceFlushReturn{Success: success}); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + } +} + // Helper function to create instrument key func createInstrumentKey(meterName, name, kind, unit, description string) string { return fmt.Sprintf("%s,%s,%s,%s,%s", meterName, strings.ToLower(strings.TrimSpace(name)), kind, unit, description) diff --git a/utils/build/docker/nodejs/parametric/server.js b/utils/build/docker/nodejs/parametric/server.js index b43b4716934..317565397cf 100644 --- a/utils/build/docker/nodejs/parametric/server.js +++ b/utils/build/docker/nodejs/parametric/server.js @@ -86,6 +86,12 @@ async function waitForMetricsLifecycle (operation, seconds) { } } +function waitForMetricsCallback (operation, seconds) { + return waitForMetricsLifecycle(new Promise((resolve, reject) => { + operation(error => error ? reject(error) : resolve()) + }), seconds) +} + app.post('/trace/span/inject_headers', (req, res) => { const request = req.body; const span = spans[request.span_id] @@ -824,14 +830,14 @@ app.post('/metrics/otel/create_asynchronous_gauge', (req, res) => { app.post('/metrics/otel/force_flush', async (req, res) => { const meterProvider = metrics.getMeterProvider(); try { - if (typeof meterProvider.forceFlush === 'function') { - await waitForMetricsLifecycle(meterProvider.forceFlush(), req.body.seconds || 10) - return res.json({ success: true }); - } if (!req.body.public_only && meterProvider.reader) { await waitForMetricsLifecycle(meterProvider.reader.forceFlush(), req.body.seconds || 10) return res.json({ success: true }); } + if (typeof meterProvider.forceFlush === 'function') { + await waitForMetricsCallback(done => meterProvider.forceFlush(done), req.body.seconds || 10) + return res.json({ success: true }); + } } catch (error) { return res.json({ success: false, message: error.message }); } @@ -844,7 +850,7 @@ app.post('/metrics/otel/shutdown', async (req, res) => { return res.json({ success: false, message: 'Shutdown not supported' }); } try { - await waitForMetricsLifecycle(meterProvider.shutdown(), req.body.seconds || 10) + await waitForMetricsCallback(done => meterProvider.shutdown(done), req.body.seconds || 10) res.json({ success: true }); } catch (error) { res.json({ success: false, message: error.message }); diff --git a/utils/build/docker/ruby/parametric/server.rb b/utils/build/docker/ruby/parametric/server.rb index c8b8db3e3b7..7ea68cf6257 100644 --- a/utils/build/docker/ruby/parametric/server.rb +++ b/utils/build/docker/ruby/parametric/server.rb @@ -714,7 +714,11 @@ def to_json(*_args) end class OtelMetricsForceFlushArgs + attr_reader :seconds, :public_only + def initialize(params) + @seconds = params.fetch('seconds', 10) + @public_only = params.fetch('public_only', false) end end @@ -1030,6 +1034,8 @@ def call(env) handle_metrics_otel_create_asynchronous_gauge(req, res) when '/metrics/otel/force_flush' handle_metrics_otel_force_flush(req, res) + when '/metrics/otel/shutdown' + handle_metrics_otel_shutdown(req, res) when '/trace/crash' handle_trace_crash(req, res) when '/otel/logger/create' @@ -1589,11 +1595,28 @@ def handle_metrics_otel_force_flush(req, res) meter_provider = OpenTelemetry.meter_provider - if meter_provider.respond_to?(:force_flush) - meter_provider.force_flush + unless args.public_only + meter_provider.force_flush if meter_provider.respond_to?(:force_flush) + return res.write(OtelMetricsForceFlushReturn.new(true).to_json) end - res.write(OtelMetricsForceFlushReturn.new(true).to_json) + success = meter_provider.respond_to?(:force_flush) + result = meter_provider.force_flush(timeout: args.seconds) if success + success &&= result == OpenTelemetry::SDK::Metrics::Export::SUCCESS + res.write(OtelMetricsForceFlushReturn.new(success).to_json) + rescue + res.write(OtelMetricsForceFlushReturn.new(false).to_json) + end + + def handle_metrics_otel_shutdown(req, res) + args = OtelMetricsForceFlushArgs.new(JSON.parse(req.body.read)) + meter_provider = OpenTelemetry.meter_provider + success = meter_provider.respond_to?(:shutdown) + result = meter_provider.shutdown(timeout: args.seconds) if success + success &&= result == OpenTelemetry::SDK::Metrics::Export::SUCCESS + res.write(OtelMetricsForceFlushReturn.new(success).to_json) + rescue + res.write(OtelMetricsForceFlushReturn.new(false).to_json) end def handle_otel_logger_create(req, res) diff --git a/utils/build/docker/rust/parametric/src/opentelemetry/mod.rs b/utils/build/docker/rust/parametric/src/opentelemetry/mod.rs index 7d6c0384311..1f320254061 100644 --- a/utils/build/docker/rust/parametric/src/opentelemetry/mod.rs +++ b/utils/build/docker/rust/parametric/src/opentelemetry/mod.rs @@ -55,6 +55,7 @@ pub fn app() -> Router { .route("/metrics/otel/create_asynchronous_updowncounter", post(otel_create_asynchronous_updowncounter)) .route("/metrics/otel/create_asynchronous_gauge", post(otel_create_asynchronous_gauge)) .route("/metrics/otel/force_flush", post(otel_metrics_force_flush)) + .route("/metrics/otel/shutdown", post(otel_metrics_shutdown)) .route("/otel/logger/create", post(otel_create_logger)) .route("/otel/logger/write", post(otel_write_log)) } @@ -786,6 +787,19 @@ async fn otel_metrics_force_flush( Json(OtelMetricsForceFlushReturn { success: result }) } +async fn otel_metrics_shutdown( + State(state): State, + Json(_args): Json, +) -> Json { + let meter_provider_guard = state.meter_provider.lock().unwrap(); + let result = if let Some(meter_provider) = meter_provider_guard.as_ref() { + meter_provider.shutdown().is_ok() + } else { + false + }; + Json(OtelMetricsForceFlushReturn { success: result }) +} + // --- Logs Handlers --- async fn otel_create_logger( diff --git a/utils/docker_fixtures/_test_clients/_test_client_parametric.py b/utils/docker_fixtures/_test_clients/_test_client_parametric.py index db7131f61a2..3f2d304c97d 100644 --- a/utils/docker_fixtures/_test_clients/_test_client_parametric.py +++ b/utils/docker_fixtures/_test_clients/_test_client_parametric.py @@ -973,12 +973,16 @@ def otel_create_asynchronous_gauge( def otel_metrics_force_flush(self, seconds: int = 10, *, public_only: bool = False) -> bool: resp = self._session.post( - self._url("/metrics/otel/force_flush"), json={"seconds": seconds, "public_only": public_only} + self._url("/metrics/otel/force_flush"), + json={"seconds": seconds, "public_only": public_only}, + timeout=seconds + 1 if public_only else None, ).json() return resp["success"] def otel_metrics_shutdown(self, seconds: int = 10) -> bool: - resp = self._session.post(self._url("/metrics/otel/shutdown"), json={"seconds": seconds}).json() + resp = self._session.post( + self._url("/metrics/otel/shutdown"), json={"seconds": seconds}, timeout=seconds + 1 + ).json() return resp["success"] def llmobs_trace( From ee90797d85da0f5e7ff9109f34eb8c58673d973d Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Tue, 8 Sep 2026 15:29:31 -0400 Subject: [PATCH 3/4] test: narrow OTel metrics lifecycle coverage to shutdown --- tests/parametric/test_otel_metrics.py | 16 ----------- .../build/docker/golang/parametric/helpers.go | 7 +++++ .../docker/golang/parametric/otel_metrics.go | 12 ++++----- .../OpenTelemetryMetricsController.java | 27 +++++++------------ .../trace/opentelemetry/dto/FlushArgs.java | 4 +-- .../build/docker/nodejs/parametric/server.js | 19 +++++-------- .../parametric/apm_test_client/server.py | 11 +++----- utils/build/docker/ruby/parametric/server.rb | 22 +++++++-------- .../_test_clients/_test_client_parametric.py | 12 +++------ 9 files changed, 47 insertions(+), 83 deletions(-) diff --git a/tests/parametric/test_otel_metrics.py b/tests/parametric/test_otel_metrics.py index 038dfbc0394..87b3875958e 100644 --- a/tests/parametric/test_otel_metrics.py +++ b/tests/parametric/test_otel_metrics.py @@ -286,22 +286,6 @@ def assert_exported_once(metric_requests: list[dict], metric_name: str) -> None: attributes=DEFAULT_MEASUREMENT_ATTRIBUTES, ) - @pytest.mark.parametrize("library_env", [{**LIFECYCLE_ENVVARS}]) - def test_force_flush_exports_pending_metric_before_return( - self, test_agent: TestAgentAPI, test_library: APMLibrary, test_id: str - ) -> None: - metric_name = f"lifecycle-force-flush-{test_id}" - self.generate_pending_counter(test_library, metric_name) - assert find_metrics_by_name(test_agent.metrics(), metric_name) == [] - - try: - success = test_library.otel_metrics_force_flush(10, public_only=True) - finally: - test_library.terminate() - - assert success - self.assert_exported_once(test_agent.metrics(), metric_name) - @pytest.mark.parametrize("library_env", [{**LIFECYCLE_ENVVARS}]) def test_shutdown_exports_pending_metric_before_return( self, test_agent: TestAgentAPI, test_library: APMLibrary, test_id: str diff --git a/utils/build/docker/golang/parametric/helpers.go b/utils/build/docker/golang/parametric/helpers.go index 92e786ce3d7..78a6bb9288b 100644 --- a/utils/build/docker/golang/parametric/helpers.go +++ b/utils/build/docker/golang/parametric/helpers.go @@ -284,6 +284,9 @@ type OtelCreateAsynchronousGaugeArgs struct { } type OtelMetricsForceFlushArgs struct { +} + +type OtelMetricsShutdownArgs struct { Seconds int `json:"seconds"` } @@ -291,6 +294,10 @@ type OtelMetricsForceFlushReturn struct { Success bool `json:"success"` } +type OtelMetricsShutdownReturn struct { + Success bool `json:"success"` +} + func (a AttributeKeyVals) ConvertToAttributes() []attribute.KeyValue { var attrs []attribute.KeyValue for k, v := range a { diff --git a/utils/build/docker/golang/parametric/otel_metrics.go b/utils/build/docker/golang/parametric/otel_metrics.go index 662ca712a4e..5c61c4b2a85 100644 --- a/utils/build/docker/golang/parametric/otel_metrics.go +++ b/utils/build/docker/golang/parametric/otel_metrics.go @@ -570,7 +570,7 @@ func (s *apmClientServer) otelMetricsForceFlushHandler(w http.ResponseWriter, r return } - success := s.OtelMetricsForceFlush(args.Seconds) + success := s.OtelMetricsForceFlush() w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(&OtelMetricsForceFlushReturn{Success: success}); err != nil { @@ -578,10 +578,10 @@ func (s *apmClientServer) otelMetricsForceFlushHandler(w http.ResponseWriter, r } } -func (s *apmClientServer) OtelMetricsForceFlush(seconds int) bool { +func (s *apmClientServer) OtelMetricsForceFlush() bool { + // Use the dd-trace-go helper to flush metrics mp := otel.GetMeterProvider() - ctx, cancel := context.WithTimeout(context.Background(), time.Duration(seconds)*time.Second) - defer cancel() + ctx := context.Background() if err := ddmetric.ForceFlush(ctx, mp); err != nil { fmt.Printf("Error flushing metrics: %v\n", err) return false @@ -590,7 +590,7 @@ func (s *apmClientServer) OtelMetricsForceFlush(seconds int) bool { } func (s *apmClientServer) otelMetricsShutdownHandler(w http.ResponseWriter, r *http.Request) { - var args OtelMetricsForceFlushArgs + var args OtelMetricsShutdownArgs if err := json.NewDecoder(r.Body).Decode(&args); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -602,7 +602,7 @@ func (s *apmClientServer) otelMetricsShutdownHandler(w http.ResponseWriter, r *h success := ddmetric.Shutdown(ctx, mp) == nil w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(&OtelMetricsForceFlushReturn{Success: success}); err != nil { + if err := json.NewEncoder(w).Encode(&OtelMetricsShutdownReturn{Success: success}); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) } } diff --git a/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/controller/OpenTelemetryMetricsController.java b/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/controller/OpenTelemetryMetricsController.java index 32786c635b5..809f488191f 100644 --- a/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/controller/OpenTelemetryMetricsController.java +++ b/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/controller/OpenTelemetryMetricsController.java @@ -14,7 +14,6 @@ import java.util.Locale; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import org.springframework.web.bind.annotation.PostMapping; @@ -217,13 +216,6 @@ public void histogramRecord(@RequestBody HistogramRecordArgs args) { public FlushResult forceFlush(@RequestBody FlushArgs args) { LOGGER.info("Flushing OTel metrics: {}", args); try { - if (invokeLifecycle("forceFlush", args.seconds())) { - MetricsController.skipNextFlush(); - return new FlushResult(true); - } - if (args.publicOnly()) { - return new FlushResult(false); - } if (GlobalTracer.get() instanceof InternalTracer internalTracer) { internalTracer.flushMetrics(); // skip the next general metrics flush, as it's covered by the same call @@ -243,27 +235,28 @@ public FlushResult forceFlush(@RequestBody FlushArgs args) { public FlushResult shutdown(@RequestBody FlushArgs args) { LOGGER.info("Shutting down OTel metrics: {}", args); try { - return new FlushResult(invokeLifecycle("shutdown", args.seconds())); + return new FlushResult(invokeShutdown(args.seconds())); } catch (Exception e) { LOGGER.warn("Failed to shut down OTel metrics", e); return new FlushResult(false); } } - private static boolean invokeLifecycle(String methodName, long seconds) throws Exception { + private static boolean invokeShutdown(long seconds) throws Exception { Class lifecycleClass; - Method method; try { - lifecycleClass = Class.forName("datadog.trace.api.metrics.OpenTelemetryMetrics"); - method = lifecycleClass.getMethod(methodName); - } catch (ClassNotFoundException | NoSuchMethodException ignored) { + lifecycleClass = Class.forName("datadog.trace.api.metrics.DatadogMeterProvider"); + } catch (ClassNotFoundException ignored) { return false; } - Object result = method.invoke(null); - if (!(result instanceof Future future)) { + Object meterProvider = GlobalOpenTelemetry.get().getMeterProvider(); + if (!lifecycleClass.isInstance(meterProvider)) { return false; } - return Boolean.TRUE.equals(future.get(seconds, TimeUnit.SECONDS)); + Object result = lifecycleClass.getMethod("shutdown").invoke(meterProvider); + Method join = result.getClass().getMethod("join", long.class, TimeUnit.class); + join.invoke(result, seconds, TimeUnit.SECONDS); + return Boolean.TRUE.equals(result.getClass().getMethod("isSuccess").invoke(result)); } /** Builds {@link Attributes} from a map of strings. */ diff --git a/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/dto/FlushArgs.java b/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/dto/FlushArgs.java index c346d8f1cb4..89554a6099d 100644 --- a/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/dto/FlushArgs.java +++ b/utils/build/docker/java/parametric/src/main/java/com/datadoghq/trace/opentelemetry/dto/FlushArgs.java @@ -1,6 +1,4 @@ package com.datadoghq.trace.opentelemetry.dto; -import com.fasterxml.jackson.annotation.JsonProperty; - -public record FlushArgs(long seconds, @JsonProperty("public_only") boolean publicOnly) { +public record FlushArgs(long seconds) { } diff --git a/utils/build/docker/nodejs/parametric/server.js b/utils/build/docker/nodejs/parametric/server.js index 317565397cf..a94f3cfd8b5 100644 --- a/utils/build/docker/nodejs/parametric/server.js +++ b/utils/build/docker/nodejs/parametric/server.js @@ -827,21 +827,14 @@ app.post('/metrics/otel/create_asynchronous_gauge', (req, res) => { res.json({}); }); -app.post('/metrics/otel/force_flush', async (req, res) => { +app.post('/metrics/otel/force_flush', (req, res) => { const meterProvider = metrics.getMeterProvider(); - try { - if (!req.body.public_only && meterProvider.reader) { - await waitForMetricsLifecycle(meterProvider.reader.forceFlush(), req.body.seconds || 10) - return res.json({ success: true }); - } - if (typeof meterProvider.forceFlush === 'function') { - await waitForMetricsCallback(done => meterProvider.forceFlush(done), req.body.seconds || 10) - return res.json({ success: true }); - } - } catch (error) { - return res.json({ success: false, message: error.message }); + if (meterProvider.reader) { + meterProvider.reader.forceFlush() + res.json({ success: true }); + } else { + res.json({ success: false, message: 'Force flush not supported' }); } - res.json({ success: false, message: 'Force flush not supported' }); }); app.post('/metrics/otel/shutdown', async (req, res) => { diff --git a/utils/build/docker/python/parametric/apm_test_client/server.py b/utils/build/docker/python/parametric/apm_test_client/server.py index 5a907079aa2..ec96817952f 100644 --- a/utils/build/docker/python/parametric/apm_test_client/server.py +++ b/utils/build/docker/python/parametric/apm_test_client/server.py @@ -1292,8 +1292,7 @@ def otel_create_asynchronous_gauge(args: OtelCreateAsynchronousGaugeArgs): class OtelMetricsForceFlushArgs(BaseModel): - seconds: int = 10 - public_only: bool = False + pass class OtelMetricsForceFlushReturn(BaseModel): @@ -1310,13 +1309,9 @@ def otel_metrics_force_flush(args: OtelMetricsForceFlushArgs): # a default _ProxyMeterProvider provided by the API which does not # have the method. if hasattr(meter_provider, "force_flush"): - try: - result = meter_provider.force_flush(timeout_millis=args.seconds * 1000) - return OtelMetricsForceFlushReturn(success=result is not False) - except Exception: - return OtelMetricsForceFlushReturn(success=False) + meter_provider.force_flush() - return OtelMetricsForceFlushReturn(success=not args.public_only) + return OtelMetricsForceFlushReturn(success=True) class OtelMetricsShutdownArgs(BaseModel): diff --git a/utils/build/docker/ruby/parametric/server.rb b/utils/build/docker/ruby/parametric/server.rb index 7ea68cf6257..2c8052477e4 100644 --- a/utils/build/docker/ruby/parametric/server.rb +++ b/utils/build/docker/ruby/parametric/server.rb @@ -714,11 +714,15 @@ def to_json(*_args) end class OtelMetricsForceFlushArgs - attr_reader :seconds, :public_only + def initialize(params) + end +end + +class OtelMetricsShutdownArgs + attr_reader :seconds def initialize(params) @seconds = params.fetch('seconds', 10) - @public_only = params.fetch('public_only', false) end end @@ -1595,21 +1599,15 @@ def handle_metrics_otel_force_flush(req, res) meter_provider = OpenTelemetry.meter_provider - unless args.public_only - meter_provider.force_flush if meter_provider.respond_to?(:force_flush) - return res.write(OtelMetricsForceFlushReturn.new(true).to_json) + if meter_provider.respond_to?(:force_flush) + meter_provider.force_flush end - success = meter_provider.respond_to?(:force_flush) - result = meter_provider.force_flush(timeout: args.seconds) if success - success &&= result == OpenTelemetry::SDK::Metrics::Export::SUCCESS - res.write(OtelMetricsForceFlushReturn.new(success).to_json) - rescue - res.write(OtelMetricsForceFlushReturn.new(false).to_json) + res.write(OtelMetricsForceFlushReturn.new(true).to_json) end def handle_metrics_otel_shutdown(req, res) - args = OtelMetricsForceFlushArgs.new(JSON.parse(req.body.read)) + args = OtelMetricsShutdownArgs.new(JSON.parse(req.body.read)) meter_provider = OpenTelemetry.meter_provider success = meter_provider.respond_to?(:shutdown) result = meter_provider.shutdown(timeout: args.seconds) if success diff --git a/utils/docker_fixtures/_test_clients/_test_client_parametric.py b/utils/docker_fixtures/_test_clients/_test_client_parametric.py index 3f2d304c97d..8979def324f 100644 --- a/utils/docker_fixtures/_test_clients/_test_client_parametric.py +++ b/utils/docker_fixtures/_test_clients/_test_client_parametric.py @@ -971,12 +971,8 @@ def otel_create_asynchronous_gauge( }, ) - def otel_metrics_force_flush(self, seconds: int = 10, *, public_only: bool = False) -> bool: - resp = self._session.post( - self._url("/metrics/otel/force_flush"), - json={"seconds": seconds, "public_only": public_only}, - timeout=seconds + 1 if public_only else None, - ).json() + def otel_metrics_force_flush(self) -> bool: + resp = self._session.post(self._url("/metrics/otel/force_flush"), json={}).json() return resp["success"] def otel_metrics_shutdown(self, seconds: int = 10) -> bool: @@ -1227,8 +1223,8 @@ def otel_create_asynchronous_gauge( ) -> None: self._client.otel_create_asynchronous_gauge(meter_name, name, unit, description, value, attributes) - def otel_metrics_force_flush(self, seconds: int = 10, *, public_only: bool = False) -> bool: - return self._client.otel_metrics_force_flush(seconds, public_only=public_only) + def otel_metrics_force_flush(self) -> bool: + return self._client.otel_metrics_force_flush() def otel_metrics_shutdown(self, seconds: int = 10) -> bool: return self._client.otel_metrics_shutdown(seconds) From 9b14aed72c9104eba6903a22cdef7f06904afad5 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Tue, 8 Sep 2026 16:03:39 -0400 Subject: [PATCH 4/4] test: gate metrics shutdown on Java 1.67 --- manifests/java.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/manifests/java.yml b/manifests/java.yml index cf1a8996ba9..ba656fddd7a 100644 --- a/manifests/java.yml +++ b/manifests/java.yml @@ -3990,8 +3990,8 @@ manifest: tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Configuration_OTLP_Exporter_Metrics_Protocol::test_otlp_protocol_grpc: incomplete_test_app (GPRC fails for system-test but works with real collector) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Host_Name::test_hostname_from_dd_hostname: irrelevant (DD_HOSTNAME is only supported in Python) tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Lifecycle: - - declaration: missing_feature (Implemented in 1.66.0) - component_version: '<=1.66.0-SNAPSHOT' + - declaration: missing_feature (Implemented in 1.67.0) + component_version: '<=1.67.0-SNAPSHOT' tests/parametric/test_otel_metrics.py::Test_Otel_Metrics_Telemetry: missing_feature (not yet implemented) tests/parametric/test_otel_span_methods.py::Test_Otel_Span_Methods::test_otel_add_event_meta_serialization: - declaration: missing_feature (Not implemented)