feat: add execution context attributes to telemetry spans

docs/docs/plugins/custom-plugins.md

@@ -130,7 +130,7 @@ This pattern allows:
 - **Shared services**:
  - **Cache management**: Access the cache service via `this.cache`. See [`CacheConfig`](../api/appkit/Interface.CacheConfig.md) for configuration.
  - **Telemetry**: Instrument your plugin with traces and metrics via `this.telemetry`. See [`ITelemetry`](../api/appkit/Interface.ITelemetry.md).
-- **Execution interceptors**: Use `execute()` and `executeStream()` with [`StreamExecutionSettings`](../api/appkit/Interface.StreamExecutionSettings.md)
+- **Execution interceptors**: Use `execute()` and `executeStream()` with [`StreamExecutionSettings`](../api/appkit/Interface.StreamExecutionSettings.md) for automatic caching, retry, timeout, and [telemetry span attributes](./execution-context.md#telemetry-span-attributes) (`execution.context`, `caller.id`)
 
 **Consuming your plugin programmatically**
 

docs/docs/plugins/execution-context.md

@@ -42,6 +42,18 @@ Exported from `@databricks/appkit`:
 - `getWarehouseId()`: `Promise<string>` (from `DATABRICKS_WAREHOUSE_ID` or auto-selected in dev)
 - `getWorkspaceId()`: `Promise<string>` (from `DATABRICKS_WORKSPACE_ID` or fetched)
 
+## Telemetry span attributes
+
+The `plugin.execute` span created by the execution interceptor chain includes these attributes:
+
+| Attribute | Type | Description |
+|-----------|------|-------------|
+| `execution.context` | `"user"` \| `"service"` | Whether the operation runs as a user (OBO) or service principal |
+| `caller.id` | `string` | The user ID (OBO) or service principal ID |
+| `execution.obo_dev_fallback` | `boolean` | Set to `true` when an OBO call falls back to service principal in development mode |
+
+These attributes are automatically added when your plugin uses `execute()` or `executeStream()`. All built-in plugins use these methods for their OBO operations. Custom plugins should do the same to get automatic telemetry instrumentation.
+
 ## Development mode behavior
 
-In local development (`NODE_ENV=development`), if `asUser(req)` is called without a user token, it logs a warning and skips user impersonation — the operation runs with the default credentials configured for the app instead.
+In local development (`NODE_ENV=development`), if `asUser(req)` is called without a user token, it logs a warning and skips user impersonation — the operation runs with the default credentials configured for the app instead. The telemetry span will show `execution.context: "service"` with `execution.obo_dev_fallback: true` to distinguish these from regular service principal calls.

packages/appkit/package.json

@@ -81,6 +81,7 @@
  "zod": "4.3.6"
  },
  "devDependencies": {
+ "@opentelemetry/context-async-hooks": "2.6.1",
  "@types/express": "4.17.25",
  "@types/json-schema": "7.0.15",
  "@types/pg": "8.16.0",

packages/appkit/src/context/execution-context.ts

@@ -81,3 +81,11 @@ export function getWarehouseId(): Promise<string> {
 export function getWorkspaceId(): Promise<string> {
  return getExecutionContext().workspaceId;
 }
+
+/**
+ * Check if currently running in a user context.
+ */
+export function isInUserContext(): boolean {
+ const ctx = executionContextStorage.getStore();
+ return ctx !== undefined;
+}

packages/appkit/src/plugin/interceptors/telemetry.ts

@@ -1,6 +1,11 @@
 import type { TelemetryConfig } from "shared";
+import {
+ getCurrentUserId,
+ isInUserContext,
+} from "../../context/execution-context";
 import type { ITelemetry, Span } from "../../telemetry";
 import { SpanStatusCode } from "../../telemetry";
+import { isDevOboFallback } from "../plugin";
 import type { ExecutionInterceptor, InterceptorContext } from "./types";
 
 export class TelemetryInterceptor implements ExecutionInterceptor {
@@ -45,6 +50,15 @@ export class TelemetryInterceptor implements ExecutionInterceptor {
  }
 
  try {
+ span.setAttribute(
+ "execution.context",
+ isInUserContext() ? "user" : "service",
+ );
+ span.setAttribute("caller.id", getCurrentUserId());
+ if (isDevOboFallback()) {
+ span.setAttribute("execution.obo_dev_fallback", true);
+ }
+
  const result = await fn();
  if (!isAborted) {
  span.setStatus({ code: SpanStatusCode.OK });

packages/appkit/src/plugin/plugin.ts

@@ -1,3 +1,4 @@
+import { createContextKey, context as otelContext } from "@opentelemetry/api";
 import type express from "express";
 import type {
  BasePlugin,
@@ -41,6 +42,20 @@ import type {
 
 const logger = createLogger("plugin");
 
+/**
+ * OTel context key for marking OBO dev mode fallback.
+ * Set when asUser() is called in development mode without a user token.
+ */
+const DEV_OBO_FALLBACK_KEY = createContextKey("appkit.devOboFallback");
+
+/**
+ * Returns true if the current execution is an OBO dev mode fallback
+ * (asUser() was called but fell back to service principal due to missing token).
+ */
+export function isDevOboFallback(): boolean {
+ return otelContext.active().getValue(DEV_OBO_FALLBACK_KEY) === true;
+}
+
 /**
  * Narrow an unknown thrown value to an Error that carries a numeric
  * `statusCode` property (e.g. `ApiError` from `@databricks/sdk-experimental`).
@@ -338,7 +353,23 @@ export abstract class Plugin<
  "asUser() called without user token in development mode. Skipping user impersonation.",
  );
 
- return this;
+ // Return a proxy that marks execution as OBO dev fallback via OTel context,
+ // so telemetry spans can distinguish intended OBO calls from regular SP calls
+ return new Proxy(this, {
+ get: (target, prop, receiver) => {
+ const value = Reflect.get(target, prop, receiver);
+ if (typeof value !== "function") return value;
+ if (typeof prop === "string" && EXCLUDED_FROM_PROXY.has(prop))
+ return value;
+
+ return (...args: unknown[]) => {
+ const ctx = otelContext
+ .active()
+ .setValue(DEV_OBO_FALLBACK_KEY, true);
+ return otelContext.with(ctx, () => value.apply(target, args));
+ };
+ },
+ }) as this;
  }
 
  if (!token) {
@@ -409,6 +440,9 @@ export abstract class Plugin<
  const effectiveUserKey = userKey ?? getCurrentUserId();
 
  const self = this;
+ // capture the active OTel context (HTTP span) before entering the async generator,
+ // where it would otherwise be lost across the async boundary
+ const parentOtelContext = otelContext.active();
 
  // wrapper function to ensure it returns a generator
  const asyncWrapperFn = async function* (streamSignal?: AbortSignal) {
@@ -428,11 +462,14 @@ export abstract class Plugin<
  return result;
  };
 
- // execute the function with interceptors
- const result = await self._executeWithInterceptors(
- wrappedFn as (signal?: AbortSignal) => Promise<T>,
- interceptors,
- context,
+ // execute the function with interceptors, restoring the parent OTel context
+ // so telemetry spans are linked as children of the HTTP request span
+ const result = await otelContext.with(parentOtelContext, () =>
+ self._executeWithInterceptors(
+ wrappedFn as (signal?: AbortSignal) => Promise<T>,
+ interceptors,
+ context,
+ ),
  );
 
  // check if result is a generator

packages/appkit/src/plugin/tests/plugin.test.ts

@@ -1,11 +1,26 @@
+import {
+ type ContextManager,
+ createContextKey,
+ context as otelContext,
+} from "@opentelemetry/api";
+import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks";
 import { createMockTelemetry, mockServiceContext } from "@tools/test-helpers";
 import type express from "express";
 import type {
  BasePluginConfig,
  IAppResponse,
  PluginExecuteConfig,
 } from "shared";
-import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+import {
+ afterAll,
+ afterEach,
+ beforeAll,
+ beforeEach,
+ describe,
+ expect,
+ test,
+ vi,
+} from "vitest";
 import { AppManager } from "../../app";
 import { CacheManager } from "../../cache";
 import { ServiceContext } from "../../context/service-context";
@@ -20,7 +35,7 @@ import { StreamManager } from "../../stream";
 import type { ITelemetry, TelemetryProvider } from "../../telemetry";
 import { TelemetryManager } from "../../telemetry";
 import type { InterceptorContext } from "../interceptors/types";
-import { Plugin } from "../plugin";
+import { isDevOboFallback, Plugin } from "../plugin";
 
 const { MockApiError } = vi.hoisted(() => {
  class MockApiError extends Error {
@@ -148,6 +163,20 @@ class PluginWithRoutes extends TestPlugin {
  }
 }
 
+class OboTestPlugin extends Plugin<BasePluginConfig> {
+ lastOboFallbackValue: boolean | undefined;
+
+ async captureOboFallback(): Promise<string> {
+ this.lastOboFallbackValue = isDevOboFallback();
+ return "captured";
+ }
+
+ syncCapture(): string {
+ this.lastOboFallbackValue = isDevOboFallback();
+ return "sync-captured";
+ }
+}
+
 describe("Plugin", () => {
  let mockTelemetry: ITelemetry;
  let mockCache: CacheManager;
@@ -916,4 +945,164 @@ describe("Plugin", () => {
  expect(result).toEqual({ ok: true, data: "integration-result" });
  });
  });
+
+ describe("asUser() dev fallback", () => {
+ let originalNodeEnv: string | undefined;
+ let contextManager: ContextManager;
+
+ beforeAll(() => {
+ otelContext.disable();
+ contextManager = new AsyncLocalStorageContextManager().enable();
+ otelContext.setGlobalContextManager(contextManager);
+ });
+
+ afterAll(() => {
+ otelContext.disable();
+ });
+
+ beforeEach(() => {
+ originalNodeEnv = process.env.NODE_ENV;
+ process.env.NODE_ENV = "development";
+ vi.useRealTimers();
+ });
+
+ afterEach(() => {
+ process.env.NODE_ENV = originalNodeEnv;
+ });
+
+ function createMockReqWithoutToken(): express.Request {
+ return {
+ header: vi.fn().mockReturnValue(undefined),
+ } as unknown as express.Request;
+ }
+
+ test("should return a Proxy (different reference) in dev mode without token", () => {
+ const plugin = new TestPlugin(config);
+ const proxied = plugin.asUser(createMockReqWithoutToken());
+
+ expect(proxied).not.toBe(plugin);
+ expect(proxied).toBeInstanceOf(TestPlugin);
+ });
+
+ test("should pass through non-function properties unchanged", () => {
+ const plugin = new TestPlugin(config);
+ const proxied = plugin.asUser(createMockReqWithoutToken());
+
+ expect(proxied.name).toBe(plugin.name);
+ });
+
+ test("should preserve return values from proxied async methods", async () => {
+ const plugin = new TestPlugin(config);
+ const proxied = plugin.asUser(createMockReqWithoutToken());
+
+ const result = await proxied.customMethod("value");
+ expect(result).toBe("processed-value");
+ });
+
+ test("should preserve return values from proxied sync methods", () => {
+ const plugin = new TestPlugin(config);
+ const proxied = plugin.asUser(createMockReqWithoutToken());
+
+ const result = proxied.syncMethod("value");
+ expect(result).toBe("sync-value");
+ });
+
+ test("should set isDevOboFallback() to true inside proxied method", async () => {
+ const plugin = new OboTestPlugin(config);
+ const proxied = plugin.asUser(createMockReqWithoutToken());
+
+ await proxied.captureOboFallback();
+
+ expect(plugin.lastOboFallbackValue).toBe(true);
+ });
+
+ test("should set isDevOboFallback() to true inside proxied sync method", () => {
+ const plugin = new OboTestPlugin(config);
+ const proxied = plugin.asUser(createMockReqWithoutToken());
+
+ proxied.syncCapture();
+
+ expect(plugin.lastOboFallbackValue).toBe(true);
+ });
+
+ test("should not set OBO fallback for excluded methods (setup)", async () => {
+ const plugin = new OboTestPlugin(config);
+ // Override setup to capture OBO fallback
+ plugin.setup = async () => {
+ plugin.lastOboFallbackValue = isDevOboFallback();
+ };
+
+ const proxied = plugin.asUser(createMockReqWithoutToken());
+ await proxied.setup();
+
+ expect(plugin.lastOboFallbackValue).toBe(false);
+ });
+
+ test("isDevOboFallback() should return false outside proxy context", () => {
+ expect(isDevOboFallback()).toBe(false);
+ });
+ });
+
+ describe("executeStream OTel context preservation", () => {
+ let contextManager: ContextManager;
+
+ beforeAll(() => {
+ otelContext.disable();
+ contextManager = new AsyncLocalStorageContextManager().enable();
+ otelContext.setGlobalContextManager(contextManager);
+ });
+
+ afterAll(() => {
+ otelContext.disable();
+ });
+
+ beforeEach(() => {
+ vi.useRealTimers();
+ });
+
+ test("should preserve parent OTel context inside async generator", async () => {
+ const plugin = new TestPlugin(config);
+ const mockResponse = {} as IAppResponse;
+
+ const TEST_KEY = createContextKey("test.parent.context");
+ const parentCtx = otelContext.active().setValue(TEST_KEY, "parent-value");
+
+ let capturedContextValue: unknown;
+
+ const mockFn = vi.fn().mockImplementation(async () => {
+ capturedContextValue = otelContext.active().getValue(TEST_KEY);
+ return "stream-result";
+ });
+
+ // Capture the generator function passed to streamManager.stream
+ let capturedGeneratorFn: any;
+ vi.mocked(mockStreamManager.stream).mockImplementation(
+ async (_res, genFn) => {
+ capturedGeneratorFn = genFn;
+ },
+ );
+
+ // Execute within the parent context
+ await otelContext.with(parentCtx, () =>
+ (plugin as any).executeStream(mockResponse, mockFn, {
+ default: {},
+ stream: {},
+ }),
+ );
+
+ // Invoke the captured generator OUTSIDE the parent context scope
+ // The generator should restore parentOtelContext internally
+ const gen = capturedGeneratorFn();
+ await gen.next();
+
+ expect(capturedContextValue).toBe("parent-value");
+ });
+
+ test("should not have parent context without the fix (baseline)", async () => {
+ const TEST_KEY = createContextKey("test.baseline.context");
+
+ // Outside any context, the value should not exist
+ expect(otelContext.active().getValue(TEST_KEY)).toBeUndefined();
+ });
+ });
 });

packages/appkit/src/plugins/analytics/analytics.ts

@@ -45,7 +45,10 @@ export class AnalyticsPlugin extends Plugin {
  }
 
  injectRoutes(router: IAppRouter) {
- // Service principal endpoints
+ // Arrow data downloads always run as service principal and bypass the
+ // interceptor chain (execute/executeStream). The original query execution
+ // handles OBO via executeStream(); this endpoint fetches pre-computed
+ // results by job ID.
  this.route(router, {
  name: "arrow",
  method: "get",