Eric Allam
0445b8ec27
## Summary A single run output, trace span, or payload carrying JSON that ClickHouse can't ingest (for example nesting past its depth limit) used to fail the whole insert batch, so unrelated runs and spans silently disappeared from the runs list, traces, and logs. This keeps the rest of the batch and handles the offending row instead of dropping everything around it. ## Fix Recovery is per-table, matched to what each table needs: - **Runs** (`task_runs_v2`) keep their status. We follow ClickHouse's failing-row hint to strip just the un-ingestable JSON column(s) so the run still lands (its output reads from Postgres on the detail page), up to a configurable limit (`RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH`, default `1`). Past the limit we stop and land the batch with `allow_errors` in a single pass, skipping the remainder. Cost stays a fixed handful of inserts no matter how large or poisoned a flush is. - **Trace events and payloads** (high volume, append-only) recover with a single `allow_errors` insert: the good rows land in one pass and only the un-ingestable rows are skipped. Before falling back, a lightweight sanitizer still repairs what it can losslessly (lone UTF-16 surrogates, out-of-range integers) so a repairable row lands in full. To read the failing-row hint we patch `@clickhouse/client-common`: its error parser truncates the server response and discards the `(at row N)` position, so the patch preserves the full text for the recovery path to read.
2926 lines
90 KiB
TypeScript
2926 lines
90 KiB
TypeScript
import type {
|
|
ClickHouse,
|
|
ClickHouseSettings,
|
|
LlmMetricsV1Input,
|
|
MetricsV1Input,
|
|
TaskEventDetailedSummaryV1Result,
|
|
TaskEventDetailsV1Result,
|
|
TaskEventSummaryV1Result,
|
|
TaskEventV1Input,
|
|
TaskEventV2Input,
|
|
} from "@internal/clickhouse";
|
|
import type { Attributes, Counter, Meter, Tracer } from "@internal/tracing";
|
|
import { getMeter, startSpan, trace } from "@internal/tracing";
|
|
|
|
import { createJsonErrorObject } from "@trigger.dev/core/v3/errors";
|
|
import { serializeTraceparent } from "@trigger.dev/core/v3/isomorphic";
|
|
import type {
|
|
AttemptFailedSpanEvent,
|
|
CancellationSpanEvent,
|
|
ExceptionSpanEvent,
|
|
OtherSpanEvent,
|
|
SpanEvents,
|
|
TaskEventStyle,
|
|
TaskRunError,
|
|
} from "@trigger.dev/core/v3/schemas";
|
|
import {
|
|
isAttemptFailedSpanEvent,
|
|
isCancellationSpanEvent,
|
|
isExceptionSpanEvent,
|
|
PRIMARY_VARIANT,
|
|
} from "@trigger.dev/core/v3/schemas";
|
|
import { SemanticInternalAttributes } from "@trigger.dev/core/v3/semanticInternalAttributes";
|
|
import { unflattenAttributes } from "@trigger.dev/core/v3/utils/flattenAttributes";
|
|
import type { TaskEventLevel } from "@trigger.dev/database";
|
|
import { logger } from "~/services/logger.server";
|
|
import { DynamicFlushScheduler } from "../dynamicFlushScheduler.server";
|
|
import { tracePubSub } from "../services/tracePubSub.server";
|
|
import type { TaskEventStoreTable } from "../taskEventStore.server";
|
|
import {
|
|
calculateDurationFromStart,
|
|
calculateDurationFromStartJsDate,
|
|
convertDateToNanoseconds,
|
|
createExceptionPropertiesFromError,
|
|
extractContextFromCarrier,
|
|
generateDeterministicSpanId,
|
|
generateSpanId,
|
|
generateTraceId,
|
|
getNowInNanoseconds,
|
|
parseEventsField,
|
|
removePrivateProperties,
|
|
} from "./common.server";
|
|
import type {
|
|
CompleteableTaskRun,
|
|
CreateEventInput,
|
|
EventBuilder,
|
|
IEventRepository,
|
|
RunPreparedEvent,
|
|
SpanDetail,
|
|
SpanDetailedSummary,
|
|
SpanOverride,
|
|
SpanSummary,
|
|
SpanSummaryCommon,
|
|
StreamedTraceEvent,
|
|
TraceAttributes,
|
|
TraceDetailedSummary,
|
|
TraceEventOptions,
|
|
TraceSummary,
|
|
} from "./eventRepository.types";
|
|
import {
|
|
insertWithBadRowSkip,
|
|
type JsonParseRecoveryOutcome,
|
|
landedNothing,
|
|
} from "./sanitizeRowsOnParseError.server";
|
|
|
|
export type ClickhouseEventRepositoryConfig = {
|
|
clickhouse: ClickHouse;
|
|
batchSize?: number;
|
|
flushInterval?: number;
|
|
insertStrategy?: "insert" | "insert_async";
|
|
waitForAsyncInsert?: boolean;
|
|
asyncInsertMaxDataSize?: number;
|
|
asyncInsertBusyTimeoutMs?: number;
|
|
tracer?: Tracer;
|
|
maximumTraceSummaryViewCount?: number;
|
|
maximumTraceDetailedSummaryViewCount?: number;
|
|
maximumLiveReloadingSetting?: number;
|
|
/**
|
|
* Maximum age in milliseconds for start_time. If start_time is older than this threshold,
|
|
* it will be clamped to the current time when creating events.
|
|
* If not provided, no clamping will be done.
|
|
*/
|
|
startTimeMaxAgeMs?: number;
|
|
/**
|
|
* The version of the ClickHouse task_events table to use.
|
|
* - "v1": Uses task_events_v1 (partitioned by start_time)
|
|
* - "v2": Uses task_events_v2 (partitioned by inserted_at to avoid "too many parts" errors)
|
|
*/
|
|
version?: "v1" | "v2";
|
|
/** LLM metrics flush scheduler config */
|
|
llmMetricsBatchSize?: number;
|
|
llmMetricsFlushInterval?: number;
|
|
llmMetricsMaxBatchSize?: number;
|
|
llmMetricsMaxConcurrency?: number;
|
|
/** OTLP / task metrics_v1 flush scheduler config */
|
|
otlpMetricsBatchSize?: number;
|
|
otlpMetricsFlushInterval?: number;
|
|
otlpMetricsMaxConcurrency?: number;
|
|
/** Inject a meter for self-observability; defaults to the global provider. */
|
|
meter?: Meter;
|
|
};
|
|
|
|
/**
|
|
* ClickHouse-based implementation of the EventRepository.
|
|
* This implementation stores events in ClickHouse for better analytics and performance.
|
|
*/
|
|
export class ClickhouseEventRepository implements IEventRepository {
|
|
private _clickhouse: ClickHouse;
|
|
private _config: ClickhouseEventRepositoryConfig;
|
|
private readonly _flushScheduler: DynamicFlushScheduler<TaskEventV1Input | TaskEventV2Input>;
|
|
private readonly _llmMetricsFlushScheduler: DynamicFlushScheduler<LlmMetricsV1Input>;
|
|
private readonly _otlpMetricsFlushScheduler: DynamicFlushScheduler<MetricsV1Input>;
|
|
private _tracer: Tracer;
|
|
private _version: "v1" | "v2";
|
|
/**
|
|
* Counts batches where every row was un-ingestable, so nothing landed. Only
|
|
* incremented when ClickHouse's summary says so exactly (`written_rows === 0`);
|
|
* expected to stay at zero, since a whole batch of un-ingestable events means
|
|
* something upstream is broken rather than one bad payload.
|
|
*/
|
|
private _permanentlyDroppedBatches = 0;
|
|
private readonly _droppedBatchesCounter: Counter;
|
|
|
|
/**
|
|
* Counts batches that took the bad-row-skip recovery path: a
|
|
* `Cannot parse JSON object` failure the sanitizer could not repair, where one
|
|
* `allow_errors` insert landed the good rows and skipped the un-ingestable
|
|
* ones. Every such batch lost at least one row, so this is the alertable
|
|
* signal for these tables.
|
|
*/
|
|
private _rowIsolationRecoveries = 0;
|
|
private readonly _rowIsolatedBatchesCounter: Counter;
|
|
|
|
/**
|
|
* Counts rows skipped as un-ingestable. A floor, not an exact count: these
|
|
* tables carry row-multiplying materialized views, so ClickHouse's insert
|
|
* summary can't separate skipped base rows from MV rows (see `droppedRowCount`).
|
|
*/
|
|
private _permanentlyDroppedRows = 0;
|
|
private readonly _rowsDroppedCounter: Counter;
|
|
|
|
constructor(config: ClickhouseEventRepositoryConfig) {
|
|
this._clickhouse = config.clickhouse;
|
|
this._config = config;
|
|
this._tracer = config.tracer ?? trace.getTracer("clickhouseEventRepo", "0.0.1");
|
|
this._version = config.version ?? "v1";
|
|
|
|
const meter = config.meter ?? getMeter("ingest-flush");
|
|
this._droppedBatchesCounter = meter.createCounter("ingest.flush.batches_dropped", {
|
|
description: "Batches permanently dropped after an unrecoverable ClickHouse JSON parse error",
|
|
unit: "batches",
|
|
});
|
|
this._rowIsolatedBatchesCounter = meter.createCounter("ingest.flush.batches_row_isolated", {
|
|
description:
|
|
"Batches recovered by skipping un-ingestable rows (landed the rest) after a ClickHouse JSON parse error; each lost at least one row",
|
|
unit: "batches",
|
|
});
|
|
this._rowsDroppedCounter = meter.createCounter("ingest.flush.rows_dropped", {
|
|
description:
|
|
"Rows skipped as un-ingestable, as a lower bound: these tables' materialized views make the exact count underivable from ClickHouse's insert summary",
|
|
unit: "rows",
|
|
});
|
|
|
|
this._flushScheduler = new DynamicFlushScheduler({
|
|
name: `task_events_${this._version}`,
|
|
batchSize: config.batchSize ?? 1000,
|
|
flushInterval: config.flushInterval ?? 1000,
|
|
callback: this.#flushBatch.bind(this),
|
|
minConcurrency: 1,
|
|
maxConcurrency: 10,
|
|
maxBatchSize: 10000,
|
|
memoryPressureThreshold: 10000,
|
|
loadSheddingThreshold: 10000,
|
|
loadSheddingEnabled: false,
|
|
isDroppableEvent: (event: TaskEventV1Input | TaskEventV2Input) => {
|
|
// Only drop LOG events during load shedding
|
|
return event.kind === "DEBUG_EVENT";
|
|
},
|
|
});
|
|
|
|
this._llmMetricsFlushScheduler = new DynamicFlushScheduler({
|
|
name: "llm_metrics",
|
|
batchSize: config.llmMetricsBatchSize ?? 5000,
|
|
flushInterval: config.llmMetricsFlushInterval ?? 2000,
|
|
callback: this.#flushLlmMetricsBatch.bind(this),
|
|
minConcurrency: 1,
|
|
maxConcurrency: config.llmMetricsMaxConcurrency ?? 2,
|
|
maxBatchSize: config.llmMetricsMaxBatchSize ?? 10000,
|
|
memoryPressureThreshold: config.llmMetricsMaxBatchSize ?? 10000,
|
|
loadSheddingEnabled: false,
|
|
});
|
|
|
|
this._otlpMetricsFlushScheduler = new DynamicFlushScheduler({
|
|
name: "otlp_metrics",
|
|
batchSize: config.otlpMetricsBatchSize ?? 10000,
|
|
flushInterval: config.otlpMetricsFlushInterval ?? 1000,
|
|
callback: this.#flushOtelMetricsBatch.bind(this),
|
|
minConcurrency: 1,
|
|
maxConcurrency: config.otlpMetricsMaxConcurrency ?? 3,
|
|
loadSheddingEnabled: false,
|
|
});
|
|
}
|
|
|
|
get version() {
|
|
return this._version;
|
|
}
|
|
|
|
get maximumLiveReloadingSetting() {
|
|
return this._config.maximumLiveReloadingSetting ?? 1000;
|
|
}
|
|
|
|
/** Exposed for tests and metrics — batches where nothing landed even after stripping JSON. */
|
|
get permanentlyDroppedBatches() {
|
|
return this._permanentlyDroppedBatches;
|
|
}
|
|
|
|
/** Exposed for tests and metrics — batches that took the bad-row-skip recovery path. */
|
|
get rowIsolationRecoveries() {
|
|
return this._rowIsolationRecoveries;
|
|
}
|
|
|
|
/** Exposed for tests and metrics — rows skipped as un-ingestable (a lower bound). */
|
|
get permanentlyDroppedRows() {
|
|
return this._permanentlyDroppedRows;
|
|
}
|
|
|
|
/**
|
|
* Clamps a start time (in nanoseconds) to now if it's too far in the past.
|
|
* Returns the clamped value as a bigint.
|
|
*/
|
|
#clampStartTimeNanoseconds(startTimeNs: bigint): bigint {
|
|
if (!this._config.startTimeMaxAgeMs) {
|
|
return startTimeNs;
|
|
}
|
|
|
|
const nowNs = getNowInNanoseconds();
|
|
const maxAgeNs = BigInt(this._config.startTimeMaxAgeMs) * 1_000_000n; // ms to ns
|
|
const minAllowedStartTime = nowNs - maxAgeNs;
|
|
|
|
if (startTimeNs < minAllowedStartTime) {
|
|
return nowNs;
|
|
}
|
|
|
|
return startTimeNs;
|
|
}
|
|
|
|
/**
|
|
* Clamps a start time string (nanoseconds as string) to now if it's too far in the past.
|
|
* Returns the formatted string for ClickHouse.
|
|
*/
|
|
#clampAndFormatStartTime(startTimeNsString: string): string {
|
|
const startTimeNs = BigInt(startTimeNsString);
|
|
const clampedNs = this.#clampStartTimeNanoseconds(startTimeNs);
|
|
return formatClickhouseDate64NanosecondsEpochString(clampedNs.toString());
|
|
}
|
|
|
|
/**
|
|
* Clamps a Date start time to now if it's too far in the past.
|
|
*/
|
|
#clampStartTimeDate(startTime: Date): Date {
|
|
if (!this._config.startTimeMaxAgeMs) {
|
|
return startTime;
|
|
}
|
|
|
|
const now = new Date();
|
|
const minAllowedStartTime = new Date(now.getTime() - this._config.startTimeMaxAgeMs);
|
|
|
|
if (startTime < minAllowedStartTime) {
|
|
return now;
|
|
}
|
|
|
|
return startTime;
|
|
}
|
|
|
|
async #flushBatch(flushId: string, events: (TaskEventV1Input | TaskEventV2Input)[]) {
|
|
await startSpan(this._tracer, "flushBatch", async (span) => {
|
|
span.setAttribute("flush_id", flushId);
|
|
span.setAttribute("event_count", events.length);
|
|
span.setAttribute("version", this._version);
|
|
|
|
const firstEvent = events[0];
|
|
|
|
if (firstEvent) {
|
|
logger.debug("ClickhouseEventRepository.flushBatch first event", {
|
|
event: firstEvent,
|
|
version: this._version,
|
|
});
|
|
}
|
|
|
|
const insertFn =
|
|
this._version === "v2"
|
|
? this._clickhouse.taskEventsV2.insert
|
|
: this._clickhouse.taskEvents.insert;
|
|
|
|
const contextLabel = `task_events_${this._version}`;
|
|
const rawInsert = async (
|
|
rows: (TaskEventV1Input | TaskEventV2Input)[],
|
|
extraSettings?: ClickHouseSettings
|
|
) => {
|
|
const [insertError, insertResult] = await insertFn(rows, {
|
|
params: {
|
|
clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings },
|
|
},
|
|
});
|
|
if (insertError) throw insertError;
|
|
return insertResult;
|
|
};
|
|
|
|
const outcome = await insertWithBadRowSkip({
|
|
rows: events,
|
|
contextLabel,
|
|
logger,
|
|
logContext: { flushId, version: this._version },
|
|
insert: (rows) => rawInsert(rows),
|
|
insertAllowingBadRows: (rows) =>
|
|
rawInsert(rows, {
|
|
async_insert: 0,
|
|
input_format_parallel_parsing: 0,
|
|
input_format_allow_errors_num: String(rows.length),
|
|
input_format_allow_errors_ratio: 1,
|
|
}),
|
|
});
|
|
this.#recordRecoveryOutcome(outcome, contextLabel, events.length);
|
|
|
|
if (landedNothing(outcome, events.length)) {
|
|
return;
|
|
}
|
|
|
|
logger.debug("ClickhouseEventRepository.flushBatch Inserted batch into clickhouse", {
|
|
events: events.length,
|
|
outcome: outcome.kind,
|
|
version: this._version,
|
|
});
|
|
|
|
this.#publishToRedis(events);
|
|
});
|
|
}
|
|
|
|
async #flushLlmMetricsBatch(flushId: string, rows: LlmMetricsV1Input[]) {
|
|
const rawInsert = async (batch: LlmMetricsV1Input[], extraSettings?: ClickHouseSettings) => {
|
|
const [insertError, insertResult] = await this._clickhouse.llmMetrics.insert(batch, {
|
|
params: {
|
|
clickhouse_settings: { ...this.#getClickhouseInsertSettings(), ...extraSettings },
|
|
},
|
|
});
|
|
if (insertError) throw insertError;
|
|
return insertResult;
|
|
};
|
|
|
|
const outcome = await insertWithBadRowSkip({
|
|
rows,
|
|
contextLabel: "llm_metrics_v1",
|
|
logger,
|
|
logContext: { flushId },
|
|
insert: (batch) => rawInsert(batch),
|
|
insertAllowingBadRows: (batch) =>
|
|
rawInsert(batch, {
|
|
async_insert: 0,
|
|
input_format_parallel_parsing: 0,
|
|
input_format_allow_errors_num: String(batch.length),
|
|
input_format_allow_errors_ratio: 1,
|
|
}),
|
|
});
|
|
this.#recordRecoveryOutcome(outcome, "llm_metrics_v1", rows.length);
|
|
|
|
if (landedNothing(outcome, rows.length)) {
|
|
return;
|
|
}
|
|
|
|
logger.debug("ClickhouseEventRepository.flushLlmMetricsBatch Inserted LLM metrics batch", {
|
|
rows: rows.length,
|
|
outcome: outcome.kind,
|
|
});
|
|
}
|
|
|
|
#recordRecoveryOutcome(
|
|
outcome: JsonParseRecoveryOutcome,
|
|
contextLabel: string,
|
|
batchSize: number
|
|
) {
|
|
if (outcome.kind !== "recovered") {
|
|
return;
|
|
}
|
|
|
|
this._rowIsolationRecoveries += 1;
|
|
this._rowIsolatedBatchesCounter.add(1, { table: contextLabel });
|
|
|
|
if (outcome.rowsDropped > 0) {
|
|
this._permanentlyDroppedRows += outcome.rowsDropped;
|
|
this._rowsDroppedCounter.add(outcome.rowsDropped, { table: contextLabel });
|
|
if (outcome.rowsDroppedExact && outcome.rowsDropped === batchSize) {
|
|
this._permanentlyDroppedBatches += 1;
|
|
this._droppedBatchesCounter.add(1, { table: contextLabel });
|
|
}
|
|
}
|
|
}
|
|
|
|
async #flushOtelMetricsBatch(flushId: string, rows: MetricsV1Input[]) {
|
|
await startSpan(this._tracer, "flushOtelMetricsBatch", async (span) => {
|
|
span.setAttribute("flush_id", flushId);
|
|
span.setAttribute("row_count", rows.length);
|
|
|
|
const [insertError] = await this._clickhouse.metrics.insert(rows, {
|
|
params: {
|
|
clickhouse_settings: this.#getClickhouseInsertSettings(),
|
|
},
|
|
});
|
|
|
|
if (insertError) {
|
|
throw insertError;
|
|
}
|
|
|
|
logger.debug("ClickhouseEventRepository.flushOtelMetricsBatch Inserted OTLP metrics batch", {
|
|
rows: rows.length,
|
|
});
|
|
});
|
|
}
|
|
|
|
#createLlmMetricsInput(event: CreateEventInput): LlmMetricsV1Input {
|
|
const llmMetrics = event._llmMetrics!;
|
|
|
|
return {
|
|
organization_id: event.organizationId,
|
|
project_id: event.projectId,
|
|
environment_id: event.environmentId,
|
|
run_id: event.runId,
|
|
task_identifier: event.taskSlug,
|
|
trace_id: event.traceId,
|
|
span_id: event.spanId,
|
|
gen_ai_system: llmMetrics.genAiSystem,
|
|
request_model: llmMetrics.requestModel,
|
|
response_model: llmMetrics.responseModel,
|
|
base_response_model: llmMetrics.baseResponseModel,
|
|
matched_model_id: llmMetrics.matchedModelId,
|
|
operation_id: llmMetrics.operationId,
|
|
finish_reason: llmMetrics.finishReason,
|
|
cost_source: llmMetrics.costSource,
|
|
pricing_tier_id: llmMetrics.pricingTierId,
|
|
pricing_tier_name: llmMetrics.pricingTierName,
|
|
input_tokens: llmMetrics.inputTokens,
|
|
output_tokens: llmMetrics.outputTokens,
|
|
total_tokens: llmMetrics.totalTokens,
|
|
usage_details: llmMetrics.usageDetails,
|
|
input_cost: llmMetrics.inputCost,
|
|
output_cost: llmMetrics.outputCost,
|
|
total_cost: llmMetrics.totalCost,
|
|
cost_details: llmMetrics.costDetails,
|
|
provider_cost: llmMetrics.providerCost,
|
|
ms_to_first_chunk: llmMetrics.msToFirstChunk,
|
|
tokens_per_second: llmMetrics.tokensPerSecond,
|
|
metadata: llmMetrics.metadata,
|
|
prompt_slug: llmMetrics.promptSlug,
|
|
prompt_version: llmMetrics.promptVersion,
|
|
start_time: this.#clampAndFormatStartTime(event.startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(event.duration ?? 0),
|
|
};
|
|
}
|
|
|
|
#getClickhouseInsertSettings() {
|
|
if (this._config.insertStrategy === "insert") {
|
|
return {};
|
|
} else {
|
|
return {
|
|
async_insert: 1 as const,
|
|
async_insert_max_data_size: this._config.asyncInsertMaxDataSize?.toString() ?? "10485760",
|
|
async_insert_busy_timeout_ms: this._config.asyncInsertBusyTimeoutMs ?? 5000,
|
|
wait_for_async_insert: this._config.waitForAsyncInsert ? (1 as const) : (0 as const),
|
|
};
|
|
}
|
|
}
|
|
|
|
async #publishToRedis(events: (TaskEventV1Input | TaskEventV2Input)[]) {
|
|
if (events.length === 0) return;
|
|
await tracePubSub.publish(events.map((e) => e.trace_id));
|
|
}
|
|
|
|
insertMany(events: CreateEventInput[]): void {
|
|
this.addToBatch(events.flatMap((event) => this.createEventToTaskEventV1Input(event)));
|
|
|
|
// Dual-write LLM metrics records for spans with cost enrichment
|
|
const llmMetricsRows = events
|
|
.filter((e) => e._llmMetrics != null)
|
|
.map((e) => this.#createLlmMetricsInput(e));
|
|
|
|
if (llmMetricsRows.length > 0) {
|
|
this._llmMetricsFlushScheduler.addToBatch(llmMetricsRows);
|
|
}
|
|
}
|
|
|
|
async insertManyImmediate(events: CreateEventInput[]): Promise<void> {
|
|
this.insertMany(events);
|
|
}
|
|
|
|
insertManyMetrics(rows: MetricsV1Input[]): void {
|
|
if (rows.length === 0) return;
|
|
this._otlpMetricsFlushScheduler.addToBatch(rows);
|
|
}
|
|
|
|
private createEventToTaskEventV1Input(event: CreateEventInput): TaskEventV1Input[] {
|
|
return [
|
|
{
|
|
environment_id: event.environmentId,
|
|
organization_id: event.organizationId,
|
|
project_id: event.projectId,
|
|
task_identifier: event.taskSlug,
|
|
run_id: event.runId,
|
|
start_time: this.#clampAndFormatStartTime(event.startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(event.duration ?? 0),
|
|
trace_id: event.traceId,
|
|
span_id: event.spanId,
|
|
parent_span_id: event.parentId ?? "",
|
|
message: event.message,
|
|
kind: this.createEventToTaskEventV1InputKind(event),
|
|
status: this.createEventToTaskEventV1InputStatus(event),
|
|
attributes: this.createEventToTaskEventV1InputAttributes(
|
|
event.properties,
|
|
event.resourceProperties
|
|
),
|
|
metadata: this.createEventToTaskEventV1InputMetadata(event),
|
|
expires_at: convertDateToClickhouseDateTime(
|
|
new Date(Date.now() + 365 * 24 * 60 * 60 * 1000) // 1 year
|
|
),
|
|
machine_id: event.machineId ?? "",
|
|
},
|
|
...this.spanEventsToTaskEventV1Input(event),
|
|
];
|
|
}
|
|
|
|
private spanEventsToTaskEventV1Input(event: CreateEventInput): TaskEventV1Input[] {
|
|
if (!event.events) return [];
|
|
|
|
const spanEvents = parseEventsField(event.events);
|
|
|
|
const records = spanEvents.map((e) => this.createTaskEventV1InputFromSpanEvent(e, event));
|
|
|
|
if (event.isPartial) {
|
|
return records;
|
|
}
|
|
|
|
// Only return events where the event start_time is greater than the span start_time
|
|
return records.filter(
|
|
(r) =>
|
|
convertClickhouseDate64NanosecondsEpochStringToBigInt(r.start_time) >
|
|
BigInt(event.startTime)
|
|
);
|
|
}
|
|
|
|
private createTaskEventV1InputFromSpanEvent(
|
|
spanEvent: SpanEvents[number],
|
|
event: CreateEventInput
|
|
): TaskEventV1Input {
|
|
if (isExceptionSpanEvent(spanEvent)) {
|
|
return this.createTaskEventV1InputFromExceptionEvent(spanEvent, event);
|
|
}
|
|
|
|
if (isCancellationSpanEvent(spanEvent)) {
|
|
return this.createTaskEventV1InputFromCancellationEvent(spanEvent, event);
|
|
}
|
|
|
|
if (isAttemptFailedSpanEvent(spanEvent)) {
|
|
return this.createTaskEventV1InputFromAttemptFailedEvent(spanEvent, event);
|
|
}
|
|
|
|
return this.createTaskEventV1InputFromOtherEvent(spanEvent, event);
|
|
}
|
|
|
|
private createTaskEventV1InputFromExceptionEvent(
|
|
spanEvent: ExceptionSpanEvent,
|
|
event: CreateEventInput
|
|
): TaskEventV1Input {
|
|
return {
|
|
environment_id: event.environmentId,
|
|
organization_id: event.organizationId,
|
|
project_id: event.projectId,
|
|
task_identifier: event.taskSlug,
|
|
run_id: event.runId,
|
|
start_time: this.#clampAndFormatStartTime(
|
|
convertDateToNanoseconds(spanEvent.time).toString()
|
|
),
|
|
duration: "0", // Events have no duration
|
|
trace_id: event.traceId,
|
|
span_id: event.spanId,
|
|
parent_span_id: event.parentId ?? "",
|
|
message: spanEvent.name,
|
|
kind: "SPAN_EVENT",
|
|
status: "ERROR",
|
|
attributes: {
|
|
error: {
|
|
message: spanEvent.properties.exception.message,
|
|
name: spanEvent.properties.exception.type,
|
|
stackTrace: spanEvent.properties.exception.stacktrace,
|
|
},
|
|
},
|
|
metadata: JSON.stringify({
|
|
exception: spanEvent.properties.exception,
|
|
}), // Events have no metadata
|
|
expires_at: convertDateToClickhouseDateTime(
|
|
new Date(Date.now() + 365 * 24 * 60 * 60 * 1000) // 1 year
|
|
),
|
|
};
|
|
}
|
|
|
|
private createTaskEventV1InputFromCancellationEvent(
|
|
spanEvent: CancellationSpanEvent,
|
|
event: CreateEventInput
|
|
): TaskEventV1Input {
|
|
return {
|
|
environment_id: event.environmentId,
|
|
organization_id: event.organizationId,
|
|
project_id: event.projectId,
|
|
task_identifier: event.taskSlug,
|
|
run_id: event.runId,
|
|
start_time: this.#clampAndFormatStartTime(
|
|
convertDateToNanoseconds(spanEvent.time).toString()
|
|
),
|
|
duration: "0", // Events have no duration
|
|
trace_id: event.traceId,
|
|
span_id: event.spanId,
|
|
parent_span_id: event.parentId ?? "",
|
|
message: spanEvent.name,
|
|
kind: "SPAN_EVENT",
|
|
status: "CANCELLED",
|
|
attributes: {},
|
|
metadata: JSON.stringify({
|
|
reason: spanEvent.properties.reason,
|
|
}), // Events have no metadata
|
|
expires_at: convertDateToClickhouseDateTime(
|
|
new Date(Date.now() + 365 * 24 * 60 * 60 * 1000) // 1 year
|
|
),
|
|
};
|
|
}
|
|
|
|
private createTaskEventV1InputFromAttemptFailedEvent(
|
|
spanEvent: AttemptFailedSpanEvent,
|
|
event: CreateEventInput
|
|
): TaskEventV1Input {
|
|
return {
|
|
environment_id: event.environmentId,
|
|
organization_id: event.organizationId,
|
|
project_id: event.projectId,
|
|
task_identifier: event.taskSlug,
|
|
run_id: event.runId,
|
|
start_time: this.#clampAndFormatStartTime(
|
|
convertDateToNanoseconds(spanEvent.time).toString()
|
|
),
|
|
duration: "0", // Events have no duration
|
|
trace_id: event.traceId,
|
|
span_id: event.spanId,
|
|
parent_span_id: event.parentId ?? "",
|
|
message: spanEvent.name,
|
|
kind: "ANCESTOR_OVERRIDE",
|
|
status: "OK",
|
|
attributes: {
|
|
error: {
|
|
message: spanEvent.properties.exception.message,
|
|
name: spanEvent.properties.exception.type,
|
|
stackTrace: spanEvent.properties.exception.stacktrace,
|
|
},
|
|
},
|
|
metadata: JSON.stringify(spanEvent.properties),
|
|
expires_at: convertDateToClickhouseDateTime(
|
|
new Date(Date.now() + 365 * 24 * 60 * 60 * 1000) // 1 year
|
|
),
|
|
};
|
|
}
|
|
|
|
private createTaskEventV1InputFromOtherEvent(
|
|
spanEvent: OtherSpanEvent,
|
|
event: CreateEventInput
|
|
): TaskEventV1Input {
|
|
return {
|
|
environment_id: event.environmentId,
|
|
organization_id: event.organizationId,
|
|
project_id: event.projectId,
|
|
task_identifier: event.taskSlug,
|
|
run_id: event.runId,
|
|
start_time: this.#clampAndFormatStartTime(
|
|
convertDateToNanoseconds(spanEvent.time).toString()
|
|
),
|
|
duration: "0", // Events have no duration
|
|
trace_id: event.traceId,
|
|
span_id: event.spanId,
|
|
parent_span_id: event.parentId ?? "",
|
|
message: spanEvent.name,
|
|
kind: "SPAN_EVENT",
|
|
status: "OK",
|
|
attributes: {},
|
|
metadata: JSON.stringify(unflattenAttributes(spanEvent.properties as Attributes)),
|
|
expires_at: convertDateToClickhouseDateTime(
|
|
new Date(Date.now() + 365 * 24 * 60 * 60 * 1000) // 1 year
|
|
),
|
|
};
|
|
}
|
|
|
|
private createEventToTaskEventV1InputKind(event: CreateEventInput): string {
|
|
if (event.kind === "UNSPECIFIED") {
|
|
return "ANCESTOR_OVERRIDE";
|
|
}
|
|
|
|
if (event.level === "TRACE") {
|
|
return "SPAN";
|
|
}
|
|
|
|
if (event.isDebug) {
|
|
return "DEBUG_EVENT";
|
|
}
|
|
|
|
return `LOG_${(event.level ?? "LOG").toUpperCase()}`;
|
|
}
|
|
|
|
private createEventToTaskEventV1InputStatus(event: CreateEventInput): string {
|
|
if (event.isPartial) {
|
|
return "PARTIAL";
|
|
}
|
|
|
|
if (event.isError) {
|
|
return "ERROR";
|
|
}
|
|
|
|
if (event.isCancelled) {
|
|
return "CANCELLED";
|
|
}
|
|
|
|
return "OK";
|
|
}
|
|
|
|
private createEventToTaskEventV1InputAttributes(
|
|
attributes: Attributes,
|
|
resourceAttributes?: Attributes
|
|
): Record<string, unknown> {
|
|
if (!attributes && !resourceAttributes) {
|
|
return {};
|
|
}
|
|
|
|
return {
|
|
...this.createAttributesToInputAttributes(attributes),
|
|
...this.createAttributesToInputAttributes(resourceAttributes, "$resource"),
|
|
};
|
|
}
|
|
|
|
private createAttributesToInputAttributes(
|
|
attributes: Attributes | undefined,
|
|
key?: string
|
|
): Record<string, unknown> {
|
|
if (!attributes) {
|
|
return {};
|
|
}
|
|
|
|
const publicAttributes = removePrivateProperties(attributes);
|
|
|
|
if (!publicAttributes) {
|
|
return {};
|
|
}
|
|
|
|
const unflattenedAttributes = unflattenAttributes(publicAttributes);
|
|
|
|
if (unflattenedAttributes && typeof unflattenedAttributes === "object") {
|
|
if (key) {
|
|
return {
|
|
[key]: unflattenedAttributes,
|
|
};
|
|
}
|
|
|
|
return {
|
|
...unflattenedAttributes,
|
|
};
|
|
}
|
|
|
|
return {};
|
|
}
|
|
|
|
private createEventToTaskEventV1InputMetadata(event: CreateEventInput): string {
|
|
return JSON.stringify({
|
|
style: event.style ? unflattenAttributes(event.style) : undefined,
|
|
attemptNumber: event.attemptNumber,
|
|
entity: this.extractEntityFromAttributes(event.properties),
|
|
});
|
|
}
|
|
|
|
private extractEntityFromAttributes(
|
|
attributes: Attributes
|
|
): { entityType: string; entityId?: string; entityMetadata?: string } | undefined {
|
|
if (!attributes || typeof attributes !== "object") {
|
|
return undefined;
|
|
}
|
|
|
|
const entityType = attributes[SemanticInternalAttributes.ENTITY_TYPE];
|
|
const entityId = attributes[SemanticInternalAttributes.ENTITY_ID];
|
|
const entityMetadata = attributes[SemanticInternalAttributes.ENTITY_METADATA];
|
|
|
|
if (typeof entityType !== "string") {
|
|
return undefined;
|
|
}
|
|
|
|
return {
|
|
entityType,
|
|
entityId: entityId as string | undefined,
|
|
entityMetadata: entityMetadata as string | undefined,
|
|
};
|
|
}
|
|
|
|
private addToBatch(events: TaskEventV1Input[] | TaskEventV1Input) {
|
|
this._flushScheduler.addToBatch(Array.isArray(events) ? events : [events]);
|
|
}
|
|
|
|
// Event recording methods
|
|
async recordEvent(
|
|
message: string,
|
|
options: TraceEventOptions & { duration?: number; parentId?: string }
|
|
): Promise<void> {
|
|
const propagatedContext = extractContextFromCarrier(options.context ?? {});
|
|
|
|
const startTime = options.startTime ?? getNowInNanoseconds();
|
|
const duration =
|
|
options.duration ??
|
|
(options.endTime
|
|
? calculateDurationFromStart(startTime, options.endTime, 100 * 1_000_000)
|
|
: 100);
|
|
|
|
const traceId = propagatedContext?.traceparent?.traceId ?? generateTraceId();
|
|
const parentId = options.parentId ?? propagatedContext?.traceparent?.spanId;
|
|
const spanId = options.spanIdSeed
|
|
? generateDeterministicSpanId(traceId, options.spanIdSeed)
|
|
: generateSpanId();
|
|
|
|
if (!options.attributes.runId) {
|
|
throw new Error("runId is required");
|
|
}
|
|
|
|
const kind = options.attributes.isDebug ? "DEBUG_EVENT" : "SPAN";
|
|
|
|
const metadata = {
|
|
style: {
|
|
icon: options.attributes.isDebug ? "warn" : "play",
|
|
},
|
|
...options.attributes.metadata,
|
|
};
|
|
|
|
const event: TaskEventV1Input = {
|
|
environment_id: options.environment.id,
|
|
organization_id: options.environment.organizationId,
|
|
project_id: options.environment.projectId,
|
|
task_identifier: options.taskSlug,
|
|
run_id: options.attributes.runId,
|
|
start_time: this.#clampAndFormatStartTime(startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(duration),
|
|
trace_id: traceId,
|
|
span_id: spanId,
|
|
parent_span_id: parentId ?? "",
|
|
message,
|
|
kind,
|
|
status: "OK",
|
|
attributes: options.attributes.properties
|
|
? this.createEventToTaskEventV1InputAttributes(options.attributes.properties)
|
|
: undefined,
|
|
metadata: JSON.stringify(metadata),
|
|
// TODO: make sure configurable and by org
|
|
expires_at: convertDateToClickhouseDateTime(new Date(Date.now() + 365 * 24 * 60 * 60 * 1000)),
|
|
};
|
|
|
|
this._flushScheduler.addToBatch([event]);
|
|
}
|
|
|
|
async traceEvent<TResult>(
|
|
message: string,
|
|
options: TraceEventOptions & { incomplete?: boolean; isError?: boolean },
|
|
callback: (
|
|
e: EventBuilder,
|
|
traceContext: Record<string, string | undefined>,
|
|
traceparent?: { traceId: string; spanId: string }
|
|
) => Promise<TResult>
|
|
): Promise<TResult> {
|
|
const propagatedContext = extractContextFromCarrier(options.context ?? {});
|
|
|
|
const start = process.hrtime.bigint();
|
|
const startTime = options.startTime ?? getNowInNanoseconds();
|
|
|
|
const traceId = options.spanParentAsLink
|
|
? generateTraceId()
|
|
: (propagatedContext?.traceparent?.traceId ?? generateTraceId());
|
|
const parentId = options.spanParentAsLink ? undefined : propagatedContext?.traceparent?.spanId;
|
|
const spanId = options.spanIdSeed
|
|
? generateDeterministicSpanId(traceId, options.spanIdSeed)
|
|
: generateSpanId();
|
|
|
|
const traceContext = {
|
|
...options.context,
|
|
traceparent: serializeTraceparent(traceId, spanId),
|
|
};
|
|
|
|
let isStopped = false;
|
|
let failedWithError: TaskRunError | undefined;
|
|
|
|
const eventBuilder = {
|
|
traceId,
|
|
spanId,
|
|
setAttribute: (key: keyof TraceAttributes, value: TraceAttributes[keyof TraceAttributes]) => {
|
|
if (value) {
|
|
// We need to merge the attributes with the existing attributes
|
|
const existingValue = options.attributes[key];
|
|
|
|
if (existingValue && typeof existingValue === "object" && typeof value === "object") {
|
|
// @ts-ignore
|
|
options.attributes[key] = { ...existingValue, ...value };
|
|
} else {
|
|
// @ts-ignore
|
|
options.attributes[key] = value;
|
|
}
|
|
}
|
|
},
|
|
stop: () => {
|
|
isStopped = true;
|
|
},
|
|
failWithError: (error: TaskRunError) => {
|
|
failedWithError = error;
|
|
},
|
|
};
|
|
|
|
const result = await callback(eventBuilder, traceContext, propagatedContext?.traceparent);
|
|
|
|
if (isStopped) {
|
|
return result;
|
|
}
|
|
|
|
const duration = process.hrtime.bigint() - start;
|
|
|
|
if (!options.attributes.runId) {
|
|
throw new Error("runId is required");
|
|
}
|
|
|
|
const metadata = {
|
|
style: {
|
|
icon: "task",
|
|
variant: PRIMARY_VARIANT,
|
|
...options.attributes.style,
|
|
},
|
|
...options.attributes.metadata,
|
|
};
|
|
|
|
const event: TaskEventV1Input = {
|
|
environment_id: options.environment.id,
|
|
organization_id: options.environment.organizationId,
|
|
project_id: options.environment.projectId,
|
|
task_identifier: options.taskSlug,
|
|
run_id: options.attributes.runId,
|
|
start_time: this.#clampAndFormatStartTime(startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(options.incomplete ? 0 : duration),
|
|
trace_id: traceId,
|
|
span_id: spanId,
|
|
parent_span_id: parentId ?? "",
|
|
message,
|
|
kind: "SPAN",
|
|
status: failedWithError ? "ERROR" : options.incomplete ? "PARTIAL" : "OK",
|
|
attributes: options.attributes.properties
|
|
? this.createEventToTaskEventV1InputAttributes(options.attributes.properties)
|
|
: {},
|
|
metadata: JSON.stringify(metadata),
|
|
// TODO: make sure configurable and by org
|
|
expires_at: convertDateToClickhouseDateTime(new Date(Date.now() + 365 * 24 * 60 * 60 * 1000)),
|
|
};
|
|
|
|
const events = [event];
|
|
|
|
if (failedWithError) {
|
|
const error = createJsonErrorObject(failedWithError);
|
|
|
|
events.push({
|
|
environment_id: options.environment.id,
|
|
organization_id: options.environment.organizationId,
|
|
project_id: options.environment.projectId,
|
|
task_identifier: options.taskSlug,
|
|
run_id: options.attributes.runId,
|
|
start_time: this.#clampAndFormatStartTime(startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(options.incomplete ? 0 : duration),
|
|
trace_id: traceId,
|
|
span_id: spanId,
|
|
parent_span_id: parentId ?? "",
|
|
message: "exception",
|
|
kind: "SPAN_EVENT",
|
|
status: "ERROR",
|
|
attributes: {
|
|
error,
|
|
},
|
|
metadata: JSON.stringify({
|
|
exception: createExceptionPropertiesFromError(failedWithError),
|
|
}),
|
|
// TODO: make sure configurable and by org
|
|
expires_at: convertDateToClickhouseDateTime(
|
|
new Date(Date.now() + 365 * 24 * 60 * 60 * 1000)
|
|
),
|
|
});
|
|
}
|
|
|
|
this._flushScheduler.addToBatch(events);
|
|
|
|
return result;
|
|
}
|
|
|
|
// Run event completion methods
|
|
async completeSuccessfulRunEvent({
|
|
run,
|
|
endTime,
|
|
}: {
|
|
run: CompleteableTaskRun;
|
|
endTime?: Date;
|
|
}): Promise<void> {
|
|
if (!run.organizationId) {
|
|
return;
|
|
}
|
|
|
|
const clampedCreatedAt = this.#clampStartTimeDate(run.createdAt);
|
|
const startTime = convertDateToNanoseconds(clampedCreatedAt);
|
|
const expiresAt = convertDateToClickhouseDateTime(
|
|
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
|
|
);
|
|
|
|
const event: TaskEventV1Input = {
|
|
environment_id: run.runtimeEnvironmentId,
|
|
organization_id: run.organizationId,
|
|
project_id: run.projectId,
|
|
task_identifier: run.taskIdentifier,
|
|
run_id: run.friendlyId,
|
|
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(
|
|
calculateDurationFromStart(startTime, endTime ?? new Date())
|
|
),
|
|
trace_id: run.traceId,
|
|
span_id: run.spanId,
|
|
parent_span_id: run.parentSpanId ?? "",
|
|
message: run.taskIdentifier,
|
|
kind: "SPAN",
|
|
status: "OK",
|
|
attributes: {},
|
|
metadata: "{}",
|
|
expires_at: expiresAt,
|
|
};
|
|
|
|
this.addToBatch(event);
|
|
}
|
|
|
|
async completeCachedRunEvent({
|
|
run,
|
|
blockedRun,
|
|
spanId,
|
|
parentSpanId,
|
|
spanCreatedAt,
|
|
isError,
|
|
endTime,
|
|
}: {
|
|
run: CompleteableTaskRun;
|
|
blockedRun: CompleteableTaskRun;
|
|
spanId: string;
|
|
parentSpanId: string;
|
|
spanCreatedAt: Date;
|
|
isError: boolean;
|
|
endTime?: Date;
|
|
}): Promise<void> {
|
|
if (!run.organizationId) {
|
|
return;
|
|
}
|
|
|
|
const clampedSpanCreatedAt = this.#clampStartTimeDate(spanCreatedAt);
|
|
const startTime = convertDateToNanoseconds(clampedSpanCreatedAt);
|
|
const expiresAt = convertDateToClickhouseDateTime(
|
|
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
|
|
);
|
|
|
|
const event: TaskEventV1Input = {
|
|
environment_id: run.runtimeEnvironmentId,
|
|
organization_id: run.organizationId,
|
|
project_id: run.projectId,
|
|
task_identifier: run.taskIdentifier,
|
|
run_id: blockedRun.friendlyId,
|
|
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(
|
|
calculateDurationFromStart(startTime, endTime ?? new Date())
|
|
),
|
|
trace_id: blockedRun.traceId,
|
|
span_id: spanId,
|
|
parent_span_id: parentSpanId,
|
|
message: run.taskIdentifier,
|
|
kind: "SPAN",
|
|
status: isError ? "ERROR" : "OK",
|
|
attributes: {},
|
|
metadata: "{}",
|
|
expires_at: expiresAt,
|
|
};
|
|
|
|
this.addToBatch(event);
|
|
}
|
|
|
|
async completeFailedRunEvent({
|
|
run,
|
|
endTime,
|
|
exception,
|
|
}: {
|
|
run: CompleteableTaskRun;
|
|
endTime?: Date;
|
|
exception: { message?: string; type?: string; stacktrace?: string };
|
|
}): Promise<void> {
|
|
if (!run.organizationId) {
|
|
return;
|
|
}
|
|
|
|
const clampedCreatedAt = this.#clampStartTimeDate(run.createdAt);
|
|
const startTime = convertDateToNanoseconds(clampedCreatedAt);
|
|
const expiresAt = convertDateToClickhouseDateTime(
|
|
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
|
|
);
|
|
|
|
const event: TaskEventV1Input = {
|
|
environment_id: run.runtimeEnvironmentId,
|
|
organization_id: run.organizationId,
|
|
project_id: run.projectId,
|
|
task_identifier: run.taskIdentifier,
|
|
run_id: run.friendlyId,
|
|
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(
|
|
calculateDurationFromStart(startTime, endTime ?? new Date())
|
|
),
|
|
trace_id: run.traceId,
|
|
span_id: run.spanId,
|
|
parent_span_id: run.parentSpanId ?? "",
|
|
message: run.taskIdentifier,
|
|
kind: "SPAN",
|
|
status: "ERROR",
|
|
attributes: {
|
|
error: {
|
|
name: exception.type,
|
|
message: exception.message,
|
|
stackTrace: exception.stacktrace,
|
|
},
|
|
},
|
|
metadata: "{}",
|
|
expires_at: expiresAt,
|
|
};
|
|
|
|
this.addToBatch(event);
|
|
}
|
|
|
|
async completeExpiredRunEvent({
|
|
run,
|
|
endTime,
|
|
ttl,
|
|
}: {
|
|
run: CompleteableTaskRun;
|
|
endTime?: Date;
|
|
ttl: string;
|
|
}): Promise<void> {
|
|
if (!run.organizationId) {
|
|
return;
|
|
}
|
|
|
|
const clampedCreatedAt = this.#clampStartTimeDate(run.createdAt);
|
|
const startTime = convertDateToNanoseconds(clampedCreatedAt);
|
|
const expiresAt = convertDateToClickhouseDateTime(
|
|
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
|
|
);
|
|
|
|
const event: TaskEventV1Input = {
|
|
environment_id: run.runtimeEnvironmentId,
|
|
organization_id: run.organizationId,
|
|
project_id: run.projectId,
|
|
task_identifier: run.taskIdentifier,
|
|
run_id: run.friendlyId,
|
|
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(
|
|
calculateDurationFromStart(startTime, endTime ?? new Date())
|
|
),
|
|
trace_id: run.traceId,
|
|
span_id: run.spanId,
|
|
parent_span_id: run.parentSpanId ?? "",
|
|
message: run.taskIdentifier,
|
|
kind: "SPAN",
|
|
status: "ERROR",
|
|
attributes: {
|
|
error: {
|
|
message: `Run expired because the TTL (${ttl}) was reached`,
|
|
},
|
|
},
|
|
metadata: "{}",
|
|
expires_at: expiresAt,
|
|
};
|
|
|
|
this.addToBatch(event);
|
|
}
|
|
|
|
async createAttemptFailedRunEvent({
|
|
run,
|
|
endTime,
|
|
attemptNumber,
|
|
exception,
|
|
}: {
|
|
run: CompleteableTaskRun;
|
|
endTime?: Date;
|
|
attemptNumber: number;
|
|
exception: { message?: string; type?: string; stacktrace?: string };
|
|
}): Promise<void> {
|
|
if (!run.organizationId) {
|
|
return;
|
|
}
|
|
|
|
const clampedEndTime = this.#clampStartTimeDate(endTime ?? new Date());
|
|
const startTime = convertDateToNanoseconds(clampedEndTime);
|
|
const expiresAt = convertDateToClickhouseDateTime(
|
|
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
|
|
);
|
|
|
|
const event: TaskEventV1Input = {
|
|
environment_id: run.runtimeEnvironmentId,
|
|
organization_id: run.organizationId,
|
|
project_id: run.projectId,
|
|
task_identifier: run.taskIdentifier,
|
|
run_id: run.friendlyId,
|
|
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
|
|
duration: "0",
|
|
trace_id: run.traceId,
|
|
span_id: run.spanId,
|
|
parent_span_id: run.parentSpanId ?? "",
|
|
message: "attempt_failed",
|
|
kind: "ANCESTOR_OVERRIDE",
|
|
status: "OK",
|
|
attributes: {},
|
|
metadata: JSON.stringify({
|
|
exception,
|
|
attemptNumber,
|
|
runId: run.friendlyId,
|
|
}),
|
|
expires_at: expiresAt,
|
|
};
|
|
|
|
this.addToBatch(event);
|
|
}
|
|
|
|
async cancelRunEvent({
|
|
reason,
|
|
run,
|
|
cancelledAt,
|
|
}: {
|
|
reason: string;
|
|
run: CompleteableTaskRun;
|
|
cancelledAt: Date;
|
|
}): Promise<void> {
|
|
if (!run.organizationId) {
|
|
return;
|
|
}
|
|
|
|
const clampedCreatedAt = this.#clampStartTimeDate(run.createdAt);
|
|
const startTime = convertDateToNanoseconds(clampedCreatedAt);
|
|
const expiresAt = convertDateToClickhouseDateTime(
|
|
new Date(run.createdAt.getTime() + 30 * 24 * 60 * 60 * 1000)
|
|
);
|
|
|
|
const event: TaskEventV1Input = {
|
|
environment_id: run.runtimeEnvironmentId,
|
|
organization_id: run.organizationId,
|
|
project_id: run.projectId,
|
|
task_identifier: run.taskIdentifier,
|
|
run_id: run.friendlyId,
|
|
start_time: formatClickhouseDate64NanosecondsEpochString(startTime.toString()),
|
|
duration: formatClickhouseUnsignedIntegerString(
|
|
calculateDurationFromStart(startTime, cancelledAt)
|
|
),
|
|
trace_id: run.traceId,
|
|
span_id: run.spanId,
|
|
parent_span_id: run.parentSpanId ?? "",
|
|
message: run.taskIdentifier,
|
|
kind: "SPAN",
|
|
status: "CANCELLED",
|
|
attributes: {},
|
|
metadata: JSON.stringify({
|
|
reason,
|
|
}),
|
|
expires_at: expiresAt,
|
|
};
|
|
|
|
this.addToBatch(event);
|
|
}
|
|
|
|
// Query methods
|
|
async getTraceSummary(
|
|
storeTable: TaskEventStoreTable,
|
|
environmentId: string,
|
|
traceId: string,
|
|
startCreatedAt: Date,
|
|
endCreatedAt?: Date,
|
|
options?: { includeDebugLogs?: boolean }
|
|
): Promise<TraceSummary | undefined> {
|
|
const limit = this._config.maximumTraceSummaryViewCount;
|
|
const records = await this.#fetchTraceSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
limit,
|
|
});
|
|
|
|
if (!records) {
|
|
return;
|
|
}
|
|
|
|
const summary = this.#buildTraceSummaryFromRecords(records);
|
|
if (!summary) {
|
|
return;
|
|
}
|
|
|
|
return {
|
|
...summary,
|
|
isTruncated: limit !== undefined && records.length >= limit,
|
|
};
|
|
}
|
|
|
|
async getTraceSubtreeSummary(
|
|
storeTable: TaskEventStoreTable,
|
|
environmentId: string,
|
|
traceId: string,
|
|
anchorSpanId: string,
|
|
startCreatedAt: Date,
|
|
endCreatedAt?: Date,
|
|
options?: { includeDebugLogs?: boolean }
|
|
): Promise<TraceSummary | undefined> {
|
|
const { records, isTruncated, missingAnchor } = await this.#fetchTraceSubtreeRecords({
|
|
environmentId,
|
|
traceId,
|
|
anchorSpanId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
limit: this._config.maximumTraceSummaryViewCount,
|
|
});
|
|
|
|
if (missingAnchor) {
|
|
return;
|
|
}
|
|
|
|
const summary = this.#buildTraceSummaryFromRecords(records, {
|
|
rootSpanId: anchorSpanId,
|
|
});
|
|
|
|
if (!summary) {
|
|
return;
|
|
}
|
|
|
|
return {
|
|
...summary,
|
|
isTruncated,
|
|
};
|
|
}
|
|
|
|
async #fetchTraceSubtreeRecords({
|
|
environmentId,
|
|
traceId,
|
|
anchorSpanId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
limit: maxRows,
|
|
}: {
|
|
environmentId: string;
|
|
traceId: string;
|
|
anchorSpanId: string;
|
|
startCreatedAt: Date;
|
|
endCreatedAt?: Date;
|
|
options?: { includeDebugLogs?: boolean };
|
|
limit?: number;
|
|
}): Promise<{
|
|
records: TaskEventSummaryV1Result[];
|
|
isTruncated: boolean;
|
|
missingAnchor: boolean;
|
|
}> {
|
|
return this.#collectTraceSubtreeRecords({
|
|
anchorSpanId,
|
|
maxRows,
|
|
// Ancestors are fetched by explicit spanIds and start before the anchor
|
|
// run's time window, so applying startCreatedAt would wrongly exclude them
|
|
// (and with it the cancellation/error overrides they propagate downward).
|
|
fetchAncestor: (batch) =>
|
|
this.#fetchTraceSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
skipTimeWindow: true,
|
|
options,
|
|
...batch,
|
|
}),
|
|
fetchDescendant: (batch) =>
|
|
this.#fetchTraceSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
...batch,
|
|
}),
|
|
});
|
|
}
|
|
|
|
async #fetchTraceDetailedSubtreeRecords({
|
|
environmentId,
|
|
traceId,
|
|
anchorSpanId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
limit: maxRows,
|
|
}: {
|
|
environmentId: string;
|
|
traceId: string;
|
|
anchorSpanId: string;
|
|
startCreatedAt: Date;
|
|
endCreatedAt?: Date;
|
|
options?: { includeDebugLogs?: boolean };
|
|
limit?: number;
|
|
}): Promise<{
|
|
records: TaskEventDetailedSummaryV1Result[];
|
|
isTruncated: boolean;
|
|
missingAnchor: boolean;
|
|
}> {
|
|
return this.#collectTraceSubtreeRecords({
|
|
anchorSpanId,
|
|
maxRows,
|
|
// Ancestors are fetched by explicit spanIds and start before the anchor
|
|
// run's time window, so applying startCreatedAt would wrongly exclude them
|
|
// (and with it the cancellation/error overrides they propagate downward).
|
|
fetchAncestor: (batch) =>
|
|
this.#fetchTraceDetailedSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
skipTimeWindow: true,
|
|
options,
|
|
...batch,
|
|
}),
|
|
fetchDescendant: (batch) =>
|
|
this.#fetchTraceDetailedSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
...batch,
|
|
}),
|
|
});
|
|
}
|
|
|
|
async #collectTraceSubtreeRecords<T extends { span_id: string; parent_span_id: string }>({
|
|
anchorSpanId,
|
|
maxRows,
|
|
fetchAncestor,
|
|
fetchDescendant,
|
|
}: {
|
|
anchorSpanId: string;
|
|
maxRows?: number;
|
|
fetchAncestor: (batch: { spanIds: string[]; limit?: number }) => Promise<T[] | undefined>;
|
|
fetchDescendant: (batch: {
|
|
spanIds?: string[];
|
|
parentSpanIds?: string[];
|
|
limit?: number;
|
|
}) => Promise<T[] | undefined>;
|
|
}): Promise<{
|
|
records: T[];
|
|
isTruncated: boolean;
|
|
missingAnchor: boolean;
|
|
}> {
|
|
const allRecords: T[] = [];
|
|
const collectedSpanIds = new Set<string>();
|
|
let isTruncated = false;
|
|
|
|
const anchorRecords = await fetchDescendant({
|
|
spanIds: [anchorSpanId],
|
|
limit: maxRows,
|
|
});
|
|
|
|
if (!anchorRecords || anchorRecords.length === 0) {
|
|
return { records: [], isTruncated: false, missingAnchor: true };
|
|
}
|
|
|
|
if (maxRows && anchorRecords.length >= maxRows) {
|
|
isTruncated = true;
|
|
}
|
|
|
|
allRecords.push(...anchorRecords);
|
|
collectedSpanIds.add(anchorSpanId);
|
|
|
|
let parentSpanId = this.#parentSpanIdFromRecords(anchorRecords, anchorSpanId);
|
|
while (parentSpanId) {
|
|
if (collectedSpanIds.has(parentSpanId)) {
|
|
break;
|
|
}
|
|
|
|
if (maxRows && allRecords.length >= maxRows) {
|
|
isTruncated = true;
|
|
break;
|
|
}
|
|
|
|
const parentRecords = await fetchAncestor({
|
|
spanIds: [parentSpanId],
|
|
limit: maxRows ? maxRows - allRecords.length : undefined,
|
|
});
|
|
|
|
if (!parentRecords || parentRecords.length === 0) {
|
|
break;
|
|
}
|
|
|
|
allRecords.push(...parentRecords);
|
|
collectedSpanIds.add(parentSpanId);
|
|
parentSpanId = this.#parentSpanIdFromRecords(parentRecords, parentSpanId);
|
|
}
|
|
|
|
// Walk descendants level-by-level rather than fetching everything after the anchor in one
|
|
// windowed query. parent_span_id isn't in the sort key, so each level rescans roughly the same
|
|
// granules - but trace depth is small in practice and repeated granule reads stay cached. A
|
|
// single broad query would pull every span after the anchor (a superset of the subtree) and
|
|
// make the maxRows cap drop real subtree spans in favour of unrelated ones.
|
|
let frontier = [anchorSpanId];
|
|
while (frontier.length > 0) {
|
|
if (maxRows && allRecords.length >= maxRows) {
|
|
isTruncated = true;
|
|
break;
|
|
}
|
|
|
|
const remaining = maxRows ? maxRows - allRecords.length : undefined;
|
|
const childRecords = await fetchDescendant({
|
|
parentSpanIds: frontier,
|
|
limit: remaining,
|
|
});
|
|
|
|
if (!childRecords || childRecords.length === 0) {
|
|
break;
|
|
}
|
|
|
|
if (remaining !== undefined && childRecords.length >= remaining) {
|
|
isTruncated = true;
|
|
}
|
|
|
|
allRecords.push(...childRecords);
|
|
|
|
const nextFrontier: string[] = [];
|
|
for (const record of childRecords) {
|
|
if (!collectedSpanIds.has(record.span_id)) {
|
|
collectedSpanIds.add(record.span_id);
|
|
nextFrontier.push(record.span_id);
|
|
}
|
|
}
|
|
|
|
frontier = nextFrontier;
|
|
}
|
|
|
|
return {
|
|
records: allRecords,
|
|
isTruncated,
|
|
missingAnchor: false,
|
|
};
|
|
}
|
|
|
|
#parentSpanIdFromRecords(
|
|
records: Array<{ span_id: string; parent_span_id: string }>,
|
|
spanId: string
|
|
): string | undefined {
|
|
const parentSpanId = records.find((record) => record.span_id === spanId)?.parent_span_id;
|
|
return parentSpanId ? parentSpanId : undefined;
|
|
}
|
|
|
|
#createTraceSummaryQueryBuilder() {
|
|
return this._version === "v2"
|
|
? this._clickhouse.taskEventsV2.traceSummaryQueryBuilder()
|
|
: this._clickhouse.taskEvents.traceSummaryQueryBuilder();
|
|
}
|
|
|
|
async #fetchTraceSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
spanIds,
|
|
parentSpanIds,
|
|
limit,
|
|
skipTimeWindow,
|
|
}: {
|
|
environmentId: string;
|
|
traceId: string;
|
|
startCreatedAt?: Date;
|
|
endCreatedAt?: Date;
|
|
options?: { includeDebugLogs?: boolean };
|
|
spanIds?: string[];
|
|
parentSpanIds?: string[];
|
|
limit?: number;
|
|
skipTimeWindow?: boolean;
|
|
}): Promise<TaskEventSummaryV1Result[] | undefined> {
|
|
const queryBuilder = this.#createTraceSummaryQueryBuilder();
|
|
|
|
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
|
|
queryBuilder.where("trace_id = {traceId: String}", { traceId });
|
|
|
|
if (!skipTimeWindow) {
|
|
if (!startCreatedAt) {
|
|
throw new Error("startCreatedAt is required when skipTimeWindow is false");
|
|
}
|
|
|
|
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 60_000);
|
|
const endCreatedAtWithBuffer = endCreatedAt
|
|
? new Date(endCreatedAt.getTime() + 60_000)
|
|
: undefined;
|
|
|
|
queryBuilder.where("start_time >= {startCreatedAt: String}", {
|
|
startCreatedAt: convertDateToNanoseconds(startCreatedAtWithBuffer).toString(),
|
|
});
|
|
|
|
if (endCreatedAtWithBuffer) {
|
|
queryBuilder.where("start_time <= {endCreatedAt: String}", {
|
|
endCreatedAt: convertDateToNanoseconds(endCreatedAtWithBuffer).toString(),
|
|
});
|
|
}
|
|
|
|
if (this._version === "v2") {
|
|
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
|
|
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
|
|
});
|
|
}
|
|
}
|
|
|
|
if (options?.includeDebugLogs === false) {
|
|
queryBuilder.where("kind != {kind: String}", { kind: "DEBUG_EVENT" });
|
|
}
|
|
|
|
if (spanIds && spanIds.length > 0) {
|
|
queryBuilder.where("span_id IN {spanIds: Array(String)}", { spanIds });
|
|
}
|
|
|
|
if (parentSpanIds && parentSpanIds.length > 0) {
|
|
queryBuilder.where("parent_span_id IN {parentSpanIds: Array(String)}", { parentSpanIds });
|
|
}
|
|
|
|
queryBuilder.orderBy("start_time ASC");
|
|
|
|
if (limit) {
|
|
queryBuilder.limit(limit);
|
|
}
|
|
|
|
const [queryError, records] = await queryBuilder.execute();
|
|
|
|
if (queryError) {
|
|
throw queryError;
|
|
}
|
|
|
|
return records;
|
|
}
|
|
|
|
#buildTraceSummaryFromRecords(
|
|
records: TaskEventSummaryV1Result[],
|
|
options?: { rootSpanId?: string }
|
|
): TraceSummary | undefined {
|
|
if (records.length === 0) {
|
|
return;
|
|
}
|
|
|
|
const recordsGroupedBySpanId: Record<string, TaskEventSummaryV1Result[]> = {};
|
|
for (const record of records) {
|
|
if (!recordsGroupedBySpanId[record.span_id]) {
|
|
recordsGroupedBySpanId[record.span_id] = [];
|
|
}
|
|
recordsGroupedBySpanId[record.span_id].push(record);
|
|
}
|
|
|
|
const spanSummaries = new Map<string, SpanSummary>();
|
|
let rootSpanId: string | undefined = options?.rootSpanId;
|
|
|
|
const metadataCache = new Map<string, Record<string, unknown>>();
|
|
|
|
for (const [spanId, spanRecords] of Object.entries(recordsGroupedBySpanId)) {
|
|
const spanSummary = this.#mergeRecordsIntoSpanSummary(spanId, spanRecords, metadataCache);
|
|
|
|
if (!spanSummary) {
|
|
continue;
|
|
}
|
|
|
|
spanSummaries.set(spanId, spanSummary);
|
|
|
|
if (!rootSpanId && !spanSummary.parentId) {
|
|
rootSpanId = spanId;
|
|
}
|
|
}
|
|
|
|
if (!rootSpanId) {
|
|
return;
|
|
}
|
|
|
|
const spans = Array.from(spanSummaries.values());
|
|
const rootSpan = spanSummaries.get(rootSpanId);
|
|
|
|
if (!rootSpan) {
|
|
return;
|
|
}
|
|
|
|
const overridesBySpanId: Record<string, SpanOverride> = {};
|
|
|
|
const finalSpans = spans.map((span) => {
|
|
return this.#applyAncestorOverrides(span, spanSummaries, overridesBySpanId);
|
|
});
|
|
|
|
return {
|
|
rootSpan,
|
|
spans: finalSpans,
|
|
overridesBySpanId,
|
|
};
|
|
}
|
|
|
|
async getSpan(
|
|
storeTable: TaskEventStoreTable,
|
|
environmentId: string,
|
|
spanId: string,
|
|
traceId: string,
|
|
startCreatedAt: Date,
|
|
endCreatedAt?: Date,
|
|
options?: { includeDebugLogs?: boolean }
|
|
): Promise<SpanDetail | undefined> {
|
|
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 60_000);
|
|
|
|
const queryBuilder =
|
|
this._version === "v2"
|
|
? this._clickhouse.taskEventsV2.spanDetailsQueryBuilder()
|
|
: this._clickhouse.taskEvents.spanDetailsQueryBuilder();
|
|
|
|
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
|
|
queryBuilder.where("trace_id = {traceId: String}", { traceId });
|
|
queryBuilder.where("span_id = {spanId: String}", { spanId });
|
|
queryBuilder.where("start_time >= {startCreatedAt: String}", {
|
|
startCreatedAt: convertDateToNanoseconds(startCreatedAtWithBuffer).toString(),
|
|
});
|
|
|
|
if (endCreatedAt) {
|
|
queryBuilder.where("start_time <= {endCreatedAt: String}", {
|
|
endCreatedAt: convertDateToNanoseconds(endCreatedAt).toString(),
|
|
});
|
|
}
|
|
|
|
// For v2, add inserted_at filtering for partition pruning
|
|
if (this._version === "v2") {
|
|
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
|
|
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
|
|
});
|
|
}
|
|
|
|
queryBuilder.orderBy("start_time ASC");
|
|
|
|
const [queryError, records] = await queryBuilder.execute();
|
|
|
|
if (queryError) {
|
|
throw queryError;
|
|
}
|
|
|
|
if (!records) {
|
|
return;
|
|
}
|
|
|
|
// Create temporary metadata cache for this query
|
|
const metadataCache = new Map<string, Record<string, unknown>>();
|
|
const span = this.#mergeRecordsIntoSpanDetail(spanId, records, metadataCache);
|
|
|
|
return span;
|
|
}
|
|
|
|
#mergeRecordsIntoSpanDetail(
|
|
spanId: string,
|
|
records: TaskEventDetailsV1Result[],
|
|
metadataCache: Map<string, Record<string, unknown>>
|
|
): SpanDetail | undefined {
|
|
if (records.length === 0) {
|
|
return undefined;
|
|
}
|
|
|
|
let span: SpanDetail | undefined;
|
|
let earliestStartTime: Date | undefined;
|
|
|
|
for (const record of records) {
|
|
const recordStartTime = convertClickhouseDateTime64ToJsDate(record.start_time);
|
|
|
|
// Track the earliest start time across all records
|
|
if (
|
|
record.kind !== "ANCESTOR_OVERRIDE" &&
|
|
record.kind !== "SPAN_EVENT" &&
|
|
(!earliestStartTime || recordStartTime < earliestStartTime)
|
|
) {
|
|
earliestStartTime = recordStartTime;
|
|
}
|
|
|
|
if (!span) {
|
|
span = {
|
|
spanId: spanId,
|
|
parentId: record.parent_span_id ? record.parent_span_id : null,
|
|
message: record.message,
|
|
isError: false,
|
|
isPartial: true, // Partial by default, can only be set to false
|
|
isCancelled: false,
|
|
level: kindToLevel(record.kind),
|
|
startTime: recordStartTime,
|
|
duration: typeof record.duration === "number" ? record.duration : Number(record.duration),
|
|
events: [],
|
|
style: {},
|
|
properties: undefined,
|
|
resourceProperties: undefined,
|
|
entity: {
|
|
type: undefined,
|
|
id: undefined,
|
|
metadata: undefined,
|
|
},
|
|
metadata: {},
|
|
};
|
|
}
|
|
|
|
if (isLogEvent(record.kind)) {
|
|
span.isPartial = false;
|
|
span.isCancelled = false;
|
|
span.isError = record.status === "ERROR";
|
|
}
|
|
|
|
const parsedMetadata = this.#parseMetadata(record.metadata, metadataCache);
|
|
|
|
if (record.kind === "SPAN_EVENT") {
|
|
// We need to add an event to the span
|
|
span.events.push({
|
|
name: record.message,
|
|
time: recordStartTime,
|
|
properties: parsedMetadata ?? {},
|
|
});
|
|
}
|
|
|
|
if (parsedMetadata && "style" in parsedMetadata && parsedMetadata.style) {
|
|
span.style = parsedMetadata.style as TaskEventStyle;
|
|
}
|
|
|
|
if (
|
|
parsedMetadata &&
|
|
"entity" in parsedMetadata &&
|
|
typeof parsedMetadata.entity === "object" &&
|
|
parsedMetadata.entity &&
|
|
"entityType" in parsedMetadata.entity &&
|
|
typeof parsedMetadata.entity.entityType === "string" &&
|
|
"entityId" in parsedMetadata.entity &&
|
|
typeof parsedMetadata.entity.entityId === "string"
|
|
) {
|
|
span.entity = {
|
|
id: parsedMetadata.entity.entityId,
|
|
type: parsedMetadata.entity.entityType,
|
|
metadata:
|
|
"entityMetadata" in parsedMetadata.entity &&
|
|
parsedMetadata.entity.entityMetadata &&
|
|
typeof parsedMetadata.entity.entityMetadata === "string"
|
|
? parsedMetadata.entity.entityMetadata
|
|
: undefined,
|
|
};
|
|
}
|
|
|
|
if (record.kind === "SPAN") {
|
|
// Prefer SPAN record message for span title (task name); SPAN_EVENT "exception" must not override it
|
|
span.message = record.message;
|
|
if (record.status === "ERROR") {
|
|
span.isError = true;
|
|
span.isPartial = false;
|
|
span.isCancelled = false;
|
|
} else if (record.status === "CANCELLED") {
|
|
span.isCancelled = true;
|
|
span.isPartial = false;
|
|
span.isError = false;
|
|
} else if (record.status === "OK") {
|
|
span.isPartial = false;
|
|
}
|
|
|
|
if (record.status !== "PARTIAL") {
|
|
span.duration =
|
|
typeof record.duration === "number" ? record.duration : Number(record.duration);
|
|
}
|
|
}
|
|
|
|
// Parse attributes from the first record that has them, then re-parse for the
|
|
// completed SPAN record. The completed record's attributes are a superset of the
|
|
// partial's (includes enriched trigger.llm.* cost data added during ingestion).
|
|
// This means at most 2x JSON.parse per span detail query, but only on this
|
|
// read path (span detail view), not on ingestion.
|
|
if (typeof record.attributes_text === "string") {
|
|
const shouldUpdate =
|
|
span.properties == null ||
|
|
(typeof span.properties === "object" && Object.keys(span.properties).length === 0) ||
|
|
(record.kind === "SPAN" && record.status !== "PARTIAL");
|
|
|
|
if (shouldUpdate) {
|
|
const parsedAttributes = this.#parseAttributes(record.attributes_text);
|
|
const resourceAttributes = parsedAttributes["$resource"];
|
|
|
|
delete parsedAttributes["$resource"];
|
|
|
|
span.properties = parsedAttributes;
|
|
span.resourceProperties = resourceAttributes as Record<string, unknown> | undefined;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Always use the earliest start time found across all records
|
|
if (span && earliestStartTime) {
|
|
span.startTime = earliestStartTime;
|
|
}
|
|
|
|
return span;
|
|
}
|
|
|
|
#parseAttributes(attributes_text: string): Record<string, unknown> {
|
|
if (!attributes_text) {
|
|
return {};
|
|
}
|
|
|
|
return JSON.parse(attributes_text) as Record<string, unknown>;
|
|
}
|
|
|
|
#applyAncestorOverrides<TSpanSummary extends SpanSummaryCommon>(
|
|
span: TSpanSummary,
|
|
spansById: Map<string, TSpanSummary>,
|
|
overridesBySpanId: Record<string, SpanOverride>
|
|
): TSpanSummary {
|
|
if (span.data.level !== "TRACE") {
|
|
return span;
|
|
}
|
|
|
|
if (!span.data.isPartial) {
|
|
return span;
|
|
}
|
|
|
|
if (!span.parentId) {
|
|
return span;
|
|
}
|
|
|
|
// Now we need to walk the ancestors of the span by span.parentId
|
|
// The first ancestor that is a TRACE span that is "closed" we will use to override the span
|
|
let parentSpanId: string | undefined = span.parentId;
|
|
let overrideSpan: TSpanSummary | undefined;
|
|
|
|
while (parentSpanId) {
|
|
const parentSpan = spansById.get(parentSpanId);
|
|
|
|
if (!parentSpan) {
|
|
break;
|
|
}
|
|
|
|
if (parentSpan.data.level === "TRACE" && !parentSpan.data.isPartial) {
|
|
overrideSpan = parentSpan;
|
|
break;
|
|
}
|
|
|
|
parentSpanId = parentSpan.parentId;
|
|
}
|
|
|
|
if (overrideSpan) {
|
|
return this.#applyAncestorToSpan(span, overrideSpan, overridesBySpanId);
|
|
}
|
|
|
|
return span;
|
|
}
|
|
|
|
#applyAncestorToSpan<TSpanSummary extends SpanSummaryCommon>(
|
|
span: TSpanSummary,
|
|
overrideSpan: TSpanSummary,
|
|
overridesBySpanId: Record<string, SpanOverride>
|
|
): TSpanSummary {
|
|
if (overridesBySpanId[span.id]) {
|
|
return span;
|
|
}
|
|
|
|
let override: SpanOverride | undefined = undefined;
|
|
|
|
const overrideEndTime = calculateEndTimeFromStartTime(
|
|
overrideSpan.data.startTime,
|
|
overrideSpan.data.duration
|
|
);
|
|
|
|
if (overrideSpan.data.isCancelled) {
|
|
override = {
|
|
isCancelled: true,
|
|
duration: calculateDurationFromStartJsDate(span.data.startTime, overrideEndTime),
|
|
};
|
|
|
|
span.data.isCancelled = true;
|
|
span.data.isPartial = false;
|
|
span.data.isError = false;
|
|
span.data.duration = calculateDurationFromStartJsDate(span.data.startTime, overrideEndTime);
|
|
|
|
const cancellationEvent = overrideSpan.data.events.find(
|
|
(event) => event.name === "cancellation"
|
|
);
|
|
|
|
if (cancellationEvent) {
|
|
span.data.events.push(cancellationEvent);
|
|
override.events = [cancellationEvent];
|
|
}
|
|
}
|
|
|
|
if (overrideSpan.data.isError && span.data.attemptNumber) {
|
|
const attemptFailedEvent = overrideSpan.data.events.find(
|
|
(event) =>
|
|
event.name === "attempt_failed" &&
|
|
event.properties.attemptNumber === span.data.attemptNumber &&
|
|
event.properties.runId === span.runId
|
|
) as AttemptFailedSpanEvent | undefined;
|
|
|
|
if (attemptFailedEvent) {
|
|
const exceptionEvent = {
|
|
name: "exception",
|
|
time: attemptFailedEvent.time,
|
|
properties: {
|
|
exception: attemptFailedEvent.properties.exception,
|
|
},
|
|
} satisfies ExceptionSpanEvent;
|
|
|
|
span.data.isError = true;
|
|
span.data.isPartial = false;
|
|
span.data.isCancelled = false;
|
|
span.data.duration = calculateDurationFromStartJsDate(span.data.startTime, overrideEndTime);
|
|
span.data.events.push(exceptionEvent);
|
|
span.data.events.push(attemptFailedEvent);
|
|
|
|
override = {
|
|
isError: true,
|
|
events: [exceptionEvent],
|
|
duration: calculateDurationFromStartJsDate(span.data.startTime, overrideEndTime),
|
|
};
|
|
}
|
|
}
|
|
|
|
if (override) {
|
|
overridesBySpanId[span.id] = override;
|
|
}
|
|
|
|
return span;
|
|
}
|
|
|
|
#mergeRecordsIntoSpanSummary(
|
|
spanId: string,
|
|
records: TaskEventSummaryV1Result[],
|
|
metadataCache: Map<string, Record<string, unknown>>
|
|
): SpanSummary | undefined {
|
|
if (records.length === 0) {
|
|
return undefined;
|
|
}
|
|
|
|
let span: SpanSummary | undefined;
|
|
let earliestStartTime: Date | undefined;
|
|
|
|
for (const record of records) {
|
|
const recordStartTime = convertClickhouseDateTime64ToJsDate(record.start_time);
|
|
|
|
// Track the earliest start time across all records, except for ancestor overrides and span events
|
|
if (
|
|
record.kind !== "ANCESTOR_OVERRIDE" &&
|
|
record.kind !== "SPAN_EVENT" &&
|
|
(!earliestStartTime || recordStartTime < earliestStartTime)
|
|
) {
|
|
earliestStartTime = recordStartTime;
|
|
}
|
|
|
|
if (!span) {
|
|
span = {
|
|
id: spanId,
|
|
parentId: record.parent_span_id ? record.parent_span_id : undefined,
|
|
runId: record.run_id,
|
|
data: {
|
|
message: record.message,
|
|
style: {},
|
|
duration:
|
|
typeof record.duration === "number" ? record.duration : Number(record.duration),
|
|
isError: false,
|
|
isPartial: true, // Partial by default, can only be set to false
|
|
isCancelled: false,
|
|
isDebug: record.kind === "DEBUG_EVENT",
|
|
startTime: recordStartTime,
|
|
level: kindToLevel(record.kind),
|
|
events: [],
|
|
},
|
|
};
|
|
}
|
|
|
|
if (isLogEvent(record.kind)) {
|
|
span.data.isPartial = false;
|
|
span.data.isCancelled = false;
|
|
span.data.isError = record.status === "ERROR";
|
|
}
|
|
|
|
const parsedMetadata = this.#parseMetadata(record.metadata, metadataCache);
|
|
|
|
if (
|
|
parsedMetadata &&
|
|
"attemptNumber" in parsedMetadata &&
|
|
typeof parsedMetadata.attemptNumber === "number"
|
|
) {
|
|
span.data.attemptNumber = parsedMetadata.attemptNumber;
|
|
}
|
|
|
|
if (record.kind === "ANCESTOR_OVERRIDE" || record.kind === "SPAN_EVENT") {
|
|
// We need to add an event to the span
|
|
span.data.events.push({
|
|
name: record.message,
|
|
time: recordStartTime,
|
|
properties: parsedMetadata ?? {},
|
|
});
|
|
}
|
|
|
|
if (parsedMetadata && "style" in parsedMetadata && parsedMetadata.style) {
|
|
const newStyle = parsedMetadata.style as TaskEventStyle;
|
|
// Merge styles: prefer the most complete value for each field
|
|
span.data.style = {
|
|
icon: newStyle.icon ?? span.data.style.icon,
|
|
variant: newStyle.variant ?? span.data.style.variant,
|
|
accessory: newStyle.accessory ?? span.data.style.accessory,
|
|
};
|
|
}
|
|
|
|
if (record.kind === "SPAN") {
|
|
// Prefer SPAN record message for span title (task name); SPAN_EVENT "exception" must not override it
|
|
span.data.message = record.message;
|
|
if (record.status === "ERROR") {
|
|
span.data.isError = true;
|
|
span.data.isPartial = false;
|
|
span.data.isCancelled = false;
|
|
} else if (record.status === "CANCELLED") {
|
|
span.data.isCancelled = true;
|
|
span.data.isPartial = false;
|
|
span.data.isError = false;
|
|
} else if (record.status === "OK") {
|
|
span.data.isPartial = false;
|
|
}
|
|
|
|
if (record.status !== "PARTIAL") {
|
|
span.data.duration =
|
|
typeof record.duration === "number" ? record.duration : Number(record.duration);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Always use the earliest start time found across all records
|
|
if (span && earliestStartTime) {
|
|
span.data.startTime = earliestStartTime;
|
|
}
|
|
|
|
return span;
|
|
}
|
|
|
|
#parseMetadata(
|
|
metadata: string,
|
|
cache: Map<string, Record<string, unknown>>
|
|
): Record<string, unknown> | undefined {
|
|
if (!metadata) {
|
|
return undefined;
|
|
}
|
|
|
|
// Check cache first
|
|
const cached = cache.get(metadata);
|
|
if (cached) {
|
|
return cached;
|
|
}
|
|
|
|
const parsed = JSON.parse(metadata);
|
|
|
|
if (typeof parsed !== "object" || parsed === null) {
|
|
return undefined;
|
|
}
|
|
|
|
const result = parsed as Record<string, unknown>;
|
|
|
|
// Cache the result - no size limit needed since cache is per-query
|
|
cache.set(metadata, result);
|
|
|
|
return result;
|
|
}
|
|
|
|
#createTraceDetailedSummaryQueryBuilder() {
|
|
return this._version === "v2"
|
|
? this._clickhouse.taskEventsV2.traceDetailedSummaryQueryBuilder()
|
|
: this._clickhouse.taskEvents.traceDetailedSummaryQueryBuilder();
|
|
}
|
|
|
|
async #fetchTraceDetailedSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
spanIds,
|
|
parentSpanIds,
|
|
limit,
|
|
skipTimeWindow,
|
|
}: {
|
|
environmentId: string;
|
|
traceId: string;
|
|
startCreatedAt?: Date;
|
|
endCreatedAt?: Date;
|
|
options?: { includeDebugLogs?: boolean };
|
|
spanIds?: string[];
|
|
parentSpanIds?: string[];
|
|
limit?: number;
|
|
skipTimeWindow?: boolean;
|
|
}): Promise<TaskEventDetailedSummaryV1Result[] | undefined> {
|
|
const queryBuilder = this.#createTraceDetailedSummaryQueryBuilder();
|
|
|
|
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
|
|
queryBuilder.where("trace_id = {traceId: String}", { traceId });
|
|
|
|
if (!skipTimeWindow) {
|
|
if (!startCreatedAt) {
|
|
throw new Error("startCreatedAt is required when skipTimeWindow is false");
|
|
}
|
|
|
|
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 1000);
|
|
|
|
queryBuilder.where("start_time >= {startCreatedAt: String}", {
|
|
startCreatedAt: convertDateToNanoseconds(startCreatedAtWithBuffer).toString(),
|
|
});
|
|
|
|
if (endCreatedAt) {
|
|
queryBuilder.where("start_time <= {endCreatedAt: String}", {
|
|
endCreatedAt: convertDateToNanoseconds(endCreatedAt).toString(),
|
|
});
|
|
}
|
|
|
|
if (this._version === "v2") {
|
|
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
|
|
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
|
|
});
|
|
}
|
|
}
|
|
|
|
if (options?.includeDebugLogs === false) {
|
|
queryBuilder.where("kind != {kind: String}", { kind: "DEBUG_EVENT" });
|
|
}
|
|
|
|
if (spanIds && spanIds.length > 0) {
|
|
queryBuilder.where("span_id IN {spanIds: Array(String)}", { spanIds });
|
|
}
|
|
|
|
if (parentSpanIds && parentSpanIds.length > 0) {
|
|
queryBuilder.where("parent_span_id IN {parentSpanIds: Array(String)}", { parentSpanIds });
|
|
}
|
|
|
|
queryBuilder.orderBy("start_time ASC");
|
|
|
|
if (limit) {
|
|
queryBuilder.limit(limit);
|
|
}
|
|
|
|
const [queryError, records] = await queryBuilder.execute();
|
|
|
|
if (queryError) {
|
|
throw queryError;
|
|
}
|
|
|
|
return records;
|
|
}
|
|
|
|
#buildTraceDetailedSummaryFromRecords(
|
|
traceId: string,
|
|
records: TaskEventDetailedSummaryV1Result[],
|
|
rootSpanId?: string
|
|
): TraceDetailedSummary | undefined {
|
|
if (records.length === 0) {
|
|
return;
|
|
}
|
|
|
|
const recordsGroupedBySpanId: Record<string, TaskEventDetailedSummaryV1Result[]> = {};
|
|
for (const record of records) {
|
|
if (!recordsGroupedBySpanId[record.span_id]) {
|
|
recordsGroupedBySpanId[record.span_id] = [];
|
|
}
|
|
recordsGroupedBySpanId[record.span_id].push(record);
|
|
}
|
|
|
|
const spanSummaries = new Map<string, SpanDetailedSummary>();
|
|
let resolvedRootSpanId: string | undefined = rootSpanId;
|
|
|
|
const metadataCache = new Map<string, Record<string, unknown>>();
|
|
|
|
for (const [spanId, spanRecords] of Object.entries(recordsGroupedBySpanId)) {
|
|
const spanSummary = this.#mergeRecordsIntoSpanDetailedSummary(
|
|
spanId,
|
|
spanRecords,
|
|
metadataCache
|
|
);
|
|
|
|
if (!spanSummary) {
|
|
continue;
|
|
}
|
|
|
|
spanSummaries.set(spanId, spanSummary);
|
|
|
|
if (!resolvedRootSpanId && !spanSummary.parentId) {
|
|
resolvedRootSpanId = spanId;
|
|
}
|
|
}
|
|
|
|
if (!resolvedRootSpanId) {
|
|
return;
|
|
}
|
|
|
|
const spans = Array.from(spanSummaries.values());
|
|
|
|
const overridesBySpanId: Record<string, SpanOverride> = {};
|
|
const spanDetailedSummaryMap = new Map<string, SpanDetailedSummary>();
|
|
|
|
const finalSpans = spans.map((span) => {
|
|
const finalSpan = this.#applyAncestorOverrides(span, spanSummaries, overridesBySpanId);
|
|
spanDetailedSummaryMap.set(span.id, finalSpan);
|
|
return finalSpan;
|
|
});
|
|
|
|
for (const finalSpan of finalSpans) {
|
|
if (finalSpan.parentId) {
|
|
const parent = spanDetailedSummaryMap.get(finalSpan.parentId);
|
|
if (parent) {
|
|
parent.children.push(finalSpan);
|
|
}
|
|
}
|
|
}
|
|
|
|
const rootSpan = spanDetailedSummaryMap.get(resolvedRootSpanId);
|
|
|
|
if (!rootSpan) {
|
|
return;
|
|
}
|
|
|
|
return {
|
|
traceId,
|
|
rootSpan,
|
|
};
|
|
}
|
|
|
|
async getTraceDetailedSummary(
|
|
storeTable: TaskEventStoreTable,
|
|
environmentId: string,
|
|
traceId: string,
|
|
startCreatedAt: Date,
|
|
endCreatedAt?: Date,
|
|
options?: { includeDebugLogs?: boolean }
|
|
): Promise<TraceDetailedSummary | undefined> {
|
|
const limit = this._config.maximumTraceDetailedSummaryViewCount;
|
|
const records = await this.#fetchTraceDetailedSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
limit,
|
|
});
|
|
|
|
if (!records) {
|
|
return;
|
|
}
|
|
|
|
const summary = this.#buildTraceDetailedSummaryFromRecords(traceId, records);
|
|
if (!summary) {
|
|
return;
|
|
}
|
|
|
|
return {
|
|
...summary,
|
|
isTruncated: limit !== undefined && records.length >= limit,
|
|
};
|
|
}
|
|
|
|
async getTraceDetailedSubtreeSummary(
|
|
storeTable: TaskEventStoreTable,
|
|
environmentId: string,
|
|
traceId: string,
|
|
anchorSpanId: string,
|
|
startCreatedAt: Date,
|
|
endCreatedAt?: Date,
|
|
options?: { includeDebugLogs?: boolean }
|
|
): Promise<TraceDetailedSummary | undefined> {
|
|
const limit = this._config.maximumTraceDetailedSummaryViewCount;
|
|
|
|
// Try one capped full-trace query first so the common case stays at a single
|
|
// round-trip; large traces pay an extra fetch before the subtree walk below.
|
|
const fullRecords = await this.#fetchTraceDetailedSummaryRecords({
|
|
environmentId,
|
|
traceId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
limit,
|
|
});
|
|
|
|
if (fullRecords && this.#canReRootDetailedRecordsAtAnchor(fullRecords, anchorSpanId)) {
|
|
const summary = this.#buildTraceDetailedSummaryFromRecords(
|
|
traceId,
|
|
fullRecords,
|
|
anchorSpanId
|
|
);
|
|
if (summary) {
|
|
return {
|
|
...summary,
|
|
isTruncated: limit !== undefined && fullRecords.length >= limit,
|
|
};
|
|
}
|
|
}
|
|
|
|
const { records, isTruncated, missingAnchor } = await this.#fetchTraceDetailedSubtreeRecords({
|
|
environmentId,
|
|
traceId,
|
|
anchorSpanId,
|
|
startCreatedAt,
|
|
endCreatedAt,
|
|
options,
|
|
limit,
|
|
});
|
|
|
|
if (missingAnchor) {
|
|
return;
|
|
}
|
|
|
|
const summary = this.#buildTraceDetailedSummaryFromRecords(traceId, records, anchorSpanId);
|
|
if (!summary) {
|
|
return;
|
|
}
|
|
|
|
return {
|
|
...summary,
|
|
isTruncated,
|
|
};
|
|
}
|
|
|
|
// Only checks the direct parent — not the full ancestor chain. Safe in practice
|
|
// because ancestors have earlier start_time and usually land inside the cap
|
|
// when the anchor does; otherwise we fall back to the subtree walk.
|
|
#canReRootDetailedRecordsAtAnchor(
|
|
records: Array<{ span_id: string; parent_span_id: string }>,
|
|
anchorSpanId: string
|
|
): boolean {
|
|
const anchorRecord = records.find((record) => record.span_id === anchorSpanId);
|
|
if (!anchorRecord) {
|
|
return false;
|
|
}
|
|
|
|
const parentSpanId = anchorRecord.parent_span_id;
|
|
if (!parentSpanId) {
|
|
return true;
|
|
}
|
|
|
|
return records.some((record) => record.span_id === parentSpanId);
|
|
}
|
|
|
|
async *streamTraceEvents(
|
|
storeTable: TaskEventStoreTable,
|
|
environmentId: string,
|
|
traceId: string,
|
|
startCreatedAt: Date,
|
|
endCreatedAt?: Date,
|
|
options?: { includeDebugLogs?: boolean }
|
|
): AsyncIterable<StreamedTraceEvent> {
|
|
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 1000);
|
|
|
|
const queryBuilder =
|
|
this._version === "v2"
|
|
? this._clickhouse.taskEventsV2.traceEventsForExportQueryBuilder()
|
|
: this._clickhouse.taskEvents.traceEventsForExportQueryBuilder();
|
|
|
|
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
|
|
queryBuilder.where("trace_id = {traceId: String}", { traceId });
|
|
queryBuilder.where("start_time >= {startCreatedAt: String}", {
|
|
startCreatedAt: convertDateToNanoseconds(startCreatedAtWithBuffer).toString(),
|
|
});
|
|
|
|
if (endCreatedAt) {
|
|
queryBuilder.where("start_time <= {endCreatedAt: String}", {
|
|
endCreatedAt: convertDateToNanoseconds(endCreatedAt).toString(),
|
|
});
|
|
}
|
|
|
|
if (this._version === "v2") {
|
|
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
|
|
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
|
|
});
|
|
}
|
|
|
|
// Admin-only debug events stay hidden unless explicitly requested.
|
|
if (options?.includeDebugLogs !== true) {
|
|
queryBuilder.where("kind != {debugKind: String}", { debugKind: "DEBUG_EVENT" });
|
|
}
|
|
|
|
// Each span is written twice: a PARTIAL start-marker (empty attributes) and
|
|
// the completed row. Keep only the completed row so the export has one line
|
|
// per span (the tree path merges these; streaming can't, so we filter).
|
|
queryBuilder.where("status != {partialStatus: String}", { partialStatus: "PARTIAL" });
|
|
|
|
// Internal trigger.dev span events (start timeline) are uninformative noise
|
|
// in the export; the tree path filters them too. Real span events such as
|
|
// exceptions are kept.
|
|
queryBuilder.where(
|
|
"(kind != {spanEventKind: String} OR NOT startsWith(message, {internalPrefix: String}))",
|
|
{ spanEventKind: "SPAN_EVENT", internalPrefix: "trigger.dev/" }
|
|
);
|
|
|
|
// ANCESTOR_OVERRIDE rows duplicate a descendant's error onto an ancestor span
|
|
// to colour the tree; they carry no event of their own. The tree path drops
|
|
// them, so the export does too (otherwise the same error shows up twice).
|
|
queryBuilder.where("kind != {ancestorKind: String}", { ancestorKind: "ANCESTOR_OVERRIDE" });
|
|
|
|
queryBuilder.orderBy("start_time ASC");
|
|
// Deliberately no LIMIT: streaming never materialises the result set, so the
|
|
// detailed-summary memory cap doesn't apply to the export.
|
|
|
|
for await (const row of queryBuilder.executeStream()) {
|
|
yield {
|
|
spanId: row.span_id,
|
|
parentSpanId: row.parent_span_id,
|
|
startTime: convertClickhouseDateTime64ToJsDate(row.start_time),
|
|
durationNs: typeof row.duration === "number" ? row.duration : Number(row.duration),
|
|
level: clickhouseKindToLevel(row.kind),
|
|
message: row.message,
|
|
isError: row.status === "ERROR",
|
|
propertiesText: row.attributes_text ?? "",
|
|
};
|
|
}
|
|
}
|
|
|
|
#mergeRecordsIntoSpanDetailedSummary(
|
|
spanId: string,
|
|
records: TaskEventDetailedSummaryV1Result[],
|
|
metadataCache: Map<string, Record<string, unknown>>
|
|
): SpanDetailedSummary | undefined {
|
|
if (records.length === 0) {
|
|
return undefined;
|
|
}
|
|
|
|
let span: SpanDetailedSummary | undefined;
|
|
let earliestStartTime: Date | undefined;
|
|
|
|
for (const record of records) {
|
|
const recordStartTime = convertClickhouseDateTime64ToJsDate(record.start_time);
|
|
|
|
// Track the earliest start time across all records
|
|
if (
|
|
record.kind !== "ANCESTOR_OVERRIDE" &&
|
|
record.kind !== "SPAN_EVENT" &&
|
|
(!earliestStartTime || recordStartTime < earliestStartTime)
|
|
) {
|
|
earliestStartTime = recordStartTime;
|
|
}
|
|
|
|
if (!span) {
|
|
span = {
|
|
id: spanId,
|
|
parentId: record.parent_span_id ? record.parent_span_id : undefined,
|
|
runId: record.run_id,
|
|
data: {
|
|
message: record.message,
|
|
taskSlug: undefined,
|
|
duration:
|
|
typeof record.duration === "number" ? record.duration : Number(record.duration),
|
|
isError: false,
|
|
isPartial: true, // Partial by default, can only be set to false
|
|
isCancelled: false,
|
|
startTime: recordStartTime,
|
|
level: kindToLevel(record.kind),
|
|
events: [],
|
|
},
|
|
children: [],
|
|
};
|
|
}
|
|
|
|
if (isLogEvent(record.kind)) {
|
|
span.data.isPartial = false;
|
|
span.data.isCancelled = false;
|
|
span.data.isError = record.status === "ERROR";
|
|
}
|
|
|
|
const parsedMetadata = this.#parseMetadata(record.metadata, metadataCache);
|
|
|
|
if (
|
|
parsedMetadata &&
|
|
"attemptNumber" in parsedMetadata &&
|
|
typeof parsedMetadata.attemptNumber === "number"
|
|
) {
|
|
span.data.attemptNumber = parsedMetadata.attemptNumber;
|
|
}
|
|
|
|
if (record.kind === "ANCESTOR_OVERRIDE" || record.kind === "SPAN_EVENT") {
|
|
// We need to add an event to the span
|
|
span.data.events.push({
|
|
name: record.message,
|
|
time: recordStartTime,
|
|
properties: parsedMetadata ?? {},
|
|
});
|
|
}
|
|
|
|
if (record.kind === "SPAN") {
|
|
// Prefer SPAN record message for span title (task name); SPAN_EVENT "exception" must not override it
|
|
span.data.message = record.message;
|
|
if (record.status === "ERROR") {
|
|
span.data.isError = true;
|
|
span.data.isPartial = false;
|
|
span.data.isCancelled = false;
|
|
} else if (record.status === "CANCELLED") {
|
|
span.data.isCancelled = true;
|
|
span.data.isPartial = false;
|
|
span.data.isError = false;
|
|
} else if (record.status === "OK") {
|
|
span.data.isPartial = false;
|
|
}
|
|
|
|
if (record.status !== "PARTIAL") {
|
|
span.data.duration =
|
|
typeof record.duration === "number" ? record.duration : Number(record.duration);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Always use the earliest start time found across all records
|
|
if (span && earliestStartTime) {
|
|
span.data.startTime = earliestStartTime;
|
|
}
|
|
|
|
return span;
|
|
}
|
|
|
|
async getRunEvents(
|
|
storeTable: TaskEventStoreTable,
|
|
environmentId: string,
|
|
traceId: string,
|
|
runId: string,
|
|
startCreatedAt: Date,
|
|
endCreatedAt?: Date
|
|
): Promise<RunPreparedEvent[]> {
|
|
const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 1000);
|
|
|
|
const queryBuilder =
|
|
this._version === "v2"
|
|
? this._clickhouse.taskEventsV2.traceSummaryQueryBuilder()
|
|
: this._clickhouse.taskEvents.traceSummaryQueryBuilder();
|
|
|
|
queryBuilder.where("environment_id = {environmentId: String}", { environmentId });
|
|
queryBuilder.where("trace_id = {traceId: String}", { traceId });
|
|
queryBuilder.where("run_id = {runId: String}", { runId });
|
|
queryBuilder.where("start_time >= {startCreatedAt: String}", {
|
|
startCreatedAt: convertDateToNanoseconds(startCreatedAtWithBuffer).toString(),
|
|
});
|
|
|
|
if (endCreatedAt) {
|
|
queryBuilder.where("start_time <= {endCreatedAt: String}", {
|
|
endCreatedAt: convertDateToNanoseconds(endCreatedAt).toString(),
|
|
});
|
|
}
|
|
|
|
// For v2, add inserted_at filtering for partition pruning
|
|
if (this._version === "v2") {
|
|
queryBuilder.where("inserted_at >= {insertedAtStart: DateTime64(3)}", {
|
|
insertedAtStart: convertDateToClickhouseDateTime(startCreatedAtWithBuffer),
|
|
});
|
|
}
|
|
|
|
queryBuilder.where("kind != {kind: String}", { kind: "DEBUG_EVENT" });
|
|
queryBuilder.orderBy("start_time ASC");
|
|
|
|
if (this._config.maximumTraceSummaryViewCount) {
|
|
queryBuilder.limit(this._config.maximumTraceSummaryViewCount);
|
|
}
|
|
|
|
const [queryError, records] = await queryBuilder.execute();
|
|
|
|
if (queryError) {
|
|
throw queryError;
|
|
}
|
|
|
|
if (!records) {
|
|
return [];
|
|
}
|
|
|
|
// O(n) grouping instead of O(n²) array spreading
|
|
const recordsGroupedBySpanId: Record<string, TaskEventSummaryV1Result[]> = {};
|
|
for (const record of records) {
|
|
if (!recordsGroupedBySpanId[record.span_id]) {
|
|
recordsGroupedBySpanId[record.span_id] = [];
|
|
}
|
|
recordsGroupedBySpanId[record.span_id].push(record);
|
|
}
|
|
|
|
const spanSummaries = new Map<string, SpanSummary>();
|
|
let rootSpanId: string | undefined;
|
|
|
|
// Create temporary metadata cache for this query
|
|
const metadataCache = new Map<string, Record<string, unknown>>();
|
|
|
|
for (const [spanId, spanRecords] of Object.entries(recordsGroupedBySpanId)) {
|
|
const spanSummary = this.#mergeRecordsIntoSpanSummary(spanId, spanRecords, metadataCache);
|
|
|
|
if (!spanSummary) {
|
|
continue;
|
|
}
|
|
|
|
spanSummaries.set(spanId, spanSummary);
|
|
|
|
// Find root span for optimized override algorithm
|
|
if (!rootSpanId && !spanSummary.parentId) {
|
|
rootSpanId = spanId;
|
|
}
|
|
}
|
|
|
|
const spans = Array.from(spanSummaries.values());
|
|
|
|
const overridesBySpanId: Record<string, SpanOverride> = {};
|
|
|
|
const finalSpans = spans.map((span) => {
|
|
return this.#applyAncestorOverrides(span, spanSummaries, overridesBySpanId);
|
|
});
|
|
|
|
const runPreparedEvents = finalSpans.map((span) => this.#spanSummaryToRunPreparedEvent(span));
|
|
|
|
return runPreparedEvents;
|
|
}
|
|
|
|
#spanSummaryToRunPreparedEvent(span: SpanSummary): RunPreparedEvent {
|
|
return {
|
|
spanId: span.id,
|
|
parentId: span.parentId ?? null,
|
|
runId: span.runId,
|
|
message: span.data.message,
|
|
style: span.data.style,
|
|
events: span.data.events,
|
|
startTime: convertDateToNanoseconds(span.data.startTime),
|
|
duration: span.data.duration,
|
|
isError: span.data.isError,
|
|
isPartial: span.data.isPartial,
|
|
isCancelled: span.data.isCancelled,
|
|
kind: "UNSPECIFIED",
|
|
attemptNumber: span.data.attemptNumber ?? null,
|
|
level: span.data.level,
|
|
};
|
|
}
|
|
}
|
|
|
|
// Precompile regex for performance (used ~30k times per trace)
|
|
const CLICKHOUSE_DATETIME_REGEX =
|
|
/^(\d{4})-(\d{2})-(\d{2})[ T](\d{2}):(\d{2}):(\d{2})(?:\.(\d{1,9}))?(?:Z|([+-])(\d{2}):?(\d{2}))?$/;
|
|
|
|
export const convertDateToClickhouseDateTime = (date: Date): string => {
|
|
// 2024-11-06T20:37:00.123Z -> 2024-11-06 21:37:00.123
|
|
return date.toISOString().replace("T", " ").replace("Z", "");
|
|
};
|
|
|
|
/**
|
|
* Convert a ClickHouse DateTime64 to nanoseconds since epoch (UTC).
|
|
* Accepts:
|
|
* - "2025-09-23 12:32:46.130262875"
|
|
* - "2025-09-23T12:32:46.13"
|
|
* - "2025-09-23 12:32:46Z"
|
|
* - "2025-09-23 12:32:46.130262875+02:00"
|
|
*/
|
|
export function convertClickhouseDateTime64ToNanosecondsEpoch(date: string): bigint {
|
|
const s = date.trim();
|
|
const m = CLICKHOUSE_DATETIME_REGEX.exec(s);
|
|
if (!m) {
|
|
throw new Error(`Invalid ClickHouse DateTime64 string: "${date}"`);
|
|
}
|
|
|
|
const year = Number(m[1]);
|
|
const month = Number(m[2]); // 1-12
|
|
const day = Number(m[3]); // 1-31
|
|
const hour = Number(m[4]);
|
|
const minute = Number(m[5]);
|
|
const second = Number(m[6]);
|
|
const fraction = m[7] ?? ""; // up to 9 digits
|
|
const sign = m[8] as "+" | "-" | undefined;
|
|
const offH = m[9] ? Number(m[9]) : 0;
|
|
const offM = m[10] ? Number(m[10]) : 0;
|
|
|
|
// Convert fractional seconds to exactly 9 digits (nanoseconds within the second).
|
|
const nsWithinSecond = Number(fraction.padEnd(9, "0")); // 0..999_999_999
|
|
|
|
// Split into millisecond part (for Date) and leftover nanoseconds.
|
|
const msPart = Math.trunc(nsWithinSecond / 1_000_000); // 0..999
|
|
const leftoverNs = nsWithinSecond - msPart * 1_000_000; // 0..999_999
|
|
|
|
// Build milliseconds since epoch in UTC using Date.UTC (avoids local TZ/DST issues).
|
|
let msEpoch = Date.UTC(year, month - 1, day, hour, minute, second, msPart);
|
|
|
|
// If an explicit offset was provided, adjust to true UTC.
|
|
if (sign) {
|
|
const offsetMinutesSigned = (sign === "+" ? 1 : -1) * (offH * 60 + offM);
|
|
msEpoch -= offsetMinutesSigned * 60_000;
|
|
}
|
|
|
|
// Combine ms to ns with leftover.
|
|
return BigInt(msEpoch) * 1_000_000n + BigInt(leftoverNs);
|
|
}
|
|
|
|
/**
|
|
* Convert a ClickHouse DateTime64 to a JS Date.
|
|
* Accepts:
|
|
* - "2025-09-23 12:32:46.130262875"
|
|
* - "2025-09-23T12:32:46.13"
|
|
* - "2025-09-23 12:32:46Z"
|
|
* - "2025-09-23 12:32:46.130262875+02:00"
|
|
*
|
|
* Optimized with fast path for common format (avoids regex for 99% of cases).
|
|
*/
|
|
// Map a ClickHouse task-event `kind` to a human display level for the streaming
|
|
// export (e.g. LOG_INFO -> INFO, SPAN -> TRACE, SPAN_EVENT -> EVENT).
|
|
function clickhouseKindToLevel(kind: string): string {
|
|
if (kind.startsWith("LOG_")) return kind.slice(4);
|
|
if (kind === "SPAN") return "TRACE";
|
|
if (kind === "SPAN_EVENT") return "EVENT";
|
|
if (kind === "DEBUG_EVENT") return "DEBUG";
|
|
return kind;
|
|
}
|
|
|
|
export function convertClickhouseDateTime64ToJsDate(date: string): Date {
|
|
// Fast path for common format: "2025-09-23 12:32:46.130262875" or "2025-09-23 12:32:46"
|
|
// This avoids the expensive regex for the common case
|
|
if (date.length >= 19 && date[4] === "-" && date[7] === "-" && date[10] === " ") {
|
|
const year = Number(date.substring(0, 4));
|
|
const month = Number(date.substring(5, 7));
|
|
const day = Number(date.substring(8, 10));
|
|
const hour = Number(date.substring(11, 13));
|
|
const minute = Number(date.substring(14, 16));
|
|
const second = Number(date.substring(17, 19));
|
|
|
|
// Parse fractional seconds if present
|
|
let ms = 0;
|
|
if (date.length > 20 && date[19] === ".") {
|
|
// Take first 3 digits after decimal (milliseconds), pad if shorter
|
|
const fracStr = date.substring(20, Math.min(23, date.length));
|
|
ms = Number(fracStr.padEnd(3, "0"));
|
|
}
|
|
|
|
return new Date(Date.UTC(year, month - 1, day, hour, minute, second, ms));
|
|
}
|
|
|
|
// Fallback to regex for other formats (T separator, timezone offsets, etc.)
|
|
const s = date.trim();
|
|
const m = CLICKHOUSE_DATETIME_REGEX.exec(s);
|
|
if (!m) {
|
|
throw new Error(`Invalid ClickHouse DateTime64 string: "${date}"`);
|
|
}
|
|
|
|
const year = Number(m[1]);
|
|
const month = Number(m[2]); // 1-12
|
|
const day = Number(m[3]); // 1-31
|
|
const hour = Number(m[4]);
|
|
const minute = Number(m[5]);
|
|
const second = Number(m[6]);
|
|
const fraction = m[7] ?? ""; // up to 9 digits
|
|
|
|
// Convert fractional seconds to exactly 9 digits (nanoseconds within the second).
|
|
const nsWithinSecond = Number(fraction.padEnd(9, "0")); // 0..999_999_999
|
|
|
|
// Split into millisecond part (for Date)
|
|
const msPart = Math.trunc(nsWithinSecond / 1_000_000); // 0..999
|
|
|
|
return new Date(Date.UTC(year, month - 1, day, hour, minute, second, msPart));
|
|
}
|
|
|
|
function kindToLevel(kind: string): TaskEventLevel {
|
|
switch (kind) {
|
|
case "DEBUG_EVENT":
|
|
case "LOG_DEBUG": {
|
|
return "DEBUG";
|
|
}
|
|
case "LOG_LOG": {
|
|
return "LOG";
|
|
}
|
|
case "LOG_INFO": {
|
|
return "INFO";
|
|
}
|
|
case "LOG_WARN": {
|
|
return "WARN";
|
|
}
|
|
case "LOG_ERROR": {
|
|
return "ERROR";
|
|
}
|
|
case "SPAN":
|
|
case "ANCESTOR_OVERRIDE":
|
|
case "SPAN_EVENT": {
|
|
return "TRACE";
|
|
}
|
|
default: {
|
|
return "TRACE";
|
|
}
|
|
}
|
|
}
|
|
|
|
function isLogEvent(kind: string): boolean {
|
|
return kind.startsWith("LOG_") || kind === "DEBUG_EVENT";
|
|
}
|
|
|
|
function calculateEndTimeFromStartTime(startTime: Date, duration: number): Date {
|
|
return new Date(startTime.getTime() + duration / 1_000_000);
|
|
}
|
|
|
|
// This will take a string like "1759427319944999936" and return "1759427319.944999936"
|
|
function formatClickhouseDate64NanosecondsEpochString(date: string): string {
|
|
if (date.length !== 19) {
|
|
return date;
|
|
}
|
|
|
|
return date.substring(0, 10) + "." + date.substring(10);
|
|
}
|
|
|
|
function convertClickhouseDate64NanosecondsEpochStringToBigInt(date: string): bigint {
|
|
const parts = date.split(".");
|
|
return BigInt(parts.join(""));
|
|
}
|
|
|
|
function formatClickhouseUnsignedIntegerString(value: number | bigint): string {
|
|
if (value < 0) {
|
|
return "0";
|
|
}
|
|
|
|
if (typeof value === "bigint") {
|
|
return value.toString();
|
|
}
|
|
|
|
return Math.floor(value).toString();
|
|
}
|