Extracted some logic out of the eventRepository for getting a trace. This will be used on the frontend
This commit is contained in:
Matt Aitken
@@ -0,0 +1,210 @@
|
||||
import { Attributes } from "@opentelemetry/api";
|
||||
import { SpanEvents, TaskEventStyle, unflattenAttributes } from "@trigger.dev/core/v3";
|
||||
import { Prisma } from "@trigger.dev/database";
|
||||
import type { PreparedEvent, QueriedEvent, SpanSummary } from "~/v3/eventRepository.server";
|
||||
|
||||
export function prepareTrace(events: QueriedEvent[]) {
|
||||
let preparedEvents: Array<PreparedEvent> = [];
|
||||
let rootSpanId: string | undefined;
|
||||
const eventsBySpanId = new Map<string, PreparedEvent>();
|
||||
|
||||
for (const event of events) {
|
||||
preparedEvents.push(prepareEvent(event));
|
||||
|
||||
if (!rootSpanId && !event.parentId) {
|
||||
rootSpanId = event.spanId;
|
||||
}
|
||||
}
|
||||
|
||||
for (const event of preparedEvents) {
|
||||
const existingEvent = eventsBySpanId.get(event.spanId);
|
||||
|
||||
if (!existingEvent) {
|
||||
eventsBySpanId.set(event.spanId, event);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (event.isCancelled || !event.isPartial) {
|
||||
eventsBySpanId.set(event.spanId, event);
|
||||
}
|
||||
}
|
||||
|
||||
preparedEvents = Array.from(eventsBySpanId.values());
|
||||
|
||||
const spansBySpanId = new Map<string, SpanSummary>();
|
||||
|
||||
const spans = preparedEvents.map((event) => {
|
||||
const ancestorCancelled = isAncestorCancelled(eventsBySpanId, event.spanId);
|
||||
const duration = calculateDurationIfAncestorIsCancelled(
|
||||
eventsBySpanId,
|
||||
event.spanId,
|
||||
event.duration
|
||||
);
|
||||
|
||||
const span = {
|
||||
recordId: event.id,
|
||||
id: event.spanId,
|
||||
parentId: event.parentId ?? undefined,
|
||||
runId: event.runId,
|
||||
idempotencyKey: event.idempotencyKey,
|
||||
data: {
|
||||
message: event.message,
|
||||
style: event.style,
|
||||
duration,
|
||||
isError: event.isError,
|
||||
isPartial: ancestorCancelled ? false : event.isPartial,
|
||||
isCancelled: event.isCancelled === true ? true : event.isPartial && ancestorCancelled,
|
||||
startTime: getDateFromNanoseconds(event.startTime),
|
||||
level: event.level,
|
||||
events: event.events,
|
||||
environmentType: event.environmentType,
|
||||
},
|
||||
};
|
||||
|
||||
spansBySpanId.set(event.spanId, span);
|
||||
|
||||
return span;
|
||||
});
|
||||
|
||||
if (!rootSpanId) {
|
||||
return;
|
||||
}
|
||||
|
||||
const rootSpan = spansBySpanId.get(rootSpanId);
|
||||
|
||||
if (!rootSpan) {
|
||||
return;
|
||||
}
|
||||
|
||||
return {
|
||||
rootSpan,
|
||||
spans,
|
||||
};
|
||||
}
|
||||
|
||||
export function prepareEvent(event: QueriedEvent): PreparedEvent {
|
||||
return {
|
||||
...event,
|
||||
duration: Number(event.duration),
|
||||
events: parseEventsField(event.events),
|
||||
style: parseStyleField(event.style),
|
||||
};
|
||||
}
|
||||
|
||||
function parseEventsField(events: Prisma.JsonValue): SpanEvents {
|
||||
const unsafe = events
|
||||
? (events as any[]).map((e) => ({
|
||||
...e,
|
||||
properties: unflattenAttributes(e.properties as Attributes),
|
||||
}))
|
||||
: undefined;
|
||||
|
||||
return unsafe as SpanEvents;
|
||||
}
|
||||
|
||||
function parseStyleField(style: Prisma.JsonValue): TaskEventStyle {
|
||||
const unsafe = unflattenAttributes(style as Attributes);
|
||||
|
||||
if (!unsafe) {
|
||||
return {};
|
||||
}
|
||||
|
||||
if (typeof unsafe === "object") {
|
||||
return Object.assign(
|
||||
{
|
||||
icon: undefined,
|
||||
variant: undefined,
|
||||
},
|
||||
unsafe
|
||||
) as TaskEventStyle;
|
||||
}
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
export function isAncestorCancelled(events: Map<string, PreparedEvent>, spanId: string) {
|
||||
const event = events.get(spanId);
|
||||
|
||||
if (!event) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (event.isCancelled) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (event.parentId) {
|
||||
return isAncestorCancelled(events, event.parentId);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
function calculateDurationIfAncestorIsCancelled(
|
||||
events: Map<string, PreparedEvent>,
|
||||
spanId: string,
|
||||
defaultDuration: number
|
||||
) {
|
||||
const event = events.get(spanId);
|
||||
|
||||
if (!event) {
|
||||
return defaultDuration;
|
||||
}
|
||||
|
||||
if (event.isCancelled) {
|
||||
return defaultDuration;
|
||||
}
|
||||
|
||||
if (!event.isPartial) {
|
||||
return defaultDuration;
|
||||
}
|
||||
|
||||
if (event.parentId) {
|
||||
const cancelledAncestor = findFirstCancelledAncestor(events, event.parentId);
|
||||
|
||||
if (cancelledAncestor) {
|
||||
// We need to get the cancellation time from the cancellation span event
|
||||
const cancellationEvent = cancelledAncestor.events.find(
|
||||
(event) => event.name === "cancellation"
|
||||
);
|
||||
|
||||
if (cancellationEvent) {
|
||||
return calculateDurationFromStart(event.startTime, cancellationEvent.time);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return defaultDuration;
|
||||
}
|
||||
|
||||
export function calculateDurationFromStart(startTime: bigint, endTime: Date = new Date()) {
|
||||
const $endtime = typeof endTime === "string" ? new Date(endTime) : endTime;
|
||||
|
||||
return Number(BigInt($endtime.getTime() * 1_000_000) - startTime);
|
||||
}
|
||||
|
||||
function findFirstCancelledAncestor(events: Map<string, PreparedEvent>, spanId: string) {
|
||||
const event = events.get(spanId);
|
||||
|
||||
if (!event) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.isCancelled) {
|
||||
return event;
|
||||
}
|
||||
|
||||
if (event.parentId) {
|
||||
return findFirstCancelledAncestor(events, event.parentId);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
export function getDateFromNanoseconds(nanoseconds: bigint) {
|
||||
return new Date(Number(nanoseconds) / 1_000_000);
|
||||
}
|
||||
|
||||
export function getNowInNanoseconds(): bigint {
|
||||
return BigInt(new Date().getTime() * 1_000_000);
|
||||
}
|
||||
@@ -32,6 +32,14 @@ import { logger } from "~/services/logger.server";
|
||||
import { singleton } from "~/utils/singleton";
|
||||
import { DynamicFlushScheduler } from "./dynamicFlushScheduler.server";
|
||||
import { startActiveSpan } from "./tracer.server";
|
||||
import {
|
||||
calculateDurationFromStart,
|
||||
getDateFromNanoseconds,
|
||||
getNowInNanoseconds,
|
||||
isAncestorCancelled,
|
||||
prepareEvent,
|
||||
prepareTrace,
|
||||
} from "~/utils/taskEvent";
|
||||
|
||||
const MAX_FLUSH_DEPTH = 5;
|
||||
|
||||
@@ -413,82 +421,7 @@ export class EventRepository {
|
||||
take: env.MAXIMUM_TRACE_SUMMARY_VIEW_COUNT,
|
||||
});
|
||||
|
||||
let preparedEvents: Array<PreparedEvent> = [];
|
||||
let rootSpanId: string | undefined;
|
||||
const eventsBySpanId = new Map<string, PreparedEvent>();
|
||||
|
||||
for (const event of events) {
|
||||
preparedEvents.push(prepareEvent(event));
|
||||
|
||||
if (!rootSpanId && !event.parentId) {
|
||||
rootSpanId = event.spanId;
|
||||
}
|
||||
}
|
||||
|
||||
for (const event of preparedEvents) {
|
||||
const existingEvent = eventsBySpanId.get(event.spanId);
|
||||
|
||||
if (!existingEvent) {
|
||||
eventsBySpanId.set(event.spanId, event);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (event.isCancelled || !event.isPartial) {
|
||||
eventsBySpanId.set(event.spanId, event);
|
||||
}
|
||||
}
|
||||
|
||||
preparedEvents = Array.from(eventsBySpanId.values());
|
||||
|
||||
const spansBySpanId = new Map<string, SpanSummary>();
|
||||
|
||||
const spans = preparedEvents.map((event) => {
|
||||
const ancestorCancelled = isAncestorCancelled(eventsBySpanId, event.spanId);
|
||||
const duration = calculateDurationIfAncestorIsCancelled(
|
||||
eventsBySpanId,
|
||||
event.spanId,
|
||||
event.duration
|
||||
);
|
||||
|
||||
const span = {
|
||||
recordId: event.id,
|
||||
id: event.spanId,
|
||||
parentId: event.parentId ?? undefined,
|
||||
runId: event.runId,
|
||||
idempotencyKey: event.idempotencyKey,
|
||||
data: {
|
||||
message: event.message,
|
||||
style: event.style,
|
||||
duration,
|
||||
isError: event.isError,
|
||||
isPartial: ancestorCancelled ? false : event.isPartial,
|
||||
isCancelled: event.isCancelled === true ? true : event.isPartial && ancestorCancelled,
|
||||
startTime: getDateFromNanoseconds(event.startTime),
|
||||
level: event.level,
|
||||
events: event.events,
|
||||
environmentType: event.environmentType,
|
||||
},
|
||||
};
|
||||
|
||||
spansBySpanId.set(event.spanId, span);
|
||||
|
||||
return span;
|
||||
});
|
||||
|
||||
if (!rootSpanId) {
|
||||
return;
|
||||
}
|
||||
|
||||
const rootSpan = spansBySpanId.get(rootSpanId);
|
||||
|
||||
if (!rootSpan) {
|
||||
return;
|
||||
}
|
||||
|
||||
return {
|
||||
rootSpan,
|
||||
spans,
|
||||
};
|
||||
return prepareTrace(events);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1240,119 +1173,6 @@ function parseTraceparent(traceparent?: string): { traceId: string; spanId: stri
|
||||
return { traceId, spanId };
|
||||
}
|
||||
|
||||
function prepareEvent(event: QueriedEvent): PreparedEvent {
|
||||
return {
|
||||
...event,
|
||||
duration: Number(event.duration),
|
||||
events: parseEventsField(event.events),
|
||||
style: parseStyleField(event.style),
|
||||
};
|
||||
}
|
||||
|
||||
function parseEventsField(events: Prisma.JsonValue): SpanEvents {
|
||||
const unsafe = events
|
||||
? (events as any[]).map((e) => ({
|
||||
...e,
|
||||
properties: unflattenAttributes(e.properties as Attributes),
|
||||
}))
|
||||
: undefined;
|
||||
|
||||
return unsafe as SpanEvents;
|
||||
}
|
||||
|
||||
function parseStyleField(style: Prisma.JsonValue): TaskEventStyle {
|
||||
const unsafe = unflattenAttributes(style as Attributes);
|
||||
|
||||
if (!unsafe) {
|
||||
return {};
|
||||
}
|
||||
|
||||
if (typeof unsafe === "object") {
|
||||
return Object.assign(
|
||||
{
|
||||
icon: undefined,
|
||||
variant: undefined,
|
||||
},
|
||||
unsafe
|
||||
) as TaskEventStyle;
|
||||
}
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
function isAncestorCancelled(events: Map<string, PreparedEvent>, spanId: string) {
|
||||
const event = events.get(spanId);
|
||||
|
||||
if (!event) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (event.isCancelled) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (event.parentId) {
|
||||
return isAncestorCancelled(events, event.parentId);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
function calculateDurationIfAncestorIsCancelled(
|
||||
events: Map<string, PreparedEvent>,
|
||||
spanId: string,
|
||||
defaultDuration: number
|
||||
) {
|
||||
const event = events.get(spanId);
|
||||
|
||||
if (!event) {
|
||||
return defaultDuration;
|
||||
}
|
||||
|
||||
if (event.isCancelled) {
|
||||
return defaultDuration;
|
||||
}
|
||||
|
||||
if (!event.isPartial) {
|
||||
return defaultDuration;
|
||||
}
|
||||
|
||||
if (event.parentId) {
|
||||
const cancelledAncestor = findFirstCancelledAncestor(events, event.parentId);
|
||||
|
||||
if (cancelledAncestor) {
|
||||
// We need to get the cancellation time from the cancellation span event
|
||||
const cancellationEvent = cancelledAncestor.events.find(
|
||||
(event) => event.name === "cancellation"
|
||||
);
|
||||
|
||||
if (cancellationEvent) {
|
||||
return calculateDurationFromStart(event.startTime, cancellationEvent.time);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return defaultDuration;
|
||||
}
|
||||
|
||||
function findFirstCancelledAncestor(events: Map<string, PreparedEvent>, spanId: string) {
|
||||
const event = events.get(spanId);
|
||||
|
||||
if (!event) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.isCancelled) {
|
||||
return event;
|
||||
}
|
||||
|
||||
if (event.parentId) {
|
||||
return findFirstCancelledAncestor(events, event.parentId);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// Prioritize spans with the same id, keeping the completed spans over partial spans
|
||||
// Completed spans are either !isPartial or isCancelled
|
||||
function removeDuplicateEvents(events: PreparedEvent[]) {
|
||||
@@ -1470,20 +1290,6 @@ function filteredAttributes(attributes: Attributes, prefix: string): Attributes
|
||||
return result;
|
||||
}
|
||||
|
||||
function calculateDurationFromStart(startTime: bigint, endTime: Date = new Date()) {
|
||||
const $endtime = typeof endTime === "string" ? new Date(endTime) : endTime;
|
||||
|
||||
return Number(BigInt($endtime.getTime() * 1_000_000) - startTime);
|
||||
}
|
||||
|
||||
function getNowInNanoseconds(): bigint {
|
||||
return BigInt(new Date().getTime() * 1_000_000);
|
||||
}
|
||||
|
||||
export function getDateFromNanoseconds(nanoseconds: bigint) {
|
||||
return new Date(Number(nanoseconds) / 1_000_000);
|
||||
}
|
||||
|
||||
function rehydrateJson(json: Prisma.JsonValue): any {
|
||||
if (json === null) {
|
||||
return undefined;
|
||||
|
||||
Reference in New Issue
Block a user