diff --git a/.server-changes/improve-global-log-search.md b/.server-changes/improve-global-log-search.md new file mode 100644 index 0000000000..f7d5dbec39 --- /dev/null +++ b/.server-changes/improve-global-log-search.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Global log search now supports faster bounded substring matching and clearer time-range expansion. diff --git a/apps/webapp/app/components/navigation/SideMenu.tsx b/apps/webapp/app/components/navigation/SideMenu.tsx index f69d0f16bc..2437c327ff 100644 --- a/apps/webapp/app/components/navigation/SideMenu.tsx +++ b/apps/webapp/app/components/navigation/SideMenu.tsx @@ -823,7 +823,7 @@ export function SideMenu({ }); } - if (isAdmin || featureFlags.hasQueryAccess) { + if (isAdmin || featureFlags.hasQueryAccess || featureFlags.hasLogsPageAccess) { staticSections.push({ id: "metrics", title: "Observability", @@ -841,55 +841,59 @@ export function SideMenu({ } satisfies SideMenuItemConfig, ] : []), - { - id: "errors", - name: "Errors", - icon: BugIcon, - activeIconColor: "text-errors", - to: v3ErrorsPath(organization, project, environment), - dataAction: "errors", - }, - { - id: "query", - name: "Query", - icon: CodeSquareIcon, - activeIconColor: "text-query", - to: queryPath(organization, project, environment), - dataAction: "query", - }, - { - id: "queues", - name: "Queues", - icon: QueuesIcon, - activeIconColor: "text-queues", - to: v3QueuesPath(organization, project, environment), - dataAction: "queues", - }, - { - id: "dashboards", - name: "Dashboards", - icon: ChartBarIcon, - activeIconColor: "text-metrics", - to: v3DashboardsLandingPath(organization, project, environment), - dataAction: "dashboards-landing", - action: ( - - ), - after: ( - - ), - }, + ...(isAdmin || featureFlags.hasQueryAccess + ? [ + { + id: "errors", + name: "Errors", + icon: BugIcon, + activeIconColor: "text-errors", + to: v3ErrorsPath(organization, project, environment), + dataAction: "errors", + }, + { + id: "query", + name: "Query", + icon: CodeSquareIcon, + activeIconColor: "text-query", + to: queryPath(organization, project, environment), + dataAction: "query", + }, + { + id: "queues", + name: "Queues", + icon: QueuesIcon, + activeIconColor: "text-queues", + to: v3QueuesPath(organization, project, environment), + dataAction: "queues", + }, + { + id: "dashboards", + name: "Dashboards", + icon: ChartBarIcon, + activeIconColor: "text-metrics", + to: v3DashboardsLandingPath(organization, project, environment), + dataAction: "dashboards-landing", + action: ( + + ), + after: ( + + ), + }, + ] + : []), ], }); } diff --git a/apps/webapp/app/components/primitives/SearchInput.tsx b/apps/webapp/app/components/primitives/SearchInput.tsx index 0c46a3d028..0ec8a4d644 100644 --- a/apps/webapp/app/components/primitives/SearchInput.tsx +++ b/apps/webapp/app/components/primitives/SearchInput.tsx @@ -14,6 +14,9 @@ export type SearchInputProps = { /** Additional URL params to reset when searching or clearing (e.g. pagination). Defaults to ["cursor", "direction"]. */ resetParams?: string[]; autoFocus?: boolean; + minLength?: number; + /** Normalize the submitted value before applying minLength validation. */ + normalizeForValidation?: (value: string) => string; /** * Controlled value. When provided alongside `onValueChange`, the input * skips URL params entirely and acts as a controlled component — useful @@ -34,6 +37,8 @@ export function SearchInput({ paramName = "search", resetParams = ["cursor", "direction"], autoFocus, + minLength, + normalizeForValidation, value: controlledValue, onValueChange, }: SearchInputProps) { @@ -70,6 +75,7 @@ export function SearchInput({ }, [isControlled, controlledValue, value, isFocused, paramName]); const updateText = (next: string) => { + inputRef.current?.setCustomValidity(""); setText(next); if (isControlled) { onValueChange?.(next); @@ -77,13 +83,25 @@ export function SearchInput({ }; const handleSubmit = () => { + const trimmedText = text.trim(); + const validationText = normalizeForValidation?.(trimmedText) ?? trimmedText; + if ( + minLength !== undefined && + trimmedText.length > 0 && + [...validationText].length < minLength + ) { + inputRef.current?.setCustomValidity(`Enter at least ${minLength} characters`); + inputRef.current?.reportValidity(); + return; + } + inputRef.current?.setCustomValidity(""); if (isControlled) { // Live updates already fired through onValueChange; submit is a no-op. return; } const resetValues = Object.fromEntries(resetParams.map((p) => [p, undefined])); - if (text.trim()) { - replace({ [paramName]: text.trim(), ...resetValues }); + if (trimmedText) { + replace({ [paramName]: trimmedText, ...resetValues }); } else { del([paramName, ...resetParams]); } @@ -116,7 +134,10 @@ export function SearchInput({ variant="secondary-small" placeholder={placeholder} value={text} - onChange={(e) => updateText(e.target.value)} + onChange={(e) => { + e.currentTarget.setCustomValidity(""); + updateText(e.target.value); + }} fullWidth autoFocus={autoFocus} className={cn("", isFocused && "placeholder:text-text-dimmed/70")} diff --git a/apps/webapp/app/entry.server.tsx b/apps/webapp/app/entry.server.tsx index c2cc31e6f2..520098c673 100644 --- a/apps/webapp/app/entry.server.tsx +++ b/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"; @@ -265,6 +266,7 @@ export const handleError = wrapHandleErrorWithSentry((error, { request }) => { initMollifierDrainerWorker(); initMollifierStaleSweepWorker(); initBillingLimitWorker(); +initLogsSearchProjectorWorker(); initQueueMetricsEmitter(); initQueueMetricsConsumer(); diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 3ce98dd521..c353125f3c 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -1925,20 +1925,48 @@ const EnvironmentSchema = z .nonnegative() .optional(), - // Logs list pagination tuning (page sizing + recent-first probe windows). + // 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), - // Days back from the page ceiling to probe before widening to the full requested window, - // comma-separated. Empty disables narrowing (a single full-window query). - LOGS_LIST_RECENT_FIRST_PROBE_DAYS: z - .string() - .default("1,7") - .transform((s) => - s - .split(",") - .map((v) => Number(v.trim())) - .filter((n) => Number.isFinite(n) && n > 0) - ), // Query feature flag QUERY_FEATURE_ENABLED: z.string().default("1"), diff --git a/apps/webapp/app/presenters/v3/LogsListPresenter.server.ts b/apps/webapp/app/presenters/v3/LogsListPresenter.server.ts index 9c19cb7571..cdc0056e68 100644 --- a/apps/webapp/app/presenters/v3/LogsListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/LogsListPresenter.server.ts @@ -1,8 +1,4 @@ -import { - type ClickHouse, - type LogsSearchListResult, - type WhereCondition, -} from "@internal/clickhouse"; +import { type ClickHouse, type WhereCondition } from "@internal/clickhouse"; import { type PrismaClientOrTransaction } from "@trigger.dev/database"; import { z } from "zod"; import { EVENT_STORE_TYPES, getConfiguredEventRepository } from "~/v3/eventRepository/index.server"; @@ -19,20 +15,18 @@ import { convertDateToClickhouseDateTime, } from "~/v3/eventRepository/clickhouseEventRepository.server"; import { ServiceValidationError } from "~/v3/services/baseService.server"; +import { + escapeClickHouseLike, + hasMinimumLogsSearchLength, + logsSearchExpansionPeriod, + LOGS_SEARCH_RETRY_OVERFETCH_FACTOR, + MIN_LOGS_SEARCH_LENGTH, + normalizeLogsSearchTerm, + prepareLogsSearchPage, +} from "~/utils/logSearch"; export type { LogLevel }; -type ErrorAttributes = { - error?: { - message?: unknown; - }; - [key: string]: unknown; -}; - -function escapeClickHouseString(val: string): string { - return val.replace(/\\/g, "\\\\").replace(/\//g, "\\/").replace(/%/g, "\\%").replace(/_/g, "\\_"); -} - export type LogsListOptions = { userId?: string; projectId: string; @@ -70,15 +64,13 @@ export const LogsListOptionsSchema = z.object({ pageSize: z.number().int().positive().max(1000).optional(), }); -const DAY_MS = 24 * 60 * 60 * 1000; - export type LogsList = Awaited>; export type LogEntry = LogsList["logs"][0]; export type LogsListAppliedFilters = LogsList["filters"]; // Bump when the cursor shape changes so stale cursors are ignored (reset to the first page) // rather than misparsed. -const LOG_CURSOR_VERSION = 2; +const LOG_CURSOR_VERSION = 4; // Cursor is a base64 encoded JSON of the pagination keys type LogCursor = { @@ -88,6 +80,7 @@ type LogCursor = { triggeredTimestamp: string; // DateTime64(9) string traceId: string; spanId: string; + projectionFingerprint?: string; }; const LogCursorSchema = z.object({ @@ -97,6 +90,7 @@ const LogCursorSchema = z.object({ triggeredTimestamp: z.string(), traceId: z.string(), spanId: z.string(), + projectionFingerprint: z.string().optional(), }); function encodeCursor(cursor: LogCursor): string { @@ -117,34 +111,6 @@ function decodeCursor(cursor: string): LogCursor | null { } } -// Ordered list of lower bounds to try, narrowest (most recent) first, ending at the user's -// requested floor (or undefined for an unbounded-below window). Because rows are returned -// newest-first, a narrow window that already fills a page returns the exact same top rows the -// full window would, so widening only happens when a page comes back short. -function buildProbeFloors( - ceil: Date, - hardFloor: Date | undefined, - stepDays: number[] -): (Date | undefined)[] { - const floors: (Date | undefined)[] = []; - - for (const days of stepDays) { - let candidate = new Date(ceil.getTime() - days * DAY_MS); - if (hardFloor && candidate <= hardFloor) { - candidate = hardFloor; - } - floors.push(candidate); - if (hardFloor && candidate.getTime() === hardFloor.getTime()) { - // Reached the requested floor; nothing wider left to probe. - return floors; - } - } - - // Final probe always covers the full requested window (or unbounded if no floor was given). - floors.push(hardFloor); - return floors; -} - // Convert display level to ClickHouse kinds and statuses function levelToKindsAndStatuses(level: LogLevel): { kinds?: string[]; statuses?: string[] } { switch (level) { @@ -262,6 +228,10 @@ export class LogsListPresenter extends BasePresenter { } const effectivePageSize = Math.min(pageSize, env.LOGS_LIST_MAX_PAGE_SIZE); + const usesV2Search = env.LOGS_SEARCH_TABLE_VERSION === "v2"; + const queryLimit = usesV2Search + ? (effectivePageSize + 1) * LOGS_SEARCH_RETRY_OVERFETCH_FACTOR + : effectivePageSize + 1; // Only honor a cursor scoped to this org+env; one copied from another scope would shift the // pagination anchor instead of resetting to the first page. @@ -273,19 +243,28 @@ export class LogsListPresenter extends BasePresenter { ? parsedCursor : null; - // Effective upper bound, always clamped to now so a probe never runs [floor, +inf). + // Effective upper bound, always clamped to now so a request never runs [floor, +inf). const now = new Date(); const clampedTo = effectiveTo !== undefined ? (effectiveTo > now ? now : effectiveTo) : now; + const rawSearchTerm = search?.trim() ?? ""; + const normalizedSearchTerm = usesV2Search + ? normalizeLogsSearchTerm(rawSearchTerm) + : rawSearchTerm.toLowerCase(); + if (rawSearchTerm !== "" && !hasMinimumLogsSearchLength(normalizedSearchTerm)) { + throw new ServiceValidationError( + `Log searches must be at least ${MIN_LOGS_SEARCH_LENGTH} characters.` + ); + } const searchTerm = - search && search.trim() !== "" - ? escapeClickHouseString(search.trim()).toLowerCase() - : undefined; + normalizedSearchTerm === "" ? undefined : escapeClickHouseLike(normalizedSearchTerm); - // Runs the full list query restricted to a single [floor, ceil] window. The recent-first - // probe loop below calls this with progressively wider floors. - const runProbe = (floor: Date | undefined) => { - const queryBuilder = this.clickhouse.taskEventsSearch.logsListQueryBuilder(); + // Run exactly one bounded query. Broadening a search window is an explicit user action; + // silently rescanning the same recent rows makes absence queries needlessly expensive. + const runQuery = () => { + const queryBuilder = this.clickhouse.taskEventsSearch.logsListQueryBuilder( + env.LOGS_SEARCH_TABLE_VERSION + ); // The materialized view excludes events without a trace_id; this guards the legacy tail. queryBuilder.where("trace_id != ''"); @@ -299,9 +278,9 @@ export class LogsListPresenter extends BasePresenter { }); } - if (floor) { + if (effectiveFrom) { queryBuilder.where("triggered_timestamp >= {triggeredAtStart: DateTime64(3)}", { - triggeredAtStart: convertDateToClickhouseDateTime(floor), + triggeredAtStart: convertDateToClickhouseDateTime(effectiveFrom), }); } @@ -315,12 +294,19 @@ export class LogsListPresenter extends BasePresenter { queryBuilder.where("run_id = {runId: String}", { runId }); } - // Case-insensitive search in message and attributes if (searchTerm !== undefined) { - queryBuilder.where( - "(lower(message) like {searchPattern: String} OR lower(attributes_text) like {searchPattern: String})", - { searchPattern: `%${searchTerm}%` } - ); + if (usesV2Search) { + // One predicate lets the text index answer substring searches without an OR across + // independently indexed columns. + queryBuilder.where("search_text LIKE {searchPattern: String}", { + searchPattern: `%${searchTerm}%`, + }); + } else { + queryBuilder.where( + "(lower(message) LIKE {searchPattern: String} OR lower(attributes_text) LIKE {searchPattern: String})", + { searchPattern: `%${searchTerm}%` } + ); + } } if (levels && levels.length > 0) { @@ -350,61 +336,58 @@ export class LogsListPresenter extends BasePresenter { queryBuilder.whereOr(conditions); } - // Keyset pagination over the full sort key. ORDER BY is DESC, so the next page is the rows - // that sort after the cursor (strictly less-than). (triggered_timestamp, trace_id) is not - // unique because spans of a trace share both, so span_id is the final tiebreaker; without - // it rows at a tie boundary could be skipped or duplicated across pages. + // Keyset pagination over the sort key. ORDER BY is DESC, so the next page is the rows + // that sort after the cursor (strictly less-than). V2 adds the projection identity as the + // final tiebreaker so retry copies and distinct rows at a span boundary paginate safely. if (decodedCursor) { + const cursorParams = { + cursorTriggeredTimestamp: decodedCursor.triggeredTimestamp, + cursorTraceId: decodedCursor.traceId, + cursorSpanId: decodedCursor.spanId, + ...(usesV2Search && decodedCursor.projectionFingerprint + ? { cursorProjectionFingerprint: decodedCursor.projectionFingerprint } + : {}), + }; queryBuilder.where( - `(triggered_timestamp < {cursorTriggeredTimestamp: String} - OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id < {cursorTraceId: String}) - OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id < {cursorSpanId: String}))`, - { - cursorTriggeredTimestamp: decodedCursor.triggeredTimestamp, - cursorTraceId: decodedCursor.traceId, - cursorSpanId: decodedCursor.spanId, - } + usesV2Search && decodedCursor.projectionFingerprint + ? `(triggered_timestamp < {cursorTriggeredTimestamp: String} + OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id < {cursorTraceId: String}) + OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id < {cursorSpanId: String}) + OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id = {cursorSpanId: String} AND projection_fingerprint < {cursorProjectionFingerprint: UInt128}))` + : `(triggered_timestamp < {cursorTriggeredTimestamp: String} + OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id < {cursorTraceId: String}) + OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id < {cursorSpanId: String}))`, + cursorParams ); } - queryBuilder.orderBy("triggered_timestamp DESC, trace_id DESC, span_id DESC"); - // Limit + 1 to check if there are more results - queryBuilder.limit(effectivePageSize + 1); + queryBuilder.orderBy( + usesV2Search + ? "triggered_timestamp DESC, trace_id DESC, span_id DESC, projection_fingerprint DESC" + : "triggered_timestamp DESC, trace_id DESC, span_id DESC" + ); + queryBuilder.limit(queryLimit); return queryBuilder.execute(); }; - // Page ceiling: the cursor (deeper pages) or the requested upper bound. Widen the lower - // bound only when a recent window doesn't fill the page. - const ceil = decodedCursor - ? convertClickhouseDateTime64ToJsDate(decodedCursor.triggeredTimestamp) - : (clampedTo ?? new Date()); - - const probeFloors = buildProbeFloors( - ceil, - effectiveFrom ?? undefined, - env.LOGS_LIST_RECENT_FIRST_PROBE_DAYS - ); - - let records: LogsSearchListResult[] = []; - for (const floor of probeFloors) { - const [queryError, probeRecords] = await runProbe(floor); - - if (queryError) { - throw queryError; - } - - records = probeRecords ?? []; - - if (records.length > effectivePageSize) { - // Page is full from this window; older rows can't be in the top page, stop widening. - break; - } + const [queryError, queryResult] = await runQuery(); + if (queryError) { + throw queryError; } - const results = records; - const hasMore = results.length > effectivePageSize; - const logs = results.slice(0, effectivePageSize); + // ClickHouse's break overflow modes can return a short prefix without a reliable completion + // marker. Keep the default throw behavior so the product never presents truncated results as + // complete. + const results = queryResult ?? []; + const page = usesV2Search + ? prepareLogsSearchPage(results, effectivePageSize, queryLimit) + : { + rows: results.slice(0, effectivePageSize), + hasMore: results.length > effectivePageSize, + }; + const hasMore = page.hasMore; + const logs = page.rows; // Build next cursor from the last item let nextCursor: string | undefined; @@ -417,6 +400,7 @@ export class LogsListPresenter extends BasePresenter { triggeredTimestamp: lastLog.triggered_timestamp, traceId: lastLog.trace_id, spanId: lastLog.span_id, + projectionFingerprint: lastLog.projection_fingerprint_string, }); } @@ -425,17 +409,10 @@ export class LogsListPresenter extends BasePresenter { const transformedLogs = logs.map((log) => { let displayMessage = log.message; - // For error logs with status ERROR, try to extract error message from attributes - if (log.status === "ERROR" && log.attributes_text) { - try { - const attributes = JSON.parse(log.attributes_text) as ErrorAttributes; - - if (attributes?.error?.message && typeof attributes.error.message === "string") { - displayMessage = attributes.error.message; - } - } catch { - // If attributes parsing fails, use the regular message - } + // The search table extracts this leaf in the materialized view, so list queries never + // need to read or parse the complete attributes blob. + if (log.status === "ERROR" && log.error_message) { + displayMessage = log.error_message; } return { @@ -457,6 +434,11 @@ export class LogsListPresenter extends BasePresenter { }; }); + const searchExpansion = + searchTerm !== undefined && time.isDefault && transformedLogs.length === 0 + ? logsSearchExpansionPeriod(effectiveFrom, clampedTo, retentionLimitDays) + : undefined; + return { logs: transformedLogs, pagination: { @@ -479,6 +461,7 @@ export class LogsListPresenter extends BasePresenter { hasFilters, hasAnyLogs: transformedLogs.length > 0, searchTerm: search, + searchExpansion: searchExpansion ? { nextPeriod: searchExpansion } : undefined, retention: retentionLimitDays !== undefined ? { diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx index ba7764cf01..431dfb7b74 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs/route.tsx @@ -1,5 +1,5 @@ import { type LoaderFunctionArgs, redirect } from "@remix-run/server-runtime"; -import { useFetcher, useNavigation, useLocation, Form } from "@remix-run/react"; +import { useFetcher, useNavigation, useLocation, useNavigate, Form } from "@remix-run/react"; import { XMarkIcon } from "@heroicons/react/20/solid"; import { ServiceValidationError } from "~/v3/services/baseService.server"; import { @@ -16,7 +16,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; import type { LogEntry } from "~/presenters/v3/LogsListPresenter.server"; import { LogsListPresenter } from "~/presenters/v3/LogsListPresenter.server"; import type { LogLevel } from "~/utils/logUtils"; -import { $replica, prisma } from "~/db.server"; +import { $replica } from "~/db.server"; import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server"; import { NavBar, PageTitle } from "~/components/primitives/PageHeader"; import { PageBody, PageContainer } from "~/components/layout/AppLayout"; @@ -41,14 +41,20 @@ import { useFrozenValue, } from "~/components/primitives/Resizable"; import { Button } from "~/components/primitives/Buttons"; -import { FEATURE_FLAG, validateFeatureFlagValue } from "~/v3/featureFlags"; import { sectionAgentPageContext } from "~/components/dashboard-agent/suggested-prompts"; import type { Handle } from "~/utils/handle"; import { pageMeta } from "~/utils/pageTitle"; +import { hasLogsPageAccess } from "~/services/logsAccess.server"; +import { MIN_LOGS_SEARCH_LENGTH, normalizeLogsSearchTerm } from "~/utils/logSearch"; // Valid log levels for filtering const validLevels: LogLevel[] = ["TRACE", "DEBUG", "INFO", "WARN", "ERROR"]; +function formatSearchPeriod(period: string): string { + const days = Number(period.replace("d", "")); + return days === 1 ? "day" : `${days} days`; +} + function parseLevelsFromUrl(url: URL): LogLevel[] | undefined { const levelParams = url.searchParams.getAll("levels").filter((v) => v.length > 0); if (levelParams.length === 0) return undefined; @@ -61,41 +67,6 @@ export const handle: Handle = { export const meta = pageMeta("Logs"); -// TODO: Move this to a more appropriate shared location -async function hasLogsPageAccess( - userId: string, - isAdmin: boolean, - isImpersonating: boolean, - organizationSlug: string -): Promise { - if (isAdmin || isImpersonating) { - return true; - } - - // Check organization feature flags - const organization = await prisma.organization.findFirst({ - where: { - slug: organizationSlug, - members: { some: { userId } }, - }, - select: { - featureFlags: true, - }, - }); - - if (!organization?.featureFlags) { - return false; - } - - const flags = organization.featureFlags as Record; - const hasLogsPageAccessResult = validateFeatureFlagValue( - FEATURE_FLAG.hasLogsPageAccess, - flags.hasLogsPageAccess - ); - - return hasLogsPageAccessResult.success && hasLogsPageAccessResult.data === true; -} - export const loader = async ({ request, params }: LoaderFunctionArgs) => { const user = await requireUser(request); const userId = user.id; @@ -156,7 +127,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { period, from, to, - defaultPeriod: "1h", + defaultPeriod: "1d", retentionLimitDays, }) .catch((error) => { @@ -168,7 +139,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { return typeddefer({ data: listPromise, - defaultPeriod: "1h", + defaultPeriod: "1d", retentionLimitDays, }); }; @@ -269,7 +240,10 @@ function FiltersBar({
{list ? ( <> - + @@ -292,7 +266,10 @@ function FiltersBar({ - + {hasFilters && (
- - +
+ {list.searchExpansion && ( + + Search last {formatSearchPeriod(list.searchExpansion.nextPeriod)} + + } + > + No matches in the last day. + + )} + + + + + + {}} + collapsedSize="0px" + collapseAnimation={RESIZABLE_PANEL_ANIMATION} + > +
+ {displayLogId && ( + + +
+ } + > + + + )} +
+ + + ); } diff --git a/apps/webapp/app/routes/admin.api.v1.logs-search-projector.ts b/apps/webapp/app/routes/admin.api.v1.logs-search-projector.ts new file mode 100644 index 0000000000..ca1a3c4055 --- /dev/null +++ b/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; + } +} diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.can-view-logs-page/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.can-view-logs-page/route.tsx index 501b4a8ad3..5412cb5a90 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.can-view-logs-page/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.can-view-logs-page/route.tsx @@ -1,43 +1,9 @@ import { type LoaderFunctionArgs } from "@remix-run/server-runtime"; import { typedjson } from "remix-typedjson"; import { requireUser } from "~/services/session.server"; -import { prisma } from "~/db.server"; -import { FEATURE_FLAG, validateFeatureFlagValue } from "~/v3/featureFlags"; +import { hasLogsPageAccess } from "~/services/logsAccess.server"; import { OrganizationParamsSchema } from "~/utils/pathBuilder"; -async function hasLogsPageAccess( - userId: string, - isAdmin: boolean, - isImpersonating: boolean, - organizationSlug: string -): Promise { - if (isAdmin || isImpersonating) { - return true; - } - - const organization = await prisma.organization.findFirst({ - where: { - slug: organizationSlug, - members: { some: { userId } }, - }, - select: { - featureFlags: true, - }, - }); - - if (!organization?.featureFlags) { - return false; - } - - const flags = organization.featureFlags as Record; - const hasLogsPageAccessResult = validateFeatureFlagValue( - FEATURE_FLAG.hasLogsPageAccess, - flags.hasLogsPageAccess - ); - - return hasLogsPageAccessResult.success && hasLogsPageAccessResult.data === true; -} - export const loader = async ({ request, params }: LoaderFunctionArgs) => { const user = await requireUser(request); const { organizationSlug } = OrganizationParamsSchema.parse(params); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx index 418cee805c..fcc607b673 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.$logId.tsx @@ -2,7 +2,7 @@ import { type LoaderFunctionArgs } from "@remix-run/server-runtime"; import { typedjson } from "remix-typedjson"; import { z } from "zod"; import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server"; -import { requireUserId } from "~/services/session.server"; +import { requireUser } from "~/services/session.server"; import { LogDetailPresenter } from "~/presenters/v3/LogDetailPresenter.server"; import { findProjectBySlug } from "~/models/project.server"; import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; @@ -10,6 +10,7 @@ import { $replica } from "~/db.server"; import { runStore } from "~/v3/runStore.server"; import { ServiceValidationError } from "~/v3/services/baseService.server"; import type { TaskRunStatus } from "@trigger.dev/database"; +import { hasLogsPageAccess } from "~/services/logsAccess.server"; const LogIdParamsSchema = z.object({ organizationSlug: z.string(), @@ -19,9 +20,14 @@ const LogIdParamsSchema = z.object({ }); export const loader = async ({ request, params }: LoaderFunctionArgs) => { - const userId = await requireUserId(request); + const user = await requireUser(request); + const userId = user.id; const { organizationSlug, projectParam, envParam, logId } = LogIdParamsSchema.parse(params); + if (!(await hasLogsPageAccess(user.id, user.admin, user.isImpersonating, organizationSlug))) { + throw new Response("Logs are not available", { status: 403 }); + } + // Validate access to project and environment const project = await findProjectBySlug(organizationSlug, projectParam, userId); if (!project) { diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts index a3425bd2da..fe593e8af9 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.logs.ts @@ -12,6 +12,7 @@ import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstan import { getCurrentPlan } from "~/services/platform.v3.server"; import { requireUser } from "~/services/session.server"; import { EnvironmentParamSchema } from "~/utils/pathBuilder"; +import { hasLogsPageAccess } from "~/services/logsAccess.server"; // Valid log levels for filtering const validLevels: LogLevel[] = ["TRACE", "DEBUG", "INFO", "WARN", "ERROR"]; @@ -27,6 +28,9 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const userId = user.id; const { projectParam, organizationSlug, envParam } = EnvironmentParamSchema.parse(params); + if (!(await hasLogsPageAccess(user.id, user.admin, user.isImpersonating, organizationSlug))) { + throw new Response("Logs are not available", { status: 403 }); + } const project = await findProjectBySlug(organizationSlug, projectParam, userId); if (!project) { @@ -69,7 +73,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { from, to, levels, - defaultPeriod: "1h", + defaultPeriod: "1d", retentionLimitDays, }) as any; // Validated by LogsListOptionsSchema at runtime diff --git a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts index 7624810efe..99b1f3740c 100644 --- a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts +++ b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts @@ -37,6 +37,28 @@ const defaultLogsClickhouseClient = singleton( initializeLogsClickhouseClient ); +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(), @@ -653,6 +675,13 @@ export function getDefaultLogsClickhouseClient(): ClickHouse { return defaultLogsClickhouseClient; } +export function getLogsSearchProjectorClickhouseClient(): ClickHouse { + return singleton( + "logsSearchProjectorClickhouseClient", + initializeLogsSearchProjectorClickhouseClient + ); +} + /** Queue-metrics client for callers with no organization in scope (the ingestion consumer). */ export function getQueueMetricsClickhouseClient(): ClickHouse { return defaultQueueMetricsClickhouseClient; diff --git a/apps/webapp/app/services/logsAccess.server.ts b/apps/webapp/app/services/logsAccess.server.ts new file mode 100644 index 0000000000..35cd2ee1d6 --- /dev/null +++ b/apps/webapp/app/services/logsAccess.server.ts @@ -0,0 +1,29 @@ +import { prisma } from "~/db.server"; +import { FEATURE_FLAG, validateFeatureFlagValue } from "~/v3/featureFlags"; + +export async function hasLogsPageAccess( + userId: string, + isAdmin: boolean, + isImpersonating: boolean, + organizationSlug: string +): Promise { + if (isAdmin || isImpersonating) { + return true; + } + + const organization = await prisma.organization.findFirst({ + where: { + slug: organizationSlug, + members: { some: { userId } }, + }, + select: { featureFlags: true }, + }); + + if (!organization?.featureFlags) { + return false; + } + + const flags = organization.featureFlags as Record; + const result = validateFeatureFlagValue(FEATURE_FLAG.hasLogsPageAccess, flags.hasLogsPageAccess); + return result.success && result.data === true; +} diff --git a/apps/webapp/app/services/logsSearchProjector.server.ts b/apps/webapp/app/services/logsSearchProjector.server.ts new file mode 100644 index 0000000000..0de7ac0643 --- /dev/null +++ b/apps/webapp/app/services/logsSearchProjector.server.ts @@ -0,0 +1,383 @@ +import { randomUUID } from "node:crypto"; +import { logger as defaultLogger } from "~/services/logger.server"; +import { + logsSearchProjectorTelemetry, + type LogsSearchProjectionMode, +} from "~/services/logsSearchProjectorTelemetry.server"; + +export const LOGS_SEARCH_PROJECTOR_STATE_ID = "task_events_search_v2"; +export const LOGS_SEARCH_PROJECTOR_WINDOW_MS = 60_000; + +export type LogsSearchProjectorState = { + id: string; + liveWatermark: Date; + historicalWatermark: Date; + backfillTarget: Date | null; + paused: boolean; + leaseToken: string | null; + leaseExpiresAt: Date | null; +}; + +export type LogsSearchProjectorWindow = { + mode: LogsSearchProjectionMode; + start: Date; + end: Date; +}; + +export type LogsSearchProjectorProjectionResult = { + queryId: string; + readRows: number; + writtenRows: number; +}; + +export type LogsSearchProjectorStatus = { + initialized: boolean; + paused: boolean; + liveWatermark: Date | null; + safeCutoff: Date | null; + liveLagMs: number | null; + liveWindowsDue: number | null; + historicalWatermark: Date | null; + backfillTarget: Date | null; + backfillWindowsRemaining: number; + leaseExpiresAt: Date | null; +}; + +export type LogsSearchProjectorConfig = { + safetyDelayMs: number; + maxWindowsPerTick: number; + leaseDurationMs: number; + backfillEnabled: boolean; + maxBackfillRangeMs: number; + maxBackfillAgeMs: number; +}; + +export type LogsSearchProjectorStateStore = { + initialize(boundary: Date): Promise; + find(): Promise; + get(): Promise; + acquireLease(token: string, leaseDurationMs: number): Promise; + renewLease(token: string, leaseDurationMs: number): Promise; + releaseLease(token: string): Promise; + advanceLive(token: string, expected: Date, next: Date): Promise; + advanceHistorical( + token: string, + expected: Date, + next: Date, + expectedTarget: Date + ): Promise; + pause(): Promise; + resume(): Promise; + setBackfillTarget(expectedHistorical: Date, target: Date): Promise; + cancelBackfill(): Promise; +}; + +export class LogsSearchProjectorConflictError extends Error {} +export class LogsSearchProjectorValidationError extends Error {} + +export class LogsSearchProjector { + constructor( + private readonly config: LogsSearchProjectorConfig, + private readonly stateStore: LogsSearchProjectorStateStore, + private readonly projectWindow: ( + window: LogsSearchProjectorWindow + ) => Promise, + private readonly clock: () => Date | Promise = () => new Date(), + private readonly logger: Pick< + typeof defaultLogger, + "debug" | "info" | "warn" | "error" + > = defaultLogger + ) {} + + async processTick(): Promise<{ processed: number; leaseAcquired: boolean }> { + const now = await this.clock(); + const initialBoundary = calculateClosedWindowBoundary(now, this.config.safetyDelayMs); + const initialState = await this.stateStore.initialize(initialBoundary); + this.updateTelemetryState(initialState, initialBoundary); + if (initialState.paused) return { processed: 0, leaseAcquired: false }; + + const leaseToken = randomUUID(); + const acquired = await this.stateStore.acquireLease(leaseToken, this.config.leaseDurationMs); + if (!acquired) { + logsSearchProjectorTelemetry.recordLeaseContention(); + return { processed: 0, leaseAcquired: false }; + } + + let processed = 0; + try { + for (let index = 0; index < this.config.maxWindowsPerTick; index++) { + const state = await this.stateStore.get(); + if (state.paused || state.leaseToken !== leaseToken) break; + + const safeCutoff = calculateClosedWindowBoundary( + await this.clock(), + this.config.safetyDelayMs + ); + const window = selectNextProjectionWindow(state, safeCutoff); + if (!window) break; + + const renewed = await this.stateStore.renewLease(leaseToken, this.config.leaseDurationMs); + if (!renewed) break; + + const startedAt = Date.now(); + let result: LogsSearchProjectorProjectionResult; + try { + result = await this.projectWindow(window); + } catch (error) { + logsSearchProjectorTelemetry.recordWindow(window.mode, "error", Date.now() - startedAt); + this.logger.error("Logs search projection window failed", { + error, + mode: window.mode, + windowStart: window.start, + windowEnd: window.end, + }); + throw error; + } + + const advanced = + window.mode === "live" + ? await this.stateStore.advanceLive(leaseToken, window.start, window.end) + : await this.stateStore.advanceHistorical( + leaseToken, + window.end, + window.start, + state.backfillTarget! + ); + + if (!advanced) { + logsSearchProjectorTelemetry.recordCasLoss(window.mode); + logsSearchProjectorTelemetry.recordWindow( + window.mode, + "cas_lost", + Date.now() - startedAt, + result.readRows, + result.writtenRows + ); + this.logger.warn("Logs search projection watermark compare-and-swap lost", { + mode: window.mode, + windowStart: window.start, + windowEnd: window.end, + queryId: result.queryId, + }); + break; + } + + processed++; + logsSearchProjectorTelemetry.recordWindow( + window.mode, + "success", + Date.now() - startedAt, + result.readRows, + result.writtenRows + ); + this.logger.info("Projected logs search window", { + mode: window.mode, + windowStart: window.start, + windowEnd: window.end, + queryId: result.queryId, + readRows: result.readRows, + writtenRows: result.writtenRows, + }); + } + } finally { + try { + await this.stateStore.releaseLease(leaseToken); + } catch (error) { + this.logger.warn("Failed to release logs search projector lease", { error }); + } + try { + const state = await this.stateStore.get(); + const safeCutoff = calculateClosedWindowBoundary( + await this.clock(), + this.config.safetyDelayMs + ); + this.updateTelemetryState(state, safeCutoff); + } catch (error) { + this.logger.warn("Failed to update logs search projector telemetry state", { error }); + } + } + + return { processed, leaseAcquired: true }; + } + + async status(): Promise { + return this.readStatus(true); + } + + async pause(): Promise { + if (!(await this.stateStore.find())) { + throw new LogsSearchProjectorConflictError("Logs search projector is not initialized"); + } + await this.stateStore.pause(); + return this.readStatus(false); + } + + async resume(): Promise { + if (!(await this.stateStore.find())) { + throw new LogsSearchProjectorConflictError("Logs search projector is not initialized"); + } + await this.stateStore.resume(); + return this.readStatus(false); + } + + async startBackfill(input: { from: Date; to: Date }): Promise { + if (!this.config.backfillEnabled) { + throw new LogsSearchProjectorConflictError("Logs search backfill is disabled"); + } + await this.ensureInitialized(); + assertMinuteBoundary(input.from, "from"); + assertMinuteBoundary(input.to, "to"); + if (input.from >= input.to) { + throw new LogsSearchProjectorValidationError("Backfill from must be before to"); + } + if (input.to.getTime() - input.from.getTime() > this.config.maxBackfillRangeMs) { + throw new LogsSearchProjectorValidationError("Backfill range exceeds the configured limit"); + } + if (input.from.getTime() < (await this.clock()).getTime() - this.config.maxBackfillAgeMs) { + throw new LogsSearchProjectorValidationError( + "Backfill start is older than the configured limit" + ); + } + + const state = await this.stateStore.get(); + if (state.backfillTarget) { + throw new LogsSearchProjectorConflictError("A logs search backfill is already active"); + } + if (input.to.getTime() !== state.historicalWatermark.getTime()) { + throw new LogsSearchProjectorConflictError( + "Backfill to must equal the current historical watermark" + ); + } + + const updated = await this.stateStore.setBackfillTarget(state.historicalWatermark, input.from); + if (!updated) { + throw new LogsSearchProjectorConflictError("Logs search projector state changed"); + } + this.logger.info("Started logs search backfill", input); + return this.status(); + } + + async cancelBackfill(): Promise { + if (!(await this.stateStore.find())) return uninitializedProjectorStatus(); + await this.stateStore.cancelBackfill(); + this.logger.info("Cancelled logs search backfill"); + return this.readStatus(false); + } + + private async readStatus(includeClickHouseClock: boolean) { + const state = await this.stateStore.find(); + if (!state) return uninitializedProjectorStatus(); + + let safeCutoff: Date | null = null; + if (includeClickHouseClock) { + try { + safeCutoff = calculateClosedWindowBoundary(await this.clock(), this.config.safetyDelayMs); + } catch (error) { + this.logger.warn("Failed to read ClickHouse clock for logs search projector status", { + error, + }); + } + } + return projectorStatus(state, safeCutoff, true); + } + + private async ensureInitialized() { + await this.stateStore.initialize( + calculateClosedWindowBoundary(await this.clock(), this.config.safetyDelayMs) + ); + } + + private updateTelemetryState(state: LogsSearchProjectorState, safeCutoff: Date) { + const status = projectorStatus(state, safeCutoff, true); + logsSearchProjectorTelemetry.updateState({ + liveLagMs: status.liveLagMs ?? 0, + backfillRemaining: status.backfillWindowsRemaining, + paused: status.paused, + }); + } +} + +export function calculateClosedWindowBoundary(now: Date, safetyDelayMs: number): Date { + return new Date( + Math.floor((now.getTime() - safetyDelayMs) / LOGS_SEARCH_PROJECTOR_WINDOW_MS) * + LOGS_SEARCH_PROJECTOR_WINDOW_MS + ); +} + +export function selectNextProjectionWindow( + state: LogsSearchProjectorState, + safeCutoff: Date +): LogsSearchProjectorWindow | null { + if (state.paused) return null; + if (state.liveWatermark < safeCutoff) { + return { + mode: "live", + start: state.liveWatermark, + end: new Date(state.liveWatermark.getTime() + LOGS_SEARCH_PROJECTOR_WINDOW_MS), + }; + } + if (state.backfillTarget && state.historicalWatermark > state.backfillTarget) { + return { + mode: "backfill", + start: new Date(state.historicalWatermark.getTime() - LOGS_SEARCH_PROJECTOR_WINDOW_MS), + end: state.historicalWatermark, + }; + } + return null; +} + +function uninitializedProjectorStatus(): LogsSearchProjectorStatus { + return { + initialized: false, + paused: false, + liveWatermark: null, + safeCutoff: null, + liveLagMs: null, + liveWindowsDue: null, + historicalWatermark: null, + backfillTarget: null, + backfillWindowsRemaining: 0, + leaseExpiresAt: null, + }; +} + +function projectorStatus( + state: LogsSearchProjectorState, + safeCutoff: Date | null, + initialized: boolean +): LogsSearchProjectorStatus { + const liveLagMs = safeCutoff + ? Math.max(0, safeCutoff.getTime() - state.liveWatermark.getTime()) + : null; + const backfillRemaining = state.backfillTarget + ? Math.max( + 0, + Math.floor( + (state.historicalWatermark.getTime() - state.backfillTarget.getTime()) / + LOGS_SEARCH_PROJECTOR_WINDOW_MS + ) + ) + : 0; + return { + initialized, + paused: state.paused, + liveWatermark: state.liveWatermark, + safeCutoff, + liveLagMs, + liveWindowsDue: + liveLagMs === null ? null : Math.floor(liveLagMs / LOGS_SEARCH_PROJECTOR_WINDOW_MS), + historicalWatermark: state.historicalWatermark, + backfillTarget: state.backfillTarget, + backfillWindowsRemaining: backfillRemaining, + leaseExpiresAt: state.leaseExpiresAt, + }; +} + +function assertMinuteBoundary(value: Date, field: string) { + if ( + !Number.isFinite(value.getTime()) || + value.getTime() % LOGS_SEARCH_PROJECTOR_WINDOW_MS !== 0 + ) { + throw new LogsSearchProjectorValidationError(`${field} must be aligned to a UTC minute`); + } +} diff --git a/apps/webapp/app/services/logsSearchProjectorInstance.server.ts b/apps/webapp/app/services/logsSearchProjectorInstance.server.ts new file mode 100644 index 0000000000..1c45d71546 --- /dev/null +++ b/apps/webapp/app/services/logsSearchProjectorInstance.server.ts @@ -0,0 +1,54 @@ +import { prisma } from "~/db.server"; +import { env } from "~/env.server"; +import { getLogsSearchProjectorClickhouseClient } from "~/services/clickhouse/clickhouseFactory.server"; +import { LogsSearchProjector } from "~/services/logsSearchProjector.server"; +import { PrismaLogsSearchProjectorStateStore } from "~/services/logsSearchProjectorStateStore.server"; +import { singleton } from "~/utils/singleton"; +import { z } from "zod"; + +function initializeLogsSearchProjector() { + const clickhouse = getLogsSearchProjectorClickhouseClient(); + const serverClockQuery = clickhouse.reader.query({ + name: "get-logs-search-projector-clock", + query: "SELECT toUnixTimestamp64Milli(now64(3)) AS now_ms", + schema: z.object({ now_ms: z.number().or(z.string()) }), + }); + const limits = { + maxExecutionTimeSeconds: env.LOGS_SEARCH_PROJECTOR_MAX_EXECUTION_TIME_SECONDS, + maxRowsToRead: env.LOGS_SEARCH_PROJECTOR_MAX_ROWS_TO_READ, + maxMemoryUsage: env.LOGS_SEARCH_PROJECTOR_MAX_MEMORY_USAGE, + maxThreads: env.LOGS_SEARCH_PROJECTOR_MAX_THREADS, + }; + + return new LogsSearchProjector( + { + safetyDelayMs: env.LOGS_SEARCH_PROJECTOR_SAFETY_DELAY_SECONDS * 1000, + maxWindowsPerTick: env.LOGS_SEARCH_PROJECTOR_MAX_WINDOWS_PER_TICK, + leaseDurationMs: env.LOGS_SEARCH_PROJECTOR_MAX_EXECUTION_TIME_SECONDS * 1000 + 60_000, + backfillEnabled: env.LOGS_SEARCH_PROJECTOR_BACKFILL_ENABLED, + maxBackfillRangeMs: env.LOGS_SEARCH_PROJECTOR_MAX_BACKFILL_RANGE_DAYS * 24 * 60 * 60 * 1000, + maxBackfillAgeMs: env.LOGS_SEARCH_PROJECTOR_MAX_BACKFILL_AGE_DAYS * 24 * 60 * 60 * 1000, + }, + new PrismaLogsSearchProjectorStateStore(prisma), + async (window) => { + const [error, result] = await clickhouse.taskEventsSearch.projectV2Window(window, limits); + if (error) throw error; + return { + queryId: result.query_id, + readRows: Number(result.summary?.read_rows ?? 0), + writtenRows: Number(result.summary?.written_rows ?? 0), + }; + }, + async () => { + const [error, rows] = await serverClockQuery({}); + if (error) throw error; + const nowMs = Number(rows[0]?.now_ms); + if (!Number.isFinite(nowMs)) throw new Error("ClickHouse returned an invalid server clock"); + return new Date(nowMs); + } + ); +} + +export function getLogsSearchProjector() { + return singleton("logsSearchProjector", initializeLogsSearchProjector); +} diff --git a/apps/webapp/app/services/logsSearchProjectorStateStore.server.ts b/apps/webapp/app/services/logsSearchProjectorStateStore.server.ts new file mode 100644 index 0000000000..6b31fb8cb9 --- /dev/null +++ b/apps/webapp/app/services/logsSearchProjectorStateStore.server.ts @@ -0,0 +1,142 @@ +import type { PrismaClient } from "@trigger.dev/database"; +import { + LOGS_SEARCH_PROJECTOR_STATE_ID, + type LogsSearchProjectorState, + type LogsSearchProjectorStateStore, +} from "~/services/logsSearchProjector.server"; + +type LogsSearchProjectorDatabase = Pick; + +export class PrismaLogsSearchProjectorStateStore implements LogsSearchProjectorStateStore { + constructor(private readonly database: LogsSearchProjectorDatabase) {} + + async initialize(boundary: Date): Promise { + return this.database.logsSearchProjectorState.upsert({ + where: { id: LOGS_SEARCH_PROJECTOR_STATE_ID }, + create: { + id: LOGS_SEARCH_PROJECTOR_STATE_ID, + liveWatermark: boundary, + historicalWatermark: boundary, + }, + update: {}, + }); + } + + async find(): Promise { + return this.database.logsSearchProjectorState.findFirst({ + where: { id: LOGS_SEARCH_PROJECTOR_STATE_ID }, + }); + } + + async get(): Promise { + const state = await this.find(); + if (!state) throw new Error("Logs search projector state is not initialized"); + return state; + } + + async acquireLease(token: string, leaseDurationMs: number): Promise { + const count = await this.database.$executeRaw` + UPDATE "LogsSearchProjectorState" + SET + "leaseToken" = ${token}, + "leaseExpiresAt" = CURRENT_TIMESTAMP + (${leaseDurationMs} * INTERVAL '1 millisecond'), + "updatedAt" = CURRENT_TIMESTAMP + WHERE "id" = ${LOGS_SEARCH_PROJECTOR_STATE_ID} + AND "paused" = false + AND ( + "leaseToken" IS NULL + OR "leaseExpiresAt" IS NULL + OR "leaseExpiresAt" <= CURRENT_TIMESTAMP + ) + `; + return count === 1; + } + + async renewLease(token: string, leaseDurationMs: number): Promise { + const count = await this.database.$executeRaw` + UPDATE "LogsSearchProjectorState" + SET + "leaseExpiresAt" = CURRENT_TIMESTAMP + (${leaseDurationMs} * INTERVAL '1 millisecond'), + "updatedAt" = CURRENT_TIMESTAMP + WHERE "id" = ${LOGS_SEARCH_PROJECTOR_STATE_ID} + AND "paused" = false + AND "leaseToken" = ${token} + `; + return count === 1; + } + + async releaseLease(token: string): Promise { + await this.database.logsSearchProjectorState.updateMany({ + where: { id: LOGS_SEARCH_PROJECTOR_STATE_ID, leaseToken: token }, + data: { leaseToken: null, leaseExpiresAt: null }, + }); + } + + async advanceLive(token: string, expected: Date, next: Date): Promise { + const result = await this.database.logsSearchProjectorState.updateMany({ + where: { + id: LOGS_SEARCH_PROJECTOR_STATE_ID, + paused: false, + leaseToken: token, + liveWatermark: expected, + }, + data: { liveWatermark: next }, + }); + return result.count === 1; + } + + async advanceHistorical( + token: string, + expected: Date, + next: Date, + expectedTarget: Date + ): Promise { + const result = await this.database.logsSearchProjectorState.updateMany({ + where: { + id: LOGS_SEARCH_PROJECTOR_STATE_ID, + paused: false, + leaseToken: token, + historicalWatermark: expected, + backfillTarget: expectedTarget, + }, + data: { + historicalWatermark: next, + ...(next.getTime() === expectedTarget.getTime() ? { backfillTarget: null } : {}), + }, + }); + return result.count === 1; + } + + async pause(): Promise { + await this.database.logsSearchProjectorState.update({ + where: { id: LOGS_SEARCH_PROJECTOR_STATE_ID }, + data: { paused: true }, + }); + } + + async resume(): Promise { + await this.database.logsSearchProjectorState.update({ + where: { id: LOGS_SEARCH_PROJECTOR_STATE_ID }, + data: { paused: false }, + }); + } + + async setBackfillTarget(expectedHistorical: Date, target: Date): Promise { + const result = await this.database.logsSearchProjectorState.updateMany({ + where: { + id: LOGS_SEARCH_PROJECTOR_STATE_ID, + historicalWatermark: expectedHistorical, + backfillTarget: null, + }, + data: { backfillTarget: target }, + }); + return result.count === 1; + } + + async cancelBackfill(): Promise { + await this.database.logsSearchProjectorState.update({ + where: { id: LOGS_SEARCH_PROJECTOR_STATE_ID }, + data: { backfillTarget: null }, + }); + } +} diff --git a/apps/webapp/app/services/logsSearchProjectorTelemetry.server.ts b/apps/webapp/app/services/logsSearchProjectorTelemetry.server.ts new file mode 100644 index 0000000000..4ffc7aeda8 --- /dev/null +++ b/apps/webapp/app/services/logsSearchProjectorTelemetry.server.ts @@ -0,0 +1,85 @@ +import { getMeter } from "@internal/tracing"; +import { singleton } from "~/utils/singleton"; + +export type LogsSearchProjectionMode = "live" | "backfill"; +export type LogsSearchProjectionOutcome = "success" | "error" | "cas_lost"; + +const telemetry = singleton("logsSearchProjectorTelemetry", () => { + const meter = getMeter("logs-search-projector"); + const values: { + liveLagMs?: number; + backfillRemaining?: number; + paused?: number; + updatedAt?: number; + } = {}; + const isFresh = () => values.updatedAt && Date.now() - values.updatedAt < 150_000; + + meter + .createObservableGauge("logs_search.projector.live_lag_ms", { + description: "Delay between the safe projection cutoff and the live watermark", + }) + .addCallback((result) => { + if (isFresh() && values.liveLagMs !== undefined) result.observe(values.liveLagMs); + }); + meter + .createObservableGauge("logs_search.projector.backfill_remaining_windows", { + description: "One-minute windows remaining in the active historical backfill", + }) + .addCallback((result) => { + if (isFresh() && values.backfillRemaining !== undefined) { + result.observe(values.backfillRemaining); + } + }); + meter + .createObservableGauge("logs_search.projector.paused", { + description: "Whether the logs search projector is paused", + }) + .addCallback((result) => { + if (isFresh() && values.paused !== undefined) result.observe(values.paused); + }); + + return { + values, + windows: meter.createCounter("logs_search.projector.windows", { + description: "Logs search projection windows by mode and outcome", + }), + duration: meter.createHistogram("logs_search.projector.window_duration_ms", { + description: "Duration of one logs search projection window", + }), + sourceRows: meter.createHistogram("logs_search.projector.source_rows", { + description: "Source rows read for one logs search projection window", + }), + destinationRows: meter.createHistogram("logs_search.projector.destination_rows", { + description: "Rows written for one logs search projection window", + }), + leaseContention: meter.createCounter("logs_search.projector.lease_contention"), + casLoss: meter.createCounter("logs_search.projector.cas_loss"), + }; +}); + +export const logsSearchProjectorTelemetry = { + recordWindow( + mode: LogsSearchProjectionMode, + outcome: LogsSearchProjectionOutcome, + durationMs: number, + sourceRows = 0, + destinationRows = 0 + ) { + telemetry.windows.add(1, { mode, outcome }); + telemetry.duration.record(durationMs, { mode, outcome }); + telemetry.sourceRows.record(sourceRows, { mode }); + telemetry.destinationRows.record(destinationRows, { mode }); + }, + recordLeaseContention() { + telemetry.leaseContention.add(1); + }, + recordCasLoss(mode: LogsSearchProjectionMode) { + telemetry.casLoss.add(1, { mode }); + }, + updateState(values: { liveLagMs: number; backfillRemaining: number; paused: boolean }) { + telemetry.values.liveLagMs = Math.max(0, values.liveLagMs); + telemetry.values.backfillRemaining = Math.max(0, values.backfillRemaining); + telemetry.values.paused = values.paused ? 1 : 0; + telemetry.values.updatedAt = Date.now(); + }, +}; diff --git a/apps/webapp/app/utils/logSearch.test.ts b/apps/webapp/app/utils/logSearch.test.ts new file mode 100644 index 0000000000..4223244f8d --- /dev/null +++ b/apps/webapp/app/utils/logSearch.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from "vitest"; +import { + escapeClickHouseLike, + hasMinimumLogsSearchLength, + logsSearchExpansionPeriod, + normalizeLogsSearchTerm, + prepareLogsSearchPage, +} from "./logSearch"; + +describe("log search normalization", () => { + it("normalizes punctuation while preserving unicode, paths, and ids", () => { + expect( + normalizeLogsSearchTerm("TypeError: Zahlungsübersicht failed, retrying (/api/orders/42)") + ).toBe("typeerror:zahlungsübersicht failed retrying /api/orders/42"); + expect(normalizeLogsSearchTerm('"status_code": 500')).toBe("status_code:500"); + expect(normalizeLogsSearchTerm("status_code:500")).toBe("status_code:500"); + }); + + it("uses the same locale-independent casing as ClickHouse", () => { + expect(normalizeLogsSearchTerm("I İ ı İSTANBUL ΟΣ")).toBe("i i ı i stanbul ος"); + }); + + it("escapes LIKE wildcards without escaping path separators", () => { + expect(escapeClickHouseLike("/api/a_b/100%")).toBe("/api/a\\_b/100\\%"); + }); + + it("requires at least three unicode characters after trimming", () => { + expect(hasMinimumLogsSearchLength("ab")).toBe(false); + expect(hasMinimumLogsSearchLength(" ab ")).toBe(false); + expect(hasMinimumLogsSearchLength("abc")).toBe(true); + expect(hasMinimumLogsSearchLength("日本語")).toBe(true); + }); + + it("only offers a strictly wider retained search range", () => { + const to = new Date("2026-08-14T12:00:00.000Z"); + + expect(logsSearchExpansionPeriod(new Date("2026-08-14T11:00:00.000Z"), to, 1)).toBe("1d"); + expect(logsSearchExpansionPeriod(new Date("2026-08-13T12:00:00.000Z"), to, 1)).toBeUndefined(); + expect(logsSearchExpansionPeriod(new Date("2026-08-13T12:00:00.000Z"), to, 7)).toBe("7d"); + }); + + it("removes projector retry copies after bounded overfetch", () => { + const row = (fingerprint: string) => ({ + projection_fingerprint_string: fingerprint, + trace_id: `trace_${fingerprint}`, + span_id: `span_${fingerprint}`, + run_id: `run_${fingerprint}`, + start_time: "2026-08-14 12:00:00.000000000", + }); + const page = prepareLogsSearchPage([row("a"), row("a"), row("b"), row("c"), row("d")], 2, 5); + + expect(page.rows.map((item) => item.projection_fingerprint_string)).toEqual(["a", "b"]); + expect(page.hasMore).toBe(true); + }); + + it("keeps pagination open when retries fill the overfetch bound", () => { + const duplicate = { + projection_fingerprint_string: "same", + trace_id: "trace", + span_id: "span", + run_id: "run", + start_time: "2026-08-14 12:00:00.000000000", + }; + + expect(prepareLogsSearchPage([duplicate, duplicate, duplicate, duplicate], 2, 4)).toEqual({ + rows: [duplicate], + hasMore: true, + }); + }); +}); diff --git a/apps/webapp/app/utils/logSearch.ts b/apps/webapp/app/utils/logSearch.ts new file mode 100644 index 0000000000..f06457b28d --- /dev/null +++ b/apps/webapp/app/utils/logSearch.ts @@ -0,0 +1,66 @@ +export const MIN_LOGS_SEARCH_LENGTH = 3; +export const LOGS_SEARCH_RETRY_OVERFETCH_FACTOR = 4; +const DAY_MS = 24 * 60 * 60 * 1000; +const RANGE_COMPARISON_TOLERANCE_MS = 1000; + +export function logsSearchExpansionPeriod( + from: Date | undefined, + to: Date, + retentionLimitDays: number | undefined +): string | undefined { + if (!from) return undefined; + + const candidateDays = Math.min(retentionLimitDays ?? 7, 7); + const currentRangeMs = Math.max(0, to.getTime() - from.getTime()); + if (candidateDays * DAY_MS <= currentRangeMs + RANGE_COMPARISON_TOLERANCE_MS) { + return undefined; + } + + return `${candidateDays}d`; +} + +type ProjectedLogIdentity = { + projection_fingerprint_string?: string; + trace_id: string; + span_id: string; + run_id: string; + start_time: string; +}; + +export function prepareLogsSearchPage( + rows: T[], + pageSize: number, + queryLimit: number +): { rows: T[]; hasMore: boolean } { + const seen = new Set(); + const uniqueRows = rows.filter((row) => { + const identity = + row.projection_fingerprint_string ?? + JSON.stringify([row.trace_id, row.span_id, row.run_id, row.start_time]); + if (seen.has(identity)) return false; + seen.add(identity); + return true; + }); + + return { + rows: uniqueRows.slice(0, pageSize), + hasMore: uniqueRows.length > pageSize || rows.length === queryLimit, + }; +} + +export function hasMinimumLogsSearchLength(value: string): boolean { + return [...value.trim()].length >= MIN_LOGS_SEARCH_LENGTH; +} + +export function escapeClickHouseLike(value: string): string { + return value.replace(/\\/g, "\\\\").replace(/%/g, "\\%").replace(/_/g, "\\_"); +} + +// Must match the scheduled ClickHouse projector normalization. +export function normalizeLogsSearchTerm(value: string): string { + return value + .toLowerCase() + .replace(/[^\p{L}\p{N}_./:@+-]+/gu, " ") + .replace(/\s*:\s*/g, ":") + .trim(); +} diff --git a/apps/webapp/app/v3/logsSearchProjectorWorker.server.ts b/apps/webapp/app/v3/logsSearchProjectorWorker.server.ts new file mode 100644 index 0000000000..69cac4624d --- /dev/null +++ b/apps/webapp/app/v3/logsSearchProjectorWorker.server.ts @@ -0,0 +1,67 @@ +import { Logger } from "@trigger.dev/core/logger"; +import { CronSchema, Worker as RedisWorker } from "@trigger.dev/redis-worker"; +import { env } from "~/env.server"; +import { logger } from "~/services/logger.server"; +import { getLogsSearchProjector } from "~/services/logsSearchProjectorInstance.server"; +import { singleton } from "~/utils/singleton"; + +function initializeWorker() { + const worker = new RedisWorker({ + name: "logs-search-projector-worker", + redisOptions: { + keyPrefix: "logs-search-projector:worker:", + host: env.COMMON_WORKER_REDIS_HOST, + port: env.COMMON_WORKER_REDIS_PORT, + username: env.COMMON_WORKER_REDIS_USERNAME, + password: env.COMMON_WORKER_REDIS_PASSWORD, + enableAutoPipelining: true, + ...(env.COMMON_WORKER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }), + }, + catalog: { + "logsSearch.projectV2": { + schema: CronSchema, + cron: "* * * * *", + jitterInMs: 5_000, + visibilityTimeoutMs: + env.LOGS_SEARCH_PROJECTOR_MAX_WINDOWS_PER_TICK * + (env.LOGS_SEARCH_PROJECTOR_MAX_EXECUTION_TIME_SECONDS + 30) * + 1000 + + 60_000, + retry: { maxAttempts: 1 }, + }, + }, + concurrency: { workers: 1, tasksPerWorker: 1, limit: 1 }, + pollIntervalMs: env.COMMON_WORKER_POLL_INTERVAL, + immediatePollIntervalMs: env.COMMON_WORKER_IMMEDIATE_POLL_INTERVAL, + shutdownTimeoutMs: env.COMMON_WORKER_SHUTDOWN_TIMEOUT_MS, + logger: new Logger("LogsSearchProjectorWorker", env.COMMON_WORKER_LOG_LEVEL), + jobs: { + "logsSearch.projectV2": async () => { + await getLogsSearchProjector().processTick(); + }, + }, + }); + + return worker; +} + +export const logsSearchProjectorWorker = singleton("logsSearchProjectorWorker", initializeWorker); + +declare global { + // eslint-disable-next-line no-var + var __logsSearchProjectorWorkerStarted__: boolean | undefined; +} + +export function initLogsSearchProjectorWorker(): void { + if ( + !env.LOGS_SEARCH_PROJECTOR_ENABLED || + env.COMMON_WORKER_ENABLED !== "true" || + global.__logsSearchProjectorWorkerStarted__ + ) { + return; + } + + logger.info("Starting logs search projector worker"); + logsSearchProjectorWorker.start(); + global.__logsSearchProjectorWorkerStarted__ = true; +} diff --git a/apps/webapp/test/logsSearchProjector.test.ts b/apps/webapp/test/logsSearchProjector.test.ts new file mode 100644 index 0000000000..39d5eea530 --- /dev/null +++ b/apps/webapp/test/logsSearchProjector.test.ts @@ -0,0 +1,81 @@ +import { describe, expect, it } from "vitest"; +import { + calculateClosedWindowBoundary, + LOGS_SEARCH_PROJECTOR_STATE_ID, + selectNextProjectionWindow, + type LogsSearchProjectorState, +} from "~/services/logsSearchProjector.server"; + +const minute = 60_000; +const at = (value: string) => new Date(value); + +function state(overrides: Partial = {}): LogsSearchProjectorState { + const boundary = overrides.liveWatermark ?? at("2026-08-14T12:05:00.000Z"); + return { + id: LOGS_SEARCH_PROJECTOR_STATE_ID, + liveWatermark: boundary, + historicalWatermark: overrides.historicalWatermark ?? boundary, + backfillTarget: overrides.backfillTarget ?? null, + paused: overrides.paused ?? false, + leaseToken: overrides.leaseToken ?? null, + leaseExpiresAt: overrides.leaseExpiresAt ?? null, + }; +} + +describe("logs search projector window selection", () => { + it("floors the safe cutoff to a closed minute", () => { + expect( + calculateClosedWindowBoundary(at("2026-08-14T12:10:59.999Z"), 2 * minute).toISOString() + ).toBe("2026-08-14T12:08:00.000Z"); + }); + + it("selects the oldest live window before historical work", () => { + expect( + selectNextProjectionWindow( + state({ + liveWatermark: at("2026-08-14T12:05:00.000Z"), + historicalWatermark: at("2026-08-14T12:04:00.000Z"), + backfillTarget: at("2026-08-14T12:02:00.000Z"), + }), + at("2026-08-14T12:08:00.000Z") + ) + ).toEqual({ + mode: "live", + start: at("2026-08-14T12:05:00.000Z"), + end: at("2026-08-14T12:06:00.000Z"), + }); + }); + + it("extends historical coverage backwards after live work catches up", () => { + expect( + selectNextProjectionWindow( + state({ + liveWatermark: at("2026-08-14T12:08:00.000Z"), + historicalWatermark: at("2026-08-14T12:04:00.000Z"), + backfillTarget: at("2026-08-14T12:02:00.000Z"), + }), + at("2026-08-14T12:08:00.000Z") + ) + ).toEqual({ + mode: "backfill", + start: at("2026-08-14T12:03:00.000Z"), + end: at("2026-08-14T12:04:00.000Z"), + }); + }); + + it("selects no work while paused or fully caught up", () => { + const safeCutoff = at("2026-08-14T12:08:00.000Z"); + expect( + selectNextProjectionWindow( + state({ liveWatermark: safeCutoff, historicalWatermark: safeCutoff }), + safeCutoff + ) + ).toBeNull(); + expect( + selectNextProjectionWindow( + state({ liveWatermark: at("2026-08-14T12:05:00.000Z"), paused: true }), + safeCutoff + ) + ).toBeNull(); + }); +}); diff --git a/apps/webapp/test/logsSearchProjectorStateStore.test.ts b/apps/webapp/test/logsSearchProjectorStateStore.test.ts new file mode 100644 index 0000000000..2097e4a62c --- /dev/null +++ b/apps/webapp/test/logsSearchProjectorStateStore.test.ts @@ -0,0 +1,52 @@ +import { postgresTest } from "@internal/testcontainers"; +import { expect } from "vitest"; +import { LOGS_SEARCH_PROJECTOR_STATE_ID } from "~/services/logsSearchProjector.server"; +import { PrismaLogsSearchProjectorStateStore } from "~/services/logsSearchProjectorStateStore.server"; + +const at = (value: string) => new Date(value); + +postgresTest( + "persists projector leases, watermarks, pause state, and backfill state", + async ({ prisma }) => { + const store = new PrismaLogsSearchProjectorStateStore(prisma); + const initial = at("2026-08-14T12:00:00.000Z"); + const next = at("2026-08-14T12:01:00.000Z"); + + await expect(store.find()).resolves.toBeNull(); + await expect(store.initialize(initial)).resolves.toMatchObject({ + id: LOGS_SEARCH_PROJECTOR_STATE_ID, + liveWatermark: initial, + historicalWatermark: initial, + paused: false, + }); + + expect(await store.acquireLease("lease-a", 60_000)).toBe(true); + expect(await store.acquireLease("lease-b", 60_000)).toBe(false); + expect(await store.advanceLive("lease-a", next, at("2026-08-14T12:02:00.000Z"))).toBe(false); + expect(await store.advanceLive("lease-a", initial, next)).toBe(true); + await store.releaseLease("lease-a"); + + await store.pause(); + expect(await store.acquireLease("lease-b", 60_000)).toBe(false); + await store.resume(); + expect(await store.acquireLease("lease-b", 60_000)).toBe(true); + + const target = at("2026-08-14T11:58:00.000Z"); + expect(await store.setBackfillTarget(initial, target)).toBe(true); + expect(await store.advanceHistorical("lease-b", next, initial, target)).toBe(false); + expect( + await store.advanceHistorical("lease-b", initial, at("2026-08-14T11:59:00.000Z"), target) + ).toBe(true); + expect( + await store.advanceHistorical("lease-b", at("2026-08-14T11:59:00.000Z"), target, target) + ).toBe(true); + + await expect(store.get()).resolves.toMatchObject({ + liveWatermark: next, + historicalWatermark: target, + backfillTarget: null, + paused: false, + leaseToken: "lease-b", + }); + } +); diff --git a/apps/webapp/test/runGetRoutes.replicaLag.guard.test.ts b/apps/webapp/test/runGetRoutes.replicaLag.guard.test.ts index f6597980f3..146aba99e1 100644 --- a/apps/webapp/test/runGetRoutes.replicaLag.guard.test.ts +++ b/apps/webapp/test/runGetRoutes.replicaLag.guard.test.ts @@ -267,7 +267,11 @@ let seq = 0; async function seedTenant(prisma: PrismaClient, suffix: string) { const organization = await prisma.organization.create({ - data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, + data: { + title: `Org ${suffix}`, + slug: `org-${suffix}`, + featureFlags: { hasLogsPageAccess: true }, + }, }); const project = await prisma.project.create({ data: { diff --git a/internal-packages/clickhouse/schema/039_create_task_events_search_v2.sql b/internal-packages/clickhouse/schema/039_create_task_events_search_v2.sql new file mode 100644 index 0000000000..80d6f74be4 --- /dev/null +++ b/internal-packages/clickhouse/schema/039_create_task_events_search_v2.sql @@ -0,0 +1,45 @@ +-- +goose Up +-- Search v2 stores bounded normalized text outside the task_events_v2 insert path. +-- The source index (idx_inserted_at_projector) is added in an earlier migration. +CREATE TABLE IF NOT EXISTS trigger_dev.task_events_search_v2 +( + environment_id String, + organization_id String, + project_id String, + triggered_timestamp DateTime64(9) CODEC(Delta(8), ZSTD(1)), + trace_id String CODEC(ZSTD(1)), + span_id String CODEC(ZSTD(1)), + run_id String CODEC(ZSTD(1)), + task_identifier String CODEC(ZSTD(1)), + start_time DateTime64(9) CODEC(Delta(8), ZSTD(1)), + inserted_at DateTime64(3), + message String CODEC(ZSTD(1)), + error_message String CODEC(ZSTD(1)), + search_text String CODEC(ZSTD(1)), + kind LowCardinality(String) CODEC(ZSTD(1)), + status LowCardinality(String) CODEC(ZSTD(1)), + duration UInt64 CODEC(ZSTD(1)), + parent_span_id String CODEC(ZSTD(1)), + projection_fingerprint UInt128 DEFAULT reinterpretAsUInt128( + sipHash128(trace_id, span_id, run_id, start_time) + ), + + INDEX idx_run_id run_id TYPE bloom_filter(0.001) GRANULARITY 1, + INDEX idx_search_text search_text + TYPE text(tokenizer = 'ngrams', preprocessor = lowerUTF8(search_text)) +) +ENGINE = ReplacingMergeTree +PARTITION BY toDate(triggered_timestamp) +ORDER BY ( + organization_id, + environment_id, + triggered_timestamp, + trace_id, + span_id, + projection_fingerprint +) +TTL toDateTime(triggered_timestamp) + INTERVAL 90 DAY +SETTINGS ttl_only_drop_parts = 1; + +-- +goose Down +DROP TABLE IF EXISTS trigger_dev.task_events_search_v2; diff --git a/internal-packages/clickhouse/src/client/client.ts b/internal-packages/clickhouse/src/client/client.ts index d6703c863c..a61598360b 100644 --- a/internal-packages/clickhouse/src/client/client.ts +++ b/internal-packages/clickhouse/src/client/client.ts @@ -13,6 +13,7 @@ import { flattenAttributes, tryCatch, type Result } from "@trigger.dev/core/v3"; import { z } from "zod"; import { InsertError, QueryError } from "./errors.js"; import type { + ClickhouseCommandFunction, ClickhouseInsertFunction, ClickhouseQueryBuilderFastFunction, ClickhouseQueryBuilderFunction, @@ -43,6 +44,7 @@ export type ClickhouseConfig = { clickhouseSettings?: ClickHouseSettings; logger?: Logger; maxOpenConnections?: number; + requestTimeoutMs?: number; logLevel?: LogLevel; compression?: { request?: boolean; @@ -66,6 +68,7 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { http_agent: config.httpAgent, compression: config.compression, max_open_connections: config.maxOpenConnections, + request_timeout: config.requestTimeoutMs, clickhouse_settings: { ...config.clickhouseSettings, output_format_json_quote_64bit_integers: 0, @@ -672,6 +675,88 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { }); } + public command>(req: { + name: string; + query: string; + params?: TSchema; + settings?: ClickHouseSettings; + }): ClickhouseCommandFunction> { + return async (params, options) => { + const queryId = randomUUID(); + + return await startSpan(this.tracer, "command", async (span) => { + span.setAttributes({ + "clickhouse.clientName": this.name, + "clickhouse.operationName": req.name, + "clickhouse.queryId": queryId, + ...flattenAttributes(req.settings, "clickhouse.settings"), + ...flattenAttributes(options?.attributes), + }); + + const validParams = req.params?.safeParse(params); + if (validParams?.error) { + recordSpanError(span, validParams.error); + return [ + new QueryError(`Bad params: ${generateErrorMessage(validParams.error.issues)}`, { + query: req.query, + }), + null, + ]; + } + + this.logger.debug("Running clickhouse command", { + clientName: this.name, + name: req.name, + query: req.query.replace(/\s+/g, " "), + settings: req.settings, + attributes: options?.attributes, + queryId, + }); + + const [clickhouseError, result] = await tryCatch( + this.client.command({ + query: req.query, + query_params: validParams?.data, + query_id: queryId, + ...options?.params, + clickhouse_settings: { + ...req.settings, + ...options?.params?.clickhouse_settings, + }, + }) + ); + + if (clickhouseError) { + this.logger.error("Error running clickhouse command", { + name: req.name, + error: clickhouseError, + query: req.query, + queryId, + }); + recordClickhouseError(span, clickhouseError); + return [ + new QueryError(`Unable to run clickhouse command: ${clickhouseError.message}`, { + query: req.query, + }), + null, + ]; + } + + span.setAttributes({ + "clickhouse.query_id": result.query_id, + "clickhouse.summary.read_rows": result.summary?.read_rows, + "clickhouse.summary.read_bytes": result.summary?.read_bytes, + "clickhouse.summary.written_rows": result.summary?.written_rows, + "clickhouse.summary.written_bytes": result.summary?.written_bytes, + "clickhouse.summary.elapsed_ns": result.summary?.elapsed_ns, + ...flattenAttributes(result.response_headers, "clickhouse.response_headers"), + }); + + return [null, result]; + }); + }; + } + public insert>(req: { name: string; table: string; diff --git a/internal-packages/clickhouse/src/client/noop.ts b/internal-packages/clickhouse/src/client/noop.ts index 00adef82c8..2e91003a5e 100644 --- a/internal-packages/clickhouse/src/client/noop.ts +++ b/internal-packages/clickhouse/src/client/noop.ts @@ -8,7 +8,7 @@ import type { QueryResultWithStats, } from "./types.js"; import type { z } from "zod"; -import type { ClickHouseSettings, InsertResult } from "@clickhouse/client"; +import type { ClickHouseSettings, CommandResult, InsertResult } from "@clickhouse/client"; import { ClickhouseQueryBuilder, ClickhouseQueryFastBuilder } from "./queryBuilder.js"; export class NoopClient implements ClickhouseReader, ClickhouseWriter { @@ -109,6 +109,41 @@ export class NoopClient implements ClickhouseReader, ClickhouseWriter { }; } + public command>(req: { + name: string; + query: string; + params?: TSchema; + settings?: ClickHouseSettings; + }): (params: z.input) => Promise> { + return async (params) => { + const validParams = req.params?.safeParse(params); + if (validParams?.error) { + return [ + new QueryError(`Bad params: ${validParams.error.message}`, { query: req.query }), + null, + ]; + } + + return [ + null, + { + query_id: "noop", + summary: { + read_rows: "0", + read_bytes: "0", + written_rows: "0", + written_bytes: "0", + total_rows_to_read: "0", + result_rows: "0", + result_bytes: "0", + elapsed_ns: "0", + }, + response_headers: {}, + }, + ]; + }; + } + public insert>(req: { name: string; table: string; diff --git a/internal-packages/clickhouse/src/client/types.ts b/internal-packages/clickhouse/src/client/types.ts index 4bfa6dc466..6cfbe35fd4 100644 --- a/internal-packages/clickhouse/src/client/types.ts +++ b/internal-packages/clickhouse/src/client/types.ts @@ -4,6 +4,7 @@ import type { InsertError, QueryError } from "./errors.js"; import { type ClickHouseSettings, type BaseQueryParams, + type CommandResult, type InsertResult, } from "@clickhouse/client"; import type { ClickhouseQueryBuilder, ClickhouseQueryFastBuilder } from "./queryBuilder.js"; @@ -237,6 +238,14 @@ export interface ClickhouseReader { close(): Promise; } +export type ClickhouseCommandFunction = ( + params: TInput, + options?: { + attributes?: Record; + params?: BaseQueryParams; + } +) => Promise>; + export type ClickhouseInsertFunction = ( events: TInput | TInput[], options?: { @@ -246,6 +255,13 @@ export type ClickhouseInsertFunction = ( ) => Promise>; export interface ClickhouseWriter { + command>(req: { + name: string; + query: string; + params?: TSchema; + settings?: ClickHouseSettings; + }): ClickhouseCommandFunction>; + insert>(req: { name: string; table: string; diff --git a/internal-packages/clickhouse/src/index.ts b/internal-packages/clickhouse/src/index.ts index 49d2d7c02b..016f389726 100644 --- a/internal-packages/clickhouse/src/index.ts +++ b/internal-packages/clickhouse/src/index.ts @@ -31,6 +31,7 @@ import { getLogDetailQueryBuilderV2, getLogsSearchListQueryBuilder, } from "./taskEvents.js"; +import { projectTaskEventsSearchV2Window } from "./taskEventsSearchProjector.js"; import { insertMetrics } from "./metrics.js"; import { insertLlmMetrics } from "./llmMetrics.js"; import { @@ -73,6 +74,7 @@ import type { Agent as HttpsAgent } from "https"; export type * from "./taskRuns.js"; export type * from "./taskEvents.js"; +export * from "./taskEventsSearchProjector.js"; export type * from "./metrics.js"; export type * from "./llmMetrics.js"; export type * from "./queueMetrics.js"; @@ -125,6 +127,7 @@ export type ClickhouseCommonConfig = { response?: boolean; }; maxOpenConnections?: number; + requestTimeoutMs?: number; }; export type ClickHouseConfig = @@ -167,6 +170,7 @@ export class ClickHouse { keepAlive: config.keepAlive, httpAgent: config.httpAgent, maxOpenConnections: config.maxOpenConnections, + requestTimeoutMs: config.requestTimeoutMs, compression: config.compression, }); this.reader = client; @@ -183,6 +187,7 @@ export class ClickHouse { keepAlive: config.keepAlive, httpAgent: config.httpAgent, maxOpenConnections: config.maxOpenConnections, + requestTimeoutMs: config.requestTimeoutMs, compression: config.compression, }); this.writer = new ClickhouseClient({ @@ -194,6 +199,7 @@ export class ClickHouse { keepAlive: config.keepAlive, httpAgent: config.httpAgent, maxOpenConnections: config.maxOpenConnections, + requestTimeoutMs: config.requestTimeoutMs, compression: config.compression, }); @@ -314,7 +320,9 @@ export class ClickHouse { get taskEventsSearch() { return { - logsListQueryBuilder: getLogsSearchListQueryBuilder(this.reader), + logsListQueryBuilder: (version: "v1" | "v2" = "v1") => + getLogsSearchListQueryBuilder(this.reader, version)(), + projectV2Window: projectTaskEventsSearchV2Window(this.writer), }; } diff --git a/internal-packages/clickhouse/src/taskEvents.ts b/internal-packages/clickhouse/src/taskEvents.ts index a1f001897b..01941d4f64 100644 --- a/internal-packages/clickhouse/src/taskEvents.ts +++ b/internal-packages/clickhouse/src/taskEvents.ts @@ -280,7 +280,7 @@ export function getTraceEventsForExportQueryBuilderV2( } // ============================================================================ -// Search Table Query Builders (for logs page, using task_events_search_v1) +// Search Table Query Builders (for logs page, using task_events_search_v2) // ============================================================================ export const LogsSearchListResult = z.object({ @@ -294,19 +294,26 @@ export const LogsSearchListResult = z.object({ span_id: z.string(), parent_span_id: z.string(), message: z.string(), + error_message: z.string(), kind: z.string(), status: z.string(), duration: z.number().or(z.string()), - attributes_text: z.string(), triggered_timestamp: z.string(), + projection_fingerprint_string: z.string().optional(), }); export type LogsSearchListResult = z.output; -export function getLogsSearchListQueryBuilder(ch: ClickhouseReader) { - return ch.queryBuilderFast({ - name: "getLogsSearchList", - table: "trigger_dev.task_events_search_v1", +export type LogsSearchTableVersion = "v1" | "v2"; + +export function getLogsSearchListQueryBuilder( + ch: ClickhouseReader, + version: LogsSearchTableVersion = "v1" +) { + const createBuilder = ch.queryBuilderFast({ + name: version === "v2" ? "getLogsSearchListV2" : "getLogsSearchListV1", + table: + version === "v2" ? "trigger_dev.task_events_search_v2" : "trigger_dev.task_events_search_v1", columns: [ "environment_id", "organization_id", @@ -318,16 +325,32 @@ export function getLogsSearchListQueryBuilder(ch: ClickhouseReader) { "span_id", "parent_span_id", { name: "message", expression: "LEFT(message, 512)" }, + { + name: "error_message", + expression: + version === "v2" + ? "error_message" + : "LEFT(JSONExtractString(attributes_text, 'error', 'message'), 2048)", + }, "kind", "status", "duration", - "attributes_text", "triggered_timestamp", + ...(version === "v2" + ? [ + { + name: "projection_fingerprint_string", + expression: "toString(projection_fingerprint)", + }, + ] + : []), ], settings: { use_query_condition_cache: 1, }, }); + + return createBuilder; } // Single log detail query builder (for side panel) diff --git a/internal-packages/clickhouse/src/taskEventsSearch.test.ts b/internal-packages/clickhouse/src/taskEventsSearch.test.ts new file mode 100644 index 0000000000..b81e2f9f3f --- /dev/null +++ b/internal-packages/clickhouse/src/taskEventsSearch.test.ts @@ -0,0 +1,255 @@ +import { clickhouseTest } from "@internal/testcontainers"; +import { randomUUID } from "node:crypto"; +import { z } from "zod"; +import { ClickHouse } from "./index.js"; + +const ORG = "org_logs_search"; +const PROJECT = "project_logs_search"; +const ENVIRONMENT = "env_logs_search"; +const LIMITS = { + maxExecutionTimeSeconds: 30, + maxRowsToRead: 1_000_000, + maxMemoryUsage: 500_000_000, + maxThreads: 1, +}; + +function clickhouseDate(value: Date) { + return value.toISOString().replace("T", " ").replace("Z", ""); +} + +function event(now: Date, overrides: Record = {}) { + const start = clickhouseDate(now); + return { + environment_id: ENVIRONMENT, + organization_id: ORG, + project_id: PROJECT, + task_identifier: "search-task", + run_id: "run_logs_search", + start_time: start, + duration: "1000000", + trace_id: "trace_logs_search", + span_id: `span_${randomUUID()}`, + parent_span_id: "", + message: "TypeError: Zahlungsübersicht failed, retrying /api/orders/42", + kind: "LOG_ERROR", + status: "ERROR", + attributes: { + request_id: "req_123", + status_code: 500, + retryable: true, + error: { message: "Payment failed, retrying" }, + }, + metadata: "{}", + expires_at: clickhouseDate(new Date(now.getTime() + 90 * 24 * 60 * 60 * 1000)), + inserted_at: start, + ...overrides, + }; +} + +async function project(ch: ClickHouse, start: Date, end: Date) { + const [error, result] = await ch.taskEventsSearch.projectV2Window({ start, end }, LIMITS); + expect(error).toBeNull(); + expect(result?.query_id).toEqual(expect.any(String)); + return result!; +} + +function searchRows(ch: ClickHouse) { + const builder = ch.taskEventsSearch.logsListQueryBuilder("v2"); + builder.where("organization_id = {organizationId: String}", { organizationId: ORG }); + builder.orderBy( + "triggered_timestamp DESC, trace_id DESC, span_id DESC, projection_fingerprint DESC" + ); + builder.limit(50); + return builder.execute(); +} + +describe("task events search v2", () => { + clickhouseTest( + "projects bounded normalized text outside the source insert path", + async ({ clickhouseContainer }) => { + const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" }); + const now = new Date("2026-08-14T10:10:30.000Z"); + const start = new Date(now.getTime() - 30_000); + const end = new Date(now.getTime() + 30_000); + const [insertError] = await ch.taskEventsV2.insert([event(now)]); + expect(insertError).toBeNull(); + + const [beforeError, beforeRows] = await searchRows(ch); + expect(beforeError).toBeNull(); + expect(beforeRows).toHaveLength(0); + + const schemaQuery = ch.reader.query({ + name: "read-search-v2-schema", + query: `SELECT name, type FROM system.data_skipping_indices + WHERE database = 'trigger_dev' AND table = 'task_events_v2' + AND name = 'idx_inserted_at_projector'`, + schema: z.object({ name: z.string(), type: z.string() }), + }); + const [schemaError, indexes] = await schemaQuery({}); + expect(schemaError).toBeNull(); + expect(indexes).toEqual([{ name: "idx_inserted_at_projector", type: "minmax" }]); + + const tableQuery = ch.reader.query({ + name: "read-search-v2-table-engine", + query: `SELECT name, engine FROM system.tables + WHERE database = 'trigger_dev' + AND name IN ('task_events_search_mv_v2', 'task_events_search_v2') + ORDER BY name`, + schema: z.object({ name: z.string(), engine: z.string() }), + }); + const [tableError, tables] = await tableQuery({}); + expect(tableError).toBeNull(); + expect(tables).toEqual([{ name: "task_events_search_v2", engine: "ReplacingMergeTree" }]); + + const firstProjection = await project(ch, start, end); + const retryProjection = await project(ch, start, end); + expect(Number(firstProjection.summary?.written_rows)).toBe(1); + expect(Number(retryProjection.summary?.written_rows)).toBe(1); + + const [preMergeReadError, preMergeRows] = await searchRows(ch); + expect(preMergeReadError).toBeNull(); + expect([1, 2]).toContain(preMergeRows?.length); + const rawQuery = ch.reader.query({ + name: "count-raw-search-v2-fixture", + query: `SELECT count() AS count FROM trigger_dev.task_events_search_v2 + WHERE organization_id = {organizationId: String}`, + params: z.object({ organizationId: z.string() }), + schema: z.object({ count: z.number() }), + }); + let [rawError, rawRows] = await rawQuery({ organizationId: ORG }); + expect(rawError).toBeNull(); + expect([1, 2]).toContain(rawRows?.[0].count); + + const optimize = ch.writer.command({ + name: "merge-search-v2-retry-fixture", + query: "OPTIMIZE TABLE trigger_dev.task_events_search_v2 FINAL", + }); + const [optimizeError] = await optimize({}); + expect(optimizeError).toBeNull(); + [rawError, rawRows] = await rawQuery({ organizationId: ORG }); + expect(rawError).toBeNull(); + expect(rawRows?.[0].count).toBe(1); + const [readError, rows] = await searchRows(ch); + expect(readError).toBeNull(); + expect(rows).toHaveLength(1); + + expect(rows?.[0].message.toLowerCase()).toContain( + "typeerror: zahlungsübersicht failed, retrying /api/orders/42" + ); + expect(rows?.[0].error_message).toBe("Payment failed, retrying"); + + const searchDataQuery = ch.reader.query({ + name: "read-search-v2-indexed-data", + query: `SELECT search_text, error_message + FROM trigger_dev.task_events_search_v2 + WHERE organization_id = {organizationId: String} + LIMIT 1`, + params: z.object({ organizationId: z.string() }), + schema: z.object({ search_text: z.string(), error_message: z.string() }), + }); + const [searchDataError, searchData] = await searchDataQuery({ organizationId: ORG }); + expect(searchDataError).toBeNull(); + expect(searchData).toHaveLength(1); + expect(searchData?.[0].search_text).toContain( + "typeerror:zahlungsübersicht failed retrying /api/orders/42" + ); + expect(searchData?.[0].search_text).toContain("status_code:500"); + expect(searchData?.[0].search_text).toContain("retryable:true"); + + await ch.close(); + } + ); + + clickhouseTest( + "uses half-open windows and deterministically clamps future timestamps", + async ({ clickhouseContainer }) => { + const ch = new ClickHouse({ url: clickhouseContainer.getConnectionUrl(), name: "test" }); + const boundary = new Date("2026-08-14T11:01:00.000Z"); + const first = new Date(boundary.getTime() - 60_000); + const second = boundary; + const end = new Date(boundary.getTime() + 60_000); + const splitUtf8Boundary = `${"x".repeat(2044)}€tail`; + const [insertError] = await ch.taskEventsV2.insert([ + event(first, { + span_id: "span_first", + duration: "18446744073709551615", + message: splitUtf8Boundary, + attributes: { + prefix: "kept-token", + payload: "x".repeat(100_000), + error: { message: splitUtf8Boundary }, + }, + }), + event(second, { span_id: "span_second" }), + ]); + expect(insertError).toBeNull(); + + await project(ch, first, boundary); + let [readError, rows] = await searchRows(ch); + expect(readError).toBeNull(); + expect(rows).toHaveLength(1); + const lengthQuery = ch.reader.query({ + name: "read-search-v2-length", + query: `SELECT + length(search_text) AS search_length, + length(error_message) AS error_message_length, + isValidUTF8(search_text) AS search_text_is_valid_utf8, + isValidUTF8(error_message) AS error_message_is_valid_utf8 + FROM trigger_dev.task_events_search_v2 + WHERE organization_id = {organizationId: String} + LIMIT 1`, + params: z.object({ organizationId: z.string() }), + schema: z.object({ + search_length: z.number(), + error_message_length: z.number(), + search_text_is_valid_utf8: z.number(), + error_message_is_valid_utf8: z.number(), + }), + }); + const [lengthError, lengths] = await lengthQuery({ organizationId: ORG }); + expect(lengthError).toBeNull(); + expect(lengths?.[0].search_length).toBeLessThanOrEqual(8192); + expect(lengths?.[0].error_message_length).toBeLessThanOrEqual(2048); + expect(lengths?.[0].search_text_is_valid_utf8).toBe(1); + expect(lengths?.[0].error_message_is_valid_utf8).toBe(1); + expect(rows?.[0].triggered_timestamp).toBeDefined(); + expect(new Date(`${rows?.[0].triggered_timestamp}Z`).getTime()).toBe( + boundary.getTime() + 5 * 60_000 + ); + + await project(ch, boundary, end); + [readError, rows] = await searchRows(ch); + expect(readError).toBeNull(); + expect(rows).toHaveLength(2); + + const cursor = rows?.[0]; + expect(cursor?.projection_fingerprint_string).toEqual(expect.any(String)); + const nextPageBuilder = ch.taskEventsSearch.logsListQueryBuilder("v2"); + nextPageBuilder.where("organization_id = {organizationId: String}", { + organizationId: ORG, + }); + nextPageBuilder.where( + `(triggered_timestamp < {cursorTriggeredTimestamp: String} + OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id < {cursorTraceId: String}) + OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id < {cursorSpanId: String}) + OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id = {cursorSpanId: String} AND projection_fingerprint < {cursorProjectionFingerprint: UInt128}))`, + { + cursorTriggeredTimestamp: cursor!.triggered_timestamp, + cursorTraceId: cursor!.trace_id, + cursorSpanId: cursor!.span_id, + cursorProjectionFingerprint: cursor!.projection_fingerprint_string!, + } + ); + nextPageBuilder.orderBy( + "triggered_timestamp DESC, trace_id DESC, span_id DESC, projection_fingerprint DESC" + ); + nextPageBuilder.limit(50); + const [nextPageError, nextPage] = await nextPageBuilder.execute(); + expect(nextPageError).toBeNull(); + expect(nextPage).toHaveLength(1); + expect(nextPage?.[0].span_id).not.toBe(cursor?.span_id); + + await ch.close(); + } + ); +}); diff --git a/internal-packages/clickhouse/src/taskEventsSearchProjector.ts b/internal-packages/clickhouse/src/taskEventsSearchProjector.ts new file mode 100644 index 0000000000..ebb08f4dfa --- /dev/null +++ b/internal-packages/clickhouse/src/taskEventsSearchProjector.ts @@ -0,0 +1,185 @@ +import type { ClickHouseSettings, CommandResult } from "@clickhouse/client"; +import type { Result } from "@trigger.dev/core/v3"; +import { z } from "zod"; +import type { QueryError } from "./client/errors.js"; +import type { ClickhouseWriter } from "./client/types.js"; + +export type TaskEventsSearchV2ProjectionWindow = { + start: Date; + end: Date; +}; + +export type TaskEventsSearchV2ProjectionLimits = { + maxExecutionTimeSeconds: number; + maxRowsToRead: number; + maxMemoryUsage: number; + maxThreads: number; +}; + +const ProjectionParams = z + .object({ + windowStart: z.string(), + windowEnd: z.string(), + }) + .refine(({ windowStart, windowEnd }) => windowStart < windowEnd, { + message: "windowStart must be before windowEnd", + }); + +const projectedColumns = ` + environment_id, + organization_id, + project_id, + triggered_timestamp, + trace_id, + span_id, + run_id, + task_identifier, + start_time, + inserted_at, + message, + error_message, + search_text, + kind, + status, + duration, + parent_span_id`; + +const projectionFingerprint = (alias: string) => `reinterpretAsUInt128(sipHash128( + ${alias}.trace_id, + ${alias}.span_id, + ${alias}.run_id, + ${alias}.start_time +))`; + +const projectionSql = ` +INSERT INTO trigger_dev.task_events_search_v2 +(${projectedColumns}, projection_fingerprint) +SELECT${projectedColumns}, + ${projectionFingerprint("candidate")} AS projection_fingerprint +FROM +( + SELECT + environment_id, + organization_id, + project_id, + fromUnixTimestamp64Nano( + toInt64( + least( + toInt128(toUnixTimestamp64Nano(start_time)) + toInt128(duration), + toInt128( + toUnixTimestamp64Nano( + {windowEnd: DateTime64(3, 'UTC')} + INTERVAL 5 MINUTE + ) + ) + ) + ) + ) AS triggered_timestamp, + trace_id, + span_id, + run_id, + task_identifier, + start_time, + inserted_at, + message, + toValidUTF8( + substring(JSONExtractString(attributes_text, 'error', 'message'), 1, 2045) + ) AS error_message, + toValidUTF8( + substring( + replaceRegexpAll( + replaceRegexpAll( + lowerUTF8( + concat( + toValidUTF8(substring(message, 1, 2045)), + ' ', + replaceAll( + toValidUTF8(substring(attributes_text, 1, 6140)), + '\\\\/', + '/' + ) + ) + ), + '[^\\\\p{L}\\\\p{N}_./:@+-]+', + ' ' + ), + '\\\\s*:\\\\s*', + ':' + ), + 1, + 8189 + ) + ) AS search_text, + kind, + status, + duration, + parent_span_id + FROM trigger_dev.task_events_v2 + WHERE + inserted_at >= {windowStart: DateTime64(3, 'UTC')} + AND inserted_at < {windowEnd: DateTime64(3, 'UTC')} + AND trace_id != '' + AND kind != 'DEBUG_EVENT' + AND status != 'PARTIAL' + AND NOT (kind = 'SPAN_EVENT' AND attributes_text = '{}') + AND kind != 'ANCESTOR_OVERRIDE' + AND message != 'trigger.dev/start' +) AS candidate +ORDER BY + organization_id, + environment_id, + triggered_timestamp, + trace_id, + span_id, + projection_fingerprint +`; + +export function projectTaskEventsSearchV2Window(writer: ClickhouseWriter) { + return async ( + window: TaskEventsSearchV2ProjectionWindow, + limits: TaskEventsSearchV2ProjectionLimits + ): Promise> => { + assertProjectionWindow(window); + const command = writer.command({ + name: "project-task-events-search-v2-window", + query: projectionSql, + params: ProjectionParams, + }); + const settings: ClickHouseSettings = { + async_insert: 0, + max_execution_time: limits.maxExecutionTimeSeconds, + max_rows_to_read: limits.maxRowsToRead.toString(), + max_memory_usage: limits.maxMemoryUsage.toString(), + max_threads: limits.maxThreads, + max_insert_threads: limits.maxThreads.toString(), + use_query_condition_cache: 0, + }; + + return command( + { + windowStart: toClickHouseDateTime64(window.start), + windowEnd: toClickHouseDateTime64(window.end), + }, + { + attributes: { + windowStart: window.start.toISOString(), + windowEnd: window.end.toISOString(), + }, + params: { clickhouse_settings: settings }, + } + ); + }; +} + +function assertProjectionWindow(window: TaskEventsSearchV2ProjectionWindow) { + if ( + !Number.isFinite(window.start.getTime()) || + !Number.isFinite(window.end.getTime()) || + window.start >= window.end + ) { + throw new Error("Invalid task events search projection window"); + } +} + +function toClickHouseDateTime64(value: Date): string { + return value.toISOString().replace("T", " ").replace("Z", ""); +} diff --git a/internal-packages/database/prisma/migrations/20260814070000_add_logs_search_projector_state/migration.sql b/internal-packages/database/prisma/migrations/20260814070000_add_logs_search_projector_state/migration.sql new file mode 100644 index 0000000000..7cffbdc3f1 --- /dev/null +++ b/internal-packages/database/prisma/migrations/20260814070000_add_logs_search_projector_state/migration.sql @@ -0,0 +1,14 @@ +-- CreateTable +CREATE TABLE "public"."LogsSearchProjectorState" ( + "id" TEXT NOT NULL, + "liveWatermark" TIMESTAMP(3) NOT NULL, + "historicalWatermark" TIMESTAMP(3) NOT NULL, + "backfillTarget" TIMESTAMP(3), + "paused" BOOLEAN NOT NULL DEFAULT false, + "leaseToken" TEXT, + "leaseExpiresAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "LogsSearchProjectorState_pkey" PRIMARY KEY ("id") +); diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index 3803e74c7f..051ce6928b 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/internal-packages/database/prisma/schema.prisma @@ -3123,6 +3123,20 @@ model PlatformNotificationInteraction { @@unique([notificationId, userId]) } +model LogsSearchProjectorState { + id String @id + + liveWatermark DateTime + historicalWatermark DateTime + backfillTarget DateTime? + paused Boolean @default(false) + leaseToken String? + leaseExpiresAt DateTime? + + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt +} + enum ErrorGroupStatus { UNRESOLVED RESOLVED