diff --git a/apps/webapp/app/routes/api.v1.runs.ts b/apps/webapp/app/routes/api.v1.runs.ts
index dca246a0c24..a9f2b8b4d2c 100644
--- a/apps/webapp/app/routes/api.v1.runs.ts
+++ b/apps/webapp/app/routes/api.v1.runs.ts
@@ -8,6 +8,7 @@ import {
createLoaderApiRoute,
everyResource,
} from "~/services/routeBuilders/apiBuilder.server";
+import { RunsListQueryError } from "~/services/runsRepository/runsRepository.server";
export const loader = createLoaderApiRoute(
{
@@ -40,13 +41,23 @@ export const loader = createLoaderApiRoute(
},
async ({ searchParams, authentication, apiVersion }) => {
const presenter = new ApiRunListPresenter();
- const result = await presenter.call(
- authentication.environment.project,
- searchParams,
- apiVersion,
- authentication.environment
- );
+ try {
+ const result = await presenter.call(
+ authentication.environment.project,
+ searchParams,
+ apiVersion,
+ authentication.environment
+ );
- return json(result);
+ return json(result);
+ } catch (error) {
+ if (error instanceof RunsListQueryError) {
+ return json(
+ { error: error.message },
+ { status: error.status, headers: { "x-should-retry": "false" } }
+ );
+ }
+ throw error;
+ }
}
);
diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts
index 4981dd19c43..02b3e014466 100644
--- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts
+++ b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts
@@ -1,4 +1,4 @@
-import { type ClickhouseQueryBuilder } from "@internal/clickhouse";
+import { type ClickhouseQueryBuilder, isClickhouseResourceLimitError } from "@internal/clickhouse";
import { ErrorId, RunId } from "@trigger.dev/core/v3/isomorphic";
import {
type FilterRunsOptions,
@@ -10,6 +10,7 @@ import {
type RunsRepositoryOptions,
type TagListOptions,
convertRunListInputOptionsToFilterRunsOptions,
+ RunsListQueryError,
} from "./runsRepository.server";
import parseDuration from "parse-duration";
import { decodeRunsCursor, encodeRunsCursor } from "./runsCursor.server";
@@ -19,6 +20,18 @@ import { type PrismaClientOrTransaction } from "~/db.server";
import { boundedIn, type Prisma } from "@trigger.dev/database";
type RunCursorRow = { runId: string; createdAt: number };
+/**
+ * Re-throws a runs-list query error, converting a ClickHouse resource-limit rejection (execution
+ * time or memory) into a typed {@link RunsListQueryError} so callers can surface an actionable 4xx
+ * instead of an opaque 500. Any other error is re-thrown unchanged.
+ */
+function rethrowRunsListQueryError(queryError: unknown): never {
+ if (isClickhouseResourceLimitError(queryError)) {
+ throw new RunsListQueryError(undefined, { cause: queryError });
+ }
+ throw queryError;
+}
+
/**
* Default hydrate select for the runs list, used when a caller does not derive
* one from the visible columns (bulk actions, the live poll). Kept in sync with
@@ -102,7 +115,7 @@ export class ClickHouseRunsRepository implements IRunsRepository {
const [queryError, result] = await queryBuilder.execute();
if (queryError) {
- throw queryError;
+ rethrowRunsListQueryError(queryError);
}
return (result?.length ?? 0) > 0;
@@ -166,7 +179,7 @@ export class ClickHouseRunsRepository implements IRunsRepository {
const [queryError, result] = await queryBuilder.execute();
if (queryError) {
- throw queryError;
+ rethrowRunsListQueryError(queryError);
}
return result.map((row) => ({ runId: row.run_id, createdAt: row.created_at_ms }));
@@ -349,7 +362,7 @@ export class ClickHouseRunsRepository implements IRunsRepository {
const [queryError, result] = await queryBuilder.execute();
if (queryError) {
- throw queryError;
+ rethrowRunsListQueryError(queryError);
}
if (result.length === 0) {
@@ -402,7 +415,7 @@ export class ClickHouseRunsRepository implements IRunsRepository {
const [queryError, result] = await queryBuilder.execute();
if (queryError) {
- throw queryError;
+ rethrowRunsListQueryError(queryError);
}
return {
diff --git a/apps/webapp/app/services/runsRepository/runsRepository.server.ts b/apps/webapp/app/services/runsRepository/runsRepository.server.ts
index 0b1049125dd..1aeeb96dbc1 100644
--- a/apps/webapp/app/services/runsRepository/runsRepository.server.ts
+++ b/apps/webapp/app/services/runsRepository/runsRepository.server.ts
@@ -13,6 +13,33 @@ import { runStore as defaultRunStore } from "~/v3/runStore.server";
import { startActiveSpan } from "~/v3/tracer.server";
import { ClickHouseRunsRepository } from "./clickhouseRunsRepository.server";
+/**
+ * User-facing message when a runs-list query exceeds a ClickHouse resource limit. It tells the
+ * caller how to recover (a narrower time range restores partition pruning), and is safe to show
+ * on the dashboard and return from the public API.
+ */
+const RUNS_LIST_QUERY_TOO_EXPENSIVE_MESSAGE =
+ "This query was too expensive to run over the selected time range. Narrow the time window (a shorter period, or a smaller createdAt from/to range) and try again.";
+
+/**
+ * Thrown when a runs-list ClickHouse query hits a server-side resource limit (execution time or
+ * memory). It is the caller's query being too broad, not a service fault, so it carries a 4xx
+ * status and a recovery message rather than surfacing as a 500.
+ */
+export class RunsListQueryError extends Error {
+ public readonly name = "RunsListQueryError";
+ public readonly status = 422;
+ constructor(
+ message: string = RUNS_LIST_QUERY_TOO_EXPENSIVE_MESSAGE,
+ options?: { cause?: unknown }
+ ) {
+ super(message);
+ if (options?.cause !== undefined) {
+ this.cause = options.cause;
+ }
+ }
+}
+
export type RunsRepositoryOptions = {
clickhouse: ClickHouse;
prisma: PrismaClientOrTransaction;
diff --git a/apps/webapp/test/clickhouseQueryMetrics.test.ts b/apps/webapp/test/clickhouseQueryMetrics.test.ts
new file mode 100644
index 00000000000..29e4bc85835
--- /dev/null
+++ b/apps/webapp/test/clickhouseQueryMetrics.test.ts
@@ -0,0 +1,93 @@
+import { ClickHouse } from "@internal/clickhouse";
+import { containerTest } from "@internal/testcontainers";
+import { describe, expect, vi } from "vitest";
+import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
+import {
+ createRun,
+ insertTaskRunV2Rows,
+ seedParents,
+} from "./helpers/apiRunListPresenterTestHelpers";
+import { createInMemoryMetrics } from "./utils/tracing";
+import { histogramCount, latestMetrics, metricSum } from "./otlpMetrics.helpers";
+
+vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
+
+vi.setConfig({ testTimeout: 90_000 });
+
+describe("clickhouse query metrics", () => {
+ containerTest(
+ "records duration + read_rows on success and an error metric with the ClickHouse error type",
+ async ({ clickhouseContainer, prisma }) => {
+ const ctx = await seedParents(prisma, "chm");
+ const run = await createRun(prisma, ctx, { friendlyId: "run_chm" });
+
+ const seedClient = new ClickHouse({
+ url: clickhouseContainer.getConnectionUrl(),
+ name: "clickhouse-metrics-seed",
+ });
+ await insertTaskRunV2Rows(seedClient, [{ ...run, createdAt: new Date() }]);
+
+ const listArgs = {
+ page: { size: 10 } as const,
+ organizationId: ctx.organizationId,
+ projectId: ctx.projectId,
+ environmentId: ctx.environmentId,
+ };
+
+ const okMetrics = createInMemoryMetrics();
+ const okClient = new ClickHouse({
+ url: clickhouseContainer.getConnectionUrl(),
+ name: "clickhouse-metrics-ok",
+ meter: okMetrics.meter,
+ });
+ const okRepo = new RunsRepository({ prisma, clickhouse: okClient });
+ const result = await okRepo.listRuns(listArgs);
+ expect(result.runs.map((r) => r.friendlyId)).toEqual(["run_chm"]);
+
+ await vi.waitFor(
+ async () => {
+ const rm = await latestMetrics(okMetrics);
+ expect(
+ histogramCount(rm, "clickhouse.query.duration", {
+ client: "clickhouse-metrics-ok",
+ status: "ok",
+ })
+ ).toBeGreaterThanOrEqual(1);
+ },
+ { timeout: 5000, interval: 50 }
+ );
+ const okRm = await latestMetrics(okMetrics);
+ expect(
+ histogramCount(okRm, "clickhouse.query.read_rows", { client: "clickhouse-metrics-ok" })
+ ).toBeGreaterThanOrEqual(1);
+ expect(
+ histogramCount(okRm, "clickhouse.query.memory_usage", { client: "clickhouse-metrics-ok" })
+ ).toBeGreaterThanOrEqual(1);
+ await okMetrics.shutdown();
+
+ const errMetrics = createInMemoryMetrics();
+ const cappedClient = new ClickHouse({
+ url: clickhouseContainer.getConnectionUrl(),
+ name: "clickhouse-metrics-capped",
+ clickhouseSettings: { max_memory_usage: "1" },
+ meter: errMetrics.meter,
+ });
+ const errRepo = new RunsRepository({ prisma, clickhouse: cappedClient });
+ await expect(errRepo.listRuns(listArgs)).rejects.toThrow();
+
+ await vi.waitFor(
+ async () => {
+ const rm = await latestMetrics(errMetrics);
+ expect(
+ metricSum(rm, "clickhouse.query.errors", {
+ client: "clickhouse-metrics-capped",
+ error_type: "MEMORY_LIMIT_EXCEEDED",
+ })
+ ).toBeGreaterThanOrEqual(1);
+ },
+ { timeout: 5000, interval: 50 }
+ );
+ await errMetrics.shutdown();
+ }
+ );
+});
diff --git a/apps/webapp/test/runsListQueryError.test.ts b/apps/webapp/test/runsListQueryError.test.ts
new file mode 100644
index 00000000000..89e0486bf72
--- /dev/null
+++ b/apps/webapp/test/runsListQueryError.test.ts
@@ -0,0 +1,52 @@
+import { ClickHouse } from "@internal/clickhouse";
+import { containerTest } from "@internal/testcontainers";
+import { describe, expect, vi } from "vitest";
+import {
+ RunsListQueryError,
+ RunsRepository,
+} from "~/services/runsRepository/runsRepository.server";
+import {
+ createRun,
+ insertTaskRunV2Rows,
+ seedParents,
+} from "./helpers/apiRunListPresenterTestHelpers";
+
+vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
+
+vi.setConfig({ testTimeout: 90_000 });
+
+describe("runs list query error handling", () => {
+ containerTest(
+ "a ClickHouse resource-limit error surfaces as RunsListQueryError",
+ async ({ clickhouseContainer, prisma }) => {
+ const ctx = await seedParents(prisma, "qerr");
+ const run = await createRun(prisma, ctx, { friendlyId: "run_qerr" });
+
+ const seedClient = new ClickHouse({
+ url: clickhouseContainer.getConnectionUrl(),
+ name: "runs-list-query-error-seed",
+ });
+ await insertTaskRunV2Rows(seedClient, [{ ...run, createdAt: new Date() }]);
+
+ const listArgs = {
+ page: { size: 10 } as const,
+ organizationId: ctx.organizationId,
+ projectId: ctx.projectId,
+ environmentId: ctx.environmentId,
+ };
+
+ const cappedClient = new ClickHouse({
+ url: clickhouseContainer.getConnectionUrl(),
+ name: "runs-list-query-error-capped",
+ clickhouseSettings: { max_memory_usage: "1" },
+ });
+ const capped = new RunsRepository({ prisma, clickhouse: cappedClient });
+ await expect(capped.listRuns(listArgs)).rejects.toBeInstanceOf(RunsListQueryError);
+ await expect(capped.countRuns(listArgs)).rejects.toBeInstanceOf(RunsListQueryError);
+
+ const ok = new RunsRepository({ prisma, clickhouse: seedClient });
+ const result = await ok.listRuns(listArgs);
+ expect(result.runs.map((r) => r.friendlyId)).toEqual(["run_qerr"]);
+ }
+ );
+});
diff --git a/internal-packages/clickhouse/src/client/client.ts b/internal-packages/clickhouse/src/client/client.ts
index 9949081d504..96db1f4a529 100644
--- a/internal-packages/clickhouse/src/client/client.ts
+++ b/internal-packages/clickhouse/src/client/client.ts
@@ -7,8 +7,8 @@ import {
type BaseQueryParams,
type InsertResult,
} from "@clickhouse/client";
-import type { Span, Tracer } from "@internal/tracing";
-import { recordSpanError, startSpan, trace } from "@internal/tracing";
+import type { Counter, Histogram, Meter, Span, Tracer, UpDownCounter } from "@internal/tracing";
+import { getMeter, recordSpanError, startSpan, trace } from "@internal/tracing";
import { flattenAttributes, tryCatch, type Result } from "@trigger.dev/core/v3";
import { z } from "zod";
import { InsertError, QueryError } from "./errors.js";
@@ -43,6 +43,7 @@ export type ClickhouseConfig = {
httpAgent?: HttpAgent | HttpsAgent;
clickhouseSettings?: ClickHouseSettings;
logger?: Logger;
+ meter?: Meter;
maxOpenConnections?: number;
requestTimeoutMs?: number;
logLevel?: LogLevel;
@@ -57,11 +58,49 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
private readonly tracer: Tracer;
private readonly name: string;
private readonly logger: Logger;
+ private readonly meter: Meter;
+ private readonly queryInFlight: UpDownCounter;
+ private readonly queryDuration: Histogram;
+ private readonly queryServerDuration: Histogram;
+ private readonly queryReadRows: Histogram;
+ private readonly queryReadBytes: Histogram;
+ private readonly queryMemoryUsage: Histogram;
+ private readonly queryErrors: Counter;
constructor(config: ClickhouseConfig) {
this.name = config.name;
this.logger = config.logger ?? new Logger("ClickhouseClient", config.logLevel ?? "info");
+ this.meter = config.meter ?? getMeter("clickhouse");
+ this.queryInFlight = this.meter.createUpDownCounter("clickhouse.query.in_flight", {
+ description: "Concurrent in-flight ClickHouse queries per client, a pool-saturation signal",
+ });
+ this.queryDuration = this.meter.createHistogram("clickhouse.query.duration", {
+ description:
+ "Wall-clock ClickHouse query duration, includes client-side connection-pool wait",
+ unit: "ms",
+ });
+ this.queryServerDuration = this.meter.createHistogram("clickhouse.query.server_duration", {
+ description: "Server-side ClickHouse query duration from the x-clickhouse-summary elapsed_ns",
+ unit: "ms",
+ });
+ this.queryReadRows = this.meter.createHistogram("clickhouse.query.read_rows", {
+ description: "Rows read by a ClickHouse query, from the x-clickhouse-summary header",
+ unit: "{row}",
+ });
+ this.queryReadBytes = this.meter.createHistogram("clickhouse.query.read_bytes", {
+ description: "Bytes read by a ClickHouse query, from the x-clickhouse-summary header",
+ unit: "By",
+ });
+ this.queryMemoryUsage = this.meter.createHistogram("clickhouse.query.memory_usage", {
+ description: "Peak memory used by a ClickHouse query, from the x-clickhouse-summary header",
+ unit: "By",
+ });
+ this.queryErrors = this.meter.createCounter("clickhouse.query.errors", {
+ description:
+ "ClickHouse query errors by type, e.g. MEMORY_LIMIT_EXCEEDED or TIMEOUT_EXCEEDED",
+ });
+
this.client = createClient({
url: config.url,
keep_alive: config.keepAlive,
@@ -87,6 +126,40 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
await this.client.close();
}
+ private recordQueryMetrics(
+ operation: string,
+ startedAt: number,
+ result: { errorType?: string; summary?: Record
}
+ ) {
+ const attributes = { client: this.name, operation };
+ this.queryDuration.record(performance.now() - startedAt, {
+ ...attributes,
+ status: result.errorType ? "error" : "ok",
+ });
+ if (result.errorType) {
+ this.queryErrors.add(1, { ...attributes, error_type: result.errorType });
+ }
+ const summary = result.summary;
+ if (summary) {
+ const elapsedNs = Number(summary.elapsed_ns);
+ if (Number.isFinite(elapsedNs) && elapsedNs > 0) {
+ this.queryServerDuration.record(elapsedNs / 1_000_000, attributes);
+ }
+ const readRows = Number(summary.read_rows);
+ if (Number.isFinite(readRows)) {
+ this.queryReadRows.record(readRows, attributes);
+ }
+ const readBytes = Number(summary.read_bytes);
+ if (Number.isFinite(readBytes)) {
+ this.queryReadBytes.record(readBytes, attributes);
+ }
+ const memoryUsage = Number(summary.memory_usage);
+ if (Number.isFinite(memoryUsage) && memoryUsage > 0) {
+ this.queryMemoryUsage.record(memoryUsage, attributes);
+ }
+ }
+ }
+
public query, TOut extends z.ZodSchema>(req: {
/**
* The name of the operation.
@@ -117,124 +190,147 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
}): ClickhouseQueryFunction, z.output> {
return async (params, options) => {
const queryId = randomUUID();
-
- return await startSpan(this.tracer, "query", async (span) => {
- this.logger.debug("Querying clickhouse", {
- name: req.name,
- query: req.query.replace(/\s+/g, " "),
- params,
- settings: req.settings,
- attributes: options?.attributes,
- queryId,
- });
-
- 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);
-
- this.logger.error("Error parsing query params", {
+ const startedAt = performance.now();
+ this.queryInFlight.add(1, { client: this.name });
+ let summary: Record | undefined;
+
+ const result = await startSpan(
+ this.tracer,
+ "query",
+ async (span): Promise[], QueryError>> => {
+ this.logger.debug("Querying clickhouse", {
name: req.name,
- error: validParams.error,
- query: req.query,
+ query: req.query.replace(/\s+/g, " "),
params,
+ settings: req.settings,
+ attributes: options?.attributes,
queryId,
});
- return [
- new QueryError(`Bad params: ${generateErrorMessage(validParams.error.issues)}`, {
- query: req.query,
- }),
- null,
- ];
- }
-
- let unparsedRows: Array = [];
-
- const [clickhouseError, res] = await tryCatch(
- this.client.query({
- query: req.query,
- query_params: validParams?.data,
- format: "JSONEachRow",
- query_id: queryId,
- ...options?.params,
- clickhouse_settings: {
- ...req.settings,
- ...options?.params?.clickhouse_settings,
- },
- })
- );
-
- if (clickhouseError) {
- const errorLogFields = {
- name: req.name,
- error: clickhouseError,
- query: req.query,
- params,
- queryId,
- };
+ span.setAttributes({
+ "clickhouse.clientName": this.name,
+ "clickhouse.operationName": req.name,
+ "clickhouse.queryId": queryId,
+ ...flattenAttributes(req.settings, "clickhouse.settings"),
+ ...flattenAttributes(options?.attributes),
+ });
- this.logger.error("Error querying clickhouse", errorLogFields);
+ const validParams = req.params?.safeParse(params);
- recordClickhouseError(span, clickhouseError);
+ if (validParams?.error) {
+ recordSpanError(span, validParams.error);
- return [
- new QueryError(`Unable to query clickhouse: ${clickhouseError.message}`, {
+ this.logger.error("Error parsing query params", {
+ name: req.name,
+ error: validParams.error,
query: req.query,
- }),
- null,
- ];
- }
+ params,
+ queryId,
+ });
+
+ return [
+ new QueryError(`Bad params: ${generateErrorMessage(validParams.error.issues)}`, {
+ query: req.query,
+ }),
+ null,
+ ];
+ }
- unparsedRows = await res.json();
+ let unparsedRows: Array = [];
- span.setAttributes({
- "clickhouse.query_id": res.query_id,
- ...flattenAttributes(res.response_headers, "clickhouse.response_headers"),
- });
+ const [clickhouseError, res] = await tryCatch(
+ this.client.query({
+ query: req.query,
+ query_params: validParams?.data,
+ format: "JSONEachRow",
+ query_id: queryId,
+ ...options?.params,
+ clickhouse_settings: {
+ ...req.settings,
+ ...options?.params?.clickhouse_settings,
+ },
+ })
+ );
+
+ if (clickhouseError) {
+ const errorLogFields = {
+ name: req.name,
+ error: clickhouseError,
+ query: req.query,
+ params,
+ queryId,
+ };
+
+ this.logger.error("Error querying clickhouse", errorLogFields);
+
+ recordClickhouseError(span, clickhouseError);
+
+ return [
+ new QueryError(
+ `Unable to query clickhouse: ${clickhouseError.message}`,
+ { query: req.query },
+ clickhouseError instanceof ClickHouseError ? clickhouseError.type : undefined
+ ),
+ null,
+ ];
+ }
- const summaryHeader = res.response_headers["x-clickhouse-summary"];
+ unparsedRows = await res.json();
- if (typeof summaryHeader === "string") {
span.setAttributes({
- ...flattenAttributes(JSON.parse(summaryHeader), "clickhouse.summary"),
+ "clickhouse.query_id": res.query_id,
+ ...flattenAttributes(res.response_headers, "clickhouse.response_headers"),
});
- }
- const parsed = z.array(req.schema).safeParse(unparsedRows);
+ const summaryHeader = res.response_headers["x-clickhouse-summary"];
- if (parsed.error) {
- this.logger.error("Error parsing clickhouse query result", {
- name: req.name,
- error: parsed.error,
- query: req.query,
- params,
- queryId,
- });
+ if (typeof summaryHeader === "string") {
+ summary = JSON.parse(summaryHeader);
+ span.setAttributes({
+ ...flattenAttributes(summary, "clickhouse.summary"),
+ });
+ }
- const queryError = new QueryError(generateErrorMessage(parsed.error.issues), {
- query: req.query,
- });
+ const parsed = z.array(req.schema).safeParse(unparsedRows);
- recordSpanError(span, queryError);
+ if (parsed.error) {
+ this.logger.error("Error parsing clickhouse query result", {
+ name: req.name,
+ error: parsed.error,
+ query: req.query,
+ params,
+ queryId,
+ });
- return [queryError, null];
- }
+ const queryError = new QueryError(generateErrorMessage(parsed.error.issues), {
+ query: req.query,
+ });
- span.setAttributes({
- "clickhouse.rows": unparsedRows.length,
- });
+ recordSpanError(span, queryError);
+
+ return [queryError, null];
+ }
- return [null, parsed.data];
+ span.setAttributes({
+ "clickhouse.rows": unparsedRows.length,
+ });
+
+ return [null, parsed.data];
+ }
+ )
+ .catch((error) => {
+ this.recordQueryMetrics(req.name, startedAt, { errorType: "exception" });
+ throw error;
+ })
+ .finally(() => this.queryInFlight.add(-1, { client: this.name }));
+
+ this.recordQueryMetrics(req.name, startedAt, {
+ errorType:
+ result[0] instanceof QueryError ? (result[0].clickhouseErrorType ?? "other") : undefined,
+ summary,
});
+
+ return result;
};
}
@@ -278,163 +374,188 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
}): ClickhouseQueryWithStatsFunction, z.output> {
return async (params, options) => {
const queryId = randomUUID();
-
- return await startSpan(this.tracer, "queryWithStats", async (span) => {
- this.logger.debug("Querying clickhouse with stats", {
- name: req.name,
- query: req.query.replace(/\s+/g, " "),
- params,
- settings: req.settings,
- attributes: options?.attributes,
- queryId,
- });
-
- 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);
-
- this.logger.error("Error parsing query params", {
+ const startedAt = performance.now();
+ this.queryInFlight.add(1, { client: this.name });
+ let summary: Record | undefined;
+
+ const result = await startSpan(
+ this.tracer,
+ "queryWithStats",
+ async (
+ span
+ ): Promise[]; stats: QueryStats }, QueryError>> => {
+ this.logger.debug("Querying clickhouse with stats", {
name: req.name,
- error: validParams.error,
- query: req.query,
+ query: req.query.replace(/\s+/g, " "),
params,
+ settings: req.settings,
+ attributes: options?.attributes,
queryId,
});
- return [
- new QueryError(`Bad params: ${generateErrorMessage(validParams.error.issues)}`, {
- query: req.query,
- }),
- null,
- ];
- }
-
- let unparsedRows: Array = [];
+ span.setAttributes({
+ "clickhouse.clientName": this.name,
+ "clickhouse.operationName": req.name,
+ "clickhouse.queryId": queryId,
+ ...flattenAttributes(req.settings, "clickhouse.settings"),
+ ...flattenAttributes(options?.attributes),
+ });
- const [clickhouseError, res] = await tryCatch(
- this.client.query({
- query: req.query,
- query_params: validParams?.data,
- format: "JSONEachRow",
- query_id: queryId,
- ...options?.params,
- clickhouse_settings: {
- ...req.settings,
- ...options?.params?.clickhouse_settings,
- },
- })
- );
+ const validParams = req.params?.safeParse(params);
- if (clickhouseError) {
- const errorLogFields = {
- ...req.logFields,
- name: req.name,
- error: clickhouseError,
- query: req.query,
- params,
- queryId,
- };
+ if (validParams?.error) {
+ recordSpanError(span, validParams.error);
- switch (classifyClickhouseError(clickhouseError, req.userAuthoredQuery)) {
- case "quota":
- this.logger.warn("Query exceeded a ClickHouse limit", errorLogFields);
- break;
- case "invalid-sql":
- this.logger.warn("ClickHouse rejected an invalid query", errorLogFields);
- break;
- default:
- this.logger.error("Error querying clickhouse", errorLogFields);
+ this.logger.error("Error parsing query params", {
+ name: req.name,
+ error: validParams.error,
+ query: req.query,
+ params,
+ queryId,
+ });
+
+ return [
+ new QueryError(`Bad params: ${generateErrorMessage(validParams.error.issues)}`, {
+ query: req.query,
+ }),
+ null,
+ ];
}
- recordClickhouseError(span, clickhouseError);
+ let unparsedRows: Array = [];
- return [
- new QueryError(`Unable to query clickhouse: ${clickhouseError.message}`, {
+ const [clickhouseError, res] = await tryCatch(
+ this.client.query({
query: req.query,
- }),
- null,
- ];
- }
+ query_params: validParams?.data,
+ format: "JSONEachRow",
+ query_id: queryId,
+ ...options?.params,
+ clickhouse_settings: {
+ ...req.settings,
+ ...options?.params?.clickhouse_settings,
+ },
+ })
+ );
+
+ if (clickhouseError) {
+ const errorLogFields = {
+ ...req.logFields,
+ name: req.name,
+ error: clickhouseError,
+ query: req.query,
+ params,
+ queryId,
+ };
+
+ switch (classifyClickhouseError(clickhouseError, req.userAuthoredQuery)) {
+ case "quota":
+ this.logger.warn("Query exceeded a ClickHouse limit", errorLogFields);
+ break;
+ case "invalid-sql":
+ this.logger.warn("ClickHouse rejected an invalid query", errorLogFields);
+ break;
+ default:
+ this.logger.error("Error querying clickhouse", errorLogFields);
+ }
- unparsedRows = await res.json();
+ recordClickhouseError(span, clickhouseError);
- span.setAttributes({
- "clickhouse.query_id": res.query_id,
- ...flattenAttributes(res.response_headers, "clickhouse.response_headers"),
- });
+ return [
+ new QueryError(
+ `Unable to query clickhouse: ${clickhouseError.message}`,
+ { query: req.query },
+ clickhouseError instanceof ClickHouseError ? clickhouseError.type : undefined
+ ),
+ null,
+ ];
+ }
- // Parse the summary header to get stats
- const summaryHeader = res.response_headers["x-clickhouse-summary"];
- let stats: QueryStats = {
- 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",
- byte_seconds: "0",
- };
+ unparsedRows = await res.json();
- if (typeof summaryHeader === "string") {
- const parsedSummary = JSON.parse(summaryHeader);
- this.logger.debug("parsedSummary", parsedSummary);
- const readBytes = parsedSummary.read_bytes ? parseInt(parsedSummary.read_bytes, 10) : 0;
- const elapsedNs = parsedSummary.elapsed_ns ? parseInt(parsedSummary.elapsed_ns, 10) : 0;
- const elapsedSeconds = elapsedNs / 1_000_000_000;
- const byteSeconds = elapsedSeconds > 0 ? readBytes / elapsedSeconds : 0;
- stats = {
- read_rows: parsedSummary.read_rows ?? "0",
- read_bytes: parsedSummary.read_bytes ?? "0",
- written_rows: parsedSummary.written_rows ?? "0",
- written_bytes: parsedSummary.written_bytes ?? "0",
- total_rows_to_read: parsedSummary.total_rows_to_read ?? "0",
- result_rows: parsedSummary.result_rows ?? "0",
- result_bytes: parsedSummary.result_bytes ?? "0",
- elapsed_ns: parsedSummary.elapsed_ns ?? "0",
- byte_seconds: byteSeconds.toString(),
- };
span.setAttributes({
- ...flattenAttributes(parsedSummary, "clickhouse.summary"),
+ "clickhouse.query_id": res.query_id,
+ ...flattenAttributes(res.response_headers, "clickhouse.response_headers"),
});
- }
- const parsed = z.array(req.schema).safeParse(unparsedRows);
+ // Parse the summary header to get stats
+ const summaryHeader = res.response_headers["x-clickhouse-summary"];
+ let stats: QueryStats = {
+ 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",
+ byte_seconds: "0",
+ };
- if (parsed.error) {
- this.logger.error("Error parsing clickhouse query result", {
- name: req.name,
- error: parsed.error,
- query: req.query,
- params,
- queryId,
- });
+ if (typeof summaryHeader === "string") {
+ const parsedSummary = JSON.parse(summaryHeader);
+ summary = parsedSummary;
+ this.logger.debug("parsedSummary", parsedSummary);
+ const readBytes = parsedSummary.read_bytes ? parseInt(parsedSummary.read_bytes, 10) : 0;
+ const elapsedNs = parsedSummary.elapsed_ns ? parseInt(parsedSummary.elapsed_ns, 10) : 0;
+ const elapsedSeconds = elapsedNs / 1_000_000_000;
+ const byteSeconds = elapsedSeconds > 0 ? readBytes / elapsedSeconds : 0;
+ stats = {
+ read_rows: parsedSummary.read_rows ?? "0",
+ read_bytes: parsedSummary.read_bytes ?? "0",
+ written_rows: parsedSummary.written_rows ?? "0",
+ written_bytes: parsedSummary.written_bytes ?? "0",
+ total_rows_to_read: parsedSummary.total_rows_to_read ?? "0",
+ result_rows: parsedSummary.result_rows ?? "0",
+ result_bytes: parsedSummary.result_bytes ?? "0",
+ elapsed_ns: parsedSummary.elapsed_ns ?? "0",
+ byte_seconds: byteSeconds.toString(),
+ };
+ span.setAttributes({
+ ...flattenAttributes(parsedSummary, "clickhouse.summary"),
+ });
+ }
- const queryError = new QueryError(generateErrorMessage(parsed.error.issues), {
- query: req.query,
- });
+ const parsed = z.array(req.schema).safeParse(unparsedRows);
+
+ if (parsed.error) {
+ this.logger.error("Error parsing clickhouse query result", {
+ name: req.name,
+ error: parsed.error,
+ query: req.query,
+ params,
+ queryId,
+ });
- recordSpanError(span, queryError);
+ const queryError = new QueryError(generateErrorMessage(parsed.error.issues), {
+ query: req.query,
+ });
- return [queryError, null];
- }
+ recordSpanError(span, queryError);
- span.setAttributes({
- "clickhouse.rows": unparsedRows.length,
- });
+ return [queryError, null];
+ }
+
+ span.setAttributes({
+ "clickhouse.rows": unparsedRows.length,
+ });
- return [null, { rows: parsed.data, stats }];
+ return [null, { rows: parsed.data, stats }];
+ }
+ )
+ .catch((error) => {
+ this.recordQueryMetrics(req.name, startedAt, { errorType: "exception" });
+ throw error;
+ })
+ .finally(() => this.queryInFlight.add(-1, { client: this.name }));
+
+ this.recordQueryMetrics(req.name, startedAt, {
+ errorType:
+ result[0] instanceof QueryError ? (result[0].clickhouseErrorType ?? "other") : undefined,
+ summary,
});
+
+ return result;
};
}
@@ -446,103 +567,126 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
}): ClickhouseQueryFunction {
return async (params, options) => {
const queryId = randomUUID();
-
- return await startSpan(this.tracer, "queryFast", async (span) => {
- this.logger.debug("Querying clickhouse fast", {
- name: req.name,
- query: req.query.replace(/\s+/g, " "),
- params,
- settings: req.settings,
- attributes: options?.attributes,
- queryId,
- });
-
- span.setAttributes({
- "clickhouse.clientName": this.name,
- "clickhouse.operationName": req.name,
- "clickhouse.queryId": queryId,
- ...flattenAttributes(req.settings, "clickhouse.settings"),
- ...flattenAttributes(options?.attributes),
- });
-
- const [clickhouseError, resultSet] = await tryCatch(
- this.client.query({
- query: req.query,
- query_params: params,
- format: "JSONCompactEachRow",
- query_id: queryId,
- ...options?.params,
- clickhouse_settings: {
- ...req.settings,
- ...options?.params?.clickhouse_settings,
- },
- })
- );
-
- if (clickhouseError) {
- const errorLogFields = {
+ const startedAt = performance.now();
+ this.queryInFlight.add(1, { client: this.name });
+ let summary: Record | undefined;
+
+ const result = await startSpan(
+ this.tracer,
+ "queryFast",
+ async (span): Promise> => {
+ this.logger.debug("Querying clickhouse fast", {
name: req.name,
- error: clickhouseError,
- query: req.query,
+ query: req.query.replace(/\s+/g, " "),
params,
+ settings: req.settings,
+ attributes: options?.attributes,
queryId,
- };
-
- this.logger.error("Error querying clickhouse", errorLogFields);
+ });
- recordClickhouseError(span, clickhouseError);
+ span.setAttributes({
+ "clickhouse.clientName": this.name,
+ "clickhouse.operationName": req.name,
+ "clickhouse.queryId": queryId,
+ ...flattenAttributes(req.settings, "clickhouse.settings"),
+ ...flattenAttributes(options?.attributes),
+ });
- return [
- new QueryError(`Unable to query clickhouse: ${clickhouseError.message}`, {
+ const [clickhouseError, resultSet] = await tryCatch(
+ this.client.query({
query: req.query,
- }),
- null,
- ];
- }
-
- span.setAttributes({
- "clickhouse.query_id": resultSet.query_id,
- ...flattenAttributes(resultSet.response_headers, "clickhouse.response_headers"),
- });
-
- const summaryHeader = resultSet.response_headers["x-clickhouse-summary"];
+ query_params: params,
+ format: "JSONCompactEachRow",
+ query_id: queryId,
+ ...options?.params,
+ clickhouse_settings: {
+ ...req.settings,
+ ...options?.params?.clickhouse_settings,
+ },
+ })
+ );
+
+ if (clickhouseError) {
+ const errorLogFields = {
+ name: req.name,
+ error: clickhouseError,
+ query: req.query,
+ params,
+ queryId,
+ };
+
+ this.logger.error("Error querying clickhouse", errorLogFields);
+
+ recordClickhouseError(span, clickhouseError);
+
+ return [
+ new QueryError(
+ `Unable to query clickhouse: ${clickhouseError.message}`,
+ { query: req.query },
+ clickhouseError instanceof ClickHouseError ? clickhouseError.type : undefined
+ ),
+ null,
+ ];
+ }
- if (typeof summaryHeader === "string") {
span.setAttributes({
- ...flattenAttributes(JSON.parse(summaryHeader), "clickhouse.summary"),
+ "clickhouse.query_id": resultSet.query_id,
+ ...flattenAttributes(resultSet.response_headers, "clickhouse.response_headers"),
});
- }
- const resultRows: Array = [];
+ const summaryHeader = resultSet.response_headers["x-clickhouse-summary"];
- for await (const rows of resultSet.stream()) {
- if (rows.length === 0) {
- continue;
+ if (typeof summaryHeader === "string") {
+ summary = JSON.parse(summaryHeader);
+ span.setAttributes({
+ ...flattenAttributes(summary, "clickhouse.summary"),
+ });
}
- for (const row of rows) {
- const rowData = row.json() as any[];
+ const resultRows: Array = [];
- const hydratedRow: Record = {};
- for (let i = 0; i < req.columns.length; i++) {
- const column = req.columns[i];
+ for await (const rows of resultSet.stream()) {
+ if (rows.length === 0) {
+ continue;
+ }
- if (typeof column === "string") {
- hydratedRow[column] = rowData[i];
- } else {
- hydratedRow[column.name] = rowData[i];
+ for (const row of rows) {
+ const rowData = row.json() as any[];
+
+ const hydratedRow: Record = {};
+ for (let i = 0; i < req.columns.length; i++) {
+ const column = req.columns[i];
+
+ if (typeof column === "string") {
+ hydratedRow[column] = rowData[i];
+ } else {
+ hydratedRow[column.name] = rowData[i];
+ }
}
+ resultRows.push(hydratedRow as TOut);
}
- resultRows.push(hydratedRow as TOut);
}
- }
- span.setAttributes({
- "clickhouse.rows": resultRows.length,
- });
+ span.setAttributes({
+ "clickhouse.rows": resultRows.length,
+ });
- return [null, resultRows];
+ return [null, resultRows];
+ }
+ )
+ .catch((error) => {
+ this.recordQueryMetrics(req.name, startedAt, { errorType: "exception" });
+ throw error;
+ })
+ .finally(() => this.queryInFlight.add(-1, { client: this.name }));
+
+ this.recordQueryMetrics(req.name, startedAt, {
+ errorType:
+ result[0] instanceof QueryError ? (result[0].clickhouseErrorType ?? "other") : undefined,
+ summary,
});
+
+ return result;
};
}
diff --git a/internal-packages/clickhouse/src/client/errors.ts b/internal-packages/clickhouse/src/client/errors.ts
index ff0be4d0d54..dd4de178055 100644
--- a/internal-packages/clickhouse/src/client/errors.ts
+++ b/internal-packages/clickhouse/src/client/errors.ts
@@ -45,10 +45,39 @@ export class InsertError extends BaseError {
export class QueryError extends BaseError<{ query: string }> {
public readonly retry = true;
public readonly name = QueryError.name;
- constructor(message: string, context: { query: string }) {
+ /**
+ * The underlying ClickHouse error type (e.g. `TIMEOUT_EXCEEDED`) when the failure came from
+ * ClickHouse rejecting the query, else undefined. Lets callers distinguish a query that hit a
+ * server-side resource limit from an unexpected failure.
+ */
+ public readonly clickhouseErrorType?: string;
+ constructor(message: string, context: { query: string }, clickhouseErrorType?: string) {
super({
message,
context,
});
+ this.clickhouseErrorType = clickhouseErrorType;
}
}
+
+/**
+ * ClickHouse error types raised when a query exceeds a server-side resource limit
+ * (`max_execution_time`, `max_memory_usage`, etc.). These mean the caller's query was too
+ * expensive, not a service fault, so callers can turn them into an actionable 4xx.
+ */
+const CLICKHOUSE_RESOURCE_LIMIT_ERROR_TYPES = new Set([
+ "MEMORY_LIMIT_EXCEEDED",
+ "TIMEOUT_EXCEEDED",
+ "TOO_SLOW",
+ "TOO_MANY_ROWS",
+ "TOO_MANY_BYTES",
+ "TOO_MANY_ROWS_OR_BYTES",
+]);
+
+export function isClickhouseResourceLimitError(error: unknown): boolean {
+ return (
+ error instanceof QueryError &&
+ error.clickhouseErrorType !== undefined &&
+ CLICKHOUSE_RESOURCE_LIMIT_ERROR_TYPES.has(error.clickhouseErrorType)
+ );
+}
diff --git a/internal-packages/clickhouse/src/index.ts b/internal-packages/clickhouse/src/index.ts
index 407c33135cc..100101a5635 100644
--- a/internal-packages/clickhouse/src/index.ts
+++ b/internal-packages/clickhouse/src/index.ts
@@ -75,6 +75,7 @@ import {
} from "./errors.js";
export { msToClickHouseInterval } from "./intervals.js";
import { Logger, type LogLevel } from "@trigger.dev/core/logger";
+import type { Meter } from "@internal/tracing";
import type { Agent as HttpAgent } from "http";
import type { Agent as HttpsAgent } from "https";
@@ -123,7 +124,7 @@ export {
export type { ColumnFormatType, OutputColumnMetadata } from "@internal/tsql";
// Errors
-export { QueryError } from "./client/errors.js";
+export { QueryError, isClickhouseResourceLimitError } from "./client/errors.js";
export type ClickhouseCommonConfig = {
keepAlive?: {
@@ -133,6 +134,7 @@ export type ClickhouseCommonConfig = {
httpAgent?: HttpAgent | HttpsAgent;
clickhouseSettings?: ClickHouseSettings;
logger?: Logger;
+ meter?: Meter;
logLevel?: LogLevel;
compression?: {
request?: boolean;
@@ -178,6 +180,7 @@ export class ClickHouse {
url: config.url,
clickhouseSettings: config.clickhouseSettings,
logger: this.logger,
+ meter: config.meter,
logLevel: config.logLevel,
keepAlive: config.keepAlive,
httpAgent: config.httpAgent,
@@ -195,6 +198,7 @@ export class ClickHouse {
url: config.readerUrl,
clickhouseSettings: config.clickhouseSettings,
logger: this.logger,
+ meter: config.meter,
logLevel: config.logLevel,
keepAlive: config.keepAlive,
httpAgent: config.httpAgent,
@@ -207,6 +211,7 @@ export class ClickHouse {
url: config.writerUrl,
clickhouseSettings: config.clickhouseSettings,
logger: this.logger,
+ meter: config.meter,
logLevel: config.logLevel,
keepAlive: config.keepAlive,
httpAgent: config.httpAgent,