Associate logs and spans with the machine id

Author: ericallam

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

@@ -266,6 +266,7 @@ export class ClickhouseEventRepository implements IEventRepository {
         expires_at: convertDateToClickhouseDateTime(
           new Date(Date.now() + 365 * 24 * 60 * 60 * 1000) // 1 year
         ),
+        machine_id: event.machineId ?? "",
       },
       ...this.spanEventsToTaskEventV1Input(event),
     ];

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

@@ -56,6 +56,7 @@ export type CreateEventInput = Omit<
   resourceProperties?: Attributes;
   metadata: Attributes | undefined;
   style: Attributes | undefined;
+  machineId?: string;
 };
 
 export type CreatableEventKind = TaskEventKind;

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

@@ -331,6 +331,7 @@ function convertLogsToCreateableEvents(
           projectId: logProperties.projectId ?? resourceProperties.projectId ?? "unknown",
           runId: logProperties.runId ?? resourceProperties.runId ?? "unknown",
           taskSlug: logProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown",
+          machineId: logProperties.machineId ?? resourceProperties.machineId,
           attemptNumber:
             extractNumberAttribute(
               log.attributes ?? [],
@@ -437,6 +438,7 @@ function convertSpansToCreateableEvents(
           projectId: spanProperties.projectId ?? resourceProperties.projectId ?? "unknown",
           runId: spanProperties.runId ?? resourceProperties.runId ?? "unknown",
           taskSlug: spanProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown",
+          machineId: spanProperties.machineId ?? resourceProperties.machineId,
           attemptNumber:
             extractNumberAttribute(
               span.attributes ?? [],
@@ -672,6 +674,7 @@ function extractEventProperties(attributes: KeyValue[], prefix?: string) {
       SemanticInternalAttributes.ATTEMPT_NUMBER,
     ]),
     taskSlug: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.TASK_SLUG]),
+    machineId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.MACHINE_ID]),
   };
 }
 

internal-packages/clickhouse/schema/018_add_machine_id_to_task_events_v2.sql

@@ -0,0 +1,7 @@
+-- +goose Up
+ALTER TABLE trigger_dev.task_events_v2
+ADD COLUMN machine_id String DEFAULT '' CODEC(ZSTD(1));
+
+-- +goose Down
+ALTER TABLE trigger_dev.task_events_v2
+DROP COLUMN machine_id;

internal-packages/clickhouse/src/taskEvents.ts

@@ -19,6 +19,7 @@ export const TaskEventV1Input = z.object({
   attributes: z.unknown(),
   metadata: z.string(),
   expires_at: z.string(),
+  machine_id: z.string().optional(),
 });
 
 export type TaskEventV1Input = z.input<typeof TaskEventV1Input>;
@@ -153,6 +154,7 @@ export const TaskEventV2Input = z.object({
   attributes: z.unknown(),
   metadata: z.string(),
   expires_at: z.string(),
+  machine_id: z.string().optional(),
   // inserted_at has a default value in the table, so it's optional for inserts
   inserted_at: z.string().optional(),
 });