stop hardcoding the triggered_at column

Author: ericallam

apps/webapp/app/services/queryService.server.ts

@@ -152,7 +152,14 @@ export async function executeQuery<TOut extends z.ZodSchema>(
     return { success: false, error: new QueryError(errorMessage, { query: options.query }) };
   }
 
-  // Build time filter fallback for triggered_at column
+  // Detect which table the query targets to determine the time column
+  // Each table schema declares its primary time column via timeConstraint
+  const matchedSchema = querySchemas.find((s) =>
+    new RegExp(`\\bFROM\\s+${s.name}\\b`, "i").test(options.query)
+  );
+  const timeColumn = matchedSchema?.timeConstraint ?? "triggered_at";
+
+  // Build time filter fallback for the table's time column
   const defaultPeriod = await getDefaultPeriod(organizationId);
   const timeFilter = timeFilters({
     period: period ?? undefined,
@@ -173,15 +180,15 @@ export async function executeQuery<TOut extends z.ZodSchema>(
   }
 
   // Build the fallback WHERE condition based on what the user specified
-  let triggeredAtFallback: WhereClauseCondition;
+  let timeFallback: WhereClauseCondition;
   if (timeFilter.from && timeFilter.to) {
-    triggeredAtFallback = { op: "between", low: timeFilter.from, high: timeFilter.to };
+    timeFallback = { op: "between", low: timeFilter.from, high: timeFilter.to };
   } else if (timeFilter.from) {
-    triggeredAtFallback = { op: "gte", value: timeFilter.from };
+    timeFallback = { op: "gte", value: timeFilter.from };
   } else if (timeFilter.to) {
-    triggeredAtFallback = { op: "lte", value: timeFilter.to };
+    timeFallback = { op: "lte", value: timeFilter.to };
   } else {
-    triggeredAtFallback = { op: "gte", value: requestedFromDate! };
+    timeFallback = { op: "gte", value: requestedFromDate! };
   }
 
   const maxQueryPeriod = await getLimit(organizationId, "queryPeriodDays", 30);
@@ -196,7 +203,7 @@ export async function executeQuery<TOut extends z.ZodSchema>(
     project_id:
       scope === "project" || scope === "environment" ? { op: "eq", value: projectId } : undefined,
     environment_id: scope === "environment" ? { op: "eq", value: environmentId } : undefined,
-    triggered_at: { op: "gte", value: maxQueryPeriodDate },
+    [timeColumn]: { op: "gte", value: maxQueryPeriodDate },
     // Optional filters for tasks and queues
     task_identifier:
       taskIdentifiers && taskIdentifiers.length > 0
@@ -238,7 +245,7 @@ export async function executeQuery<TOut extends z.ZodSchema>(
       enforcedWhereClause,
       fieldMappings,
       whereClauseFallback: {
-        triggered_at: triggeredAtFallback,
+        [timeColumn]: timeFallback,
       },
       timeRange,
       clickhouseSettings: {

apps/webapp/app/v3/querySchemas.ts

@@ -434,10 +434,176 @@ export const runsSchema: TableSchema = {
   },
 };
 
+/**
+ * Schema definition for the metrics table (trigger_dev.metrics_v1)
+ */
+export const metricsSchema: TableSchema = {
+  name: "metrics",
+  clickhouseName: "trigger_dev.metrics_v1",
+  description: "Host and runtime metrics collected during task execution",
+  timeConstraint: "bucket_start",
+  tenantColumns: {
+    organizationId: "organization_id",
+    projectId: "project_id",
+    environmentId: "environment_id",
+  },
+  columns: {
+    environment: {
+      name: "environment",
+      clickhouseName: "environment_id",
+      ...column("String", { description: "The environment slug", example: "prod" }),
+      fieldMapping: "environment",
+      customRenderType: "environment",
+    },
+    project: {
+      name: "project",
+      clickhouseName: "project_id",
+      ...column("String", {
+        description: "The project reference, they always start with `proj_`.",
+        example: "proj_howcnaxbfxdmwmxazktx",
+      }),
+      fieldMapping: "project",
+      customRenderType: "project",
+    },
+    metric_name: {
+      name: "metric_name",
+      ...column("LowCardinality(String)", {
+        description: "The name of the metric (e.g. process.cpu.utilization, system.memory.usage)",
+        example: "process.cpu.utilization",
+        coreColumn: true,
+      }),
+    },
+    metric_type: {
+      name: "metric_type",
+      ...column("LowCardinality(String)", {
+        description: "The type of metric",
+        allowedValues: ["gauge", "sum", "histogram"],
+        example: "gauge",
+      }),
+    },
+    machine_id: {
+      name: "machine_id",
+      clickhouseName: "metric_subject",
+      ...column("String", {
+        description: "The machine ID that produced this metric",
+        example: "machine-abc123",
+      }),
+    },
+    bucket_start: {
+      name: "bucket_start",
+      ...column("DateTime", {
+        description: "The start of the 10-second aggregation bucket",
+        example: "2024-01-15 09:30:00",
+        coreColumn: true,
+      }),
+    },
+    count: {
+      name: "count",
+      ...column("UInt64", {
+        description: "Number of data points in this bucket",
+        example: "6",
+      }),
+    },
+    sum_value: {
+      name: "sum_value",
+      ...column("Float64", {
+        description: "Sum of values in this bucket",
+        example: "0.45",
+      }),
+    },
+    max_value: {
+      name: "max_value",
+      ...column("Float64", {
+        description: "Maximum value in this bucket",
+        example: "0.85",
+        coreColumn: true,
+      }),
+    },
+    min_value: {
+      name: "min_value",
+      ...column("Float64", {
+        description: "Minimum value in this bucket",
+        example: "0.12",
+      }),
+    },
+    last_value: {
+      name: "last_value",
+      ...column("Float64", {
+        description: "Last recorded value in this bucket",
+        example: "0.42",
+        coreColumn: true,
+      }),
+    },
+
+    // Trigger context columns (from attributes.trigger.* JSON subpaths)
+    run_id: {
+      name: "run_id",
+      ...column("String", {
+        description: "The run ID associated with this metric",
+        customRenderType: "runId",
+        example: "run_cm1a2b3c4d5e6f7g8h9i",
+      }),
+      expression: "attributes.trigger.run_id",
+    },
+    task_identifier: {
+      name: "task_identifier",
+      ...column("String", {
+        description: "Task identifier/slug",
+        example: "my-background-task",
+        coreColumn: true,
+      }),
+      expression: "attributes.trigger.task_slug",
+    },
+    attempt_number: {
+      name: "attempt_number",
+      ...column("String", {
+        description: "The attempt number for this metric",
+        example: "1",
+      }),
+      expression: "attributes.trigger.attempt_number",
+    },
+    machine_name: {
+      name: "machine_name",
+      ...column("String", {
+        description: "The machine preset used for execution",
+        allowedValues: [...MACHINE_PRESETS],
+        example: "small-1x",
+      }),
+      expression: "attributes.trigger.machine_name",
+    },
+    environment_type: {
+      name: "environment_type",
+      ...column("String", {
+        description: "Environment type",
+        allowedValues: [...ENVIRONMENT_TYPES],
+        customRenderType: "environmentType",
+        example: "PRODUCTION",
+      }),
+      expression: "attributes.trigger.environment_type",
+    },
+    worker_id: {
+      name: "worker_id",
+      ...column("String", {
+        description: "The worker ID that produced this metric",
+        example: "worker-abc123",
+      }),
+      expression: "attributes.trigger.worker_id",
+    },
+    worker_version: {
+      name: "worker_version",
+      ...column("String", {
+        description: "The worker version that produced this metric",
+        example: "20240115.1",
+      }),
+      expression: "attributes.trigger.worker_version",
+    },
+  },
+};
+
 /**
  * All available schemas for the query editor
  */
-export const querySchemas: TableSchema[] = [runsSchema];
+export const querySchemas: TableSchema[] = [runsSchema, metricsSchema];
 
 /**
  * Default query for the query editor