feat(webapp): split logs search ClickHouse URL into reader and writer

Mono-RevId: 65c2b9b96bc9a32a351ab93a70eb65773b6ede6e

Author: carderne

apps/webapp/app/env.server.ts

@@ -2226,8 +2226,8 @@ const EnvironmentSchema = z
       .nonnegative()
       .optional(),
 
-    // Scheduled logs-search projection. Disabled by default. LOGS_CLICKHOUSE_URL, or the
-    // CLICKHOUSE_URL fallback, must reach both source and destination tables and allow writes.
+    // Scheduled logs-search projection. Disabled by default. LOGS_SEARCH_WRITER_CLICKHOUSE_URL,
+    // or the CLICKHOUSE_URL fallback, must reach both source and destination tables and allow writes.
     LOGS_SEARCH_PROJECTOR_ENABLED: BoolEnv.default(false),
     LOGS_SEARCH_PROJECTOR_PREVIEW_ENABLED: BoolEnv.default(false),
     LOGS_SEARCH_PROJECTOR_MAX_WINDOWS_PER_TICK: z.coerce.number().int().min(1).max(20).default(5),
@@ -2249,6 +2249,11 @@ const EnvironmentSchema = z
     LOGS_SEARCH_DUAL_WRITE_ORGANIZATION_IDS: z.string().default(""),
     LOGS_SEARCH_DUAL_WRITE_MAX_CONCURRENCY: z.coerce.number().int().min(1).max(10).default(2),
     LOGS_SEARCH_DUAL_WRITE_MAX_PENDING: z.coerce.number().int().min(1).max(100).default(4),
+    // Logs-search write side (dual writer, projector). Must share storage with
+    // EVENTS_CLICKHOUSE_URL and allow writes. Unset: the dual writer uses the events client.
+    LOGS_SEARCH_WRITER_CLICKHOUSE_URL: z.string().optional(),
+    // Logs-search read side (logs page). Falls back to CLICKHOUSE_READER_URL, then CLICKHOUSE_URL.
+    LOGS_SEARCH_READER_CLICKHOUSE_URL: z.string().optional(),
 
     // Logs list pagination tuning.
     LOGS_LIST_DEFAULT_PAGE_SIZE: z.coerce.number().int().positive().default(50),
@@ -2260,9 +2265,6 @@ const EnvironmentSchema = z
     // AI features (Prompts, Models, AI Metrics sidebar section)
     AI_FEATURES_ENABLED: z.string().default("0"),
 
-    // Logs page ClickHouse URL (for logs queries)
-    LOGS_CLICKHOUSE_URL: z.string().optional(),
-
     // Query page ClickHouse limits (for TSQL queries)
     QUERY_CLICKHOUSE_URL: z
       .string()

apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts

@@ -38,7 +38,7 @@ const defaultLogsClickhouseClient = singleton(
 );
 
 function initializeLogsSearchProjectorClickhouseClient() {
-  const url = new URL(env.LOGS_CLICKHOUSE_URL ?? env.CLICKHOUSE_URL);
+  const url = new URL(env.LOGS_SEARCH_WRITER_CLICKHOUSE_URL ?? env.CLICKHOUSE_URL);
   url.searchParams.delete("secure");
 
   return new ClickHouse({
@@ -80,7 +80,9 @@ function getLogsListClickhouseSettings() {
 }
 
 function initializeLogsClickhouseClient() {
-  const url = new URL(env.LOGS_CLICKHOUSE_URL ?? env.CLICKHOUSE_READER_URL ?? env.CLICKHOUSE_URL);
+  const url = new URL(
+    env.LOGS_SEARCH_READER_CLICKHOUSE_URL ?? env.CLICKHOUSE_READER_URL ?? env.CLICKHOUSE_URL
+  );
   url.searchParams.delete("secure");
 
   return new ClickHouse({
@@ -466,6 +468,39 @@ function initializeEventsClickhouseClient(): ClickHouse {
   });
 }
 
+/**
+ * Logs-search dual-write destination (`LOGS_SEARCH_WRITER_CLICKHOUSE_URL`). Shares storage with the
+ * events warehouse; a separate service keeps the extra insert stream off the events writer.
+ * Undefined when unset, in which case the dual writer uses the events client.
+ */
+const defaultLogsSearchClickhouseClient = singleton(
+  "logsSearchClickhouseClient",
+  initializeLogsSearchClickhouseClient
+);
+
+function initializeLogsSearchClickhouseClient(): ClickHouse | undefined {
+  if (!env.LOGS_SEARCH_WRITER_CLICKHOUSE_URL) {
+    return undefined;
+  }
+
+  const url = new URL(env.LOGS_SEARCH_WRITER_CLICKHOUSE_URL);
+  url.searchParams.delete("secure");
+
+  return new ClickHouse({
+    name: "logs-search-writer",
+    url: url.toString(),
+    keepAlive: {
+      enabled: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
+      idleSocketTtl: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+    },
+    logLevel: env.EVENTS_CLICKHOUSE_LOG_LEVEL,
+    compression: {
+      request: env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST === "1",
+    },
+    maxOpenConnections: env.LOGS_SEARCH_DUAL_WRITE_MAX_CONCURRENCY,
+  });
+}
+
 // ---------------------------------------------------------------------------
 // Helpers
 // ---------------------------------------------------------------------------
@@ -732,7 +767,11 @@ export class ClickhouseFactory {
       let defaultRepo = this._eventRepositoryCache.get(defaultKey);
       if (!defaultRepo) {
         const eventsClickhouse = getEventsClickhouseClient();
-        defaultRepo = buildEventRepository(store, eventsClickhouse);
+        defaultRepo = buildEventRepository(
+          store,
+          eventsClickhouse,
+          defaultLogsSearchClickhouseClient
+        );
         this._eventRepositoryCache.set(defaultKey, defaultRepo);
       }
       return { key: defaultKey, repository: defaultRepo };
@@ -786,11 +825,16 @@ const logsSearchDualWriteOrganizationIds = new Set(
     .filter(Boolean)
 );
 
-function buildEventRepository(store: string, clickhouse: ClickHouse): ClickhouseEventRepository {
+function buildEventRepository(
+  store: string,
+  clickhouse: ClickHouse,
+  logsSearchClickhouse?: ClickHouse
+): ClickhouseEventRepository {
   switch (store) {
     case "clickhouse": {
       return new ClickhouseEventRepository({
         clickhouse,
+        logsSearchClickhouse,
         batchSize: env.EVENTS_CLICKHOUSE_BATCH_SIZE,
         flushInterval: env.EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS,
         maximumTraceSummaryViewCount: clampToEmergencySpanCap(
@@ -822,6 +866,7 @@ function buildEventRepository(store: string, clickhouse: ClickHouse): Clickhouse
     case "clickhouse_v2": {
       return new ClickhouseEventRepository({
         clickhouse: clickhouse,
+        logsSearchClickhouse,
         batchSize: env.EVENTS_CLICKHOUSE_BATCH_SIZE,
         flushInterval: env.EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS,
         maximumTraceSummaryViewCount: clampToEmergencySpanCap(

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

@@ -100,6 +100,8 @@ export function logsSearchRolloutSelectedRowCount(
 
 export type ClickhouseEventRepositoryConfig = {
   clickhouse: ClickHouse;
+  /** Destination for logs-search dual writes; defaults to `clickhouse`. */
+  logsSearchClickhouse?: ClickHouse;
   batchSize?: number;
   flushInterval?: number;
   insertStrategy?: "insert" | "insert_async";
@@ -145,6 +147,7 @@ export type ClickhouseEventRepositoryConfig = {
  */
 export class ClickhouseEventRepository implements IEventRepository {
   private _clickhouse: ClickHouse;
+  private _logsSearchClickhouse: ClickHouse;
   private _config: ClickhouseEventRepositoryConfig;
   private readonly _flushScheduler: DynamicFlushScheduler<TaskEventV1Input | TaskEventV2Input>;
   private readonly _llmMetricsFlushScheduler: DynamicFlushScheduler<LlmMetricsV1Input>;
@@ -194,6 +197,7 @@ export class ClickhouseEventRepository implements IEventRepository {
 
   constructor(config: ClickhouseEventRepositoryConfig) {
     this._clickhouse = config.clickhouse;
+    this._logsSearchClickhouse = config.logsSearchClickhouse ?? config.clickhouse;
     this._config = config;
     this._tracer = config.tracer ?? trace.getTracer("clickhouseEventRepo", "0.0.1");
     this._version = config.version ?? "v1";
@@ -574,7 +578,7 @@ export class ClickhouseEventRepository implements IEventRepository {
     let lastError: { clickhouseErrorType?: string } | undefined;
 
     for (let attempt = 1; attempt <= 2; attempt++) {
-      const [error] = await this._clickhouse.taskEventsSearch.insert(rows, {
+      const [error] = await this._logsSearchClickhouse.taskEventsSearch.insert(rows, {
         params: {
           clickhouse_settings: {
             async_insert: 0,