From ccdac92e4cb312297bea08313048d4617daeb59d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 7 Feb 2026 10:04:25 +0000 Subject: [PATCH] Generalize metrics pipeline: MetricDefinition + MetricsStreamConsumer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract the Redis Stream → ClickHouse pipeline into a generic @internal/metrics-pipeline package. Queue metrics becomes the first MetricDefinition implementation. The generic pipeline provides: - MetricDefinition interface (parse, dimension key, aggregate, insert) - MetricsStreamConsumer (XREADGROUP, pre-aggregate, INSERT, XACK) - MetricsStreamEmitter (convenience wrapper for Node.js producers) Any part of the app can now ship metrics to ClickHouse by: 1. Creating a ClickHouse table 2. Implementing MetricDefinition 3. Emitting XADD entries (from Lua or Node.js) 4. Registering a consumer at startup Includes worker health metrics as a second example definition. https://claude.ai/code/session_01PkVYeZ8tQzgcT4YQsctGo1 --- docs/queue-metrics-design.md | 414 +++++++++++++++++++++++++++++------ 1 file changed, 350 insertions(+), 64 deletions(-) diff --git a/docs/queue-metrics-design.md b/docs/queue-metrics-design.md index 182505013..519d16119 100644 --- a/docs/queue-metrics-design.md +++ b/docs/queue-metrics-design.md @@ -503,9 +503,346 @@ These are the new user-facing metrics enabled by this system: --- -## 6. Implementation Plan +## 6. Generic Metrics Pipeline -### Phase 1: Lua script changes +The transport layer (Redis Stream → Consumer → ClickHouse) is not queue-specific. It should be built as a generic pipeline that any part of the application can use to ship metrics to ClickHouse. Queue metrics is the first consumer. + +### 6.1 Architecture + +``` +┌──────────────────────┐ ┌──────────────────────┐ ┌──────────────────────┐ +│ Queue Lua Scripts │ │ Worker Health │ │ Future: API Metrics│ +│ (XADD in Lua) │ │ (XADD from Node.js) │ │ (XADD from Node.js)│ +└──────────┬───────────┘ └──────────┬───────────┘ └──────────┬───────────┘ + │ │ │ + ▼ ▼ ▼ +┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ +│ metrics:queue:0 │ │ metrics:worker:0 │ │ metrics:api:0 │ +│ metrics:queue:1 │ │ metrics:worker:1 │ │ metrics:api:1 │ +│ (Redis Streams) │ │ (Redis Streams) │ │ (Redis Streams) │ +└──────────┬───────┘ └──────────┬───────┘ └──────────┬───────┘ + │ │ │ + └───────────┬───────────┘───────────────────────┘ + ▼ + ┌──────────────────────────────┐ + │ MetricsStreamConsumer │ + │ (generic, one per metric │ + │ definition) │ + │ │ + │ - XREADGROUP per shard │ + │ - pre-aggregate via │ + │ MetricDefinition │ + │ - INSERT into target table │ + │ - XACK on success │ + └──────────────────────────────┘ +``` + +### 6.2 MetricDefinition interface + +Each metric type registers a definition that tells the pipeline how to parse, aggregate, and store its data: + +```typescript +/** + * Defines a metric type for the generic Redis Stream → ClickHouse pipeline. + * + * The pipeline handles: stream consumption, consumer groups, PEL recovery, + * retry with backoff, batching, and graceful shutdown. + * + * The metric definition handles: what the data looks like, how to aggregate + * it, and where it goes. + */ +interface MetricDefinition { + /** Unique name for this metric (used in stream keys, consumer groups) */ + name: string; + + /** Target ClickHouse table for inserts */ + clickhouseTable: string; + + /** Number of stream shards (streams are named `metrics:{name}:{shard}`) */ + shardCount: number; + + /** MAXLEN for each stream shard */ + maxStreamLength: number; + + /** Bucket size in milliseconds for pre-aggregation */ + bucketSizeMs: number; + + /** + * Parse a raw Redis Stream entry (string key-value pairs) + * into a typed entry. Return null to skip/filter the entry. + */ + parseEntry(fields: Record, streamId: string): TEntry | null; + + /** + * Extract the dimension key for grouping. + * Entries with the same dimension key and time bucket are aggregated together. + * Returns a string that uniquely identifies the dimension combination. + */ + dimensionKey(entry: TEntry): string; + + /** + * Extract the timestamp from a parsed entry (ms since epoch). + * Used to assign entries to time buckets. + */ + timestamp(entry: TEntry): number; + + /** + * Aggregate a batch of entries that share the same dimension key + * and time bucket into a single row for ClickHouse insertion. + */ + aggregate(dimensionKey: string, bucketStart: Date, entries: TEntry[]): TAggregated; + + /** + * Convert aggregated rows into the format expected by the ClickHouse client. + * Returns column names and values for JSONEachRow insert. + */ + toInsertRow(row: TAggregated): Record; +} +``` + +### 6.3 MetricsStreamConsumer (generic pipeline) + +```typescript +/** + * Generic consumer that reads from Redis Streams and inserts into ClickHouse. + * One instance per MetricDefinition. + */ +class MetricsStreamConsumer { + constructor(options: { + redis: RedisOptions; + clickhouse: ClickHouseClient; + definition: MetricDefinition; + consumerGroup: string; + consumerId: string; + pollIntervalMs?: number; // default: 1000 + batchSize?: number; // default: 1000 + }) {} + + async start(): Promise { + // For each shard: + // 1. XGROUP CREATE metrics:{name}:{shard} {consumerGroup} $ MKSTREAM + // 2. Start polling loop + } + + private async pollShard(shard: number): Promise { + // 1. Read pending entries first (PEL recovery): XREADGROUP ... 0 + // - INSERT these as a separate batch (enables CH insert dedup) + // - XACK on success + // 2. Read new entries: XREADGROUP ... > + // - Parse via definition.parseEntry() + // - Group by definition.dimensionKey() + time bucket + // - Aggregate via definition.aggregate() + // - Convert via definition.toInsertRow() + // - INSERT batch into definition.clickhouseTable + // - XACK on success + // 3. On failure: back off, retry from PEL next iteration + } + + async stop(): Promise { + // Signal shutdown, drain in-flight batches + } +} +``` + +### 6.4 MetricsStreamEmitter (convenience for Node.js producers) + +For metrics emitted from Node.js (not Lua), provide a thin helper: + +```typescript +/** + * Emits metric entries to a Redis Stream. For use in Node.js code. + * Lua scripts use XADD directly — this is for non-Lua producers. + */ +class MetricsStreamEmitter { + constructor(options: { + redis: Redis; + streamPrefix: string; // e.g., "metrics" + metricName: string; // e.g., "worker_health" + shardCount: number; + maxStreamLength?: number; // default: 100000 + }) {} + + /** + * Emit a metric entry to the appropriate shard. + * Shard selection can be based on a dimension value (e.g., envId) + * for locality, or round-robin. + */ + async emit( + fields: Record, + shardKey?: string + ): Promise { + const shard = shardKey + ? jumpHash(shardKey, this.shardCount) + : this.roundRobinShard(); + const streamKey = `${this.streamPrefix}:${this.metricName}:${shard}`; + await this.redis.xadd( + streamKey, 'MAXLEN', '~', this.maxStreamLength.toString(), '*', + ...Object.entries(fields).flat() + ); + } +} +``` + +### 6.5 Queue metrics as the first MetricDefinition + +```typescript +const queueMetricsDefinition: MetricDefinition = { + name: "queue", + clickhouseTable: "queue_metrics_5s_v1", + shardCount: 2, // match RunQueue shard count + maxStreamLength: 100_000, + bucketSizeMs: 5_000, // 5 seconds + + parseEntry(fields, streamId) { + return { + organizationId: fields.org, + projectId: fields.proj, + environmentId: fields.env, + queueName: fields.queue, + operation: fields.op, + timestamp: redisStreamIdToMs(streamId), + queueLength: parseInt(fields.ql ?? "0"), + concurrencyCurrent: parseInt(fields.cc ?? "0"), + envQueueLength: parseInt(fields.eql ?? "0"), + envConcurrency: parseInt(fields.ec ?? "0"), + oldestMessageAgeMs: parseInt(fields.age ?? "0"), + enqueueCount: parseInt(fields.eq ?? "0"), + dequeueCount: parseInt(fields.dq ?? "0"), + ackCount: parseInt(fields.ak ?? "0"), + nackCount: parseInt(fields.nk ?? "0"), + dlqCount: parseInt(fields.dlq ?? "0"), + ttlExpireCount: parseInt(fields.ttl ?? "0"), + waitDurationMs: parseInt(fields.wd ?? "0"), + }; + }, + + dimensionKey(entry) { + return `${entry.organizationId}:${entry.projectId}:${entry.environmentId}:${entry.queueName}`; + }, + + timestamp(entry) { + return entry.timestamp; + }, + + aggregate(dimensionKey, bucketStart, entries) { + const [orgId, projId, envId, queue] = dimensionKey.split(":"); + return { + organization_id: orgId, + project_id: projId, + environment_id: envId, + queue_name: queue, + bucket_start: bucketStart, + enqueue_count: sum(entries, "enqueueCount"), + dequeue_count: sum(entries, "dequeueCount"), + ack_count: sum(entries, "ackCount"), + nack_count: sum(entries, "nackCount"), + dlq_count: sum(entries, "dlqCount"), + ttl_expire_count: sum(entries, "ttlExpireCount"), + max_queue_length: max(entries, "queueLength"), + max_concurrency_current: max(entries, "concurrencyCurrent"), + max_env_queue_length: max(entries, "envQueueLength"), + max_env_concurrency: max(entries, "envConcurrency"), + max_oldest_message_age_ms: max(entries, "oldestMessageAgeMs"), + total_wait_duration_ms: sum(entries, "waitDurationMs"), + wait_duration_count: countNonZero(entries, "waitDurationMs"), + }; + }, + + toInsertRow(row) { + return { ...row, bucket_start: formatDateTime(row.bucket_start) }; + }, +}; +``` + +### 6.6 Example: adding a second metric type + +To ship a new metric to ClickHouse, you only need: + +1. **A ClickHouse table** (+ optional MVs for rollup) +2. **A MetricDefinition** implementation +3. **XADD calls** at the emission points (Lua or Node.js) +4. **Register the consumer** at startup + +For example, worker health metrics: + +```typescript +const workerHealthDefinition: MetricDefinition = { + name: "worker_health", + clickhouseTable: "worker_health_5s_v1", + shardCount: 1, + maxStreamLength: 50_000, + bucketSizeMs: 5_000, + + parseEntry(fields, streamId) { + return { + workerId: fields.wid, + environmentId: fields.env, + timestamp: redisStreamIdToMs(streamId), + cpuPercent: parseFloat(fields.cpu ?? "0"), + memoryMb: parseInt(fields.mem ?? "0"), + activeConnections: parseInt(fields.conn ?? "0"), + }; + }, + + dimensionKey(entry) { return `${entry.environmentId}:${entry.workerId}`; }, + timestamp(entry) { return entry.timestamp; }, + + aggregate(dimensionKey, bucketStart, entries) { + const [envId, workerId] = dimensionKey.split(":"); + return { + environment_id: envId, + worker_id: workerId, + bucket_start: bucketStart, + max_cpu_percent: max(entries, "cpuPercent"), + max_memory_mb: max(entries, "memoryMb"), + max_active_connections: max(entries, "activeConnections"), + sample_count: entries.length, + }; + }, + + toInsertRow(row) { return { ...row, bucket_start: formatDateTime(row.bucket_start) }; }, +}; + +// At startup: +const workerHealthConsumer = new MetricsStreamConsumer({ + redis: redisOptions, + clickhouse: clickhouseClient, + definition: workerHealthDefinition, + consumerGroup: "worker_health_cg", + consumerId: `consumer_${process.pid}`, +}); +await workerHealthConsumer.start(); +``` + +### 6.7 Where to put the generic pipeline + +``` +internal-packages/ + metrics-pipeline/ # NEW package: @internal/metrics-pipeline + src/ + types.ts # MetricDefinition interface + consumer.ts # MetricsStreamConsumer + emitter.ts # MetricsStreamEmitter + helpers.ts # sum(), max(), countNonZero(), redisStreamIdToMs() + index.ts # public exports + + run-engine/ + src/run-queue/ + queueMetrics.ts # queueMetricsDefinition (implements MetricDefinition) + index.ts # Lua scripts with XADD emission +``` + +The generic pipeline lives in its own internal package so it can be used by any app (webapp, supervisor) without depending on run-engine. + +--- + +## 7. Queue-Specific Implementation Plan + +### Phase 1: Generic pipeline package + +Create `@internal/metrics-pipeline` with `MetricDefinition`, `MetricsStreamConsumer`, `MetricsStreamEmitter`, and helpers. This is framework code with no queue-specific logic. + +### Phase 2: Lua script changes Modify each Lua script to accept an additional KEYS entry (the metrics stream key) and emit an XADD at the end. The additional KEYS/ARGV entries to pass: @@ -528,66 +865,15 @@ pcall(function() end) ``` -### Phase 2: Stream consumer +### Phase 3: Queue metric definition + consumer wiring -Create a new service class `QueueMetricsConsumer` in `internal-packages/run-engine/src/run-queue/`: +Create `queueMetricsDefinition` (section 6.5) and wire a `MetricsStreamConsumer` for it in the webapp startup. The queue metric definition specifies the 5s bucket size, the target ClickHouse table, and the aggregation logic. -```typescript -export class QueueMetricsConsumer { - constructor(options: { - redis: RedisOptions; - clickhouse: ClickHouseClient; - shardCount: number; - consumerGroup: string; - consumerId: string; - pollIntervalMs?: number; // default: 1000 - batchSize?: number; // default: 1000 - maxRetries?: number; // default: 3 - }) {} - - async start(): Promise { - // 1. Create consumer group if not exists (XGROUP CREATE ... MKSTREAM) - // 2. Start polling loop for each shard - } - - private async pollShard(shard: number): Promise { - // 1. First, check for pending (unacked) entries: XREADGROUP ... 0 - // 2. Then read new entries: XREADGROUP ... > - // 3. Pre-aggregate entries into 5s buckets in memory (see below) - // 4. Bulk INSERT aggregated rows into queue_metrics_5s_v1 - // 5. On success: XACK all processed IDs - // 6. On failure: back off, retry from PEL next iteration - } - - /** - * Pre-aggregates raw stream entries into 5-second buckets. - * - * Groups entries by (org, project, env, queue, floor(timestamp / 5000)) - * and computes: - * - Counters: sum of enqueue_count, dequeue_count, etc. - * - Gauges: max of queue_length, concurrency_current, etc. - * - Latency: sum of wait_duration_ms, count of non-zero waits - * - * This reduces N raw stream entries into M << N aggregated rows - * (one per active queue per 5s window in the batch). - * - * SummingMergeTree handles the case where two consumer batches - * produce rows for the same 5s bucket — they merge correctly - * on background merge. - */ - private preAggregate(entries: StreamEntry[]): AggregatedRow[] { ... } - - async stop(): Promise { - // Graceful shutdown - } -} -``` - -### Phase 3: ClickHouse migration +### Phase 4: ClickHouse migration Add migration `016_add_queue_metrics.sql` with the 5s table, minute/hour tables, and the two materialized views (5s→minute, minute→hour) from section 3. -### Phase 4: API and presenters +### Phase 5: API and presenters - New `QueueMetricsPresenter` that queries the 5s/minute/hour tables (auto-selects based on time range) - New API endpoint `GET /api/v1/queues/:queueParam/metrics` @@ -595,7 +881,7 @@ Add migration `016_add_queue_metrics.sql` with the 5s table, minute/hour tables, --- -## 7. Alerting Architecture +## 8. Alerting Architecture ### How alerts fit in @@ -699,7 +985,7 @@ When the condition is no longer met, the evaluator can optionally auto-resolve: --- -## 8. Risks +## 9. Risks ### Risk 1: Double-counting on consumer crash (MEDIUM) @@ -765,7 +1051,7 @@ Adding a metrics stream key (e.g., `queue_metrics:shard:0`) to these Lua scripts --- -## 9. Metric Importance Ranking +## 10. Metric Importance Ranking Ranked by user value — how directly the metric answers questions users actually ask. @@ -827,7 +1113,7 @@ Ranked by user value — how directly the metric answers questions users actuall --- -## 10. Performance Considerations +## 11. Performance Considerations ### Redis impact @@ -872,7 +1158,7 @@ The critical scaling property: a queue that processes 1 event/5s and a queue tha --- -## 11. Failure Modes and Recovery +## 12. Failure Modes and Recovery | Failure | Impact | Recovery | |---------|--------|----------| @@ -886,7 +1172,7 @@ The key invariant: **queue operations (enqueue/dequeue/ack) are never blocked or --- -## 12. Migration and Rollout Strategy +## 13. Migration and Rollout Strategy 1. **Feature flag**: Pass a metrics-enabled flag as an ARGV to Lua scripts from Node.js (see Risk 5 — avoids an extra GET on every Lua invocation). The Node.js layer caches the flag from a Redis key (`queue_metrics:enabled`) and refreshes every 10s: ```lua @@ -906,7 +1192,7 @@ The key invariant: **queue operations (enqueue/dequeue/ack) are never blocked or --- -## 13. Summary of Tradeoffs +## 14. Summary of Tradeoffs | Decision | Alternative | Why This Choice | |----------|-------------|-----------------|