Expose OpenTelemetry metrics shutdown through MeterProvider

dd-java-agent/agent-otel/otel-shim/src/main/java/datadog/opentelemetry/shim/metrics/OtelMeterProvider.java

@@ -1,5 +1,8 @@
 package datadog.opentelemetry.shim.metrics;
 
+import datadog.trace.api.CompletableResultCode;
+import datadog.trace.api.metrics.DatadogMeterProvider;
+import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
 import datadog.trace.bootstrap.otel.metrics.data.OtelMetricStorage;
 import datadog.trace.util.Strings;
@@ -15,7 +18,7 @@
 import org.slf4j.LoggerFactory;
 
 @ParametersAreNonnullByDefault
-public final class OtelMeterProvider implements MeterProvider {
+public final class OtelMeterProvider implements MeterProvider, DatadogMeterProvider {
   private static final Logger LOGGER = LoggerFactory.getLogger(OtelMeterProvider.class);
   private static final String DEFAULT_METER_NAME = "unknown";
 
@@ -43,6 +46,11 @@ public MeterBuilder meterBuilder(String instrumentationScopeName) {
     return new OtelMeterBuilder(this, instrumentationScopeName);
   }
 
+  @Override
+  public CompletableResultCode shutdown() {
+    return AgentTracer.get().shutdownOtelMetrics();
+  }
+
   OtelMeter getMeterShim(
       String instrumentationScopeName,
       @Nullable String instrumentationScopeVersion,

dd-java-agent/instrumentation/opentelemetry/opentelemetry-1.47/src/test/java/opentelemetry147/metrics/OpenTelemetryMetricsLifecycleForkedTest.java

@@ -0,0 +1,42 @@
+package opentelemetry147.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+import datadog.trace.agent.test.AbstractInstrumentationTest;
+import datadog.trace.api.CompletableResultCode;
+import datadog.trace.api.metrics.DatadogMeterProvider;
+import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
+import datadog.trace.test.junit.utils.config.WithConfig;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import java.lang.reflect.Proxy;
+import org.junit.jupiter.api.Test;
+
+@WithConfig(key = "metrics.otel.enabled", value = "true")
+class OpenTelemetryMetricsLifecycleForkedTest extends AbstractInstrumentationTest {
+
+  @Test
+  void globalMeterProviderExposesDatadogShutdown() {
+    DatadogMeterProvider meterProvider =
+        assertInstanceOf(DatadogMeterProvider.class, GlobalOpenTelemetry.get().getMeterProvider());
+    AgentTracer.TracerAPI originalAgentTracer = AgentTracer.get();
+    Object expected = new CompletableResultCode();
+    AgentTracer.TracerAPI replacementAgentTracer =
+        (AgentTracer.TracerAPI)
+            Proxy.newProxyInstance(
+                AgentTracer.TracerAPI.class.getClassLoader(),
+                new Class<?>[] {AgentTracer.TracerAPI.class},
+                (proxy, method, arguments) ->
+                    method.getName().equals("shutdownOtelMetrics") ? expected : null);
+
+    Object result;
+    try {
+      AgentTracer.forceRegister(replacementAgentTracer);
+      result = meterProvider.shutdown();
+    } finally {
+      AgentTracer.forceRegister(originalAgentTracer);
+    }
+
+    assertSame(expected, result);
+  }
+}

dd-trace-api/src/main/java/datadog/trace/api/CompletableResultCode.java

@@ -0,0 +1,205 @@
+package datadog.trace.api;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+/** Replacement for java.util.concurrent.CompletableFuture without the FJP side effects. */
+public final class CompletableResultCode {
+  private static final CompletableResultCode SUCCESS = new CompletableResultCode(true);
+  private static final CompletableResultCode FAILURE = new CompletableResultCode(false);
+
+  private final SharedState sharedState;
+  private final boolean resultView;
+
+  private volatile Boolean resultViewSuccess;
+  private List<Runnable> callbacks;
+
+  @SuppressFBWarnings(
+      value = "SING_SINGLETON_HAS_NONPRIVATE_CONSTRUCTOR",
+      justification = "Not a singleton")
+  public CompletableResultCode() {
+    this(new SharedState(), false);
+  }
+
+  private CompletableResultCode(SharedState sharedState, boolean resultView) {
+    this.sharedState = sharedState;
+    this.resultView = resultView;
+  }
+
+  private CompletableResultCode(boolean success) {
+    this();
+    complete(success);
+  }
+
+  public static CompletableResultCode ofSuccess() {
+    return SUCCESS;
+  }
+
+  public static CompletableResultCode ofFailure() {
+    return FAILURE;
+  }
+
+  /**
+   * Creates an independent view onto this result's outcome, to hand to a separate caller. A view
+   * observes this result's completion, or may complete first on its own, without either side
+   * holding a reference to the other.
+   *
+   * @return a new view sharing this result's outcome
+   */
+  public CompletableResultCode newResultView() {
+    return new CompletableResultCode(sharedState, true);
+  }
+
+  public CompletableResultCode succeed() {
+    return complete(true);
+  }
+
+  public CompletableResultCode fail() {
+    return complete(false);
+  }
+
+  public boolean isSuccess() {
+    return Boolean.TRUE.equals(outcome());
+  }
+
+  public boolean isDone() {
+    return outcome() != null;
+  }
+
+  /**
+   * Registers an action to run on the completing thread. If this result is already complete, the
+   * action runs immediately on the calling thread.
+   *
+   * @param callback action to run after completion
+   * @return this result
+   * @throws NullPointerException if {@code callback} is {@code null}
+   */
+  public CompletableResultCode whenComplete(Runnable callback) {
+    Objects.requireNonNull(callback, "callback");
+    if (outcome() == null) {
+      synchronized (sharedState) {
+        if (outcome() == null) {
+          if (callbacks == null) {
+            callbacks = new ArrayList<>();
+            if (sharedState.callbackResults == null) {
+              sharedState.callbackResults = new ArrayList<>();
+            }
+            sharedState.callbackResults.add(this);
+          }
+          callbacks.add(callback);
+          return this;
+        }
+      }
+    }
+    callback.run();
+    return this;
+  }
+
+  /**
+   * Waits up to the timeout for completion and returns this result. A timeout does not complete or
+   * cancel the operation; use {@link #isDone()} and {@link #isSuccess()} to inspect the outcome.
+   *
+   * @param timeout maximum time to wait
+   * @param unit unit of the timeout
+   * @return this result, which may still be incomplete after the timeout
+   */
+  public CompletableResultCode join(long timeout, TimeUnit unit) {
+    if (outcome() == null) {
+      synchronized (sharedState) {
+        if (outcome() == null) {
+          long remainingNanos = Objects.requireNonNull(unit, "unit").toNanos(timeout);
+          while (outcome() == null && remainingNanos > 0) {
+            long start = System.nanoTime();
+            try {
+              NANOSECONDS.timedWait(sharedState, remainingNanos);
+            } catch (InterruptedException ignored) {
+              Thread.currentThread().interrupt();
+              break;
+            }
+            remainingNanos -= Math.max(1, System.nanoTime() - start);
+          }
+        }
+      }
+    }
+    return this;
+  }
+
+  private CompletableResultCode complete(boolean succeeded) {
+    List<Runnable> completionCallbacks;
+    if (outcome() != null) {
+      return this;
+    }
+    synchronized (sharedState) {
+      if (outcome() != null) {
+        return this;
+      }
+
+      if (resultView) {
+        resultViewSuccess = succeeded;
+        completionCallbacks = callbacks;
+        callbacks = null;
+        removeCallbackResult();
+      } else {
+        sharedState.success = succeeded;
+        completionCallbacks = collectCallbacks();
+      }
+      sharedState.notifyAll();
+    }
+
+    Throwable firstFailure = null;
+    if (completionCallbacks != null) {
+      for (Runnable callback : completionCallbacks) {
+        try {
+          callback.run();
+        } catch (RuntimeException | Error failure) {
+          if (firstFailure == null) {
+            firstFailure = failure;
+          }
+        }
+      }
+    }
+    if (firstFailure instanceof RuntimeException) {
+      throw (RuntimeException) firstFailure;
+    }
+    if (firstFailure != null) {
+      throw (Error) firstFailure;
+    }
+    return this;
+  }
+
+  private Boolean outcome() {
+    return resultView && resultViewSuccess != null ? resultViewSuccess : sharedState.success;
+  }
+
+  private List<Runnable> collectCallbacks() {
+    if (sharedState.callbackResults == null) {
+      return null;
+    }
+    List<Runnable> completionCallbacks = new ArrayList<>();
+    for (CompletableResultCode result : sharedState.callbackResults) {
+      completionCallbacks.addAll(result.callbacks);
+      result.callbacks = null;
+    }
+    sharedState.callbackResults = null;
+    return completionCallbacks;
+  }
+
+  private void removeCallbackResult() {
+    if (sharedState.callbackResults != null) {
+      sharedState.callbackResults.remove(this);
+      if (sharedState.callbackResults.isEmpty()) {
+        sharedState.callbackResults = null;
+      }
+    }
+  }
+
+  private static final class SharedState {
+    private volatile Boolean success;
+    private List<CompletableResultCode> callbackResults;
+  }
+}

dd-trace-api/src/main/java/datadog/trace/api/metrics/DatadogMeterProvider.java

@@ -0,0 +1,20 @@
+package datadog.trace.api.metrics;
+
+import datadog.trace.api.CompletableResultCode;
+
+/**
+ * Datadog lifecycle controls implemented by the {@code MeterProvider} returned from {@code
+ * GlobalOpenTelemetry} when Datadog OpenTelemetry metrics support is enabled.
+ */
+public interface DatadogMeterProvider {
+
+  /**
+   * Performs a final export and stops Datadog's OpenTelemetry metrics pipeline. Repeated calls
+   * observe the first result.
+   *
+   * <p>A timed join bounds only the caller and does not cancel shutdown.
+   *
+   * @return the shutdown result; an unavailable or disabled pipeline succeeds as a no-op
+   */
+  CompletableResultCode shutdown();
+}