Drive ClickHouse metrics DDL from .sql templates + preinit bulker metrics destination

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

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

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();
   }