feat: (tracing) - Capturing otel via Zipkin for evaluating tracing and span capture

README.md

@@ -121,6 +121,58 @@ const state = await client.waitForWorkflowCompletion(instanceId);
 await harness.stop();
 ```
 
+### Capturing workflow traces
+
+Enable runtime telemetry capture before starting the harness:
+
+```typescript
+const harness = new WorkflowHarness().enableTelemetryCapture();
+try {
+  await harness.start();
+  const runtime = harness.createWorkflowRuntime();
+  runtime.registerWorkflow(myWorkflow);
+  runtime.registerActivity(myActivity);
+  await runtime.start();
+
+  const client = harness.createWorkflowClient();
+  const instanceId = await client.scheduleNewWorkflow(myWorkflow, "input");
+  await client.waitForWorkflowCompletion(instanceId);
+
+  const telemetry = harness.getTelemetryCollector()!;
+  const workflowSpan = await telemetry.waitForZipkinSpan((span) => span.name === "orchestration||myworkflow", 30_000);
+  const activitySpan = await telemetry.waitForZipkinSpan(
+    (span) => span.name === "activity||myactivity" && span.traceId === workflowSpan.traceId,
+    30_000
+  );
+  console.log(workflowSpan.traceId, activitySpan.parentId);
+  console.log(telemetry.getZipkinSpans());
+} finally {
+  await harness.stop();
+}
+```
+
+Alternatively, pass `{ telemetryCapture: true }` to `WorkflowHarness`, or use
+`new DaprContainer().withNetwork(network).withTelemetryCapture()` for any Dapr building block.
+Both the builder and started container expose `getTelemetryCollector()` (undefined when capture is disabled).
+
+Like the .NET testcontainer, `DaprTelemetryCollector` receives Dapr's OpenTelemetry-generated runtime spans
+through the **Zipkin v2 JSON exporter**, not OTLP. Capture is opt-in and sets sampling to `"1"`, replacing any
+configured Zipkin endpoint while preserving other configuration, including OTLP export and actor features.
+The receiver binds an available host port, makes it reachable through Testcontainers host-port forwarding,
+and stops with the sidecar, including when startup fails. No external tracing backend is needed.
+
+`getZipkinSpans()` returns independent snapshots with trace/span/parent IDs, names, kinds, endpoint metadata,
+tags, and microsecond timestamps/durations. Export is asynchronous, so use `waitForZipkinSpan(predicate, timeoutMs, signal?)`
+rather than inspecting immediately after a workflow finishes. Waits support `AbortSignal` cancellation and report
+captured span details on timeout. Zipkin exports span names in lowercase; workflow and activity names above
+correspond to functions named `myWorkflow` and `myActivity`. Filter by trace ID or instance tags when running multiple workflows.
+Snapshots remain available after shutdown; collection is in-memory and intended for bounded test runs.
+
+The collector can also be started independently with `await new DaprTelemetryCollector().start()` and disposed
+with `stop()` or `await using`. Its local endpoint is available via `getZipkinEndpointAddress()`.
+This captures **runtime-exported spans only**; it does not install a Node OpenTelemetry provider or emulate
+the .NET SDK's in-process `ActivitySource` listener.
+
 ## Dapr Actor Testing
 
 You can use `ActorHarness` or `.withActors()` on `DaprContainer` to test Dapr Actors backed by the Redis actor state store, placement service, and scheduler service:

src/DaprContainer.ts

@@ -22,10 +22,11 @@ import {
   StartedTestContainer,
   StopOptions,
   StoppedTestContainer,
+  TestContainers,
   Wait,
 } from "testcontainers";
 import { Component } from "./Component";
-import { Configuration } from "./Configuration";
+import { Configuration, TracingConfigurationSettings, ZipkinTracingConfigurationSettings } from "./Configuration";
 import {
   DAPR_RUNTIME_VERSION_ENV_VAR,
   DaprComponentNames,
@@ -37,6 +38,7 @@ import {
 } from "./Constants";
 import { DaprPlacementContainer } from "./DaprPlacementContainer";
 import { DaprSchedulerContainer } from "./DaprSchedulerContainer";
+import { DaprTelemetryCollector } from "./DaprTelemetryCollector";
 import { HttpEndpoint } from "./HttpEndpoint";
 import { MetadataEntry } from "./MetadataEntry";
 import { OLLAMA_DEFAULT_MODEL, OLLAMA_DEFAULT_PORT, OllamaContainer, StartedOllamaContainer } from "./OllamaContainer";
@@ -180,6 +182,8 @@ export class DaprContainer extends GenericContainer {
   private distributedLockOptions?: DistributedLockOptions;
   private startedNetwork?: StartedNetwork;
   private configuration?: Configuration;
+  private telemetryCollector?: DaprTelemetryCollector;
+  private telemetryConfigurationLocked = false;
   private components: Component[] = [];
   private subscriptions: Subscription[] = [];
   private httpEndpoints: HttpEndpoint[] = [];
@@ -339,12 +343,19 @@ export class DaprContainer extends GenericContainer {
       }
     }
 
+    this.telemetryConfigurationLocked = true;
     try {
       const containers = await Promise.all(startTasks);
-      return new StartedDaprContainer(await super.start(), containers);
+      if (this.telemetryCollector) {
+        await this.telemetryCollector.start();
+        await TestContainers.exposeHostPorts(this.telemetryCollector.getPort());
+      }
+      return new StartedDaprContainer(await super.start(), containers, this.telemetryCollector);
     } catch (error) {
       await Promise.allSettled(startTasks);
       await Promise.allSettled(startedContainers.map((container) => container.stop()));
+      await this.telemetryCollector?.stop();
+      this.telemetryConfigurationLocked = false;
       throw error;
     }
   }
@@ -361,6 +372,20 @@ export class DaprContainer extends GenericContainer {
       }
     }
 
+    const configuration = this.telemetryCollector
+      ? new Configuration(
+          this.configuration?.name ?? "telemetry",
+          new TracingConfigurationSettings(
+            "1",
+            this.configuration?.tracing?.stdout,
+            this.configuration?.tracing?.otel,
+            new ZipkinTracingConfigurationSettings(this.telemetryCollector.getZipkinEndpointAddressForDapr())
+          ),
+          this.configuration?.appHttpPipeline,
+          this.configuration?.features
+        )
+      : this.configuration;
+
     const cmds = [
       "./daprd",
       "--app-id",
@@ -394,21 +419,21 @@ export class DaprContainer extends GenericContainer {
       cmds.push("--enable-api-logging");
     }
 
-    if (this.configuration) {
-      cmds.push("--config", `/dapr-resources/${this.configuration.name}.yaml`);
+    if (configuration) {
+      cmds.push("--config", `/dapr-resources/${configuration.name}.yaml`);
     }
 
     log.info("> `daprd` Command: \n");
     log.info(`\t${JSON.stringify(cmds, undefined, 2)}\n`);
 
     this.withCommand(cmds);
 
-    if (this.configuration) {
-      const configurationYaml = this.configuration.toYaml();
+    if (configuration) {
+      const configurationYaml = configuration.toYaml();
       log.info("> Configuration YAML: \n");
       log.info(`\t\n${DaprContainer.redactYaml(configurationYaml)}\n`);
       this.withCopyContentToContainer([
-        { content: configurationYaml, target: `/dapr-resources/${this.configuration.name}.yaml` },
+        { content: configurationYaml, target: `/dapr-resources/${configuration.name}.yaml` },
       ]);
     }
 
@@ -703,6 +728,21 @@ export class DaprContainer extends GenericContainer {
     return this;
   }
 
+  /** Captures runtime spans with 100% sampling. Must be enabled before start(). */
+  withTelemetryCapture(): this {
+    if (!this.telemetryCollector) {
+      if (this.telemetryConfigurationLocked) {
+        throw new Error("Telemetry capture must be enabled before starting DaprContainer.");
+      }
+      this.telemetryCollector = new DaprTelemetryCollector();
+    }
+    return this;
+  }
+
+  getTelemetryCollector(): DaprTelemetryCollector | undefined {
+    return this.telemetryCollector;
+  }
+
   withPlacementService(placementService: string): this {
     this.placementService = placementService;
     return this;
@@ -949,15 +989,24 @@ export class DaprContainer extends GenericContainer {
 export class StartedDaprContainer extends AbstractStartedContainer {
   constructor(
     startedTestContainer: StartedTestContainer,
-    private readonly containers: StartedTestContainer[]
+    private readonly containers: StartedTestContainer[],
+    private readonly telemetryCollector?: DaprTelemetryCollector
   ) {
     super(startedTestContainer);
   }
 
   async stop(options?: Partial<StopOptions>): Promise<StoppedTestContainer> {
-    const stoppedTestContainer = await super.stop(options);
-    await Promise.all(this.containers.map((container) => container.stop(options)));
-    return stoppedTestContainer;
+    try {
+      const stoppedTestContainer = await super.stop(options);
+      await Promise.all(this.containers.map((container) => container.stop(options)));
+      return stoppedTestContainer;
+    } finally {
+      await this.telemetryCollector?.stop();
+    }
+  }
+
+  getTelemetryCollector(): DaprTelemetryCollector | undefined {
+    return this.telemetryCollector;
   }
 
   getHost(): string {

src/DaprTelemetryCollector.test.ts

@@ -0,0 +1,173 @@
+/*
+Copyright 2026 The Dapr Authors
+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 { DaprTelemetryCollector, ZipkinSpan } from "./DaprTelemetryCollector";
+
+const span: ZipkinSpan = {
+  traceId: "0123456789abcdef0123456789abcdef",
+  id: "0123456789abcdef",
+  parentId: "abcdef0123456789",
+  name: "workflow",
+  kind: "SERVER",
+  timestamp: 1790000000000000,
+  duration: 1234,
+  localEndpoint: { serviceName: "workflow-app", ipv4: "127.0.0.1", port: 3500 },
+  remoteEndpoint: { ipv6: "::1" },
+  tags: { "dapr.workflow.instance_id": "instance-1" },
+};
+
+async function post(collector: DaprTelemetryCollector, body: string): Promise<Response> {
+  return fetch(collector.getZipkinEndpointAddress(), {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body,
+  });
+}
+
+describe("DaprTelemetryCollector", () => {
+  it("captures multiple batches and returns independent snapshots", async () => {
+    await using collector = await new DaprTelemetryCollector().start();
+    expect(collector.getZipkinEndpointAddressForDapr()).toBe(
+      `http://host.testcontainers.internal:${collector.getPort()}/api/v2/spans`
+    );
+    expect((await post(collector, JSON.stringify([span]))).status).toBe(202);
+    expect((await post(collector, JSON.stringify([{ ...span, name: "activity" }]))).status).toBe(202);
+    expect(collector.getZipkinSpans()).toEqual([span, { ...span, name: "activity" }]);
+
+    const snapshot = collector.getZipkinSpans();
+    snapshot[0].tags = { changed: "true" };
+    snapshot[0].localEndpoint!.serviceName = "changed";
+    snapshot.pop();
+    expect(collector.getZipkinSpans()).toEqual([span, { ...span, name: "activity" }]);
+    await expect(collector.waitForZipkinSpan((item) => item.name === "activity", 0)).resolves.toMatchObject({
+      name: "activity",
+    });
+  });
+
+  it("waits for a future matching span, ignoring unrelated spans", async () => {
+    await using collector = await new DaprTelemetryCollector().start();
+    const waiting = collector.waitForZipkinSpan((item) => item.name === "activity", 2000);
+    await post(collector, JSON.stringify([span]));
+    await post(collector, JSON.stringify([{ ...span, name: "activity" }]));
+    await expect(waiting).resolves.toMatchObject({ name: "activity", traceId: span.traceId });
+  });
+
+  it("includes captured span details in timeout errors", async () => {
+    await using collector = await new DaprTelemetryCollector().start();
+    await post(collector, JSON.stringify([span]));
+    await expect(collector.waitForZipkinSpan(() => false, 20)).rejects.toThrow(
+      `Captured Zipkin spans: 1; spans: ${span.traceId}/workflow/SERVER`
+    );
+  });
+
+  it("supports cancellation before and during a wait", async () => {
+    await using collector = await new DaprTelemetryCollector().start();
+    const controller = new AbortController();
+    const waiting = collector.waitForZipkinSpan(() => false, 60_000, controller.signal);
+    const rejected = expect(waiting).rejects.toMatchObject({ name: "AbortError" });
+    controller.abort();
+    await rejected;
+    await expect(collector.waitForZipkinSpan(() => true, 60_000, controller.signal)).rejects.toMatchObject({
+      name: "AbortError",
+    });
+  });
+
+  it("propagates predicate failures", async () => {
+    await using collector = await new DaprTelemetryCollector().start();
+    await post(collector, JSON.stringify([span]));
+    await expect(
+      collector.waitForZipkinSpan(() => {
+        throw new Error("predicate failed");
+      }, 1000)
+    ).rejects.toThrow("predicate failed");
+  });
+
+  it.each([
+    "{",
+    "null",
+    "{}",
+    "[null]",
+    JSON.stringify([span, { id: "bad" }]),
+    JSON.stringify([{ ...span, traceId: "bad" }]),
+    JSON.stringify([{ ...span, tags: { number: 1 } }]),
+    JSON.stringify([{ ...span, localEndpoint: { port: -1 } }]),
+    JSON.stringify([{ ...span, duration: "123" }]),
+  ])("rejects invalid payloads atomically: %s", async (body) => {
+    await using collector = await new DaprTelemetryCollector().start();
+    const response = await post(collector, body);
+    expect(response.status).toBe(400);
+    expect(await response.text()).not.toBe("");
+    expect(collector.getZipkinSpans()).toEqual([]);
+    expect((await post(collector, JSON.stringify([span]))).status).toBe(202);
+  });
+
+  it("accepts empty batches and rejects unknown routes and methods", async () => {
+    await using collector = await new DaprTelemetryCollector().start();
+    expect((await post(collector, "[]")).status).toBe(202);
+    const response = await fetch(collector.getZipkinEndpointAddress());
+    expect(response.status).toBe(405);
+    expect(response.headers.get("allow")).toBe("POST");
+    expect((await fetch(`http://127.0.0.1:${collector.getPort()}/other`)).status).toBe(404);
+  });
+
+  it("bounds request size", async () => {
+    await using collector = await new DaprTelemetryCollector().start();
+    expect((await post(collector, " ".repeat(4 * 1024 * 1024 + 1))).status).toBe(413);
+    expect(collector.getZipkinSpans()).toEqual([]);
+  });
+
+  it("starts and stops idempotently, releases its port, and preserves captured spans", async () => {
+    const collector = new DaprTelemetryCollector();
+    expect(() => collector.getPort()).toThrow("has not been started");
+    await collector.stop();
+    await Promise.all([collector.start(), collector.start()]);
+    const port = collector.getPort();
+    expect(await collector.start()).toBe(collector);
+    expect(collector.getPort()).toBe(port);
+    await post(collector, JSON.stringify([span]));
+    await Promise.all([collector.stop(), collector.stop()]);
+    expect(() => collector.getPort()).toThrow("has not been started");
+    expect(collector.getZipkinSpans()).toEqual([span]);
+    await using replacement = await new DaprTelemetryCollector(port).start();
+    expect(replacement.getPort()).toBe(port);
+  });
+
+  it("surfaces bind failures and permits retrying", async () => {
+    const first = await new DaprTelemetryCollector().start();
+    await using second = new DaprTelemetryCollector(first.getPort());
+    try {
+      await expect(second.start()).rejects.toMatchObject({ code: "EADDRINUSE" });
+    } finally {
+      await first.stop();
+    }
+    await second.start();
+    expect((await post(second, "[]")).status).toBe(202);
+  });
+
+  it("terminates pending waits when stopped", async () => {
+    const collector = await new DaprTelemetryCollector().start();
+    const waiting = expect(collector.waitForZipkinSpan(() => false, 60_000)).rejects.toThrow("not running");
+    await collector.stop();
+    await waiting;
+  });
+
+  it.each([-1, 65536, 1.5, NaN, Infinity])("rejects invalid port %s", (port) => {
+    expect(() => new DaprTelemetryCollector(port)).toThrow("port must be an integer");
+  });
+
+  it.each([-1, NaN, Infinity])("rejects invalid timeout %s", async (timeout) => {
+    await expect(new DaprTelemetryCollector().waitForZipkinSpan(() => false, timeout)).rejects.toThrow(
+      "timeout must be a finite"
+    );
+  });
+});