refactor(appkit): let plugins contribute otel span processors

packages/appkit/package.json

@@ -73,8 +73,8 @@
     "@opentelemetry/resources": "2.8.0",
     "@opentelemetry/sdk-logs": "0.219.0",
     "@opentelemetry/sdk-metrics": "2.8.0",
-    "@opentelemetry/sdk-node": "0.219.0",
     "@opentelemetry/sdk-trace-base": "2.8.0",
+    "@opentelemetry/sdk-trace-node": "2.8.0",
     "@opentelemetry/semantic-conventions": "1.38.0",
     "@types/semver": "7.7.1",
     "apache-arrow": "21.1.0",

packages/appkit/src/core/appkit.ts

@@ -222,6 +222,12 @@ export class AppKit<TPlugins extends InputPluginMap> {
     const instance = new AppKit(mergedConfig);
 
     await Promise.all(instance.#setupPromises);
+
+    // Build the global tracer provider now that every plugin's setup() has run
+    // and contributed any span processors. Deferred to here so a single provider
+    // carries all processors (OTLP + plugin-contributed); see TelemetryManager.
+    TelemetryManager.start();
+
     await instance.#context.emitLifecycle("setup:complete");
 
     const handle = instance as unknown as PluginMap<T>;

packages/appkit/src/core/tests/appkit-as-user-exports.test.ts

@@ -44,6 +44,8 @@ vi.mock("../../telemetry", async () => {
     ...actual,
     TelemetryManager: {
       initialize: vi.fn(),
+      start: vi.fn(),
+      registerSpanProcessor: vi.fn(),
       getProvider: () => ({
         getTracer: () => ({
           startActiveSpan: vi.fn((_name: string, fn: (span: any) => any) =>

packages/appkit/src/telemetry/telemetry-manager.ts

@@ -1,3 +1,5 @@
+import { metrics } from "@opentelemetry/api";
+import { logs } from "@opentelemetry/api-logs";
 import { getNodeAutoInstrumentations } from "@opentelemetry/auto-instrumentations-node";
 import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-proto";
 import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-proto";
@@ -14,9 +16,19 @@ import {
   type Resource,
   resourceFromAttributes,
 } from "@opentelemetry/resources";
-import { BatchLogRecordProcessor } from "@opentelemetry/sdk-logs";
-import { PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
-import { NodeSDK } from "@opentelemetry/sdk-node";
+import {
+  BatchLogRecordProcessor,
+  LoggerProvider,
+} from "@opentelemetry/sdk-logs";
+import {
+  MeterProvider,
+  PeriodicExportingMetricReader,
+} from "@opentelemetry/sdk-metrics";
+import {
+  BatchSpanProcessor,
+  type SpanProcessor,
+} from "@opentelemetry/sdk-trace-base";
+import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
 import {
   ATTR_SERVICE_NAME,
   ATTR_SERVICE_VERSION,
@@ -29,12 +41,34 @@ import type { TelemetryConfig } from "./types";
 
 const logger = createLogger("telemetry");
 
+/**
+ * Owns the app's OpenTelemetry providers, split into two phases so plugins can
+ * contribute trace span processors before the tracer provider is built.
+ *
+ * - `initialize()` runs at app bootstrap, before plugin setup. It registers the
+ *   meter and logger providers eagerly, because OTel's metrics API has no lazy
+ *   proxy: a counter/histogram bound against the NoOp meter (as every connector
+ *   and the cache do in their constructors) stays NoOp for the process lifetime.
+ *   It does NOT register a tracer provider.
+ * - `registerSpanProcessor()` is called by plugins during `setup()` to add a
+ *   span processor (e.g. an MLflow exporter) to the not-yet-built tracer.
+ * - `start()` runs after all plugin `setup()` completes. It builds the single
+ *   global tracer provider with the OTLP processor (if configured) plus every
+ *   contributed processor. Deferring is safe for traces: OTel's ProxyTracer
+ *   rebinds tracers obtained before registration, and no span is emitted during
+ *   setup.
+ */
 export class TelemetryManager {
   private static readonly DEFAULT_EXPORT_INTERVAL_MS = 10000;
   private static readonly DEFAULT_FALLBACK_APP_NAME = "databricks-app";
 
   private static instance?: TelemetryManager;
-  private sdk?: NodeSDK;
+  private resource?: Resource;
+  private meterProvider?: MeterProvider;
+  private loggerProvider?: LoggerProvider;
+  private tracerProvider?: NodeTracerProvider;
+  private readonly spanProcessors: SpanProcessor[] = [];
+  private started = false;
   private shutdownPromise?: Promise<void>;
 
   /**
@@ -66,42 +100,122 @@ export class TelemetryManager {
     instance._initialize(config);
   }
 
+  /**
+   * Contribute a span processor to the not-yet-built tracer provider. Called by
+   * plugins during `setup()`. No-op with a warning once `start()` has run, since
+   * a started provider's processors are immutable in OTel JS 2.x.
+   */
+  static registerSpanProcessor(processor: SpanProcessor): void {
+    TelemetryManager.getInstance()._registerSpanProcessor(processor);
+  }
+
+  private _registerSpanProcessor(processor: SpanProcessor): void {
+    if (this.started) {
+      logger.warn(
+        "registerSpanProcessor called after start(); processor ignored. " +
+          "Contribute span processors during plugin setup().",
+      );
+      return;
+    }
+    this.spanProcessors.push(processor);
+  }
+
+  /**
+   * Phase 1: register the meter and logger providers eagerly (before plugin
+   * setup), so metric instruments bound in connector/cache constructors attach
+   * to real meters. The tracer provider is deferred to `start()`.
+   *
+   * When no OTLP endpoint is configured, meter/logger registration is skipped;
+   * a contributed span processor can still bring up tracing in `start()`.
+   */
   private _initialize(config: Partial<TelemetryConfig>): void {
-    if (this.sdk) return;
+    if (this.resource) return;
+    this.resource = this.createResource(config);
 
+    // OTLP exporters need an endpoint. Without one there is nothing to export
+    // metrics/logs to, so skip those providers — but still capture the resource
+    // and let `start()` bring up a tracer if a plugin contributed a processor.
     if (!process.env.OTEL_EXPORTER_OTLP_ENDPOINT) {
       return;
     }
 
     try {
-      this.sdk = new NodeSDK({
-        resource: this.createResource(config),
-        autoDetectResources: false,
-        sampler: new AppKitSampler(),
-        traceExporter: new OTLPTraceExporter({ headers: config.headers }),
-        metricReaders: [
+      this.meterProvider = new MeterProvider({
+        resource: this.resource,
+        readers: [
           new PeriodicExportingMetricReader({
             exporter: new OTLPMetricExporter({ headers: config.headers }),
             exportIntervalMillis:
               config.exportIntervalMs ||
               TelemetryManager.DEFAULT_EXPORT_INTERVAL_MS,
           }),
         ],
-        logRecordProcessors: [
+      });
+      metrics.setGlobalMeterProvider(this.meterProvider);
+
+      this.loggerProvider = new LoggerProvider({
+        resource: this.resource,
+        processors: [
           new BatchLogRecordProcessor(
             new OTLPLogExporter({ headers: config.headers }),
           ),
         ],
-        instrumentations: this.getDefaultInstrumentations(),
       });
+      logs.setGlobalLoggerProvider(this.loggerProvider);
+
+      // The OTLP trace exporter is the first span processor; contributed
+      // processors join it in `start()`.
+      this.spanProcessors.push(
+        new BatchSpanProcessor(
+          new OTLPTraceExporter({ headers: config.headers }),
+        ),
+      );
 
-      this.sdk.start();
-      logger.debug("Initialized successfully");
+      this.registerInstrumentations(this.getDefaultInstrumentations());
+      logger.debug("Meter/logger providers initialized");
     } catch (error) {
       logger.error("Failed to initialize: %O", error);
     }
   }
 
+  /**
+   * Phase 2: build and register the global tracer provider. Called by core
+   * after every plugin's `setup()` completes, so all contributed span
+   * processors are known. No-op when nothing needs tracing (no OTLP endpoint
+   * and no contributed processor), preserving "no telemetry unless configured".
+   *
+   * `NodeTracerProvider.register()` installs the async-hooks context manager and
+   * W3C propagators — the same wiring `NodeSDK.start()` did — so span nesting
+   * across awaits is preserved.
+   */
+  static start(): void {
+    TelemetryManager.getInstance()._start();
+  }
+
+  private _start(): void {
+    if (this.started) return;
+    this.started = true;
+
+    if (this.spanProcessors.length === 0) {
+      return;
+    }
+
+    try {
+      this.tracerProvider = new NodeTracerProvider({
+        resource: this.resource,
+        sampler: new AppKitSampler(),
+        spanProcessors: this.spanProcessors,
+      });
+      this.tracerProvider.register();
+      logger.debug(
+        "Tracer provider started with %d span processor(s)",
+        this.spanProcessors.length,
+      );
+    } catch (error) {
+      logger.error("Failed to start tracer provider: %O", error);
+    }
+  }
+
   /**
    * Register OpenTelemetry instrumentations.
    * Can be called at any time, but recommended to call in plugin constructor.
@@ -159,23 +273,34 @@ export class TelemetryManager {
   }
 
   /**
-   * Flush and shut down the OpenTelemetry SDK.
+   * Flush and shut down the tracer, meter, and logger providers.
    *
-   * Idempotent: the SDK reference is cleared synchronously and concurrent
+   * Idempotent: the provider references are cleared synchronously and concurrent
    * or repeated calls await the same in-flight flush. Awaited by the core
    * lifecycle manager during graceful shutdown — that manager owns the
    * process signal handlers, so telemetry no longer registers its own.
    */
   async shutdown(): Promise<void> {
-    if (this.sdk) {
-      const sdk = this.sdk;
-      this.sdk = undefined;
+    const providers = [
+      this.tracerProvider,
+      this.meterProvider,
+      this.loggerProvider,
+    ].filter((p): p is NonNullable<typeof p> => p !== undefined);
+
+    if (providers.length > 0) {
+      this.tracerProvider = undefined;
+      this.meterProvider = undefined;
+      this.loggerProvider = undefined;
       this.shutdownPromise = (async () => {
-        try {
-          await sdk.shutdown();
-        } catch (error) {
-          logger.error("Error shutting down: %O", error);
-        }
+        await Promise.all(
+          providers.map(async (provider) => {
+            try {
+              await provider.shutdown();
+            } catch (error) {
+              logger.error("Error shutting down: %O", error);
+            }
+          }),
+        );
       })();
     }