implemented both detailed trace view and events/download logs in clickhouse

Author: ericallam

apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts

@@ -75,6 +75,7 @@ export class ApiRetrieveRunPresenter {
       },
       select: {
         ...commonRunSelect,
+        traceId: true,
         payload: true,
         payloadType: true,
         output: true,

apps/webapp/app/routes/api.v1.runs.$runId.events.ts

@@ -2,8 +2,8 @@ import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
 import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { eventRepository } from "~/v3/eventRepository/eventRepository.server";
 import { ApiRetrieveRunPresenter } from "~/presenters/v3/ApiRetrieveRunPresenter.server";
+import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 
 const ParamsSchema = z.object({
   runId: z.string(), // This is the run friendly ID
@@ -31,9 +31,12 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ resource: run, authentication }) => {
+    const eventRepository = resolveEventRepositoryForStore(run.taskEventStore);
+
     const runEvents = await eventRepository.getRunEvents(
       getTaskEventStoreTableForRun(run),
       authentication.environment.id,
+      run.traceId,
       run.friendlyId,
       run.createdAt,
       run.completedAt ?? undefined

apps/webapp/app/routes/api.v1.runs.$runId.trace.ts

@@ -3,7 +3,7 @@ import { BatchId } from "@trigger.dev/core/v3/isomorphic";
 import { z } from "zod";
 import { $replica } from "~/db.server";
 import { createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server";
-import { eventRepository } from "~/v3/eventRepository/eventRepository.server";
+import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 
 const ParamsSchema = z.object({
@@ -36,6 +36,8 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ resource: run, authentication }) => {
+    const eventRepository = resolveEventRepositoryForStore(run.taskEventStore);
+
     const traceSummary = await eventRepository.getTraceDetailedSummary(
       getTaskEventStoreTableForRun(run),
       authentication.environment.id,

apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts

@@ -2,13 +2,13 @@ import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { prisma } from "~/db.server";
 import { requireUser } from "~/services/session.server";
 import { v3RunParamsSchema } from "~/utils/pathBuilder";
-import { eventRepository } from "~/v3/eventRepository/eventRepository.server";
 import type { RunPreparedEvent } from "~/v3/eventRepository/eventRepository.types";
 import { createGzip } from "zlib";
 import { Readable } from "stream";
 import { formatDurationMilliseconds } from "@trigger.dev/core/v3/utils/durations";
 import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 import { TaskEventKind } from "@trigger.dev/database";
+import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 
 export async function loader({ params, request }: LoaderFunctionArgs) {
   const user = await requireUser(request);
@@ -33,9 +33,12 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
     return new Response("Not found", { status: 404 });
   }
 
+  const eventRepository = resolveEventRepositoryForStore(run.taskEventStore);
+
   const runEvents = await eventRepository.getRunEvents(
     getTaskEventStoreTableForRun(run),
     run.runtimeEnvironmentId,
+    run.traceId,
     run.friendlyId,
     run.createdAt,
     run.completedAt ?? undefined

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts

@@ -1512,11 +1512,92 @@ export class ClickhouseEventRepository implements IEventRepository {
   async getRunEvents(
     storeTable: TaskEventStoreTable,
     environmentId: string,
+    traceId: string,
     runId: string,
     startCreatedAt: Date,
     endCreatedAt?: Date
   ): Promise<RunPreparedEvent[]> {
-    throw new Error("ClickhouseEventRepository.getRunEvents not implemented");
+    const startCreatedAtWithBuffer = new Date(startCreatedAt.getTime() - 1000);
+
+    const queryBuilder = 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(),
+      });
+    }
+
+    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 [];
+    }
+
+    const recordsGroupedBySpanId = records.reduce((acc, record) => {
+      acc[record.span_id] = [...(acc[record.span_id] ?? []), record];
+      return acc;
+    }, {} as Record<string, TaskEventSummaryV1Result[]>);
+
+    const spanSummaries = new Map<string, SpanSummary>();
+
+    for (const [spanId, spanRecords] of Object.entries(recordsGroupedBySpanId)) {
+      const spanSummary = this.#mergeRecordsIntoSpanSummary(spanId, spanRecords);
+
+      if (!spanSummary) {
+        continue;
+      }
+
+      spanSummaries.set(spanId, spanSummary);
+    }
+
+    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,
+    };
   }
 }
 

apps/webapp/app/v3/eventRepository/eventRepository.server.ts

@@ -681,6 +681,7 @@ export class EventRepository implements IEventRepository {
   public async getRunEvents(
     storeTable: TaskEventStoreTable,
     environmentId: string,
+    traceId: string,
     runId: string,
     startCreatedAt: Date,
     endCreatedAt?: Date

apps/webapp/app/v3/eventRepository/eventRepository.types.ts

@@ -373,6 +373,7 @@ export interface IEventRepository {
   getRunEvents(
     storeTable: TaskEventStoreTable,
     environmentId: string,
+    traceId: string,
     runId: string,
     startCreatedAt: Date,
     endCreatedAt?: Date