feat(webapp): Queue size and webhook limit metrics

Adds queue-size and webhook limit metrics to the `metrics` query table
so alerts can watch rate limits, concurrency, and queue capacity.

Mono-RevId: 5d2716bcd252bd59d921fddd08e57ef4c2774eb6

Author: kathiekiwi

.server-changes/limit-metrics.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+The `metrics` query table now includes queue size (`queue.size.*`) and webhook limit metrics (`webhook.waiters.*`, `webhook.processing.*`, `webhook.ingress.*`), so you can chart and alert on queue capacity and webhook rate limits.

apps/webapp/app/entry.server.tsx

@@ -15,6 +15,7 @@ import { initAlertEngineWorker } from "~/v3/alertEngineWorker.server";
 import { initLogsSearchProjectorWorker } from "~/v3/logsSearchProjectorWorker.server";
 import { initQueueMetricsConsumer, initQueueMetricsEmitter } from "~/v3/queueMetrics.server";
 import { initApiRateLimitMetrics } from "~/services/apiRateLimitMetrics.server";
+import { initWebhookMetrics } from "~/services/webhookMetrics.server";
 import { bootstrap } from "./bootstrap";
 import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import type { OperatingSystemPlatform } from "./components/primitives/OperatingSystemProvider";
@@ -301,6 +302,7 @@ initLogsSearchProjectorWorker();
 initQueueMetricsEmitter();
 initQueueMetricsConsumer();
 initApiRateLimitMetrics();
+initWebhookMetrics();
 
 bootstrap().catch((error) => {
   logError(error);

apps/webapp/app/env.server.ts

@@ -839,6 +839,18 @@ const EnvironmentSchema = z
       .positive()
       .default(10_000),
 
+    QUEUE_SIZE_METRICS_INTERVAL_SEC: z.coerce
+      .number()
+      .int()
+      .positive()
+      .refine((seconds) => 60 % seconds === 0, { message: "must divide 60" })
+      .default(30),
+    QUEUE_SIZE_METRICS_MAX_ENVS: z.coerce.number().int().positive().default(2000),
+
+    WEBHOOK_METRICS_FLUSH_INTERVAL_MS: z.coerce.number().int().positive().default(10_000),
+    WEBHOOK_METRICS_MAX_ENTRIES: z.coerce.number().int().positive().default(10_000),
+    WEBHOOK_METRICS_MAX_ENDPOINTS: z.coerce.number().int().positive().default(500),
+
     API_RATE_LIMIT_JWT_WINDOW: z.string().default("1m"),
     API_RATE_LIMIT_JWT_TOKENS: z.coerce.number().int().default(60),
 

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

@@ -3,7 +3,10 @@ import { env } from "~/env.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { singleton } from "~/utils/singleton";
 import { meter } from "~/v3/tracer.server";
-import { ApiRateLimitMetricsAggregator } from "./apiRateLimitMetricsAggregator.server";
+import {
+  createApiRateLimitMetricsAggregator,
+  type ApiRateLimitMetricsAggregator,
+} from "./apiRateLimitMetricsAggregator.server";
 import {
   apiRateLimitMetricsInsertSettings,
   exportApiRateLimitMetricRows,
@@ -13,7 +16,6 @@ import type {
   RateLimitTenant,
 } from "./authorizationRateLimitMiddleware.server";
 import { logger } from "./logger.server";
-import { signalsEmitter } from "./signals.server";
 
 function enabledByEnv(): boolean {
   return env.API_RATE_LIMIT_METRICS_ENABLED !== "0";
@@ -86,68 +88,16 @@ function createAggregator(): ApiRateLimitMetricsAggregator {
       "API rate limit metric rows lost because ClickHouse rejected the insert request; failures while writing a queued async insert are not visible here",
   });
 
-  const aggregator = new ApiRateLimitMetricsAggregator({
+  const aggregator = createApiRateLimitMetricsAggregator({
     bucketSeconds: env.API_RATE_LIMIT_METRICS_BUCKET_SECONDS,
     maxEntries: env.API_RATE_LIMIT_METRICS_MAX_ENTRIES,
     isEnabled: createRuntimeGate(),
     sink: (rows) => exportRows(rows, (failed) => insertFailedCounter.add(failed)),
-    onDropped: (count) => droppedCounter.add(count, { reason: "cap" }),
+    counters: { dropped: droppedCounter, rowsFlushed: flushedCounter },
+    flushIntervalMs: env.API_RATE_LIMIT_METRICS_FLUSH_INTERVAL_MS,
+    ready: clickhouseFactory.isReady(),
   });
-
-  let routingReady = false;
-  clickhouseFactory
-    .isReady()
-    .then(() => {
-      routingReady = true;
-    })
-    .catch((error) => {
-      logger.error("api rate limit metrics: data store registry never became ready", {
-        error: error instanceof Error ? error.message : String(error),
-      });
-    });
-
-  let deferredWarned = false;
-  const flush = () => {
-    if (!routingReady) {
-      if (aggregator.size > 0 && !deferredWarned) {
-        deferredWarned = true;
-        logger.warn("api rate limit metrics: flush deferred, data store routing not ready", {
-          pendingEntries: aggregator.size,
-        });
-      }
-      return;
-    }
-    try {
-      const flushed = aggregator.flush();
-      if (flushed > 0) {
-        flushedCounter.add(flushed);
-      }
-    } catch (error) {
-      logger.error("api rate limit metrics: flush failed", {
-        error: error instanceof Error ? error.message : String(error),
-      });
-    }
-  };
-
-  const interval = setInterval(flush, env.API_RATE_LIMIT_METRICS_FLUSH_INTERVAL_MS);
-  interval.unref();
-
-  const shutdown = () => {
-    clearInterval(interval);
-    if (!routingReady) {
-      const dropped = aggregator.discard();
-      if (dropped > 0) {
-        droppedCounter.add(dropped, { reason: "shutdown_before_ready" });
-        logger.warn("api rate limit metrics: shutting down before routing was ready, dropping", {
-          droppedObservations: dropped,
-        });
-      }
-      return;
-    }
-    flush();
-  };
-  signalsEmitter.on("SIGTERM", shutdown);
-  signalsEmitter.on("SIGINT", shutdown);
+  aggregator.start();
 
   return aggregator;
 }

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

@@ -1,10 +1,13 @@
-import type { MetricsV1Input } from "@internal/clickhouse";
 import type {
   RateLimiterConfig,
   RateLimitObservation,
   RateLimitTenant,
 } from "./authorizationRateLimitMiddleware.server";
-import { logger } from "./logger.server";
+import {
+  createMetricsAggregator,
+  type MetricsAggregatorOptions,
+  type MetricUpdate,
+} from "./metricsAggregator.server";
 import { parseDuration } from "./realtime/duration.server";
 
 export const API_RATE_LIMIT_METRIC_NAMES = {
@@ -53,192 +56,61 @@ export function rateLimitConfigToLimits(config: RateLimiterConfig): RateLimitLim
   return { perSecond: config.tokens / windowSeconds, burst: config.tokens };
 }
 
-export type ApiRateLimitMetricsAggregatorOptions = {
-  bucketSeconds: number;
-  /** Hard cap on distinct (environment, bucket) entries held between flushes. */
-  maxEntries: number;
-  /** Runtime gate, evaluated per observation once a tenant is known. */
-  isEnabled: (tenant: RateLimitTenant) => boolean;
-  /** Receives every flushed batch. A rejected promise or a throw is logged and the batch dropped. */
-  sink: (rows: MetricsV1Input[]) => void | Promise<void>;
-  now?: () => number;
-  onDropped?: (count: number) => void;
-};
-
-type Entry = {
-  organizationId: string;
-  projectId: string;
-  environmentId: string;
-  bucketStartMs: number;
-  allowed: number;
-  denied: number;
-  remainingMin: number;
-  limits: RateLimitLimits | undefined;
-};
+export type ApiRateLimitMetricsAggregatorOptions = Omit<
+  MetricsAggregatorOptions<RateLimitTenant>,
+  "name"
+>;
 
 /**
- * In-process counter for the API rate limiter's allow/deny decisions, keyed by
- * (environment, time bucket). Every flush emits the counts accumulated since the previous
- * flush as delta rows, so several replicas (or one replica across a flush boundary) may
- * emit rows for the same key and sum() at read time stays exact.
+ * Counts the API rate limiter's allow/deny decisions per (environment, time bucket) as
+ * delta rows, so sum() at read time stays exact across replicas and flushes.
  */
-export class ApiRateLimitMetricsAggregator {
-  private readonly entries = new Map<string, Entry>();
-  private readonly bucketMs: number;
-  private readonly maxEntries: number;
-  private readonly isEnabled: (tenant: RateLimitTenant) => boolean;
-  private readonly sink: (rows: MetricsV1Input[]) => void | Promise<void>;
-  private readonly now: () => number;
-  private readonly onDropped?: (count: number) => void;
-
-  constructor(options: ApiRateLimitMetricsAggregatorOptions) {
-    this.bucketMs = options.bucketSeconds * 1000;
-    this.maxEntries = options.maxEntries;
-    this.isEnabled = options.isEnabled;
-    this.sink = options.sink;
-    this.now = options.now ?? Date.now;
-    this.onDropped = options.onDropped;
-  }
-
-  get size(): number {
-    return this.entries.size;
-  }
+export function createApiRateLimitMetricsAggregator(options: ApiRateLimitMetricsAggregatorOptions) {
+  const metrics = createMetricsAggregator({ ...options, name: "api rate limit metrics" });
 
-  record(observation: RateLimitObservation): void {
+  const record = (observation: RateLimitObservation): void => {
     const tenant = observation.tenant;
-    if (!tenant || !this.isEnabled(tenant)) {
+    if (!tenant) {
       return;
     }
+    metrics.record(tenant, observationUpdates(observation), () => limitUpdates(observation.config));
+  };
 
-    const bucketStartMs = Math.floor(this.now() / this.bucketMs) * this.bucketMs;
-    const key = `${tenant.environmentId}\0${bucketStartMs}`;
-    const remaining = Math.max(0, observation.remaining);
-
-    let entry = this.entries.get(key);
-    if (!entry) {
-      if (this.entries.size >= this.maxEntries) {
-        this.onDropped?.(1);
-        return;
-      }
-      entry = {
-        organizationId: tenant.organizationId,
-        projectId: tenant.projectId,
-        environmentId: tenant.environmentId,
-        bucketStartMs,
-        allowed: 0,
-        denied: 0,
-        remainingMin: remaining,
-        limits: rateLimitConfigToLimits(observation.config),
-      };
-      this.entries.set(key, entry);
-    }
-
-    if (observation.success) {
-      entry.allowed++;
-    } else {
-      entry.denied++;
-    }
-    entry.remainingMin = Math.min(entry.remainingMin, remaining);
-  }
-
-  /**
-   * Throws away everything held without emitting it and returns how many observations
-   * were folded into the discarded entries, so callers can account for them as dropped.
-   */
-  discard(): number {
-    let observations = 0;
-    for (const entry of this.entries.values()) {
-      observations += entry.allowed + entry.denied;
-    }
-    this.entries.clear();
-    return observations;
-  }
-
-  flush(): number {
-    if (this.entries.size === 0) {
-      return 0;
-    }
-
-    const rows: MetricsV1Input[] = [];
-    for (const entry of this.entries.values()) {
-      rows.push(...entryToRows(entry));
-    }
-    this.entries.clear();
-
-    const dropBatch = (error: unknown) => {
-      logger.error("api rate limit metrics: sink failed, dropping batch", {
-        rows: rows.length,
-        error: error instanceof Error ? error.message : String(error),
-      });
-    };
-    try {
-      const result = this.sink(rows);
-      if (result instanceof Promise) {
-        result.catch(dropBatch);
-      }
-    } catch (error) {
-      dropBatch(error);
-    }
-
-    return rows.length;
-  }
-}
-
-function entryToRows(entry: Entry): MetricsV1Input[] {
-  const base = {
-    organization_id: entry.organizationId,
-    project_id: entry.projectId,
-    environment_id: entry.environmentId,
-    metric_subject: "",
-    bucket_start: formatBucketStart(entry.bucketStartMs),
-    attributes: {},
+  return {
+    record,
+    size: metrics.size,
+    discard: metrics.discard,
+    flush: metrics.flush,
+    start: metrics.start,
   };
+}
 
-  const rows: MetricsV1Input[] = [];
-  if (entry.allowed > 0) {
-    rows.push({
-      ...base,
-      metric_name: API_RATE_LIMIT_METRIC_NAMES.allowed,
-      metric_type: "sum",
-      value: entry.allowed,
-    });
-  }
-  if (entry.denied > 0) {
-    rows.push({
-      ...base,
-      metric_name: API_RATE_LIMIT_METRIC_NAMES.denied,
-      metric_type: "sum",
-      value: entry.denied,
-    });
-  }
-  rows.push({
-    ...base,
-    metric_name: API_RATE_LIMIT_METRIC_NAMES.remainingMin,
-    metric_type: "gauge",
-    value: entry.remainingMin,
-  });
-  if (entry.limits) {
-    rows.push(
-      {
-        ...base,
-        metric_name: API_RATE_LIMIT_METRIC_NAMES.limitPerSecond,
-        metric_type: "gauge",
-        value: entry.limits.perSecond,
-      },
-      {
-        ...base,
-        metric_name: API_RATE_LIMIT_METRIC_NAMES.limitBurst,
-        metric_type: "gauge",
-        value: entry.limits.burst,
-      }
-    );
-  }
-  return rows;
+export type ApiRateLimitMetricsAggregator = ReturnType<typeof createApiRateLimitMetricsAggregator>;
+
+function observationUpdates(observation: RateLimitObservation): MetricUpdate[] {
+  return [
+    {
+      name: observation.success
+        ? API_RATE_LIMIT_METRIC_NAMES.allowed
+        : API_RATE_LIMIT_METRIC_NAMES.denied,
+      reducer: "sum",
+      value: 1,
+    },
+    {
+      name: API_RATE_LIMIT_METRIC_NAMES.remainingMin,
+      reducer: "min",
+      value: Math.max(0, observation.remaining),
+    },
+  ];
 }
 
-function formatBucketStart(epochMs: number): string {
-  return new Date(epochMs)
-    .toISOString()
-    .replace("T", " ")
-    .replace(/\.\d{3}Z$/, "");
+function limitUpdates(config: RateLimiterConfig): MetricUpdate[] {
+  const limits = rateLimitConfigToLimits(config);
+  if (!limits) {
+    return [];
+  }
+  return [
+    { name: API_RATE_LIMIT_METRIC_NAMES.limitPerSecond, reducer: "first", value: limits.perSecond },
+    { name: API_RATE_LIMIT_METRIC_NAMES.limitBurst, reducer: "first", value: limits.burst },
+  ];
 }