From 7ecacc06f755ab8ae256138d981b1dae436891b5 Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Sun, 19 Jul 2026 14:53:01 +0400 Subject: [PATCH] feat(console,helm): drive ClickHouse DDL from .sql templates + bulker metrics destination Console: make prisma/{events_log,metrics}.clickhouse.sql the single source of truth for the metrics ClickHouse DDL, rendered and executed by clickhouse-init.ts instead of hardcoded query strings. - Rename events_log.sql / metrics.sql -> *.clickhouse.sql and templatize them (${db}, ${onCluster}, ${engine:Family:znode}, ${user}/${password}, and a "-- @ch-settings" directive for per-statement ClickHouse settings). - clickhouse-init.ts renders + splits + runs the templates; the events_log DDL now includes task_log/dead_letter (the old file had drifted and lacked them). - Add initMetricsTables and wire it into admin/events-log-init; the integration test harness now runs the same templates instead of an inlined DDL copy. - Add a vitest raw-sql loader and a *.sql module type declaration. - Drop the unused mv_active_incoming3 / to_mv_active_incoming3 objects. Helm: render BULKER_INTERNAL_METRICS_DESTINATION into the bulker Deployment so the metrics ClickHouse destination is preinitialized (mirrors prod). Its connection resolves per field as: explicit values.bulker.internalMetricsDestination > derived from env.common.CLICKHOUSE_URL (hostname, userinfo, /database, and scheme->protocol) > in-cluster helm-deps fallback. An external CLICKHOUSE_URL therefore moves the metrics destination with it, avoiding a split-brain where bulker writes metrics to one ClickHouse while the console reads from another. Co-Authored-By: Claude Opus 4.8 --- helm/templates/bulker.yaml | 40 ++- helm/values.yaml | 37 ++ .../__tests__/integration/support/setup.ts | 40 +-- webapps/console/lib/server/clickhouse-init.ts | 333 +++++++++--------- .../pages/api/admin/events-log-init.ts | 10 +- .../console/prisma/events_log.clickhouse.sql | 102 ++++++ webapps/console/prisma/events_log.sql | 86 ----- webapps/console/prisma/metrics.clickhouse.sql | 147 ++++++++ webapps/console/prisma/metrics.sql | 156 -------- webapps/console/sql.d.ts | 6 + webapps/console/vitest.config.ts | 22 +- 11 files changed, 528 insertions(+), 451 deletions(-) create mode 100644 webapps/console/prisma/events_log.clickhouse.sql delete mode 100644 webapps/console/prisma/events_log.sql create mode 100644 webapps/console/prisma/metrics.clickhouse.sql delete mode 100644 webapps/console/prisma/metrics.sql create mode 100644 webapps/console/sql.d.ts diff --git a/helm/templates/bulker.yaml b/helm/templates/bulker.yaml index 5434aba19..ba828cde5 100644 --- a/helm/templates/bulker.yaml +++ b/helm/templates/bulker.yaml @@ -85,8 +85,44 @@ spec: name: jitsu-deps-urls optional: true env: - {{- include "jitsu-dev.env" (dict "ctx" . "service" "bulker" - "extra" (dict "HTTP_PORT" "3042")) | nindent 12 }} + {{- $extra := dict "HTTP_PORT" "3042" }} + {{- with .Values.bulker.internalMetricsDestination }} + {{- if .enabled }} + {{- /* Resolve the metrics ClickHouse connection. Per field: + explicit value here > derived from env.common.CLICKHOUSE_URL > + in-cluster helm-deps fallback. */ -}} + {{- $hosts := .hosts }} + {{- $database := .database }} + {{- $username := .username }} + {{- $password := .password }} + {{- $protocol := .protocol }} + {{- $chUrl := "" }} + {{- if $.Values.env }}{{- with $.Values.env.common }}{{- $chUrl = (.CLICKHOUSE_URL | default "") }}{{- end }}{{- end }} + {{- if $chUrl }} + {{- $u := urlParse $chUrl }} + {{- $host := splitList ":" $u.host | first }} + {{- if not $hosts }}{{- $hosts = list $host }}{{- end }} + {{- if not $database }}{{- $database = trimPrefix "/" $u.path }}{{- end }} + {{- with $u.userinfo }} + {{- $parts := splitList ":" . }} + {{- if not $username }}{{- $username = first $parts }}{{- end }} + {{- if and (not $password) (gt (len $parts) 1) }}{{- $password = index $parts 1 }}{{- end }} + {{- end }} + {{- if not $protocol }}{{- $protocol = ternary "clickhouse-secure" "clickhouse" (eq $u.scheme "https") }}{{- end }} + {{- end }} + {{- /* In-cluster helm-deps ClickHouse fallback for anything still unset. */ -}} + {{- if not $hosts }}{{- $hosts = list "clickhouse" }}{{- end }} + {{- $database = default "default" $database }} + {{- $username = default "default" $username }} + {{- $password = default "jitsu-dev" $password }} + {{- $protocol = default "clickhouse" $protocol }} + {{- $creds := dict "hosts" $hosts "database" $database "username" $username "password" $password "protocol" $protocol "loadAsJson" .loadAsJson }} + {{- if .cluster }}{{- $_ := set $creds "cluster" .cluster }}{{- end }} + {{- $dest := dict "id" "metrics" "special" "metrics" "usesBulker" true "type" "clickhouse" "options" .options "credentials" $creds }} + {{- $_ := set $extra "BULKER_INTERNAL_METRICS_DESTINATION" ($dest | toJson) }} + {{- end }} + {{- end }} + {{- include "jitsu-dev.env" (dict "ctx" . "service" "bulker" "extra" $extra) | nindent 12 }} volumeMounts: - name: build mountPath: /build diff --git a/helm/values.yaml b/helm/values.yaml index e23c1b37c..4705666cc 100644 --- a/helm/values.yaml +++ b/helm/values.yaml @@ -53,3 +53,40 @@ scaling: replicas: 1 console: replicas: 1 + +# Bulker configuration +bulker: + # Internal "metrics" ClickHouse destination preinitialized in bulker + # (special: "metrics"). Bulker writes pipeline metrics + active-user pings + # here and auto-creates the predefined tables (metrics, active_incoming). It + # is rendered into the bulker Deployment as the BULKER_INTERNAL_METRICS_DESTINATION + # env var (bulker loads every BULKER_INTERNAL* var as a destination config). + # + # Connection resolution, per field (highest precedence first): + # 1. an explicit value set below; + # 2. derived from env.common.CLICKHOUSE_URL when set — hostname, the + # user:password userinfo, and the /database path; scheme https -> the + # `clickhouse-secure` protocol, otherwise `clickhouse`; + # 3. the in-cluster helm-deps ClickHouse fallback + # (clickhouse / default / default / jitsu-dev / clickhouse). + # So pointing the stack at an external ClickHouse via env.common.CLICKHOUSE_URL + # moves the metrics destination with it — no split-brain where bulker writes + # metrics to one ClickHouse while the console reads them from another. Override + # a field only when the URL can't express it: a native port/protocol the HTTP + # URL doesn't imply, a URL-encoded password, or a differing metrics database + # (prod: newjitsu_metrics). `cluster` empty = single-node (no ON CLUSTER); + # prod uses `jitsu-cluster`. Set enabled: false to omit the env var entirely. + internalMetricsDestination: + enabled: true + hosts: [] + protocol: "" + database: "" + username: "" + password: "" + cluster: "" + loadAsJson: true + options: + mode: batch + frequency: 1 + deduplicate: false + batchSize: 100000 diff --git a/webapps/console/__tests__/integration/support/setup.ts b/webapps/console/__tests__/integration/support/setup.ts index 01982d684..afc1f7a82 100644 --- a/webapps/console/__tests__/integration/support/setup.ts +++ b/webapps/console/__tests__/integration/support/setup.ts @@ -79,41 +79,15 @@ process.env.JITSU_PUBLIC_URL = "http://console.test.local"; // (database + events_log/task_log/dead_letter + retention machinery), pointed // at this file's database. Imported dynamically — env above is already final. { - const { initEventsLogTables } = await import("../../../lib/server/clickhouse-init"); + const { initEventsLogTables, initMetricsTables } = await import("../../../lib/server/clickhouse-init"); const chAdmin = createClient({ url: chUrl }); try { - await initEventsLogTables({ clickhouse: chAdmin, database: chDb, username: "default", password: "" }); - // The metrics pipeline has no code counterpart (prisma/metrics.sql is - // cluster-only DDL applied manually in prod) — recreate it here with - // ON CLUSTER dropped and the replicated engine de-replicated. - for (const statement of [ - `create table ${chDb}.metrics - ( - timestamp DateTime, messageId String, - workspaceId LowCardinality(String), streamId LowCardinality(String), - connectionId LowCardinality(String), functionId LowCardinality(String), - destinationId LowCardinality(String), status LowCardinality(String), - events Int64, eventIndex UInt32 - ) ENGINE = Null`, - `create table ${chDb}.mv_metrics - ( - timestamp DateTime, - workspaceId LowCardinality(String), streamId LowCardinality(String), - connectionId LowCardinality(String), functionId LowCardinality(String), - destinationId LowCardinality(String), status LowCardinality(String), - events AggregateFunction(sum, Int64) - ) engine = AggregatingMergeTree() - ORDER BY (timestamp, workspaceId, streamId, connectionId, functionId, destinationId, status)`, - `CREATE MATERIALIZED VIEW ${chDb}.to_mv_metrics TO ${chDb}.mv_metrics - AS SELECT - date_trunc('minute', timestamp) as timestamp, - workspaceId, streamId, connectionId, functionId, destinationId, status, - sumState(events) AS events - FROM ${chDb}.metrics - GROUP BY timestamp, workspaceId, streamId, connectionId, functionId, destinationId, status`, - ]) { - await chAdmin.command({ query: statement }); - } + // Same DDL (and single source) the admin/events-log-init route runs in + // prod, pointed at this file's database. No cluster → ON CLUSTER dropped + // and replicated engines de-replicated by the templates' single-node path. + const initOpts = { clickhouse: chAdmin, database: chDb, username: "default", password: "" }; + await initEventsLogTables(initOpts); + await initMetricsTables(initOpts); } finally { await chAdmin.close(); } diff --git a/webapps/console/lib/server/clickhouse-init.ts b/webapps/console/lib/server/clickhouse-init.ts index 8fdb3588f..0c424eb5a 100644 --- a/webapps/console/lib/server/clickhouse-init.ts +++ b/webapps/console/lib/server/clickhouse-init.ts @@ -1,9 +1,14 @@ import type { ClickHouseClient } from "@clickhouse/client"; import { getServerLog } from "./log"; +// Raw SQL templates (loaded as strings via raw-loader in Next / the raw-sql +// plugin in vitest). These are the single source of truth for the ClickHouse +// DDL — do not duplicate the statements here. +import eventsLogSqlTemplate from "../../prisma/events_log.clickhouse.sql"; +import metricsSqlTemplate from "../../prisma/metrics.clickhouse.sql"; -const log = getServerLog("events-log-init"); +const log = getServerLog("clickhouse-init"); -export interface EventsLogInitOptions { +export interface ClickhouseInitOptions { clickhouse: ClickHouseClient; /** Target database (aka metrics schema), e.g. `newjitsu_metrics`. */ database: string; @@ -14,189 +19,179 @@ export interface EventsLogInitOptions { password: string; } +/** @deprecated use {@link ClickhouseInitOptions}. Kept for backwards compatibility. */ +export type EventsLogInitOptions = ClickhouseInitOptions; + +interface TemplateVars { + db: string; + cluster?: string | undefined; + username: string; + password: string; +} + +interface PreparedStatement { + sql: string; + /** Per-statement ClickHouse settings, from a `-- @ch-settings` directive. */ + settings?: Record; +} + +// Escape single quotes for values interpolated into single-quoted SQL literals. +// The values come from trusted config, not user input. Note the rendered DDL — +// including the dictionary credentials — is visible in ClickHouse query logs +// and SHOW CREATE; a named collection is the follow-up to remove that exposure. +const sqlLit = (s: string): string => s.replace(/'/g, "''"); + /** - * Create the events-log ClickHouse objects: the database, the `events_log` / - * `task_log` / `dead_letter` tables, and the events_log retention machinery - * (cutoff tables + dictionary + dictGet-driven TTL). Used by the - * `admin/events-log-init` route against the prod cluster and by the test - * harness against a per-test-file database — one source for the DDL. - * - * Throws when the database can't be created; accumulates per-object failures - * and throws a combined error at the end otherwise. + * Resolve the template placeholders for one rendering target: + * ${onCluster} -> " ON CLUSTER ``" (cluster) or "" (single-node) + * ${engine:Family:znode} -> Replicated(znode path) (cluster) or () (single-node) + * ${user} / ${password} -> escaped credentials + * ${db} -> target database */ -export async function initEventsLogTables(opts: EventsLogInitOptions): Promise { - const { clickhouse, database: metricsSchema } = opts; - // Backtick-quote the cluster name: the k8s metrics cluster is `jitsu-cluster` - // (the Altinity CRD forbids underscores), and an unquoted hyphen breaks the - // ON CLUSTER DDL ("syntax error at -cluster"). - const onCluster = opts.cluster ? ` ON CLUSTER \`${opts.cluster}\`` : ""; - const mergeTree = (name: string) => - opts.cluster - ? `ReplicatedMergeTree('/clickhouse/tables/{shard}/${metricsSchema}/${name}', '{replica}')` - : "MergeTree()"; +function renderTemplate(fragment: string, vars: TemplateVars): string { + const onCluster = vars.cluster ? ` ON CLUSTER \`${vars.cluster}\`` : ""; + return fragment + .replace(/\$\{onCluster\}/g, onCluster) + .replace(/\$\{engine:([A-Za-z]+):([A-Za-z0-9_]+)\}/g, (_m, family: string, znode: string) => + vars.cluster + ? `Replicated${family}('/clickhouse/tables/{shard}/${vars.db}/${znode}', '{replica}')` + : `${family}()` + ) + .replace(/\$\{user\}/g, sqlLit(vars.username)) + .replace(/\$\{password\}/g, sqlLit(vars.password)) + .replace(/\$\{db\}/g, vars.db); +} - const createDbQuery: string = `create database IF NOT EXISTS ${metricsSchema}${onCluster}`; - try { - await clickhouse.command({ - query: createDbQuery, - }); - log.atInfo().log(`Database ${metricsSchema} created or already exists`); - } catch (e: any) { - log.atError().withCause(e).log(`Failed to create ${metricsSchema} database.`); - throw new Error(`Failed to create ${metricsSchema} database.`); - } - const errors: Error[] = []; - const createEventsLogTableQuery: string = `create table IF NOT EXISTS ${metricsSchema}.events_log ${onCluster} - ( - timestamp DateTime64(3), - actorId LowCardinality(String), - type LowCardinality(String), - level LowCardinality(String), - message String - ) - engine = ${mergeTree("events_log")} - PARTITION BY toYYYYMM(timestamp) - ORDER BY (actorId, type, timestamp)`; +/** + * Split a `.clickhouse.sql` template into executable statements. Line comments + * (`-- ...`) are stripped; statements are separated by `;`. A directive line + * `-- @ch-settings k=v, k2=v2` attaches those settings to the next statement. + * The templates keep `;` out of statement bodies, so a naive top-level split is + * safe (no string/paren-aware parsing needed). + */ +function prepareStatements(template: string, vars: TemplateVars): PreparedStatement[] { + const statements: PreparedStatement[] = []; + let buffer = ""; + let pendingSettings: Record | undefined; + let currentSettings: Record | undefined; - try { - await clickhouse.command({ - query: createEventsLogTableQuery, - }); - log.atInfo().log(`Table ${metricsSchema}.events_log created or already exists`); - } catch (e: any) { - log.atError().withCause(e).log(`Failed to create ${metricsSchema}.events_log table.`); - errors.push(new Error(`Failed to create ${metricsSchema}.events_log table.`)); - } - const createTaskLogTableQuery: string = `create table IF NOT EXISTS ${metricsSchema}.task_log ${onCluster} - ( - task_id String, - sync_id LowCardinality(String), - timestamp DateTime64(3), - level LowCardinality(String), - logger LowCardinality(String), - message String - ) - engine = ${mergeTree("task_log")} - PARTITION BY toYYYYMM(timestamp) - ORDER BY (task_id, sync_id, timestamp) - TTL toDateTime(timestamp) + INTERVAL 3 MONTH DELETE`; + const flush = () => { + const trimmed = buffer.trim(); + if (trimmed) { + statements.push({ sql: renderTemplate(trimmed, vars), settings: currentSettings }); + } + buffer = ""; + currentSettings = undefined; + }; - try { - await clickhouse.command({ - query: createTaskLogTableQuery, - }); - log.atInfo().log(`Table ${metricsSchema}.task_log created or already exists`); - } catch (e: any) { - log.atError().withCause(e).log(`Failed to create ${metricsSchema}.task_log table.`); - errors.push(new Error(`Failed to create ${metricsSchema}.task_log table.`)); + for (const rawLine of template.split("\n")) { + const directive = rawLine.match(/^\s*--\s*@ch-settings\s+(.+)$/); + if (directive) { + pendingSettings = pendingSettings ?? {}; + for (const pair of directive[1].split(",")) { + const [k, v] = pair.split("=").map(s => s.trim()); + if (k) pendingSettings[k] = Number(v); + } + continue; + } + const line = rawLine.replace(/--.*$/, ""); // strip ordinary line comments + if (buffer.trim() === "" && line.trim() === "") { + continue; // blank space between statements + } + if (buffer.trim() === "") { + // starting a new statement: latch any pending settings onto it + currentSettings = pendingSettings; + pendingSettings = undefined; + } + buffer += (buffer ? "\n" : "") + line; + if (line.includes(";")) { + const semi = buffer.lastIndexOf(";"); + const rest = buffer.slice(semi + 1); + buffer = buffer.slice(0, semi); + flush(); + buffer = rest; + } } - const createDeadLetterTableQuery: string = `create table IF NOT EXISTS ${metricsSchema}.dead_letter ${onCluster} - ( - timestamp DateTime64(3), - workspaceId LowCardinality(String), - actorId LowCardinality(String), - type LowCardinality(String), - payload String, - error String - ) - engine = ${mergeTree("dead_letter")} - ORDER BY (workspaceId, actorId, timestamp) - TTL toDateTime(timestamp) + INTERVAL 1 MONTH DELETE`; + flush(); + return statements; +} + +function describeStatement(sql: string): string { + const firstLine = (sql.trim().split("\n")[0] ?? "").trim(); + return firstLine.length > 80 ? firstLine.slice(0, 77) + "..." : firstLine; +} +async function ensureDatabase(opts: ClickhouseInitOptions): Promise { + const onCluster = opts.cluster ? ` ON CLUSTER \`${opts.cluster}\`` : ""; try { - await clickhouse.command({ - query: createDeadLetterTableQuery, - }); - log.atInfo().log(`Table ${metricsSchema}.dead_letter or already exists`); + await opts.clickhouse.command({ query: `create database IF NOT EXISTS ${opts.database}${onCluster}` }); + log.atInfo().log(`Database ${opts.database} created or already exists`); } catch (e: any) { - log.atError().withCause(e).log(`Failed to create ${metricsSchema}.dead_letter table.`); - errors.push(new Error(`Failed to create ${metricsSchema}.dead_letter table.`)); + log.atError().withCause(e).log(`Failed to create ${opts.database} database.`); + throw new Error(`Failed to create ${opts.database} database.`); } - // --- events_log retention: cutoff dictionary + dictGet-driven TTL --- - // Goal: keep the newest EVENTS_LOG_SIZE rows per (actorId, type, is_error). - // The dictionary holds, per entity, the timestamp of the N-th newest row - // (its retention cutoff); events_log's TTL deletes anything older on merge. - // This replaces the old full-table scan + lightweight-delete trim and, - // unlike lightweight deletes, TTL DELETE physically reclaims disk on merge. +} - // The cutoffs live in their own small table rather than being computed by - // the dictionary directly from events_log: a dictionary that sourced from - // events_log while events_log's TTL references that dictionary would be a - // cyclic dependency (ClickHouse rejects it). events-log-trim recomputes the - // contents of this table (one row per over-cap entity). - // The `_staging` twin lets the trim job rebuild cutoffs and swap them in - // atomically (EXCHANGE TABLES), so a transient recompute failure never - // leaves the live table empty (which would pause retention). - const cutoffTable = (name: string): string => `create table IF NOT EXISTS ${metricsSchema}.${name} ${onCluster} - ( - actorId String, - type String, - is_error UInt8, - cutoff DateTime64(3) - ) - engine = ${mergeTree(name)} - ORDER BY (actorId, type, is_error)`; - for (const name of ["events_log_cutoff_src", "events_log_cutoff_staging"]) { +/** + * Render a template against the init target and run every statement, one per + * object. Failures are accumulated (so one bad object doesn't hide the rest) + * and rethrown as a single combined error at the end. + */ +async function applyTemplate(opts: ClickhouseInitOptions, template: string): Promise { + const statements = prepareStatements(template, { + db: opts.database, + cluster: opts.cluster, + username: opts.username, + password: opts.password, + }); + const errors: Error[] = []; + for (const { sql, settings } of statements) { + const label = describeStatement(sql); try { - await clickhouse.command({ query: cutoffTable(name) }); - log.atInfo().log(`Table ${metricsSchema}.${name} created or already exists`); + await opts.clickhouse.command({ + query: sql, + ...(settings ? { clickhouse_settings: settings as any } : {}), + }); + log.atInfo().log(`Applied: ${label}`); } catch (e: any) { - log.atError().withCause(e).log(`Failed to create ${metricsSchema}.${name} table.`); - errors.push(new Error(`Failed to create ${metricsSchema}.${name} table.`)); + log.atError().withCause(e).log(`Failed: ${label}`); + errors.push(new Error(`${label}: ${e?.message ?? e}`)); } } - - // Credentials are interpolated into the DDL, so escape single quotes (the - // values come from trusted config, not user input). Note the rendered DDL - // — including these creds — is visible in ClickHouse query logs and - // SHOW CREATE; a named collection is the follow-up to remove that exposure. - const sqlLit = (s: string): string => s.replace(/'/g, "''"); - const chUser = sqlLit(opts.username); - const chPass = sqlLit(opts.password); - const createCutoffDictQuery: string = `create dictionary IF NOT EXISTS ${metricsSchema}.events_log_cutoff ${onCluster} - ( - actorId String, - type String, - is_error UInt8, - cutoff DateTime64(3) - ) - PRIMARY KEY actorId, type, is_error - SOURCE(CLICKHOUSE( - user '${chUser}' password '${chPass}' db '${metricsSchema}' table 'events_log_cutoff_src' - )) - LAYOUT(COMPLEX_KEY_HASHED()) - LIFETIME(MIN 1800 MAX 3600)`; - try { - await clickhouse.command({ query: createCutoffDictQuery }); - log.atInfo().log(`Dictionary ${metricsSchema}.events_log_cutoff created or already exists`); - } catch (e: any) { - log.atError().withCause(e).log(`Failed to create ${metricsSchema}.events_log_cutoff dictionary.`); - errors.push(new Error(`Failed to create ${metricsSchema}.events_log_cutoff dictionary.`)); + if (errors.length > 0) { + throw new Error("Failed to initialize tables: " + errors.map(e => e.message).join("; ")); } +} - // Attach the retention TTL. allow_suspicious_ttl_expressions is required - // because dictGet is non-deterministic; materialize_ttl_after_modify=0 - // avoids an immediate full-table mutation on deploy — enforcement happens - // on background merges and via the events-log-trim cron's MATERIALIZE TTL. - const modifyTtlQuery: string = `alter table ${metricsSchema}.events_log ${onCluster} modify TTL toDateTime( - if(timestamp < dictGetOrDefault('${metricsSchema}.events_log_cutoff', 'cutoff', (actorId, type, toUInt8(level = 'error')), toDateTime64('1970-01-01 00:00:00', 3)), - toDateTime('2000-01-01 00:00:00'), - toDateTime('2099-01-01 00:00:00'))) DELETE`; - try { - await clickhouse.command({ - query: modifyTtlQuery, - clickhouse_settings: { - allow_suspicious_ttl_expressions: 1, - materialize_ttl_after_modify: 0, - }, - }); - log.atInfo().log(`Retention TTL set on ${metricsSchema}.events_log`); - } catch (e: any) { - log.atError().withCause(e).log(`Failed to set retention TTL on ${metricsSchema}.events_log.`); - errors.push(new Error(`Failed to set retention TTL on ${metricsSchema}.events_log.`)); - } +/** + * Create the events-log ClickHouse objects: the database, the `events_log` / + * `task_log` / `dead_letter` tables, and the events_log retention machinery + * (cutoff tables + dictionary + dictGet-driven TTL). The DDL lives in + * `prisma/events_log.clickhouse.sql`; this renders it for the target (cluster + * vs single-node, injected credentials) and executes it. Used by the + * `admin/events-log-init` route against the prod cluster and by the test + * harness against a per-test-file database — one source for the DDL. + * + * Throws when the database can't be created; accumulates per-object failures + * and throws a combined error at the end otherwise. + */ +export async function initEventsLogTables(opts: ClickhouseInitOptions): Promise { + await ensureDatabase(opts); + await applyTemplate(opts, eventsLogSqlTemplate); +} - if (errors.length > 0) { - throw new Error("Failed to initialize tables: " + errors.map(e => e.message).join(", ")); - } +/** + * Create the metrics ClickHouse objects: the per-minute event `metrics` + * aggregation (`metrics` -> `to_mv_metrics` -> `mv_metrics`) and the + * active-users pipeline (`active_incoming` -> `mv_active_incoming2/3` -> + * `active_incoming_agg` -> `active_incoming_agg_view`). The DDL lives in + * `prisma/metrics.clickhouse.sql`; statements are ordered dependencies-first so + * they can be applied top to bottom. + * + * Throws when the database can't be created; accumulates per-object failures + * and throws a combined error at the end otherwise. + */ +export async function initMetricsTables(opts: ClickhouseInitOptions): Promise { + await ensureDatabase(opts); + await applyTemplate(opts, metricsSqlTemplate); } diff --git a/webapps/console/pages/api/admin/events-log-init.ts b/webapps/console/pages/api/admin/events-log-init.ts index 912b18883..8b03c5d6d 100644 --- a/webapps/console/pages/api/admin/events-log-init.ts +++ b/webapps/console/pages/api/admin/events-log-init.ts @@ -4,7 +4,7 @@ import { clickhouse } from "../../../lib/server/clickhouse"; import { z } from "zod"; import { getServerLog } from "../../../lib/server/log"; import { getServerEnv } from "../../../lib/server/serverEnv"; -import { initEventsLogTables } from "../../../lib/server/clickhouse-init"; +import { initEventsLogTables, initMetricsTables } from "../../../lib/server/clickhouse-init"; const log = getServerLog("events-log-init"); @@ -33,14 +33,16 @@ export default createRoute() } await verifyAdmin(user); } - log.atInfo().log(`Init events log`); + log.atInfo().log(`Init events log and metrics tables`); const chConfig = getClickhouseConfig(serverEnv); - await initEventsLogTables({ + const initOpts = { clickhouse, database: requireDefined(chConfig.database, "ClickHouse database is not configured"), cluster: serverEnv.CLICKHOUSE_METRICS_CLUSTER || serverEnv.CLICKHOUSE_CLUSTER, username: chConfig.username, password: chConfig.password, - }); + }; + await initEventsLogTables(initOpts); + await initMetricsTables(initOpts); }) .toNextApiHandler(); diff --git a/webapps/console/prisma/events_log.clickhouse.sql b/webapps/console/prisma/events_log.clickhouse.sql new file mode 100644 index 000000000..5002108d6 --- /dev/null +++ b/webapps/console/prisma/events_log.clickhouse.sql @@ -0,0 +1,102 @@ +-- ClickHouse DDL for the events-log / task-log / dead-letter tables and the +-- events_log retention machinery (cutoff tables + dictionary + dictGet TTL). +-- +-- NOT managed by prisma, and NOT plain SQL you paste into a client: this is a +-- template rendered and executed by lib/server/clickhouse-init.ts +-- (initEventsLogTables) — in prod against the metrics cluster, in tests +-- against a single-node container. Placeholders: +-- ${db} target database (e.g. newjitsu_metrics) +-- ${onCluster} " ON CLUSTER ``" on a cluster, empty single-node +-- ${engine:Family:znode} Replicated('/clickhouse/tables/{shard}//', '{replica}') +-- on a cluster, plain () single-node +-- ${user} / ${password} dictionary source credentials (single-quote escaped) +-- A directive line of the form "-- @ch-settings k=v, k2=v2" applies those +-- ClickHouse settings to the NEXT statement only. Statements are separated by +-- ';' — keep ';' out of a statement body (only at its end). + +create table IF NOT EXISTS ${db}.events_log${onCluster} +( + timestamp DateTime64(3), + actorId LowCardinality(String), + type LowCardinality(String), + level LowCardinality(String), + message String +) +engine = ${engine:MergeTree:events_log} +PARTITION BY toYYYYMM(timestamp) +ORDER BY (actorId, type, timestamp); + +create table IF NOT EXISTS ${db}.task_log${onCluster} +( + task_id String, + sync_id LowCardinality(String), + timestamp DateTime64(3), + level LowCardinality(String), + logger LowCardinality(String), + message String +) +engine = ${engine:MergeTree:task_log} +PARTITION BY toYYYYMM(timestamp) +ORDER BY (task_id, sync_id, timestamp) +TTL toDateTime(timestamp) + INTERVAL 3 MONTH DELETE; + +create table IF NOT EXISTS ${db}.dead_letter${onCluster} +( + timestamp DateTime64(3), + workspaceId LowCardinality(String), + actorId LowCardinality(String), + type LowCardinality(String), + payload String, + error String +) +engine = ${engine:MergeTree:dead_letter} +ORDER BY (workspaceId, actorId, timestamp) +TTL toDateTime(timestamp) + INTERVAL 1 MONTH DELETE; + +-- events_log retention: cutoff dictionary + dictGet-driven TTL. The cutoffs +-- live in their own table (not sourced directly from events_log — that would +-- be a cyclic dependency ClickHouse rejects). events-log-trim rebuilds cutoffs +-- in the _staging twin and swaps them in atomically (EXCHANGE TABLES). +create table IF NOT EXISTS ${db}.events_log_cutoff_src${onCluster} +( + actorId String, + type String, + is_error UInt8, + cutoff DateTime64(3) +) +engine = ${engine:MergeTree:events_log_cutoff_src} +ORDER BY (actorId, type, is_error); + +create table IF NOT EXISTS ${db}.events_log_cutoff_staging${onCluster} +( + actorId String, + type String, + is_error UInt8, + cutoff DateTime64(3) +) +engine = ${engine:MergeTree:events_log_cutoff_staging} +ORDER BY (actorId, type, is_error); + +create dictionary IF NOT EXISTS ${db}.events_log_cutoff${onCluster} +( + actorId String, + type String, + is_error UInt8, + cutoff DateTime64(3) +) +PRIMARY KEY actorId, type, is_error +SOURCE(CLICKHOUSE( + user '${user}' password '${password}' db '${db}' table 'events_log_cutoff_src' +)) +LAYOUT(COMPLEX_KEY_HASHED()) +LIFETIME(MIN 1800 MAX 3600); + +-- dictGet in a TTL is non-deterministic (allow_suspicious_ttl_expressions); +-- materialize_ttl_after_modify=0 avoids an immediate full-table mutation on +-- deploy — enforcement happens on background merges and via events-log-trim. +-- @ch-settings allow_suspicious_ttl_expressions=1, materialize_ttl_after_modify=0 +alter table ${db}.events_log${onCluster} +modify TTL toDateTime( + if(timestamp < dictGetOrDefault('${db}.events_log_cutoff', 'cutoff', (actorId, type, toUInt8(level = 'error')), toDateTime64('1970-01-01 00:00:00', 3)), + toDateTime('2000-01-01 00:00:00'), + toDateTime('2099-01-01 00:00:00'))) DELETE; diff --git a/webapps/console/prisma/events_log.sql b/webapps/console/prisma/events_log.sql deleted file mode 100644 index 4808100e4..000000000 --- a/webapps/console/prisma/events_log.sql +++ /dev/null @@ -1,86 +0,0 @@ --- not managed by prisma -create table IF NOT EXISTS newjitsu_metrics.events_log ---ON CLUSTER jitsu_cluster -( - timestamp DateTime64(3), - actorId LowCardinality(String), - type LowCardinality(String), - level LowCardinality(String), - message String -) - engine = MergeTree() - --engine = ReplicatedMergeTree('/clickhouse/tables/{shard}/newjitsu_metrics/events_log3', '{replica}') - PARTITION BY toYYYYMM(timestamp) - ORDER BY (actorId, type, timestamp) - SETTINGS index_granularity = 8192; - --- Retention: keep the newest EVENTS_LOG_SIZE (default 200000) rows per --- (actorId, type, is_error). A cutoff dictionary holds, per entity, the --- timestamp of the N-th newest row; the TTL below deletes anything older on --- merge. Unlike the old lightweight-delete trim, TTL DELETE physically --- reclaims disk. Created/maintained by pages/api/admin/events-log-init.ts and --- enforced by pages/api/admin/events-log-trim.ts (drop floor + reload dict + --- MATERIALIZE TTL). dictGet in a TTL requires allow_suspicious_ttl_expressions=1. - --- Cutoffs live in their own table (NOT computed by the dictionary directly --- from events_log): a dictionary sourcing from events_log while events_log's --- TTL references that dictionary is a cyclic dependency that ClickHouse --- rejects. events-log-trim.ts rebuilds cutoffs in the _staging twin and swaps --- them in atomically, so a transient failure never leaves the live table empty: --- truncate table newjitsu_metrics.events_log_cutoff_staging; --- insert into newjitsu_metrics.events_log_cutoff_staging --- select actorId, type, is_error, cutoff from ( --- select actorId, type, toUInt8(level = 'error') as is_error, timestamp as cutoff, --- row_number() over (partition by actorId, type, level = 'error' order by timestamp desc) as rn --- from newjitsu_metrics.events_log) --- where rn = 200000; -- row_number streams; avoids groupArray materializing every group --- exchange tables newjitsu_metrics.events_log_cutoff_src and newjitsu_metrics.events_log_cutoff_staging; -create table IF NOT EXISTS newjitsu_metrics.events_log_cutoff_src ---ON CLUSTER jitsu_cluster -( - actorId String, - type String, - is_error UInt8, - cutoff DateTime64(3) -) - engine = MergeTree() - --engine = ReplicatedMergeTree('/clickhouse/tables/{shard}/newjitsu_metrics/events_log_cutoff_src', '{replica}') - ORDER BY (actorId, type, is_error); - --- staging twin for atomic cutoff swaps (same schema) -create table IF NOT EXISTS newjitsu_metrics.events_log_cutoff_staging ---ON CLUSTER jitsu_cluster -( - actorId String, - type String, - is_error UInt8, - cutoff DateTime64(3) -) - engine = MergeTree() - --engine = ReplicatedMergeTree('/clickhouse/tables/{shard}/newjitsu_metrics/events_log_cutoff_staging', '{replica}') - ORDER BY (actorId, type, is_error); - -create dictionary IF NOT EXISTS newjitsu_metrics.events_log_cutoff ---ON CLUSTER jitsu_cluster -( - actorId String, - type String, - is_error UInt8, - cutoff DateTime64(3) -) -PRIMARY KEY actorId, type, is_error -SOURCE(CLICKHOUSE( - -- no host/port: reads the local server in-process; auth still required - user 'default' password '' db 'newjitsu_metrics' table 'events_log_cutoff_src' -)) -LAYOUT(COMPLEX_KEY_HASHED()) -LIFETIME(MIN 1800 MAX 3600); - --- SET allow_suspicious_ttl_expressions = 1, materialize_ttl_after_modify = 0; -alter table newjitsu_metrics.events_log ---ON CLUSTER jitsu_cluster - modify TTL toDateTime( - if(timestamp < dictGetOrDefault('newjitsu_metrics.events_log_cutoff', 'cutoff', (actorId, type, toUInt8(level = 'error')), toDateTime64('1970-01-01 00:00:00', 3)), - toDateTime('2000-01-01 00:00:00'), - toDateTime('2099-01-01 00:00:00'))) DELETE; - diff --git a/webapps/console/prisma/metrics.clickhouse.sql b/webapps/console/prisma/metrics.clickhouse.sql new file mode 100644 index 000000000..536faf598 --- /dev/null +++ b/webapps/console/prisma/metrics.clickhouse.sql @@ -0,0 +1,147 @@ +-- ClickHouse DDL for the metrics pipeline: the per-minute event `metrics` +-- aggregation (metrics -> to_mv_metrics -> mv_metrics) and the active-users +-- pipeline (active_incoming -> mv_active_incoming2/3 -> active_incoming_agg -> +-- active_incoming_agg_view, refreshed by mv_active_incoming_agg). +-- +-- NOT managed by prisma, and NOT plain SQL you paste into a client: this is a +-- template rendered and executed by lib/server/clickhouse-init.ts +-- (initMetricsTables) — in prod against the metrics cluster, in tests against a +-- single-node container. Placeholders: +-- ${db} target database (e.g. newjitsu_metrics) +-- ${onCluster} " ON CLUSTER ``" on a cluster, empty single-node +-- ${engine:Family:znode} Replicated('/clickhouse/tables/{shard}//', '{replica}') +-- on a cluster, plain () single-node +-- A directive line "-- @ch-settings k=v" applies those ClickHouse settings to +-- the NEXT statement only. Statements are separated by ';' and run top to +-- bottom, so objects are ordered dependencies-first. + +-- ── active-users pipeline ─────────────────────────────────────────────────── + +-- Ingest entry point: rotor writes active-user pings here; the Null engine +-- keeps nothing and only fans out to the materialized views below. +create table ${db}.active_incoming${onCluster} +( + timestamp DateTime, + workspaceId LowCardinality(String), + messageId String +) +engine = Null; + +CREATE MATERIALIZED VIEW ${db}.mv_active_incoming2${onCluster} +( + `workspaceId` LowCardinality(String), + `timestamp` DateTime, + `count` AggregateFunction(uniq, String) +) +ENGINE = ${engine:AggregatingMergeTree:mv_active_incoming2_0} +ORDER BY (workspaceId, timestamp) +PARTITION BY toYYYYMM(timestamp) +SETTINGS index_granularity = 8192 +AS +SELECT + workspaceId, + timestamp, + uniqState(messageId) AS count +FROM ${db}.active_incoming +GROUP BY + workspaceId, + timestamp; + +CREATE TABLE ${db}.active_incoming_agg${onCluster} +( + `workspaceId` LowCardinality(String), + `timestamp` DateTime, + `insertedAt` DateTime DEFAULT now(), + `count` UInt64 +) +ENGINE = ReplacingMergeTree() +ORDER BY (workspaceId, timestamp) +PARTITION BY toYYYYMM(timestamp); + +-- REFRESH ... APPEND materialized views are still experimental in 25.4. +-- @ch-settings allow_experimental_refreshable_materialized_view=1 +CREATE MATERIALIZED VIEW ${db}.mv_active_incoming_agg${onCluster} +REFRESH EVERY 5 MINUTE APPEND TO ${db}.active_incoming_agg AS +select + workspaceId, + timestamp, + now() as insertedAt, + uniqMerge(count) as "count" +from ${db}.mv_active_incoming2 +where + timestamp >= date_trunc('day', now() - interval 3 day) +group by workspaceId, timestamp order by workspaceId, timestamp; + +CREATE VIEW ${db}.active_incoming_agg_view${onCluster} AS +select + timestamp, + workspaceId, + count +from ${db}.active_incoming_agg +ORDER BY workspaceId, timestamp DESC, insertedAt DESC +LIMIT 1 BY workspaceId, timestamp; + +-- ── per-minute event metrics ──────────────────────────────────────────────── + +-- Ingest entry point for per-event metrics; Null engine fans out to to_mv_metrics. +CREATE TABLE IF NOT EXISTS ${db}.metrics${onCluster} +( + timestamp DateTime, + messageId String, + workspaceId LowCardinality(String), + streamId LowCardinality(String), + connectionId LowCardinality(String), + functionId LowCardinality(String), + destinationId LowCardinality(String), + status LowCardinality(String), + events Int64, + eventIndex UInt32 +) +ENGINE = Null; + +create table ${db}.mv_metrics${onCluster} +( + timestamp DateTime, + workspaceId LowCardinality(String), + streamId LowCardinality(String), + connectionId LowCardinality(String), + functionId LowCardinality(String), + destinationId LowCardinality(String), + status LowCardinality(String), + events AggregateFunction(sum, Int64) +) +engine = ${engine:AggregatingMergeTree:mv_metrics} +ORDER BY (timestamp, workspaceId, streamId, connectionId, functionId, destinationId, status) +SETTINGS index_granularity = 8192; + +CREATE MATERIALIZED VIEW ${db}.to_mv_metrics${onCluster} +TO ${db}.mv_metrics +( + `timestamp` DateTime, + `workspaceId` LowCardinality(String), + `streamId` LowCardinality(String), + `connectionId` LowCardinality(String), + `functionId` LowCardinality(String), + `destinationId` LowCardinality(String), + `status` LowCardinality(String), + `events` AggregateFunction(sum, Int64) +) +AS +SELECT + date_trunc('minute', timestamp) as timestamp, + workspaceId, + streamId, + connectionId, + functionId, + destinationId, + status, + sumState(events) AS events +FROM ${db}.metrics +GROUP BY + timestamp, + workspaceId, + streamId, + connectionId, + functionId, + destinationId, + status; diff --git a/webapps/console/prisma/metrics.sql b/webapps/console/prisma/metrics.sql deleted file mode 100644 index ffdeb073d..000000000 --- a/webapps/console/prisma/metrics.sql +++ /dev/null @@ -1,156 +0,0 @@ --- not managed by prisma -create table newjitsu_metrics.active_incoming on cluster `jitsu-cluster` -( - timestamp DateTime, - workspaceId LowCardinality(String), - messageId String -) - engine = Null; - -CREATE TABLE newjitsu_metrics.active_incoming_agg on cluster `jitsu-cluster` - ( - `workspaceId` LowCardinality(String), - `timestamp` DateTime, - `insertedAt` DateTime DEFAULT now(), - `count` UInt64 - ) - ENGINE = ReplacingMergeTree() ORDER BY (workspaceId, timestamp) - PARTITION BY toYYYYMM(timestamp) -; - -CREATE MATERIALIZED VIEW newjitsu_metrics.mv_active_incoming_agg on cluster `jitsu-cluster` - REFRESH EVERY 5 MINUTE APPEND TO newjitsu_metrics.active_incoming_agg AS -select - workspaceId, - timestamp, - now() as insertedAt, - uniqMerge(count) as "count" -from newjitsu_metrics.mv_active_incoming2 -where - timestamp >= date_trunc('day', now() - interval 3 day) -group by workspaceId, timestamp order by workspaceId, timestamp; - -CREATE VIEW newjitsu_metrics.active_incoming_agg_view ON CLUSTER `jitsu-cluster` as select - timestamp, - workspaceId, - count -from newjitsu_metrics.active_incoming_agg -ORDER BY workspaceId, timestamp DESC, insertedAt DESC -LIMIT 1 BY workspaceId, timestamp; - -CREATE MATERIALIZED VIEW newjitsu_metrics.mv_active_incoming2 on cluster `jitsu-cluster` - ( - `workspaceId` LowCardinality(String), - `timestamp` DateTime, - `count` AggregateFunction(uniq, String) - ) - ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/{shard}/newjitsu_metrics/mv_active_incoming2_0', - '{replica}') - ORDER BY (workspaceId, timestamp) - PARTITION BY toYYYYMM(timestamp) - SETTINGS index_granularity = 8192 -AS -SELECT - workspaceId, - timestamp, - uniqState(messageId) AS count -FROM newjitsu_metrics.active_incoming -GROUP BY - workspaceId, - timestamp; - - - -create table newjitsu_metrics.mv_active_incoming3 on cluster `jitsu-cluster` -( - `workspaceId` LowCardinality(String), - `timestamp` DateTime, - `messageId` String -) - engine = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/newjitsu_metrics/mv_active_incoming3_1', '{replica}') - ORDER BY (workspaceId, timestamp, messageId) - PRIMARY KEY (workspaceId, timestamp) - PARTITION BY toYYYYMM(timestamp) -; - ---drop table newjitsu_metrics.mv_active_incoming3 on cluster `jitsu-cluster`; ---drop view newjitsu_metrics.to_mv_active_incoming3 on cluster `jitsu-cluster`; - -CREATE MATERIALIZED VIEW newjitsu_metrics.to_mv_active_incoming3 on cluster `jitsu-cluster` - TO newjitsu_metrics.mv_active_incoming3 - ( - `workspaceId` LowCardinality(String), - `timestamp` DateTime, - `messageId` String - ) -AS -SELECT - workspaceId, - timestamp, - messageId -FROM newjitsu_metrics.active_incoming; - - -CREATE TABLE IF NOT EXISTS newjitsu_metrics.metrics on cluster `jitsu-cluster` -( - timestamp DateTime, - messageId String, - workspaceId LowCardinality(String), - streamId LowCardinality(String), - connectionId LowCardinality(String), - functionId LowCardinality(String), - destinationId LowCardinality(String), - status LowCardinality(String), - events Int64, - eventIndex UInt32 - ) - ENGINE = Null; - -create table newjitsu_metrics.mv_metrics on cluster `jitsu-cluster` -( - timestamp DateTime, - workspaceId LowCardinality(String), - streamId LowCardinality(String), - connectionId LowCardinality(String), - functionId LowCardinality(String), - destinationId LowCardinality(String), - status LowCardinality(String), - events AggregateFunction(sum, Int64) -) - engine = ReplicatedAggregatingMergeTree('/clickhouse/tables/{shard}/newjitsu_metrics/mv_metrics', '{replica}') - ORDER BY (timestamp, workspaceId, streamId, connectionId, functionId, destinationId, status) - SETTINGS index_granularity = 8192; - - -CREATE MATERIALIZED VIEW newjitsu_metrics.to_mv_metrics on cluster `jitsu-cluster` - TO newjitsu_metrics.mv_metrics - ( - `timestamp` DateTime, - `workspaceId` LowCardinality(String), - `streamId` LowCardinality(String), - `connectionId` LowCardinality(String), - `functionId` LowCardinality(String), - `destinationId` LowCardinality(String), - `status` LowCardinality(String), - `events` AggregateFunction(sum, Int64) - ) -AS -SELECT - date_trunc('minute', timestamp) as timestamp, - workspaceId, - streamId, - connectionId, - functionId, - destinationId, - status, - sumState(events) AS events -FROM newjitsu_metrics.metrics -GROUP BY - timestamp, - workspaceId, - streamId, - connectionId, - functionId, - destinationId, - status; - diff --git a/webapps/console/sql.d.ts b/webapps/console/sql.d.ts new file mode 100644 index 000000000..41cb74fd3 --- /dev/null +++ b/webapps/console/sql.d.ts @@ -0,0 +1,6 @@ +// `.sql` files are imported as raw strings — via raw-loader in Next (see +// next.config.js) and via the raw-sql plugin in vitest (see vitest.config.ts). +declare module "*.sql" { + const content: string; + export default content; +} diff --git a/webapps/console/vitest.config.ts b/webapps/console/vitest.config.ts index 7c7cd49aa..45628eb9d 100644 --- a/webapps/console/vitest.config.ts +++ b/webapps/console/vitest.config.ts @@ -1,4 +1,23 @@ -import { defineConfig } from "vitest/config"; +import { readFileSync } from "node:fs"; +import { defineConfig, type Plugin } from "vitest/config"; + +// Load `.sql` files as raw string modules (`export default ""`), +// mirroring Next's raw-loader. clickhouse-init.ts imports the .clickhouse.sql +// DDL templates this way; without this the integration setup (which imports it) +// would fail to resolve them. vite:esbuild ignores the .sql extension, so the +// JS we return here passes through untransformed. +function rawSql(): Plugin { + return { + name: "raw-sql", + load(id) { + const file = id.split("?")[0]; + if (file.endsWith(".sql")) { + return `export default ${JSON.stringify(readFileSync(file, "utf-8"))};`; + } + return null; + }, + }; +} // Two projects: // unit — pure tests, no external dependencies. Runs anywhere, instantly. @@ -10,6 +29,7 @@ import { defineConfig } from "vitest/config"; // Running only unit tests (vitest run --project unit, or a file filter that // matches nothing under __tests__/integration/) starts no containers. export default defineConfig({ + plugins: [rawSql()], // Use the automatic JSX runtime so tests that transitively import `.tsx` files // (e.g. the destinations catalog → icon components) don't fail with // "React is not defined". The app compiles JSX via Next's automatic runtime;