feat(webapp,clickhouse): decouple log search indexing from event inserts

Project closed source windows with durable watermarks and leases. Keep v2 reads and backfill disabled by default until sufficient history exists.

Author: carderne

.server-changes/improve-global-log-search.md

@@ -3,4 +3,4 @@ area: webapp
 type: improvement
 ---
 
-Global log search now supports a bounded search index and clearer time-range expansion while keeping existing search history available during rollout.
+Global log search now supports faster bounded substring matching and clearer time-range expansion. Existing search remains the default while the new index builds sufficient history.

apps/webapp/app/entry.server.tsx

@@ -10,6 +10,7 @@ import { PassThrough } from "stream";
 import { initMollifierDrainerWorker } from "~/v3/mollifierDrainerWorker.server";
 import { initMollifierStaleSweepWorker } from "~/v3/mollifierStaleSweepWorker.server";
 import { initBillingLimitWorker } from "~/v3/billingLimitWorker.server";
+import { initLogsSearchProjectorWorker } from "~/v3/logsSearchProjectorWorker.server";
 import { initQueueMetricsConsumer, initQueueMetricsEmitter } from "~/v3/queueMetrics.server";
 import { bootstrap } from "./bootstrap";
 import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
@@ -277,6 +278,7 @@ export const handleError = wrapHandleErrorWithSentry((error, { request }) => {
 initMollifierDrainerWorker();
 initMollifierStaleSweepWorker();
 initBillingLimitWorker();
+initLogsSearchProjectorWorker();
 initQueueMetricsEmitter();
 initQueueMetricsConsumer();
 

apps/webapp/app/env.server.ts

@@ -2094,10 +2094,45 @@ const EnvironmentSchema = z
       .nonnegative()
       .optional(),
 
-    // v2 is populated forward-only. Keep reads on v1 until v2 has enough history or has been
-    // backfilled, then opt in explicitly per deployment.
+    // Keep reads on v1 until the scheduled v2 projector has enough history.
     LOGS_SEARCH_TABLE_VERSION: z.enum(["v1", "v2"]).default("v1"),
 
+    // Scheduled logs-search projection. Disabled by default. The writer URL must reach both the
+    // task_events_v2 source and task_events_search_v2 destination tables.
+    LOGS_SEARCH_PROJECTOR_ENABLED: BoolEnv.default(false),
+    LOGS_SEARCH_PROJECTOR_CLICKHOUSE_URL: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.EVENTS_CLICKHOUSE_URL ?? process.env.CLICKHOUSE_URL),
+    LOGS_SEARCH_PROJECTOR_SAFETY_DELAY_SECONDS: z.coerce
+      .number()
+      .int()
+      .min(60)
+      .max(3600)
+      .default(120),
+    LOGS_SEARCH_PROJECTOR_MAX_WINDOWS_PER_TICK: z.coerce.number().int().min(1).max(20).default(5),
+    LOGS_SEARCH_PROJECTOR_MAX_EXECUTION_TIME_SECONDS: z.coerce
+      .number()
+      .int()
+      .min(1)
+      .max(300)
+      .default(120),
+    LOGS_SEARCH_PROJECTOR_MAX_ROWS_TO_READ: z.coerce.number().int().positive().default(10_000_000),
+    LOGS_SEARCH_PROJECTOR_MAX_MEMORY_USAGE: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(1_500_000_000),
+    LOGS_SEARCH_PROJECTOR_MAX_THREADS: z.coerce.number().int().min(1).max(8).default(2),
+    LOGS_SEARCH_PROJECTOR_BACKFILL_ENABLED: BoolEnv.default(false),
+    LOGS_SEARCH_PROJECTOR_MAX_BACKFILL_RANGE_DAYS: z.coerce
+      .number()
+      .int()
+      .min(1)
+      .max(90)
+      .default(7),
+    LOGS_SEARCH_PROJECTOR_MAX_BACKFILL_AGE_DAYS: z.coerce.number().int().min(1).max(90).default(90),
+
     // Logs list pagination tuning.
     LOGS_LIST_DEFAULT_PAGE_SIZE: z.coerce.number().int().positive().default(50),
     LOGS_LIST_MAX_PAGE_SIZE: z.coerce.number().int().positive().default(100),

apps/webapp/app/routes/admin.api.v1.logs-search-projector.ts

@@ -0,0 +1,67 @@
+import { type ActionFunctionArgs, type LoaderFunctionArgs, json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import {
+  LogsSearchProjectorConflictError,
+  LogsSearchProjectorValidationError,
+} from "~/services/logsSearchProjector.server";
+import { getLogsSearchProjector } from "~/services/logsSearchProjectorInstance.server";
+import { logger } from "~/services/logger.server";
+import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
+
+const Body = z.discriminatedUnion("action", [
+  z.object({ action: z.literal("pause") }),
+  z.object({ action: z.literal("resume") }),
+  z.object({ action: z.literal("cancelBackfill") }),
+  z.object({
+    action: z.literal("startBackfill"),
+    from: z
+      .string()
+      .datetime()
+      .transform((value) => new Date(value)),
+    to: z
+      .string()
+      .datetime()
+      .transform((value) => new Date(value)),
+  }),
+]);
+
+export async function loader({ request }: LoaderFunctionArgs) {
+  await requireAdminApiRequest(request);
+  return json(await getLogsSearchProjector().status());
+}
+
+export async function action({ request }: ActionFunctionArgs) {
+  const user = await requireAdminApiRequest(request);
+
+  try {
+    const body = Body.parse(await request.json());
+    const logsSearchProjector = getLogsSearchProjector();
+    logger.info("Updating logs search projector", { userId: user.id, action: body.action });
+
+    switch (body.action) {
+      case "pause":
+        return json(await logsSearchProjector.pause());
+      case "resume":
+        return json(await logsSearchProjector.resume());
+      case "cancelBackfill":
+        return json(await logsSearchProjector.cancelBackfill());
+      case "startBackfill":
+        return json(await logsSearchProjector.startBackfill(body));
+    }
+  } catch (error) {
+    if (error instanceof LogsSearchProjectorConflictError) {
+      return json({ error: error.message }, { status: 409 });
+    }
+    if (
+      error instanceof LogsSearchProjectorValidationError ||
+      error instanceof z.ZodError ||
+      error instanceof SyntaxError
+    ) {
+      return json(
+        { error: error instanceof Error ? error.message : String(error) },
+        { status: 400 }
+      );
+    }
+    throw error;
+  }
+}

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

@@ -37,6 +37,33 @@ const defaultLogsClickhouseClient = singleton(
   initializeLogsClickhouseClient
 );
 
+const logsSearchProjectorClickhouseClient = singleton(
+  "logsSearchProjectorClickhouseClient",
+  initializeLogsSearchProjectorClickhouseClient
+);
+
+function initializeLogsSearchProjectorClickhouseClient() {
+  if (!env.LOGS_SEARCH_PROJECTOR_CLICKHOUSE_URL) {
+    throw new Error("LOGS_SEARCH_PROJECTOR_CLICKHOUSE_URL is not set");
+  }
+
+  const url = new URL(env.LOGS_SEARCH_PROJECTOR_CLICKHOUSE_URL);
+  url.searchParams.delete("secure");
+
+  return new ClickHouse({
+    url: url.toString(),
+    name: "logs-search-projector",
+    keepAlive: {
+      enabled: env.CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
+      idleSocketTtl: env.CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+    },
+    logLevel: env.CLICKHOUSE_LOG_LEVEL,
+    compression: { request: true },
+    maxOpenConnections: Math.min(env.CLICKHOUSE_MAX_OPEN_CONNECTIONS, 2),
+    requestTimeoutMs: (env.LOGS_SEARCH_PROJECTOR_MAX_EXECUTION_TIME_SECONDS + 30) * 1000,
+  });
+}
+
 function getLogsListClickhouseSettings() {
   return {
     max_memory_usage: env.CLICKHOUSE_LOGS_LIST_MAX_MEMORY_USAGE.toString(),
@@ -678,6 +705,10 @@ export function getAdminClickhouse(): ClickHouse {
   return defaultAdminClickhouseClient;
 }
 
+export function getLogsSearchProjectorClickhouseClient(): ClickHouse {
+  return logsSearchProjectorClickhouseClient;
+}
+
 /** Queue-metrics client for callers with no organization in scope (the ingestion consumer). */
 export function getQueueMetricsClickhouseClient(): ClickHouse {
   return defaultQueueMetricsClickhouseClient;