feat(js): add open telementry fetch provider

js/core/src/config.ts

@@ -15,6 +15,7 @@
  */
 
 import { logger } from './logging';
+import type { TelemetryProvider } from './tracing.js';
 
 const CONFIG_KEY = '__GENKIT_RUNTIME_CONFIG__';
 
@@ -34,6 +35,17 @@ export interface GenkitRuntimeConfig {
    * If true, features that require access to the file system or spawning processes (like the Reflection API) will be disabled.
    */
   sandboxedRuntime?: boolean;
+
+  /**
+   * Optional custom telemetry provider.
+   * When set, this provider is used instead of the default Node.js OpenTelemetry setup.
+   * Use this in worker-type environments (e.g. Cloudflare Workers) where Node SDK is not
+   * available; use a fetch-compatible provider such as {@link FetchTelemetryProvider}.
+   *
+   * Must be set before importing flows/actions (e.g. call setGenkitRuntimeConfig before
+   * any other genkit imports).
+   */
+  telemetry?: TelemetryProvider;
 }
 
 function getConfig(): GenkitRuntimeConfig {
@@ -64,6 +76,7 @@ export function getGenkitRuntimeConfig(): GenkitRuntimeConfig {
   return {
     jsonSchemaMode: config.jsonSchemaMode ?? 'compile',
     sandboxedRuntime: config.sandboxedRuntime ?? false,
+    telemetry: config.telemetry,
   };
 }
 

js/core/src/node.ts

@@ -14,10 +14,36 @@
  * limitations under the License.
  */
 
+import { getGenkitRuntimeConfig } from './config.js';
 import { initNodeAsyncContext } from './node-async-context.js';
+import {
+  setTelemetryProvider,
+  setTelemetryProviderInitializer,
+} from './tracing.js';
 import { initNodeTelemetryProvider } from './tracing/node-telemetry-provider.js';
 
+export { initNodeTelemetryProvider };
+
+/**
+ * Node-specific runtime setup (e.g. async_hooks for context propagation).
+ */
 export function initNodeFeatures() {
   initNodeAsyncContext();
+}
+
+/**
+ * Ensures a telemetry provider is set: uses config.telemetry if set,
+ * otherwise the default Node provider. Called lazily when telemetry is first needed.
+ */
+function ensureTelemetryProvider() {
+  const config = getGenkitRuntimeConfig();
+  if (config.telemetry) {
+    setTelemetryProvider(config.telemetry);
+    return;
+  }
   initNodeTelemetryProvider();
 }
+
+// Register so that any code path that loads core/node gets the initializer.
+// This ensures telemetry works even when reflection or other core code runs before the genkit package has finished loading.
+setTelemetryProviderInitializer(ensureTelemetryProvider);

js/core/src/reflection.ts

@@ -26,7 +26,7 @@ import { GENKIT_REFLECTION_API_SPEC_VERSION, GENKIT_VERSION } from './index.js';
 import { logger } from './logging.js';
 import type { Registry } from './registry.js';
 import { toJsonSchema } from './schema.js';
-import { flushTracing, setTelemetryServerUrl } from './tracing.js';
+import { flushTracing, setTelemetryServerConfig } from './tracing.js';
 
 // TODO: Move this to common location for schemas.
 export const RunActionResponseSchema = z.object({
@@ -373,7 +373,7 @@ export class ReflectionServer {
       const { telemetryServerUrl, reflectionApiSpecVersion } = request.body;
       if (!process.env.GENKIT_TELEMETRY_SERVER) {
         if (typeof telemetryServerUrl === 'string') {
-          setTelemetryServerUrl(telemetryServerUrl);
+          setTelemetryServerConfig({ url: telemetryServerUrl });
           logger.debug(
             `Connected to telemetry server on ${telemetryServerUrl}`
           );

js/core/src/tracing.ts

@@ -19,13 +19,26 @@ import { logger } from './logging.js';
 import type { TelemetryConfig } from './telemetryTypes.js';
 
 export * from './tracing/exporter.js';
+export * from './tracing/fetch-telemetry-provider.js';
 export * from './tracing/instrumentation.js';
 export * from './tracing/types.js';
 
 const oTelInitializationKey = '__GENKIT_DISABLE_GENKIT_OTEL_INITIALIZATION';
 const instrumentationKey = '__GENKIT_TELEMETRY_INSTRUMENTED';
 const telemetryProviderKey = '__GENKIT_TELEMETRY_PROVIDER';
 
+let telemetryProviderInitializer: (() => void) | null = null;
+
+/**
+ * Registers a function to run when the telemetry provider is first needed.
+ * This allows config (e.g. custom telemetry) to be set after importing genkit
+ * and still take effect. Called by the genkit package on load.
+ * @hidden
+ */
+export function setTelemetryProviderInitializer(fn: () => void) {
+  telemetryProviderInitializer = fn;
+}
+
 /**
  * @hidden
  */
@@ -81,6 +94,13 @@ function getTelemetryProvider(): TelemetryProvider {
   if (global[telemetryProviderKey]) {
     return global[telemetryProviderKey];
   }
+  if (telemetryProviderInitializer) {
+    telemetryProviderInitializer();
+
+    if (global[telemetryProviderKey]) {
+      return global[telemetryProviderKey];
+    }
+  }
   throw new GenkitError({
     status: 'FAILED_PRECONDITION',
     message: 'TelemetryProvider is not initialized.',

js/core/src/tracing/exporter.ts

@@ -27,12 +27,29 @@ import { deleteUndefinedProps } from '../utils.js';
 import type { SpanData, TraceData } from './types.js';
 
 export let telemetryServerUrl: string | undefined;
+export let telemetryServerHeaders: Record<string, string> | undefined;
 
 /**
+ * Configuration for the telemetry server used by TraceServerExporter.
  * @hidden
  */
-export function setTelemetryServerUrl(url: string) {
-  telemetryServerUrl = url;
+export interface TelemetryServerConfig {
+  url?: string;
+  headers?: Record<string, string>;
+}
+
+/**
+ * Set telemetry server URL and/or headers in one call.
+ * @hidden
+ */
+export function setTelemetryServerConfig(config: TelemetryServerConfig) {
+  if (config.url !== undefined) {
+    telemetryServerUrl = config.url;
+  }
+  if (config.headers !== undefined) {
+    telemetryServerHeaders =
+      Object.keys(config.headers).length > 0 ? config.headers : undefined;
+  }
 }
 
 /**
@@ -158,16 +175,19 @@ export class TraceServerExporter implements SpanExporter {
         data.endTime = convertedSpan.endTime;
       }
     }
+
+    const headers: Record<string, string> = {
+      Accept: 'application/json',
+      'Content-Type': 'application/json',
+      ...telemetryServerHeaders,
+    };
     // Suppress tracing to prevent infinite loops when auto-instrumentation
     // (e.g., undici) is enabled. Without this, the fetch call would be traced,
     // creating new spans that trigger more exports, causing stack overflow.
     await context.with(suppressTracing(context.active()), () =>
       fetch(`${telemetryServerUrl}/api/traces`, {
         method: 'POST',
-        headers: {
-          Accept: 'application/json',
-          'Content-Type': 'application/json',
-        },
+        headers,
         body: JSON.stringify(data),
       })
     );

js/core/src/tracing/fetch-telemetry-provider.ts

@@ -0,0 +1,138 @@
+/**
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { context, trace } from '@opentelemetry/api';
+import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks';
+import {
+  BasicTracerProvider,
+  BatchSpanProcessor,
+  SimpleSpanProcessor,
+  type SpanProcessor,
+} from '@opentelemetry/sdk-trace-base';
+import type { TelemetryConfig } from '../telemetryTypes.js';
+import type { TelemetryProvider } from '../tracing.js';
+import { TraceServerExporter, setTelemetryServerConfig } from './exporter.js';
+
+/**
+ * Options for the fetch-compatible telemetry provider.
+ * Uses only fetch and standard APIs (no Node.js SDK), suitable for workers.
+ */
+export interface FetchTelemetryProviderOptions {
+  /**
+   * URL of the Genkit telemetry server (e.g. from dev UI or your own collector).
+   * If not set, GENKIT_TELEMETRY_SERVER env var is used when enableTelemetry runs.
+   */
+  serverUrl?: string;
+  /**
+   * If true, use SimpleSpanProcessor for more real-time export (e.g. in dev).
+   * Default: true. Set to false for BatchSpanProcessor (production).
+   */
+  realtime?: boolean;
+  /**
+   * Optional headers to send with each trace export request (e.g. Authorization).
+   */
+  headers?: Record<string, string>;
+}
+
+/**
+ * Telemetry provider that uses only fetch-compatible APIs.
+ * Use this in worker environments (Cloudflare Workers, Vercel Edge, etc.) where
+ * the Node.js OpenTelemetry SDK is not available.
+ *
+ * When the runtime has AsyncLocalStorage (e.g. Cloudflare Workers with `nodejs_compat`),
+ * OTel context propagation is enabled so parent-child span links are preserved across
+ * async boundaries and traces are fully hierarchical.
+ *
+ * Set via runtime config before any other genkit imports:
+ *
+ * @example
+ * ```ts
+ * import { setGenkitRuntimeConfig } from 'genkit';
+ * import { FetchTelemetryProvider } from 'genkit/tracing';
+ *
+ * setGenkitRuntimeConfig({
+ *   jsonSchemaMode: 'interpret',
+ *   sandboxedRuntime: true,
+ *   telemetry: new FetchTelemetryProvider({
+ *     serverUrl: 'https://your-telemetry-server.example.com',
+ *   }),
+ * });
+ *
+ * // Then import flows/actions
+ * import { flow, run } from 'genkit';
+ * ```
+ */
+export class FetchTelemetryProvider implements TelemetryProvider {
+  private readonly options: FetchTelemetryProviderOptions;
+  private spanProcessors: SpanProcessor[] = [];
+
+  constructor(options: FetchTelemetryProviderOptions = {}) {
+    this.options = options;
+  }
+
+  async enableTelemetry(
+    telemetryConfig: TelemetryConfig | Promise<TelemetryConfig>
+  ): Promise<void> {
+    const config =
+      telemetryConfig instanceof Promise
+        ? await telemetryConfig
+        : telemetryConfig;
+    const serverUrl =
+      this.options.serverUrl ??
+      (typeof process !== 'undefined'
+        ? process.env?.GENKIT_TELEMETRY_SERVER
+        : undefined);
+    setTelemetryServerConfig({
+      ...(serverUrl && { url: serverUrl }),
+      ...(this.options.headers &&
+        Object.keys(this.options.headers).length > 0 && {
+          headers: this.options.headers,
+        }),
+    });
+
+    const exporter = new TraceServerExporter();
+    const processor: SpanProcessor =
+      this.options.realtime !== false
+        ? new SimpleSpanProcessor(exporter)
+        : new BatchSpanProcessor(exporter);
+
+    this.spanProcessors = [processor];
+
+    if (config?.spanProcessors?.length) {
+      this.spanProcessors.push(...config.spanProcessors);
+    }
+
+    // When AsyncLocalStorage is available (e.g. Node, Cloudflare Workers with nodejs_compat),
+    // enable OTel context propagation so parent-child span links work across await.
+    try {
+      const contextManager = new AsyncLocalStorageContextManager();
+      contextManager.enable();
+      context.setGlobalContextManager(contextManager);
+    } catch {
+      // No async_hooks / AsyncLocalStorage (e.g. some edge runtimes); spans still work, hierarchy may be flat.
+    }
+
+    const provider = new BasicTracerProvider();
+    for (const p of this.spanProcessors) {
+      provider.addSpanProcessor(p);
+    }
+    trace.setGlobalTracerProvider(provider);
+  }
+
+  async flushTracing(): Promise<void> {
+    await Promise.all(this.spanProcessors.map((p) => p.forceFlush()));
+  }
+}

js/core/src/tracing/node-telemetry-provider.ts

@@ -24,7 +24,7 @@ import { logger } from '../logging.js';
 import type { TelemetryConfig } from '../telemetryTypes.js';
 import { setTelemetryProvider } from '../tracing.js';
 import { isDevEnv } from '../utils.js';
-import { TraceServerExporter, setTelemetryServerUrl } from './exporter.js';
+import { TraceServerExporter, setTelemetryServerConfig } from './exporter.js';
 import { RealtimeSpanProcessor } from './realtime-span-processor.js';
 
 let telemetrySDK: NodeSDK | null = null;
@@ -44,7 +44,7 @@ async function enableTelemetry(
   telemetryConfig: TelemetryConfig | Promise<TelemetryConfig>
 ) {
   if (process.env.GENKIT_TELEMETRY_SERVER) {
-    setTelemetryServerUrl(process.env.GENKIT_TELEMETRY_SERVER);
+    setTelemetryServerConfig({ url: process.env.GENKIT_TELEMETRY_SERVER });
   }
 
   telemetryConfig =

js/core/tests/action_test.ts

@@ -18,10 +18,11 @@ import * as assert from 'assert';
 import { beforeEach, describe, it } from 'node:test';
 import { z } from 'zod';
 import { action, defineAction } from '../src/action.js';
-import { initNodeFeatures } from '../src/node.js';
+import { initNodeFeatures, initNodeTelemetryProvider } from '../src/node.js';
 import { Registry } from '../src/registry.js';
 
 initNodeFeatures();
+initNodeTelemetryProvider();
 
 describe('action', () => {
   var registry: Registry;

js/core/tests/dynamic-action-provider_test.ts

@@ -23,10 +23,11 @@ import {
   defineDynamicActionProvider,
   isDynamicActionProvider,
 } from '../src/dynamic-action-provider.js';
-import { initNodeFeatures } from '../src/node.js';
+import { initNodeFeatures, initNodeTelemetryProvider } from '../src/node.js';
 import { Registry } from '../src/registry.js';
 
 initNodeFeatures();
+initNodeTelemetryProvider();
 
 describe('dynamic action provider', () => {
   let registry: Registry;