feat: replicate task runs to clickhouse to power dashboard improvements (#2035)

* WIP clickhouse package with test containers setup

* More clickhouse client setup now with otel and real tests, and the v1 of raw run events

* Add some additional columns to raw_run_events_v1

* WIP runs dashboard service

* Create a new run engine event bus event for the runs dashboard to hook into

* Track run events in the run engine

* make sure engine v1 runs get synced to CH

* Update the attemptNumber of v3 task runs

* Restructure the run events to be more sparse

* emit more stuff

* Setup replication package

* scaffold the replication package

* replication wip

* resolve conflicts

* more replication stuff

* Add ability to drop the replication slot completely on teardown

* Use the new single replacingmergetree task events table for replication

* get it working

* insert payloads into their own table only on insert and then join

* prepare for using clickhouse cloud and now running ch migrations during boot in the entrypoint.sh

* Handover WIP and tests

* Testing the replication service

* Remove the runs dashboard stuff that we aren't using anymore

* Added a test for large payloads

* hacky typecheck fix

* Fix new internal package typecheck issues and start adding telemetry to the replication service

* tracing over spans, some other improvements

* Improvements to the runs replication service, now ready for testing

* Some fixes and cleanups

* Don't need this code anymore

* move transaction types into the runs replication service

* only send spans where there are transaction events

* A couple of suggested tweaks

Author: ericallam

apps/webapp/app/env.server.ts

@@ -725,6 +725,47 @@ const EnvironmentSchema = z.object({
   // BetterStack
   BETTERSTACK_API_KEY: z.string().optional(),
   BETTERSTACK_STATUS_PAGE_ID: z.string().optional(),
+
+  RUN_REPLICATION_REDIS_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_HOST),
+  RUN_REPLICATION_REDIS_READER_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_READER_HOST),
+  RUN_REPLICATION_REDIS_READER_PORT: z.coerce
+    .number()
+    .optional()
+    .transform(
+      (v) =>
+        v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
+    ),
+  RUN_REPLICATION_REDIS_PORT: z.coerce
+    .number()
+    .optional()
+    .transform((v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)),
+  RUN_REPLICATION_REDIS_USERNAME: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_USERNAME),
+  RUN_REPLICATION_REDIS_PASSWORD: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_PASSWORD),
+  RUN_REPLICATION_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
+
+  RUN_REPLICATION_CLICKHOUSE_URL: z.string().optional(),
+  RUN_REPLICATION_ENABLED: z.string().default("0"),
+  RUN_REPLICATION_SLOT_NAME: z.string().default("task_runs_to_clickhouse_v1"),
+  RUN_REPLICATION_PUBLICATION_NAME: z.string().default("task_runs_to_clickhouse_v1_publication"),
+  RUN_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(100),
+  RUN_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
+  RUN_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
+  RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
+  RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
+  RUN_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
+  RUN_REPLICATION_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/metrics.server.ts

@@ -4,7 +4,9 @@ import { env } from "./env.server";
 
 export const metricsRegister = singleton("metricsRegister", initializeMetricsRegister);
 
-function initializeMetricsRegister() {
+export type MetricsRegister = Registry<OpenMetricsContentType>;
+
+function initializeMetricsRegister(): MetricsRegister {
   const registry = new Registry<OpenMetricsContentType>();
 
   register.setDefaultLabels({