Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/external-trace-id-per-run.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@trigger.dev/core": patch
---

Runs that don't continue an incoming trace are no longer merged into one trace when they execute on the same warm worker process. Each run now appears as its own trace in your external observability tool, so per-run cost and latency attribution works again.
83 changes: 64 additions & 19 deletions packages/core/src/v3/otel/tracingSDK.ts
Original file line number Diff line number Diff line change
Expand Up @@ -162,12 +162,13 @@ export class TracingSDK {
)
);

const externalTraceId = idGenerator.generateTraceId();
// Shared by every wrapper below so a run's spans and logs agree on the id.
const fallbackTraceId = new FallbackExternalTraceId(idGenerator.generateTraceId());

for (const exporter of config.exporters ?? []) {
spanProcessors.push(
getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId), {
? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceId), {
maxExportBatchSize: parseInt(
getEnvVar("TRIGGER_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64"
),
Expand All @@ -179,7 +180,7 @@ export class TracingSDK {
),
maxQueueSize: parseInt(getEnvVar("TRIGGER_OTEL_SPAN_MAX_QUEUE_SIZE") ?? "512"),
})
: new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId))
: new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, fallbackTraceId))
);
}

Expand Down Expand Up @@ -231,7 +232,7 @@ export class TracingSDK {
logProcessors.push(
getEnvVar("TRIGGER_OTEL_BATCH_PROCESSING_ENABLED") === "1"
? new BatchLogRecordProcessor(
new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId),
new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceId),
{
maxExportBatchSize: parseInt(
getEnvVar("TRIGGER_OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64"
Expand All @@ -246,7 +247,7 @@ export class TracingSDK {
}
)
: new SimpleLogRecordProcessor(
new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId)
new ExternalLogRecordExporterWrapper(externalLogExporter, fallbackTraceId)
)
);
}
Expand Down Expand Up @@ -393,21 +394,64 @@ function setLogLevel(level: TracingDiagnosticLogLevel) {
diag.setLogger(new DiagConsoleLogger(), diagLogLevel);
}

/**
* The external trace id used by runs that carry no external trace context,
* minted once per run.
*
* It has to change per run for the same reason the wrappers read the external
* context live: with `processKeepAlive` the `TracingSDK` — and so the wrappers
* — outlive the run, so an id captured at construction merges every run on the
* process into one trace.
*
* One instance is shared by every wrapper, so a run's spans and logs still
* agree on the id after a remint.
*/
export class FallbackExternalTraceId {
private traceId: string;
private seenEpoch: number;

constructor(
private seed: string,
private traceIdGenerator: Pick<RandomIdGenerator, "generateTraceId"> = idGenerator
) {
this.traceId = seed;
this.seenEpoch = traceContext.getTraceContextEpoch();
}

forCurrentRun(): string {
// An empty seed means external export is disabled — leave it that way
// rather than minting an id and switching the feature on.
if (!this.seed) {
return this.seed;
}

const epoch = traceContext.getTraceContextEpoch();

if (epoch !== this.seenEpoch) {
this.seenEpoch = epoch;
this.traceId = this.traceIdGenerator.generateTraceId();
}

return this.traceId;
}
}

export class ExternalSpanExporterWrapper {
constructor(
private underlyingExporter: SpanExporter,
private externalTraceId: string
private fallback: FallbackExternalTraceId
) {}

private transformSpan(span: ReadableSpan): ReadableSpan | undefined {
// Read external context live, so per-run reassignment of
// standardTraceContextManager.traceContext is honoured on warm-started
// workers that reuse a single TracingSDK across runs.
const externalTraceContext = traceContext.getExternalTraceContext();
const fallbackTraceId = this.fallback.forCurrentRun();

const isExternallySampled = externalTraceContext
? isTraceFlagSampled(externalTraceContext.traceFlags)
: !!this.externalTraceId;
: !!fallbackTraceId;

if (!isExternallySampled) {
return;
Expand All @@ -417,9 +461,7 @@ export class ExternalSpanExporterWrapper {
return;
}

const externalTraceId = externalTraceContext
? externalTraceContext.traceId
: this.externalTraceId;
const externalTraceId = externalTraceContext ? externalTraceContext.traceId : fallbackTraceId;

const isAttemptSpan = span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT];

Expand Down Expand Up @@ -477,26 +519,29 @@ export class ExternalSpanExporterWrapper {
}
}

class ExternalLogRecordExporterWrapper {
export class ExternalLogRecordExporterWrapper {
constructor(
private underlyingExporter: LogRecordExporter,
private externalTraceId: string
private fallback: FallbackExternalTraceId
) {}

export(logs: any[], resultCallback: (result: any) => void): void {
const externalTraceContext = traceContext.getExternalTraceContext();
const fallbackTraceId = this.fallback.forCurrentRun();

const isExternallySampled = externalTraceContext
? isTraceFlagSampled(externalTraceContext.traceFlags)
: !!this.externalTraceId;
: !!fallbackTraceId;

if (!isExternallySampled) {
this.underlyingExporter.export([], resultCallback);

return;
}

const modifiedLogs = logs.map((log) => this.transformLogRecord(log, externalTraceContext));
const modifiedLogs = logs.map((log) =>
this.transformLogRecord(log, externalTraceContext, fallbackTraceId)
);

this.underlyingExporter.export(modifiedLogs, resultCallback);
}
Expand All @@ -517,13 +562,13 @@ class ExternalLogRecordExporterWrapper {
logRecord: ReadableLogRecord,
externalTraceContext:
| { traceId: string; spanId: string; tracestate?: string; traceFlags: number }
| undefined
| undefined,
fallbackTraceId: string
): ReadableLogRecord {
// Capture externalTraceId for use within the proxy's scope.
// Use externalTraceContext.traceId if available, otherwise fall back to generated externalTraceId
const externalTraceId = externalTraceContext
? externalTraceContext.traceId
: this.externalTraceId;
// Use externalTraceContext.traceId if available, otherwise fall back to the
// per-run generated id.
const externalTraceId = externalTraceContext ? externalTraceContext.traceId : fallbackTraceId;

// If there's no spanContext, or if the externalTraceId is not set, return the original logRecord.
if (!logRecord.spanContext || !externalTraceId) {
Expand Down
9 changes: 9 additions & 0 deletions packages/core/src/v3/traceContext/api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,11 @@ class NoopTraceContextManager implements TraceContextManager {
return {};
}

// Never advances: with no manager registered there are no runs to separate.
getTraceContextEpoch() {
return 0;
}

reset() {}

getExternalTraceContext() {
Expand Down Expand Up @@ -57,6 +62,10 @@ export class TraceContextAPI implements TraceContextManager {
return this.#getManager().getTraceContext();
}

public getTraceContextEpoch() {
return this.#getManager().getTraceContextEpoch();
}

public getExternalTraceContext() {
return this.#getManager().getExternalTraceContext();
}
Expand Down
19 changes: 18 additions & 1 deletion packages/core/src/v3/traceContext/manager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,29 @@ import { parseTraceParent } from "@opentelemetry/core";
import type { TraceContextManager } from "./types.js";

export class StandardTraceContextManager implements TraceContextManager {
public traceContext: Record<string, unknown> = {};
#traceContext: Record<string, unknown> = {};
#epoch = 0;

// An accessor rather than a plain field so that replacing the context, which
// is what starting a run does, is what advances the epoch. Call sites are
// unchanged.
get traceContext(): Record<string, unknown> {
return this.#traceContext;
}

set traceContext(value: Record<string, unknown>) {
this.#traceContext = value;
this.#epoch++;
}
Comment on lines +17 to +20

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔍 Epoch bumps per execution message, so retry attempts of one run get different fallback trace ids

The run boundary is defined as "the trace context object was replaced". In the workers, standardTraceContextManager.traceContext = traceContext runs once per EXECUTE_TASK_RUN message (packages/cli-v3/src/entryPoints/managed-run-worker.ts:403, packages/cli-v3/src/entryPoints/dev-run-worker.ts:426), and reset() also assigns (packages/core/src/v3/traceContext/manager.ts:31), so the epoch advances at least twice per execution message. Since each retry attempt of the same run arrives as its own EXECUTE_TASK_RUN, a run that retries on the same warm process will mint a new fallback external trace id per attempt, i.e. the granularity is per-attempt rather than per-run (the changeset text says "Each run now appears as its own trace"). This is consistent with the cold-start behaviour (a fresh process always generated a fresh id), so it is not a regression, but it's worth confirming the intended granularity — the runs whose attempts land on the same warm process will no longer be grouped, and there is currently no signal in the trace context to distinguish "new attempt of the same run" from "new run".

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.


getTraceContext() {
return this.traceContext;
}

getTraceContextEpoch() {
return this.#epoch;
}

reset() {
this.traceContext = {};
}
Expand Down
6 changes: 6 additions & 0 deletions packages/core/src/v3/traceContext/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,12 @@ import type { Context } from "@opentelemetry/api";

export interface TraceContextManager {
getTraceContext(): Record<string, unknown>;
/**
* Increments every time the trace context is replaced, which on a worker that
* reuses one process across runs is the run boundary. Long-lived consumers
* compare it to tell "still the same run" from "a new run started".
*/
getTraceContextEpoch(): number;
extractContext(): Context;
reset(): void;
getExternalTraceContext():
Expand Down
Loading
Loading