Add env vars for configuring the llm_metrics_v1 flush scheudler settings

Author: ericallam

apps/webapp/app/env.server.ts

@@ -1288,6 +1288,10 @@ const EnvironmentSchema = z
     LLM_PRICING_RELOAD_INTERVAL_MS: z.coerce.number().int().default(5 * 60 * 1000), // 5 minutes
     LLM_PRICING_SEED_ON_STARTUP: BoolEnv.default(false),
     LLM_PRICING_READY_TIMEOUT_MS: z.coerce.number().int().default(500),
+    LLM_METRICS_BATCH_SIZE: z.coerce.number().int().default(5000),
+    LLM_METRICS_FLUSH_INTERVAL_MS: z.coerce.number().int().default(2000),
+    LLM_METRICS_MAX_BATCH_SIZE: z.coerce.number().int().default(10000),
+    LLM_METRICS_MAX_CONCURRENCY: z.coerce.number().int().default(2),
 
     // Bootstrap
     TRIGGER_BOOTSTRAP_ENABLED: z.string().default("0"),

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts

@@ -86,6 +86,11 @@ export type ClickhouseEventRepositoryConfig = {
    * - "v2": Uses task_events_v2 (partitioned by inserted_at to avoid "too many parts" errors)
    */
   version?: "v1" | "v2";
+  /** LLM metrics flush scheduler config */
+  llmMetricsBatchSize?: number;
+  llmMetricsFlushInterval?: number;
+  llmMetricsMaxBatchSize?: number;
+  llmMetricsMaxConcurrency?: number;
 };
 
 /**
@@ -123,13 +128,13 @@ export class ClickhouseEventRepository implements IEventRepository {
     });
 
     this._llmMetricsFlushScheduler = new DynamicFlushScheduler({
-      batchSize: 5000,
-      flushInterval: 2000,
+      batchSize: config.llmMetricsBatchSize ?? 5000,
+      flushInterval: config.llmMetricsFlushInterval ?? 2000,
       callback: this.#flushLlmMetricsBatch.bind(this),
       minConcurrency: 1,
-      maxConcurrency: 2,
-      maxBatchSize: 10000,
-      memoryPressureThreshold: 10000,
+      maxConcurrency: config.llmMetricsMaxConcurrency ?? 2,
+      maxBatchSize: config.llmMetricsMaxBatchSize ?? 10000,
+      memoryPressureThreshold: config.llmMetricsMaxBatchSize ?? 10000,
       loadSheddingEnabled: false,
     });
   }

apps/webapp/app/v3/eventRepository/clickhouseEventRepositoryInstance.server.ts

@@ -64,6 +64,10 @@ function initializeClickhouseRepository() {
     asyncInsertMaxDataSize: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE,
     asyncInsertBusyTimeoutMs: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS,
     startTimeMaxAgeMs: env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS,
+    llmMetricsBatchSize: env.LLM_METRICS_BATCH_SIZE,
+    llmMetricsFlushInterval: env.LLM_METRICS_FLUSH_INTERVAL_MS,
+    llmMetricsMaxBatchSize: env.LLM_METRICS_MAX_BATCH_SIZE,
+    llmMetricsMaxConcurrency: env.LLM_METRICS_MAX_CONCURRENCY,
     version: "v1",
   });
 
@@ -97,6 +101,10 @@ function initializeClickhouseRepositoryV2() {
     waitForAsyncInsert: env.EVENTS_CLICKHOUSE_WAIT_FOR_ASYNC_INSERT === "1",
     asyncInsertMaxDataSize: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE,
     asyncInsertBusyTimeoutMs: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS,
+    llmMetricsBatchSize: env.LLM_METRICS_BATCH_SIZE,
+    llmMetricsFlushInterval: env.LLM_METRICS_FLUSH_INTERVAL_MS,
+    llmMetricsMaxBatchSize: env.LLM_METRICS_MAX_BATCH_SIZE,
+    llmMetricsMaxConcurrency: env.LLM_METRICS_MAX_CONCURRENCY,
     version: "v2",
   });