diff --git a/README.md b/README.md index 0128918..ae7acca 100644 --- a/README.md +++ b/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: diff --git a/src/DaprContainer.ts b/src/DaprContainer.ts index a5c4068..f185b34 100644 --- a/src/DaprContainer.ts +++ b/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,8 +419,8 @@ 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"); @@ -403,12 +428,12 @@ export class DaprContainer extends GenericContainer { 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): Promise { - 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 { diff --git a/src/DaprTelemetryCollector.test.ts b/src/DaprTelemetryCollector.test.ts new file mode 100644 index 0000000..ab28d2d --- /dev/null +++ b/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 { + 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" + ); + }); +}); diff --git a/src/DaprTelemetryCollector.ts b/src/DaprTelemetryCollector.ts new file mode 100644 index 0000000..9526e1c --- /dev/null +++ b/src/DaprTelemetryCollector.ts @@ -0,0 +1,243 @@ +/* +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 { createServer, IncomingMessage, Server, ServerResponse } from "node:http"; +import { setTimeout as delay } from "node:timers/promises"; +import { log } from "testcontainers"; + +export type ZipkinEndpoint = { + serviceName?: string; + ipv4?: string; + ipv6?: string; + port?: number; +}; + +/** A Zipkin v2 span exported by daprd. Timestamps and durations are in microseconds. */ +export type ZipkinSpan = { + traceId: string; + id: string; + parentId?: string; + name?: string; + kind?: string; + timestamp?: number; + duration?: number; + localEndpoint?: ZipkinEndpoint; + remoteEndpoint?: ZipkinEndpoint; + tags?: Record; +}; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function isEndpoint(value: unknown): value is ZipkinEndpoint { + return ( + isRecord(value) && + ["serviceName", "ipv4", "ipv6"].every((key) => value[key] === undefined || typeof value[key] === "string") && + (value.port === undefined || + (typeof value.port === "number" && Number.isInteger(value.port) && value.port >= 0 && value.port <= 65535)) + ); +} + +function isZipkinSpan(value: unknown): value is ZipkinSpan { + return ( + isRecord(value) && + typeof value.traceId === "string" && + /^(?:[0-9a-f]{16}|[0-9a-f]{32})$/i.test(value.traceId) && + typeof value.id === "string" && + /^[0-9a-f]{16}$/i.test(value.id) && + (value.parentId === undefined || (typeof value.parentId === "string" && /^[0-9a-f]{16}$/i.test(value.parentId))) && + ["name", "kind"].every((key) => value[key] === undefined || typeof value[key] === "string") && + ["timestamp", "duration"].every( + (key) => + value[key] === undefined || + (typeof value[key] === "number" && Number.isSafeInteger(value[key]) && value[key] >= 0) + ) && + (value.localEndpoint === undefined || isEndpoint(value.localEndpoint)) && + (value.remoteEndpoint === undefined || isEndpoint(value.remoteEndpoint)) && + (value.tags === undefined || + (isRecord(value.tags) && Object.values(value.tags).every((tag) => typeof tag === "string"))) + ); +} + +/** + * In-process Zipkin v2 receiver for Dapr runtime tracing. No external collector is required. + * Host ports are exposed to Docker automatically when used with DaprContainer.withTelemetryCapture(). + */ +export class DaprTelemetryCollector { + private readonly spans: ZipkinSpan[] = []; + private server?: Server; + private starting?: Promise; + private stopping?: Promise; + + /** Use port 0 (the default) to bind an available port atomically. */ + constructor(private readonly port = 0) { + if (!Number.isInteger(port) || port < 0 || port > 65535) { + throw new Error("Telemetry collector port must be an integer between 0 and 65535."); + } + } + + public getPort(): number { + const address = this.server?.address(); + if (!address || typeof address === "string") { + throw new Error("DaprTelemetryCollector has not been started. Call start() first."); + } + return address.port; + } + + public getZipkinEndpointAddress(): string { + return `http://127.0.0.1:${this.getPort()}/api/v2/spans`; + } + + public getZipkinEndpointAddressForDapr(): string { + return `http://host.testcontainers.internal:${this.getPort()}/api/v2/spans`; + } + + /** Returns a snapshot that can be filtered or modified without changing captured spans. */ + public getZipkinSpans(): ZipkinSpan[] { + return structuredClone(this.spans); + } + + public async start(): Promise { + await this.stopping; + if (this.starting) { + return this.starting; + } + if (this.server?.listening) { + return this; + } + const server = createServer((request, response) => { + void this.receive(request, response).catch((error: unknown) => { + log.error(`Telemetry collector request failed: ${String(error)}`); + if (!response.destroyed) { + response.writeHead(500).end("Telemetry collector request failed."); + } + }); + }); + this.server = server; + this.starting = new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(this.port, "0.0.0.0", () => { + server.removeListener("error", reject); + resolve(this); + }); + }); + try { + return await this.starting; + } catch (error) { + this.server = undefined; + throw error; + } finally { + this.starting = undefined; + } + } + + public async stop(): Promise { + await this.starting; + if (this.stopping) { + return this.stopping; + } + const server = this.server; + if (!server) { + return; + } + this.server = undefined; + this.stopping = new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + server.closeAllConnections(); + }); + try { + await this.stopping; + } finally { + this.stopping = undefined; + } + } + + /** Waits for an existing or future matching span; timeoutMs is in milliseconds. */ + public async waitForZipkinSpan( + predicate: (span: ZipkinSpan) => boolean, + timeoutMs: number, + signal?: AbortSignal + ): Promise { + if (!Number.isFinite(timeoutMs) || timeoutMs < 0) { + throw new Error("Telemetry timeout must be a finite, non-negative number of milliseconds."); + } + const deadline = performance.now() + timeoutMs; + for (;;) { + signal?.throwIfAborted(); + const span = this.getZipkinSpans().find(predicate); + if (span) { + return span; + } + const remaining = deadline - performance.now(); + if (remaining <= 0) { + const diagnostics = this.spans + .slice(0, 10) + .map((item) => `${item.traceId}/${item.name ?? ""}/${item.kind ?? ""}`) + .join(", "); + throw new Error( + `Telemetry condition was not satisfied within ${timeoutMs}ms. ` + + `Captured Zipkin spans: ${this.spans.length}; spans: ${diagnostics}` + ); + } + if (!this.server?.listening) { + throw new Error("Telemetry collector is not running."); + } + await delay(Math.min(100, remaining), undefined, { signal }); + } + } + + public async [Symbol.asyncDispose](): Promise { + await this.stop(); + } + + private async receive(request: IncomingMessage, response: ServerResponse): Promise { + if (request.url !== "/api/v2/spans") { + response.writeHead(404).end("Not found."); + return; + } + if (request.method !== "POST") { + response.writeHead(405, { Allow: "POST" }).end("Use POST to export spans."); + return; + } + const chunks: Buffer[] = []; + let size = 0; + for await (const chunk of request) { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + size += buffer.length; + if (size > 4 * 1024 * 1024) { + response.writeHead(413).end("Telemetry payload exceeds 4 MiB."); + return; + } + chunks.push(buffer); + } + let payload: unknown; + try { + payload = JSON.parse(Buffer.concat(chunks).toString("utf8")); + } catch (error) { + if (!(error instanceof SyntaxError)) { + throw error; + } + response.writeHead(400).end("Invalid Zipkin JSON."); + return; + } + if (!Array.isArray(payload) || !payload.every(isZipkinSpan)) { + response.writeHead(400).end("Expected an array of Zipkin v2 spans."); + return; + } + for (const span of payload) { + this.spans.push(span); + } + response.writeHead(202).end(); + } +} diff --git a/src/TelemetryCapture.test.ts b/src/TelemetryCapture.test.ts new file mode 100644 index 0000000..1254670 --- /dev/null +++ b/src/TelemetryCapture.test.ts @@ -0,0 +1,143 @@ +/* +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 YAML from "yaml"; +import { Network, TestContainers } from "testcontainers"; +import { + AppHttpPipeline, + Configuration, + OtelTracingConfigurationSettings, + TracingConfigurationSettings, + ZipkinTracingConfigurationSettings, +} from "./Configuration"; +import { DaprContainer } from "./DaprContainer"; +import { DaprPlacementContainer } from "./DaprPlacementContainer"; +import { DaprSchedulerContainer } from "./DaprSchedulerContainer"; +import { WorkflowHarness } from "./WorkflowHarness"; + +class PreparedDaprContainer extends DaprContainer { + public prepare(): Promise { + return this.beforeContainerCreated(); + } +} + +describe("Telemetry capture configuration", () => { + it("is opt-in and enabling it repeatedly preserves the collector", () => { + const harness = new WorkflowHarness(); + expect(harness.getTelemetryCollector()).toBeUndefined(); + expect(harness.enableTelemetryCapture()).toBe(harness); + const collector = harness.getTelemetryCollector(); + expect(collector).toBeDefined(); + expect(harness.enableTelemetryCapture().getTelemetryCollector()).toBe(collector); + expect(harness.getDaprContainer().getTelemetryCollector()).toBe(collector); + expect(new WorkflowHarness({ telemetryCapture: true }).getTelemetryCollector()).toBeDefined(); + }); + + it.each([false, true])("preserves custom configuration with capture enabled first: %s", async (captureFirst) => { + const dapr = new PreparedDaprContainer() + .withPlacementContainer(new DaprPlacementContainer()) + .withSchedulerContainer(new DaprSchedulerContainer()); + const original = new Configuration( + "custom", + new TracingConfigurationSettings( + "0.25", + true, + new OtelTracingConfigurationSettings("otel:4317", false, "grpc"), + new ZipkinTracingConfigurationSettings("http://original/api/v2/spans") + ), + new AppHttpPipeline([{ name: "middleware", type: "middleware.http.uppercase" }]), + [{ name: "ActorStateTTL", enabled: true }] + ); + if (captureFirst) { + dapr.withTelemetryCapture(); + } + dapr.withConfiguration(original).withTelemetryCapture(); + await using collector = await dapr.getTelemetryCollector()!.start(); + const copy = jest.spyOn(dapr, "withCopyContentToContainer"); + const command = jest.spyOn(dapr, "withCommand"); + await dapr.prepare(); + const configurationCopy = copy.mock.calls + .flatMap(([contents]) => contents) + .find((entry) => entry.target === "/dapr-resources/custom.yaml"); + expect(configurationCopy).toBeDefined(); + const config = YAML.parse(configurationCopy!.content.toString()); + expect(config.spec).toEqual({ + tracing: { + samplingRate: "1", + stdout: true, + otel: original.tracing?.otel, + zipkin: { endpointAddress: collector.getZipkinEndpointAddressForDapr() }, + }, + appHttpPipeline: original.appHttpPipeline, + features: original.features, + }); + expect(command).toHaveBeenCalledWith(expect.arrayContaining(["--config", "/dapr-resources/custom.yaml"])); + expect(dapr.getConfiguration()).toBe(original); + expect(original.tracing?.samplingRate).toBe("0.25"); + expect(original.tracing?.zipkin?.endpointAddress).toBe("http://original/api/v2/spans"); + }); + + it("preserves implicit actor features", async () => { + const dapr = new PreparedDaprContainer() + .withPlacementContainer(new DaprPlacementContainer()) + .withSchedulerContainer(new DaprSchedulerContainer()) + .withActors() + .withTelemetryCapture(); + await using _collector = await dapr.getTelemetryCollector()!.start(); + const copy = jest.spyOn(dapr, "withCopyContentToContainer"); + await dapr.prepare(); + const config = copy.mock.calls + .flatMap(([contents]) => contents) + .find((entry) => entry.target === "/dapr-resources/actorConfig.yaml"); + expect(YAML.parse(config!.content.toString()).spec.features).toEqual([{ name: "ActorStateTTL", enabled: true }]); + }); + + it("captures runtime spans with propagated trace IDs and stops the receiver with the sidecar", async () => { + await using network = await new Network().start(); + const dapr = new DaprContainer().withNetwork(network).withTelemetryCapture(); + const started = await dapr.start(); + const collector = started.getTelemetryCollector()!; + try { + expect(collector).toBe(dapr.getTelemetryCollector()); + const response = await fetch(`${started.getHttpEndpoint()}/v1.0/state/kvstore/key`, { + headers: { traceparent: "00-0123456789abcdef0123456789abcdef-abcdef0123456789-01" }, + }); + expect(response.ok).toBe(true); + const span = await collector.waitForZipkinSpan( + (item) => item.traceId === "0123456789abcdef0123456789abcdef", + 30_000 + ); + expect(span.parentId).toBe("abcdef0123456789"); + expect(span.localEndpoint?.serviceName).toBe("dapr-app"); + expect(span.id).toMatch(/^[0-9a-f]{16}$/); + expect(span.duration).toBeGreaterThanOrEqual(0); + } finally { + await started.stop(); + } + expect(() => collector.getPort()).toThrow("has not been started"); + expect(collector.getZipkinSpans().length).toBeGreaterThan(0); + }, 120_000); + + it("cleans up capture on startup failure", async () => { + await using network = await new Network().start(); + const dapr = new DaprContainer().withNetwork(network).withTelemetryCapture(); + const expose = jest.spyOn(TestContainers, "exposeHostPorts").mockRejectedValue(new Error("port forwarding failed")); + try { + await expect(dapr.start()).rejects.toThrow("port forwarding failed"); + expect(expose).toHaveBeenCalled(); + expect(() => dapr.getTelemetryCollector()!.getPort()).toThrow("has not been started"); + } finally { + expose.mockRestore(); + } + }, 120_000); +}); diff --git a/src/WorkflowHarness.test.ts b/src/WorkflowHarness.test.ts index cd0860b..f8b94c4 100644 --- a/src/WorkflowHarness.test.ts +++ b/src/WorkflowHarness.test.ts @@ -88,6 +88,7 @@ describe("WorkflowHarness and Workflow Support", () => { appId: "workflow-test-app", daprLogLevel: "info", network, + telemetryCapture: true, }); try { @@ -111,6 +112,21 @@ describe("WorkflowHarness and Workflow Support", () => { expect(state).toBeDefined(); expect(state?.runtimeStatus).toBe(WorkflowRuntimeStatus.COMPLETED); expect(state?.serializedOutput).toBe(JSON.stringify("Hello, World!")); + + const collector = harness.getTelemetryCollector()!; + const workflowSpan = await collector.waitForZipkinSpan( + (span) => span.name === "orchestration||helloworkflow", + 30_000 + ); + const activitySpan = await collector.waitForZipkinSpan( + (span) => span.name === "activity||helloactivity" && span.traceId === workflowSpan.traceId, + 30_000 + ); + expect(workflowSpan.traceId).toMatch(/^[0-9a-f]{32}$/); + expect(activitySpan.id).not.toBe(workflowSpan.id); + expect(activitySpan.parentId).toBe(workflowSpan.id); + expect(workflowSpan.duration).toBeGreaterThan(0); + expect(activitySpan.duration).toBeGreaterThan(0); } finally { await harness.stop(); } diff --git a/src/WorkflowHarness.ts b/src/WorkflowHarness.ts index ee3a18c..10ce3ae 100644 --- a/src/WorkflowHarness.ts +++ b/src/WorkflowHarness.ts @@ -15,6 +15,7 @@ import { DaprWorkflowClient, WorkflowRuntime } from "@dapr/dapr"; import { Network, StartedNetwork } from "testcontainers"; import { DaprComponentNames } from "./Constants"; import { DaprContainer, StartedDaprContainer, WorkflowOptions } from "./DaprContainer"; +import { DaprTelemetryCollector } from "./DaprTelemetryCollector"; import { RedisContainer } from "./RedisContainer"; export type WorkflowHarnessOptions = { @@ -28,6 +29,7 @@ export type WorkflowHarnessOptions = { redisContainer?: RedisContainer; redisHost?: string; network?: StartedNetwork; + telemetryCapture?: boolean; }; /** @@ -62,6 +64,18 @@ export class WorkflowHarness { if (options.appChannelAddress) { this.daprContainer.withAppChannelAddress(options.appChannelAddress); } + if (options.telemetryCapture) { + this.enableTelemetryCapture(); + } + } + + public enableTelemetryCapture(): this { + this.daprContainer.withTelemetryCapture(); + return this; + } + + public getTelemetryCollector(): DaprTelemetryCollector | undefined { + return this.daprContainer.getTelemetryCollector(); } public getDaprContainer(): DaprContainer { @@ -78,7 +92,15 @@ export class WorkflowHarness { } this.daprContainer.withNetwork(this.network); - this.startedDaprContainer = await this.daprContainer.start(); + try { + this.startedDaprContainer = await this.daprContainer.start(); + } catch (error) { + if (this.ownsNetwork) { + await this.network.stop(); + this.network = undefined; + } + throw error; + } return this; } diff --git a/src/index.ts b/src/index.ts index 22ac763..2f167e9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -22,6 +22,7 @@ export * from "./DaprContainer"; export * from "./DaprPlacementContainer"; export * from "./DaprRuntimeVersionGate"; export * from "./DaprSchedulerContainer"; +export * from "./DaprTelemetryCollector"; export * from "./DistributedLockHarness"; export * from "./HttpEndpoint"; export * from "./LocalStorageCryptographyContainer";