diff --git a/.server-changes/sentry-request-attribution.md b/.server-changes/sentry-request-attribution.md new file mode 100644 index 00000000000..d42abf78d09 --- /dev/null +++ b/.server-changes/sentry-request-attribution.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Fixed error reports being attributed to the wrong request when several requests were in flight at once. diff --git a/.server-changes/tsql-query-error-log-levels.md b/.server-changes/tsql-query-error-log-levels.md new file mode 100644 index 00000000000..1aaedba3b76 --- /dev/null +++ b/.server-changes/tsql-query-error-log-levels.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Invalid queries sent to the query API are no longer treated as internal errors, and a query that does fail is now recorded together with the query text that produced it. diff --git a/apps/webapp/app/v3/tracer.server.ts b/apps/webapp/app/v3/tracer.server.ts index 3b924ff8a19..87e0877091f 100644 --- a/apps/webapp/app/v3/tracer.server.ts +++ b/apps/webapp/app/v3/tracer.server.ts @@ -1,6 +1,7 @@ import { type Attributes, type Context, + context as otelContext, createContextKey, DiagConsoleLogger, DiagLogLevel, @@ -14,6 +15,7 @@ import { metrics, type Meter, } from "@opentelemetry/api"; +import sentryRemix from "@sentry/remix"; import { logs, SeverityNumber } from "@opentelemetry/api-logs"; import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http"; import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http"; @@ -209,10 +211,32 @@ function getResource() { return baseResource.merge(detectedResource); } +/** + * Sentry's `withIsolationScope` only marks the OTel context; the fork itself is + * done by Sentry's context manager. We pass `skipOpenTelemetrySetup: true` to + * `Sentry.init` because we run our own OTel pipeline, which also skips the + * `setGlobalContextManager(new SentryContextManager())` that Sentry would + * otherwise do. Registering it here is what keeps per-request scopes (and so + * the request attributed to each Sentry event) from leaking between concurrent + * requests. It extends `AsyncLocalStorageContextManager`, so OTel behaviour is + * unchanged. + * + * Reached through the default export because `@sentry/remix` is CommonJS and + * Node's ESM loader does not detect this transitively re-exported name, so a + * named import resolves at build time and then fails when the server boots. + */ +function createContextManager() { + return new sentryRemix.SentryContextManager(); +} + function setupTelemetry() { if (env.INTERNAL_OTEL_TRACE_DISABLED === "1") { console.log(`🔦 Tracer disabled, returning a noop tracer`); + const contextManager = createContextManager(); + contextManager.enable(); + otelContext.setGlobalContextManager(contextManager); + return { tracer: trace.getTracer("trigger.dev", "3.3.12"), logger: logs.getLogger("trigger.dev", "3.3.12"), @@ -300,7 +324,7 @@ function setupTelemetry() { ); } - provider.register(); + provider.register({ contextManager: createContextManager() }); let instrumentations: Instrumentation[] = [ new AwsSdkInstrumentation({ diff --git a/apps/webapp/test/sentryRequestIsolation.test.ts b/apps/webapp/test/sentryRequestIsolation.test.ts new file mode 100644 index 00000000000..ebcb0af6929 --- /dev/null +++ b/apps/webapp/test/sentryRequestIsolation.test.ts @@ -0,0 +1,51 @@ +import { context } from "@opentelemetry/api"; +import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; +import * as Sentry from "@sentry/remix"; +import sentryRemix from "@sentry/remix"; +import { afterEach, beforeAll, describe, expect, it } from "vitest"; + +/** + * Two overlapping requests, each tagging its own isolation scope, mirroring what + * `SentryHttpInstrumentation` does per incoming request. Returns what each one + * reads back after the other has started. + */ +async function raceTwoRequests(): Promise> { + const observed: Record = {}; + + const handleRequest = (name: string, holdMs: number) => + Sentry.withIsolationScope(async () => { + Sentry.getIsolationScope().setTag("request", name); + await new Promise((resolve) => setTimeout(resolve, holdMs)); + observed[name] = Sentry.getIsolationScope().getScopeData().tags.request; + }); + + await Promise.all([handleRequest("slow", 30), handleRequest("fast", 5)]); + + return observed; +} + +describe("Sentry request isolation", () => { + beforeAll(() => { + Sentry.init({ dsn: undefined, defaultIntegrations: false, skipOpenTelemetrySetup: true }); + }); + + afterEach(() => { + context.disable(); + }); + + it("leaks the isolation scope between concurrent requests without SentryContextManager", async () => { + new NodeTracerProvider().register(); + + const observed = await raceTwoRequests(); + + expect(observed).toEqual({ slow: "fast", fast: "fast" }); + }); + + it("keeps each request's isolation scope separate with SentryContextManager", async () => { + new NodeTracerProvider().register({ contextManager: new sentryRemix.SentryContextManager() }); + + const observed = await raceTwoRequests(); + + expect(observed).toEqual({ slow: "slow", fast: "fast" }); + }); +}); diff --git a/internal-packages/clickhouse/src/client/client.ts b/internal-packages/clickhouse/src/client/client.ts index d91f86b428f..49c594c296e 100644 --- a/internal-packages/clickhouse/src/client/client.ts +++ b/internal-packages/clickhouse/src/client/client.ts @@ -171,13 +171,19 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { ); if (clickhouseError) { - this.logger.error("Error querying clickhouse", { + const errorLogFields = { name: req.name, error: clickhouseError, query: req.query, params, queryId, - }); + }; + + if (isClickhouseQuotaError(clickhouseError)) { + this.logger.warn("Query exceeded a ClickHouse limit", errorLogFields); + } else { + this.logger.error("Error querying clickhouse", errorLogFields); + } recordClickhouseError(span, clickhouseError); @@ -260,6 +266,11 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { * These will be merged with the default settings. */ settings?: ClickHouseSettings; + /** + * Extra fields to attach to the error log if the query fails. Use this to + * record what produced the SQL, e.g. the TSQL a caller actually wrote. + */ + logFields?: Record; }): ClickhouseQueryWithStatsFunction, z.output> { return async (params, options) => { const queryId = randomUUID(); @@ -320,13 +331,20 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { ); if (clickhouseError) { - this.logger.error("Error querying clickhouse", { + const errorLogFields = { + ...req.logFields, name: req.name, error: clickhouseError, query: req.query, params, queryId, - }); + }; + + if (isClickhouseQuotaError(clickhouseError)) { + this.logger.warn("Query exceeded a ClickHouse limit", errorLogFields); + } else { + this.logger.error("Error querying clickhouse", errorLogFields); + } recordClickhouseError(span, clickhouseError); @@ -453,13 +471,19 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { ); if (clickhouseError) { - this.logger.error("Error querying clickhouse", { + const errorLogFields = { name: req.name, error: clickhouseError, query: req.query, params, queryId, - }); + }; + + if (isClickhouseQuotaError(clickhouseError)) { + this.logger.warn("Query exceeded a ClickHouse limit", errorLogFields); + } else { + this.logger.error("Error querying clickhouse", errorLogFields); + } recordClickhouseError(span, clickhouseError); @@ -599,13 +623,19 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { span.setAttributes({ "clickhouse.rows": rowCount }); } catch (error) { - self.logger.error("Error streaming clickhouse", { + const errorLogFields = { name: req.name, error, query: req.query, params, queryId, - }); + }; + + if (error instanceof Error && isClickhouseQuotaError(error)) { + self.logger.warn("Streamed query exceeded a ClickHouse limit", errorLogFields); + } else { + self.logger.error("Error streaming clickhouse", errorLogFields); + } if (error instanceof Error) { recordClickhouseError(span, error); @@ -1001,6 +1031,29 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter { } } +/** + * ClickHouse error types raised by a query that is valid but asks for more than + * the caller is allowed to spend. The caller gets a 4xx and there is nothing on + * our side to fix, so these are logged at warn rather than error. + */ +const CLICKHOUSE_QUOTA_ERROR_TYPES = new Set([ + "MEMORY_LIMIT_EXCEEDED", + "TIMEOUT_EXCEEDED", + "TOO_SLOW", + "TOO_MANY_ROWS", + "TOO_MANY_BYTES", + "TOO_MANY_ROWS_OR_BYTES", + "QUERY_WAS_CANCELLED", +]); + +function isClickhouseQuotaError(error: Error): boolean { + return ( + error instanceof ClickHouseError && + error.type !== undefined && + CLICKHOUSE_QUOTA_ERROR_TYPES.has(error.type) + ); +} + function recordClickhouseError(span: Span, error: Error): void { if (error instanceof ClickHouseError) { span.setAttributes({ diff --git a/internal-packages/clickhouse/src/client/tsql.ts b/internal-packages/clickhouse/src/client/tsql.ts index c712820812f..36daa636463 100644 --- a/internal-packages/clickhouse/src/client/tsql.ts +++ b/internal-packages/clickhouse/src/client/tsql.ts @@ -8,6 +8,7 @@ import type { ClickHouseSettings } from "@clickhouse/client"; import { compileTSQL, + ExposedTSQLError, type OutputColumnMetadata, sanitizeErrorMessage, transformResults, @@ -207,6 +208,7 @@ export async function executeTSQL( // EXPLAIN returns rows with an 'explain' column schema: isExplain ? z.object({ explain: z.string() }) : options.schema, settings: options.clickhouseSettings, + logFields: { tsql: options.query }, }); const [error, result] = await queryFn(params); @@ -240,6 +242,7 @@ export async function executeTSQL( params: z.record(z.any()), schema: z.object({ explain: z.string() }), settings: options.clickhouseSettings, + logFields: { tsql: options.query }, }); const [additionalError, additionalResult] = await additionalQueryFn(params); @@ -297,14 +300,21 @@ export async function executeTSQL( } catch (error) { const errorMessage = error instanceof Error ? error.message : "Unknown error"; - // Log TSQL compilation or unexpected errors (with original message for debugging) - logger.error("[TSQL] Query error", { + const logFields = { name: options.name, error: errorMessage, tsql: options.query, generatedSql: generatedSql ?? "(compilation failed)", generatedParams: generatedParams ?? {}, - }); + }; + + const callerWroteABadQuery = error instanceof ExposedTSQLError; + + if (callerWroteABadQuery) { + logger.warn("[TSQL] Invalid query", logFields); + } else { + logger.error("[TSQL] Query error", logFields); + } // Sanitize error message to show TSQL names instead of ClickHouse internals const sanitizedMessage = sanitizeErrorMessage(errorMessage, options.tableSchema); diff --git a/internal-packages/clickhouse/src/client/types.ts b/internal-packages/clickhouse/src/client/types.ts index d785ab34ebb..5c674776370 100644 --- a/internal-packages/clickhouse/src/client/types.ts +++ b/internal-packages/clickhouse/src/client/types.ts @@ -135,6 +135,11 @@ export interface ClickhouseReader { * These will be merged with the default settings. */ settings?: ClickHouseSettings; + /** + * Extra fields to attach to the error log if the query fails. Use this to + * record what produced the SQL, e.g. the TSQL a caller actually wrote. + */ + logFields?: Record; }): ClickhouseQueryWithStatsFunction, z.output>; queryFast, TParams extends Record>(req: { diff --git a/internal-packages/clickhouse/src/tsql.test.ts b/internal-packages/clickhouse/src/tsql.test.ts index b0a93bfe704..937b2a2dcae 100644 --- a/internal-packages/clickhouse/src/tsql.test.ts +++ b/internal-packages/clickhouse/src/tsql.test.ts @@ -1,4 +1,5 @@ import { clickhouseTest } from "@internal/testcontainers"; +import type { MockInstance } from "vitest"; import { z } from "zod"; import { ClickhouseClient } from "./client/client.js"; import { executeTSQL, createTSQLExecutor, type TableSchema } from "./client/tsql.js"; @@ -1609,3 +1610,115 @@ describe("Field Mapping Tests", () => { expect(result?.rows?.map((r) => r.run_id).sort()).toEqual(["run_fm_in1", "run_fm_in2"]); }); }); + +describe("TSQL Error Log Levels", () => { + let warnSpy: MockInstance; + let errorSpy: MockInstance; + + beforeEach(() => { + warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + function logged(spy: MockInstance): string { + return spy.mock.calls.map(([line]) => String(line)).join("\n"); + } + + clickhouseTest("logs an unknown column as a warning", async ({ clickhouseContainer }) => { + const client = new ClickhouseClient({ + name: "test", + url: clickhouseContainer.getConnectionUrl(), + }); + + const [error] = await executeTSQL(client, { + name: "test-unknown-column", + query: "SELECT nope FROM task_runs", + schema: z.object({ nope: z.string() }), + enforcedWhereClause: { + organization_id: { op: "eq", value: "org_tenant1" }, + }, + tableSchema: [taskRunsSchema], + }); + + expect(error).not.toBeNull(); + expect(logged(warnSpy)).toContain("[TSQL] Invalid query"); + expect(logged(errorSpy)).not.toContain("[TSQL] Query error"); + }); + + clickhouseTest("logs a syntax error as a warning", async ({ clickhouseContainer }) => { + const client = new ClickhouseClient({ + name: "test", + url: clickhouseContainer.getConnectionUrl(), + }); + + const [error] = await executeTSQL(client, { + name: "test-syntax-error", + query: "SELECT FROM WHERE", + schema: z.object({ run_id: z.string() }), + enforcedWhereClause: { + organization_id: { op: "eq", value: "org_tenant1" }, + }, + tableSchema: [taskRunsSchema], + }); + + expect(error).not.toBeNull(); + expect(logged(warnSpy)).toContain("[TSQL] Invalid query"); + expect(logged(errorSpy)).not.toContain("[TSQL] Query error"); + }); + + clickhouseTest( + "logs a query ClickHouse rejects at execution as an error, with the TSQL that produced it", + async ({ clickhouseContainer }) => { + const client = new ClickhouseClient({ + name: "test", + url: clickhouseContainer.getConnectionUrl(), + }); + + const [error] = await executeTSQL(client, { + name: "test-execution-error", + query: "SELECT toDateTime(tags) AS bad FROM task_runs", + schema: z.object({ bad: z.string() }), + enforcedWhereClause: { + organization_id: { op: "eq", value: "org_tenant1" }, + }, + tableSchema: [taskRunsSchema], + }); + + expect(error).not.toBeNull(); + expect(logged(errorSpy)).toContain("Error querying clickhouse"); + expect(logged(errorSpy)).toContain("SELECT toDateTime(tags) AS bad FROM task_runs"); + } + ); + + clickhouseTest("logs a ClickHouse limit breach as a warning", async ({ clickhouseContainer }) => { + const client = new ClickhouseClient({ + name: "test", + url: clickhouseContainer.getConnectionUrl(), + }); + + await insertTaskRuns(client, { async_insert: 0 })([ + createTaskRun({ run_id: "run_limit1" }), + createTaskRun({ run_id: "run_limit2" }), + createTaskRun({ run_id: "run_limit3" }), + ]); + + const [error] = await executeTSQL(client, { + name: "test-resource-limit", + query: "SELECT run_id FROM task_runs", + schema: z.object({ run_id: z.string() }), + enforcedWhereClause: { + organization_id: { op: "eq", value: "org_tenant1" }, + }, + tableSchema: [taskRunsSchema], + clickhouseSettings: { max_rows_to_read: "1" }, + }); + + expect(error).not.toBeNull(); + expect(logged(warnSpy)).toContain("Query exceeded a ClickHouse limit"); + expect(logged(errorSpy)).not.toContain("Error querying clickhouse"); + }); +});