feat(server-utils): Rewrite @opentelemetry/instrumentation-dataloader to orchestrion

Replaces the `InstrumentationBase`-based OTel dataloader instrumentation with a
diagnostics-channel listener, with orchestrion injecting the channels into
`dataloader`'s constructor and prototype methods.

`dataloaderIntegration` is opt-in (never a default), so unlike prior migrations
the default-swap in `_init` can't apply (a user's explicit instance wins dedup).
Instead the `@sentry/node` factory switches to the channel version itself when
diagnostics-channel injection is enabled, keeping it opt-in in both modes.

Author: logaretm

dev-packages/node-integration-tests/suites/tracing/dataloader/test.ts

@@ -1,7 +1,11 @@
 import { afterAll, describe, expect } from 'vitest';
+import { isOrchestrionEnabled } from '../../../utils';
 import { cleanupChildProcesses, createEsmAndCjsTests } from '../../../utils/runner';
 
-const ORIGIN = 'auto.db.otel.dataloader';
+// The span origin depends on which instrumentation is active. When the generic orchestrion run is
+// enabled (via INJECT_ORCHESTRION) the OTel `Dataloader` integration is swapped for the
+// diagnostics-channel one, which stamps a different origin.
+const ORIGIN = isOrchestrionEnabled() ? 'auto.db.orchestrion.dataloader' : 'auto.db.otel.dataloader';
 const CACHE_GET_OP = 'cache.get';
 
 describe('dataloader auto-instrumentation', () => {

packages/node/src/integrations/tracing/dataloader/index.ts

@@ -1,13 +1,36 @@
-import { DataloaderInstrumentation } from './vendored/instrumentation';
-import type { IntegrationFn } from '@sentry/core';
+import type { Integration, IntegrationFn } from '@sentry/core';
 import { defineIntegration } from '@sentry/core';
 import { generateInstrumentOnce } from '@sentry/node-core';
+import {
+  isDiagnosticsChannelInjectionEnabled,
+  resolveDiagnosticsChannelInjection,
+} from '../../../sdk/diagnosticsChannelInjection';
+import { DataloaderInstrumentation } from './vendored/instrumentation';
 
 const INTEGRATION_NAME = 'Dataloader' as const;
 
 export const instrumentDataloader = generateInstrumentOnce(INTEGRATION_NAME, () => new DataloaderInstrumentation());
 
+// When the user has opted into diagnostics-channel injection, resolve the orchestrion channel
+// integration (kept behind the loader indirection so this file never imports orchestrion directly,
+// preserving the tree-shaking boundary). Returns `undefined` otherwise.
+function getDataloaderChannelIntegration(): Integration | undefined {
+  if (!isDiagnosticsChannelInjectionEnabled()) {
+    return undefined;
+  }
+
+  return resolveDiagnosticsChannelInjection()?.optInChannelIntegrations?.[INTEGRATION_NAME]?.();
+}
+
 const _dataloaderIntegration = (() => {
+  // Unlike the default auto-performance integrations (swapped in `_init`), `dataloaderIntegration` is
+  // opt-in: the user lists it explicitly, and an explicit instance wins integration dedup over a
+  // default one. So the swap to the orchestrion channel integration has to happen here in the factory.
+  const channelIntegration = getDataloaderChannelIntegration();
+  if (channelIntegration) {
+    return channelIntegration;
+  }
+
   return {
     name: INTEGRATION_NAME,
     setupOnce() {

packages/node/src/sdk/diagnosticsChannelInjection.ts

@@ -17,6 +17,13 @@ export interface DiagnosticsChannelInjection {
   integrations: Integration[] | readonly Integration[];
   /** OTel integration names these replace; filtered out of the default set. */
   replacedOtelIntegrationNames: string[];
+  /**
+   * Opt-in channel integrations keyed by their public (OTel-parity) factory name. Unlike
+   * `integrations`, these are NOT auto-appended to the default set: they replace OTel integrations
+   * that are themselves opt-in (e.g. `Dataloader`), so the matching `@sentry/node` factory looks its
+   * channel version up here when injection is enabled, keeping it opt-in in both modes.
+   */
+  optInChannelIntegrations?: Record<string, () => Integration>;
   /** Installs the module hooks that inject the diagnostics channels. */
   register: () => void;
   /** Warns (DEBUG only) about missing or doubled channel injection. */

packages/node/src/sdk/experimentalUseDiagnosticsChannelInjection.ts

@@ -1,5 +1,6 @@
 import {
   channelIntegrations,
+  dataloaderChannelIntegration,
   ioredisChannelIntegration,
   redisChannelIntegration,
   detectOrchestrionSetup,
@@ -67,6 +68,12 @@ export function experimentalUseDiagnosticsChannelInjection(
         redisChannelIntegration({ responseHook: cacheResponseHook }),
       ],
       replacedOtelIntegrationNames,
+      // Opt-in channel integrations, looked up by name from their `@sentry/node` factory rather than
+      // auto-appended. `Dataloader` is opt-in (never a default), so `dataloaderIntegration()` swaps
+      // to the channel version itself when injection is enabled.
+      optInChannelIntegrations: {
+        Dataloader: dataloaderChannelIntegration,
+      },
       register: () => registerDiagnosticsChannelInjection(options),
       detect: detectOrchestrionSetup,
     };

packages/node/test/integrations/tracing/dataloader-gating.test.ts

@@ -0,0 +1,60 @@
+import type { Integration } from '@sentry/core';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+const { instrumentCalls, injection } = vi.hoisted(() => ({
+  instrumentCalls: [] as string[],
+  injection: { enabled: false, channelIntegration: undefined as Integration | undefined },
+}));
+
+// Control the gating flag and the channel integration the loader hands back.
+vi.mock('../../../src/sdk/diagnosticsChannelInjection', () => ({
+  isDiagnosticsChannelInjectionEnabled: () => injection.enabled,
+  resolveDiagnosticsChannelInjection: () =>
+    injection.channelIntegration
+      ? { optInChannelIntegrations: { Dataloader: () => injection.channelIntegration } }
+      : undefined,
+}));
+
+// Record which instrumentations actually get generated, without registering real
+// OTel module hooks (the creator is never invoked).
+vi.mock('@sentry/node-core', async importOriginal => {
+  const actual = (await importOriginal()) as Record<string, unknown>;
+  return {
+    ...actual,
+    generateInstrumentOnce: (name: string) => Object.assign(() => instrumentCalls.push(name), { id: name }),
+  };
+});
+
+import { dataloaderIntegration } from '../../../src/integrations/tracing/dataloader';
+
+describe('dataloaderIntegration orchestrion gating', () => {
+  beforeEach(() => {
+    instrumentCalls.length = 0;
+    injection.enabled = false;
+    injection.channelIntegration = undefined;
+  });
+
+  it('uses the OTel instrumentation when diagnostics-channel injection is disabled', () => {
+    injection.enabled = false;
+
+    const integration = dataloaderIntegration();
+    expect(integration.name).toBe('Dataloader');
+
+    integration.setupOnce?.();
+    expect(instrumentCalls).toContain('Dataloader');
+  });
+
+  it('uses the orchestrion channel integration when diagnostics-channel injection is enabled', () => {
+    injection.enabled = true;
+    const channelIntegration: Integration = { name: 'Dataloader', setupOnce: vi.fn() };
+    injection.channelIntegration = channelIntegration;
+
+    const integration = dataloaderIntegration();
+    expect(integration).toBe(channelIntegration);
+
+    integration.setupOnce?.();
+    // The OTel instrumentation is never generated in this mode.
+    expect(instrumentCalls).not.toContain('Dataloader');
+    expect(channelIntegration.setupOnce).toHaveBeenCalledTimes(1);
+  });
+});

packages/server-utils/src/integrations/tracing-channel/dataloader.ts

@@ -0,0 +1,168 @@
+import * as diagnosticsChannel from 'node:diagnostics_channel';
+import type { IntegrationFn, Span, StartSpanOptions } from '@sentry/core';
+import {
+  debug,
+  defineIntegration,
+  getActiveSpan,
+  SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
+  SPAN_KIND,
+  startInactiveSpan,
+  startSpan,
+  waitForTracingChannelBinding,
+} from '@sentry/core';
+import { DEBUG_BUILD } from '../../debug-build';
+import type { ChannelName } from '../../orchestrion/channels';
+import { CHANNELS } from '../../orchestrion/channels';
+import { bindTracingChannelToSpan } from '../../tracing-channel';
+
+// NOTE: this uses the same name as the OTel integration by design.
+// When enabled, the OTel 'Dataloader' integration is omitted from the default set.
+const INTEGRATION_NAME = 'Dataloader' as const;
+
+const MODULE_NAME = 'dataloader';
+const ORIGIN = 'auto.db.orchestrion.dataloader';
+
+// `load`, `loadMany` and `batch` are cache reads; the rest are cache mutations that get no `op`.
+const CACHE_GET_OP = 'cache.get';
+
+type Operation = 'load' | 'loadMany' | 'batch' | 'prime' | 'clear' | 'clearAll';
+
+// The link shape shared between a `load` span and the `batch` span it triggers.
+type DataLoaderSpanLink = { context: ReturnType<Span['spanContext']> };
+
+// The private batch object `dataloader` stores on the loader. We stash the pending `load` span links
+// here (matching the vendored OTel instrumentation) so the batch span can link back to them.
+interface DataLoaderBatch {
+  spanLinks?: DataLoaderSpanLink[];
+}
+
+interface DataLoaderInstance {
+  name?: string | null;
+  _batch?: DataLoaderBatch | null;
+}
+
+/**
+ * The shape orchestrion's transform attaches to the tracing-channel `context`. Documented here rather
+ * than imported because orchestrion's runtime doesn't export it.
+ */
+interface DataLoaderChannelContext {
+  arguments: unknown[];
+  self?: DataLoaderInstance;
+  result?: unknown;
+  error?: unknown;
+}
+
+// Marks a wrapped `batchLoadFn` so a re-used loader (or a double construct) isn't wrapped twice.
+const WRAPPED = Symbol('sentry.dataloader.wrapped');
+
+function getSpanName(loader: DataLoaderInstance | undefined, operation: Operation): string {
+  const name = loader?.name;
+
+  return name ? `${MODULE_NAME}.${operation} ${name}` : `${MODULE_NAME}.${operation}`;
+}
+
+function makeSpanOptions(loader: DataLoaderInstance | undefined, operation: Operation): StartSpanOptions {
+  const isCacheGet = operation === 'load' || operation === 'loadMany' || operation === 'batch';
+
+  return {
+    name: getSpanName(loader, operation),
+    // The batch runs off a deferred tick and has no obvious network peer, so only `load`/`loadMany`
+    // get a client kind, matching the vendored OTel instrumentation.
+    ...(operation === 'batch' ? {} : { kind: SPAN_KIND.CLIENT }),
+    ...(isCacheGet ? { op: CACHE_GET_OP } : {}),
+    onlyIfParent: true,
+    attributes: {
+      [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: ORIGIN,
+    },
+  };
+}
+
+let subscribed = false;
+
+const _dataloaderChannelIntegration = (() => {
+  return {
+    name: INTEGRATION_NAME,
+    setupOnce() {
+      // `tracingChannel` is unavailable before Node 18.19 so do nothing in that case.
+      if (!diagnosticsChannel.tracingChannel || subscribed) {
+        return;
+      }
+      subscribed = true;
+
+      DEBUG_BUILD && debug.log('[orchestrion:dataloader] subscribing to dataloader tracing channels');
+
+      waitForTracingChannelBinding(() => {
+        subscribeConstruct();
+        subscribeLoad();
+        subscribeSimpleOperation(CHANNELS.DATALOADER_LOAD_MANY, 'loadMany');
+        subscribeSimpleOperation(CHANNELS.DATALOADER_PRIME, 'prime');
+        subscribeSimpleOperation(CHANNELS.DATALOADER_CLEAR, 'clear');
+        subscribeSimpleOperation(CHANNELS.DATALOADER_CLEAR_ALL, 'clearAll');
+      });
+    },
+  };
+}) satisfies IntegrationFn;
+
+/**
+ * Wraps the user's `batchLoadFn` (constructor arg 0) so the batch span opens when it runs on the
+ * deferred dispatch tick. The span links back to the `load` calls that populated the batch.
+ */
+function subscribeConstruct(): void {
+  diagnosticsChannel
+    .tracingChannel<DataLoaderChannelContext>(CHANNELS.DATALOADER_CONSTRUCT)
+    .start.subscribe(message => {
+      const data = message as DataLoaderChannelContext;
+      const batchLoadFn = data.arguments[0];
+      if (typeof batchLoadFn !== 'function' || (batchLoadFn as { [WRAPPED]?: boolean })[WRAPPED]) {
+        return;
+      }
+
+      const original = batchLoadFn as (...args: unknown[]) => unknown;
+      const wrapped = function (this: DataLoaderInstance, ...args: unknown[]): unknown {
+        return startSpan({ ...makeSpanOptions(this, 'batch'), links: this._batch?.spanLinks }, () =>
+          original.apply(this, args),
+        );
+      };
+      (wrapped as { [WRAPPED]?: boolean })[WRAPPED] = true;
+      data.arguments[0] = wrapped;
+    });
+}
+
+/**
+ * `load` is a cache read that additionally records its span so the batch it feeds into can link back.
+ * The link is recorded on span end (not creation) because `dataloader` only assigns the batch the
+ * span belongs to inside `load`'s body, after our `start` hook has already run.
+ */
+function subscribeLoad(): void {
+  bindTracingChannelToSpan(
+    diagnosticsChannel.tracingChannel<DataLoaderChannelContext>(CHANNELS.DATALOADER_LOAD),
+    data => (getActiveSpan() ? startInactiveSpanFor(data.self, 'load') : undefined),
+    {
+      beforeSpanEnd(span, data) {
+        const batch = data.self?._batch;
+        if (batch && span.isRecording()) {
+          (batch.spanLinks ??= []).push({ context: span.spanContext() });
+        }
+      },
+    },
+  );
+}
+
+function subscribeSimpleOperation(channelName: ChannelName, operation: Operation): void {
+  bindTracingChannelToSpan(diagnosticsChannel.tracingChannel<DataLoaderChannelContext>(channelName), data =>
+    getActiveSpan() ? startInactiveSpanFor(data.self, operation) : undefined,
+  );
+}
+
+function startInactiveSpanFor(loader: DataLoaderInstance | undefined, operation: Operation): Span {
+  return startInactiveSpan(makeSpanOptions(loader, operation));
+}
+
+/**
+ * EXPERIMENTAL: orchestrion-driven `dataloader` integration.
+ *
+ * Subscribes to the `orchestrion:dataloader:*` diagnostics_channels that the orchestrion code
+ * transform injects into `dataloader`'s constructor and prototype methods. Requires the orchestrion
+ * runtime hook or bundler plugin to be active.
+ */
+export const dataloaderChannelIntegration = defineIntegration(_dataloaderChannelIntegration);

packages/server-utils/src/orchestrion/config/dataloader.ts

@@ -1,6 +1,52 @@
 import type { InstrumentationConfig } from '@apm-js-collab/code-transformer';
 
-// TODO: Stub for the `dataloader` orchestrion integration (ports `@opentelemetry/instrumentation-dataloader`).
-export const dataloaderConfig: InstrumentationConfig[] = [];
+// `dataloader` ships a single transpiled CommonJS `index.js`. Its class methods are emitted as
+// `_proto.<name> = function <name>() {}` (named function *expressions*), so they match on
+// `expressionName` rather than `methodName`. The constructor is a named function declaration.
+// The version range mirrors `supportedVersions` in the vendored OTel instrumentation.
+const module = { name: 'dataloader', versionRange: '>=2.0.0 <3', filePath: 'index.js' } as const;
 
-export const dataloaderChannels = {} as const;
+export const dataloaderConfig = [
+  // Wrap the constructor so the subscriber can wrap the user's `batchLoadFn` (arg 0). The batch span
+  // is opened when that wrapped function actually runs (on the deferred dispatch tick), mirroring the
+  // vendored OTel instrumentation which also wraps `batchLoadFn` at construction time.
+  {
+    channelName: 'construct',
+    module,
+    functionQuery: { functionName: 'DataLoader', kind: 'Sync' },
+  },
+  {
+    channelName: 'load',
+    module,
+    functionQuery: { expressionName: 'load', kind: 'Sync' },
+  },
+  {
+    channelName: 'loadMany',
+    module,
+    functionQuery: { expressionName: 'loadMany', kind: 'Sync' },
+  },
+  {
+    channelName: 'prime',
+    module,
+    functionQuery: { expressionName: 'prime', kind: 'Sync' },
+  },
+  {
+    channelName: 'clear',
+    module,
+    functionQuery: { expressionName: 'clear', kind: 'Sync' },
+  },
+  {
+    channelName: 'clearAll',
+    module,
+    functionQuery: { expressionName: 'clearAll', kind: 'Sync' },
+  },
+] satisfies InstrumentationConfig[];
+
+export const dataloaderChannels = {
+  DATALOADER_CONSTRUCT: 'orchestrion:dataloader:construct',
+  DATALOADER_LOAD: 'orchestrion:dataloader:load',
+  DATALOADER_LOAD_MANY: 'orchestrion:dataloader:loadMany',
+  DATALOADER_PRIME: 'orchestrion:dataloader:prime',
+  DATALOADER_CLEAR: 'orchestrion:dataloader:clear',
+  DATALOADER_CLEAR_ALL: 'orchestrion:dataloader:clearAll',
+} as const;

packages/server-utils/src/orchestrion/index.ts

@@ -1,5 +1,6 @@
 import { amqplibChannelIntegration } from '../integrations/tracing-channel/amqplib';
 import { anthropicChannelIntegration } from '../integrations/tracing-channel/anthropic';
+import { dataloaderChannelIntegration } from '../integrations/tracing-channel/dataloader';
 import { googleGenAIChannelIntegration } from '../integrations/tracing-channel/google-genai';
 import {
   graphqlChannelIntegration,
@@ -20,6 +21,7 @@ export { detectOrchestrionSetup, isOrchestrionInjected } from './detect';
 export {
   amqplibChannelIntegration,
   anthropicChannelIntegration,
+  dataloaderChannelIntegration,
   googleGenAIChannelIntegration,
   graphqlChannelIntegration,
   hapiChannelIntegration,
@@ -54,6 +56,11 @@ export type * from '../integrations/tracing-channel/graphql/graphql-types';
  * NOTE: `ioredisChannelIntegration` and `redisChannelIntegration` are intentionally NOT here. They
  * only partially replace the composite OTel `Redis` integration and need the node SDK's redis cache
  * `responseHook` (which can't live in `server-utils`), so `@sentry/node` wires them up separately.
+ *
+ * NOTE: `dataloaderChannelIntegration` is also NOT here. Everything in this map is auto-appended to
+ * the default integrations, but the OTel `Dataloader` integration is opt-in (never a default), so
+ * `@sentry/node`'s `dataloaderIntegration()` factory swaps to the channel version itself when the
+ * user has opted into diagnostics-channel injection, keeping it opt-in in both modes.
  */
 export const channelIntegrations = {
   postgresIntegration: postgresChannelIntegration,