diff --git a/manifests/dotnet.yml b/manifests/dotnet.yml index 774815390e7..48e6e5ac079 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 (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 1041600d993..2faaabc4748 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: 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 f56d208458c..ba656fddd7a 100644 --- a/manifests/java.yml +++ b/manifests/java.yml @@ -3989,6 +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: + - 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) diff --git a/manifests/nodejs.yml b/manifests/nodejs.yml index 0a412f144a5..347471e5585 100644 --- a/manifests/nodejs.yml +++ b/manifests/nodejs.yml @@ -2417,6 +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: + - 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/tests/parametric/test_otel_metrics.py b/tests/parametric/test_otel_metrics.py index b4a075e9436..87b3875958e 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( @@ -228,6 +230,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 +253,56 @@ 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(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"], + "AGGREGATION_TEMPORALITY_DELTA", + is_monotonic=True, + value=42, + attributes=DEFAULT_MEASUREMENT_ATTRIBUTES, + ) + + @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: + 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.metrics(), metric_name) + + @scenarios.parametric @features.otel_metrics_api class Test_Otel_Metrics_Configuration_Enabled: diff --git a/utils/build/docker/golang/parametric/helpers.go b/utils/build/docker/golang/parametric/helpers.go index b4958d24186..78a6bb9288b 100644 --- a/utils/build/docker/golang/parametric/helpers.go +++ b/utils/build/docker/golang/parametric/helpers.go @@ -286,10 +286,18 @@ type OtelCreateAsynchronousGaugeArgs struct { type OtelMetricsForceFlushArgs struct { } +type OtelMetricsShutdownArgs struct { + Seconds int `json:"seconds"` +} + 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/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..5c61c4b2a85 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" @@ -588,6 +589,24 @@ func (s *apmClientServer) OtelMetricsForceFlush() bool { return true } +func (s *apmClientServer) otelMetricsShutdownHandler(w http.ResponseWriter, r *http.Request) { + var args OtelMetricsShutdownArgs + 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(&OtelMetricsShutdownReturn{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/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..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 @@ -10,9 +10,11 @@ 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.TimeUnit; import java.util.function.Consumer; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; @@ -229,6 +231,34 @@ 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(invokeShutdown(args.seconds())); + } catch (Exception e) { + LOGGER.warn("Failed to shut down OTel metrics", e); + return new FlushResult(false); + } + } + + private static boolean invokeShutdown(long seconds) throws Exception { + Class lifecycleClass; + try { + lifecycleClass = Class.forName("datadog.trace.api.metrics.DatadogMeterProvider"); + } catch (ClassNotFoundException ignored) { + return false; + } + Object meterProvider = GlobalOpenTelemetry.get().getMeterProvider(); + if (!lifecycleClass.isInstance(meterProvider)) { + return false; + } + 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. */ private static Attributes fromMap(Map map) { AttributesBuilder builder = Attributes.builder(); diff --git a/utils/build/docker/nodejs/parametric/server.js b/utils/build/docker/nodejs/parametric/server.js index 10fb172b8b5..a94f3cfd8b5 100644 --- a/utils/build/docker/nodejs/parametric/server.js +++ b/utils/build/docker/nodejs/parametric/server.js @@ -72,6 +72,26 @@ 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) + } +} + +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] @@ -817,6 +837,19 @@ app.post('/metrics/otel/force_flush', (req, res) => { } }); +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 waitForMetricsCallback(done => meterProvider.shutdown(done), req.body.seconds || 10) + res.json({ success: true }); + } catch (error) { + res.json({ success: false, message: error.message }); + } +}); + // add LLM Observability routes const addLlmObsRoutes = require('./llmobs'); addLlmObsRoutes(app); 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..ec96817952f 100644 --- a/utils/build/docker/python/parametric/apm_test_client/server.py +++ b/utils/build/docker/python/parametric/apm_test_client/server.py @@ -1314,6 +1314,27 @@ def otel_metrics_force_flush(args: OtelMetricsForceFlushArgs): return OtelMetricsForceFlushReturn(success=True) +class OtelMetricsShutdownArgs(BaseModel): + seconds: int = 10 + + +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): name: str level: str diff --git a/utils/build/docker/ruby/parametric/server.rb b/utils/build/docker/ruby/parametric/server.rb index c8b8db3e3b7..2c8052477e4 100644 --- a/utils/build/docker/ruby/parametric/server.rb +++ b/utils/build/docker/ruby/parametric/server.rb @@ -718,6 +718,14 @@ def initialize(params) end end +class OtelMetricsShutdownArgs + attr_reader :seconds + + def initialize(params) + @seconds = params.fetch('seconds', 10) + end +end + class OtelMetricsForceFlushReturn attr_accessor :success @@ -1030,6 +1038,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' @@ -1596,6 +1606,17 @@ def handle_metrics_otel_force_flush(req, res) res.write(OtelMetricsForceFlushReturn.new(true).to_json) end + def handle_metrics_otel_shutdown(req, res) + 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 + 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) args = LogCreateLoggerArgs.new(JSON.parse(req.body.read)) if OTEL_LOGGERS[args.name] 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 703b808ec75..8979def324f 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") @@ -971,6 +975,12 @@ 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: + resp = self._session.post( + self._url("/metrics/otel/shutdown"), json={"seconds": seconds}, timeout=seconds + 1 + ).json() + return resp["success"] + def llmobs_trace( self, trace_structure_request: SpanRequest | LlmObsAnnotationContextRequest, *, raise_on_error: bool = True ) -> dict | str | None: @@ -1216,6 +1226,12 @@ def otel_create_asynchronous_gauge( 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) + + def terminate(self) -> None: + self._client.terminate() + def is_alive(self) -> bool: return self._client.is_alive()