From 19d9dc975f932e5432c5f756eed10afebd02b25b Mon Sep 17 00:00:00 2001 From: Drew Davis Date: Mon, 31 Aug 2026 09:58:49 -0400 Subject: [PATCH] feat: Accept optional time bounds on Prometheus label values endpoint --- .changeset/promql-label-values-time-range.md | 6 + .../__tests__/timeseriesEngine.int.test.ts | 244 +++++++++++ .../api/src/controllers/timeseriesEngine.ts | 137 ++++++ packages/api/src/fixtures.ts | 90 ++++ .../api/__tests__/prometheus.int.test.ts | 390 +++++++++++++++++- packages/api/src/routers/api/prometheus.ts | 154 +++++-- packages/app/src/api.ts | 4 + .../src/__tests__/metadata.int.test.ts | 166 ++++++++ packages/common-utils/src/core/metadata.ts | 33 ++ 9 files changed, 1191 insertions(+), 33 deletions(-) create mode 100644 .changeset/promql-label-values-time-range.md create mode 100644 packages/api/src/controllers/__tests__/timeseriesEngine.int.test.ts create mode 100644 packages/api/src/controllers/timeseriesEngine.ts diff --git a/.changeset/promql-label-values-time-range.md b/.changeset/promql-label-values-time-range.md new file mode 100644 index 0000000000..c03a945711 --- /dev/null +++ b/.changeset/promql-label-values-time-range.md @@ -0,0 +1,6 @@ +--- +'@hyperdx/api': patch +'@hyperdx/app': patch +--- + +feat: Accept optional time bounds on Prometheus label values endpoint diff --git a/packages/api/src/controllers/__tests__/timeseriesEngine.int.test.ts b/packages/api/src/controllers/__tests__/timeseriesEngine.int.test.ts new file mode 100644 index 0000000000..f8262b8996 --- /dev/null +++ b/packages/api/src/controllers/__tests__/timeseriesEngine.int.test.ts @@ -0,0 +1,244 @@ +import { ClickhouseClient } from '@/clickhouse'; +import * as config from '@/config'; +import { queryLabelValues } from '@/controllers/timeseriesEngine'; +import { + closeTestFixtureClickHouseClient, + DEFAULT_DATABASE, + dropTimeSeriesTable, + executeSqlCommand, + seedTimeSeriesTagsTable, +} from '@/fixtures'; + +const CONNECTION_ID = 'timeseries-engine-int-test'; +const BOUNDED_TABLE = 'test_ts_label_values'; +const UNBOUNDED_TABLE = 'test_ts_label_values_no_bounds'; + +const HOUR_SEC = 3600; +const OLD_START_SEC = 1600000000; +const RECENT_START_SEC = 1700000000; +const sec = (s: number) => s * 1000; + +// Three series: one only in the distant past, two only in the recent window +// sharing a metric name (so DISTINCT has something to collapse). +const SERIES = [ + { + metricName: 'old_metric', + tags: { job: 'batch' }, + startSec: OLD_START_SEC, + endSec: OLD_START_SEC + HOUR_SEC, + }, + { + metricName: 'recent_metric', + tags: { job: 'api' }, + startSec: RECENT_START_SEC, + endSec: RECENT_START_SEC + HOUR_SEC, + }, + { + metricName: 'recent_metric', + tags: { job: 'web' }, + startSec: RECENT_START_SEC, + endSec: RECENT_START_SEC + HOUR_SEC, + }, +]; + +describe('timeseriesEngine controller', () => { + let client: ClickhouseClient; + + const labelValues = ( + args: Partial[0]> & { + labelName: string; + }, + ) => + queryLabelValues({ + client, + connectionId: CONNECTION_ID, + databaseName: DEFAULT_DATABASE, + tableName: BOUNDED_TABLE, + ...args, + }); + + beforeAll(async () => { + client = new ClickhouseClient({ + host: config.CLICKHOUSE_HOST, + username: config.CLICKHOUSE_USER, + password: config.CLICKHOUSE_PASSWORD, + }); + await seedTimeSeriesTagsTable({ table: BOUNDED_TABLE, series: SERIES }); + await seedTimeSeriesTagsTable({ + table: UNBOUNDED_TABLE, + series: SERIES, + storeTimeBounds: false, + }); + }); + + afterAll(async () => { + for (const table of [BOUNDED_TABLE, UNBOUNDED_TABLE]) { + await dropTimeSeriesTable({ table }); + } + await client.close(); + await closeTestFixtureClickHouseClient(); + }); + + describe('queryLabelValues', () => { + describe('__name__', () => { + it('returns every metric name, deduplicated and sorted', async () => { + expect(await labelValues({ labelName: '__name__' })).toEqual([ + 'old_metric', + 'recent_metric', + ]); + }); + + it('keeps only metrics whose series overlap the window', async () => { + expect( + await labelValues({ + labelName: '__name__', + startMs: sec(RECENT_START_SEC), + endMs: sec(RECENT_START_SEC + HOUR_SEC), + }), + ).toEqual(['recent_metric']); + + expect( + await labelValues({ + labelName: '__name__', + startMs: sec(OLD_START_SEC), + endMs: sec(OLD_START_SEC + HOUR_SEC), + }), + ).toEqual(['old_metric']); + }); + + // A lone `end` bounds min_time (nothing may start after the window) and a + // lone `start` bounds max_time (nothing may end before it). Getting the + // pairing backwards still filters, just wrongly, so both directions are + // pinned. + it('applies a lone end bound against min_time', async () => { + expect( + await labelValues({ + labelName: '__name__', + endMs: sec(OLD_START_SEC + HOUR_SEC), + }), + ).toEqual(['old_metric']); + }); + + it('applies a lone start bound against max_time', async () => { + expect( + await labelValues({ + labelName: '__name__', + startMs: sec(RECENT_START_SEC), + }), + ).toEqual(['recent_metric']); + }); + + it('returns nothing for a window no series covers', async () => { + expect( + await labelValues({ + labelName: '__name__', + startMs: sec(OLD_START_SEC + 2 * HOUR_SEC), + endMs: sec(OLD_START_SEC + 3 * HOUR_SEC), + }), + ).toEqual([]); + }); + }); + + describe('tag labels', () => { + it('returns the label values, deduplicated and sorted', async () => { + expect(await labelValues({ labelName: 'job' })).toEqual([ + 'api', + 'batch', + 'web', + ]); + }); + + it('scopes label values to the window', async () => { + expect( + await labelValues({ + labelName: 'job', + startMs: sec(RECENT_START_SEC), + endMs: sec(RECENT_START_SEC + HOUR_SEC), + }), + ).toEqual(['api', 'web']); + }); + + it('returns nothing for a label no series carries', async () => { + expect(await labelValues({ labelName: 'instance' })).toEqual([]); + }); + + // The label name reaches both the WHERE clause and the value expression. + // Parameterized, this is an ordinary miss; interpolated, it is a syntax + // error — so the assertion is "empty", not "throws". + it('parameterizes the label name rather than interpolating it', async () => { + expect(await labelValues({ labelName: `job') OR 1=1 --` })).toEqual([]); + }); + }); + + describe('limit', () => { + it('caps the number of values returned, keeping the ordering', async () => { + expect(await labelValues({ labelName: '__name__', limit: 1 })).toEqual([ + 'old_metric', + ]); + }); + + // Prometheus reads limit=0 as unlimited, so it must not become `LIMIT 0`. + it('treats a zero limit as unlimited', async () => { + expect(await labelValues({ labelName: '__name__', limit: 0 })).toEqual([ + 'old_metric', + 'recent_metric', + ]); + }); + + it('returns everything when the limit exceeds the result set', async () => { + expect( + await labelValues({ labelName: '__name__', limit: 100 }), + ).toEqual(['old_metric', 'recent_metric']); + }); + }); + + // store_min_time_and_max_time = 0 leaves the tags table without the columns + // the predicate reads, and referencing a missing column is a hard error. + // Time bounds are best-effort narrowing, so the request widens instead. + describe('table without min_time/max_time', () => { + it('ignores the bounds instead of failing', async () => { + expect( + await queryLabelValues({ + client, + connectionId: CONNECTION_ID, + databaseName: DEFAULT_DATABASE, + tableName: UNBOUNDED_TABLE, + labelName: '__name__', + startMs: sec(RECENT_START_SEC), + endMs: sec(RECENT_START_SEC + HOUR_SEC), + }), + ).toEqual(['old_metric', 'recent_metric']); + }); + + it('still filters on the label itself', async () => { + expect( + await queryLabelValues({ + client, + connectionId: CONNECTION_ID, + databaseName: DEFAULT_DATABASE, + tableName: UNBOUNDED_TABLE, + labelName: 'job', + }), + ).toEqual(['api', 'batch', 'web']); + }); + }); + + it('rejects when the table is not a TimeSeries table', async () => { + await executeSqlCommand( + `CREATE OR REPLACE TABLE ${DEFAULT_DATABASE}.test_ts_plain (ts DateTime) ENGINE = MergeTree ORDER BY ts`, + ); + try { + await expect( + labelValues({ + labelName: '__name__', + tableName: 'test_ts_plain', + }), + ).rejects.toThrow(/TimeSeries table only/); + } finally { + await executeSqlCommand( + `DROP TABLE IF EXISTS ${DEFAULT_DATABASE}.test_ts_plain`, + ); + } + }); + }); +}); diff --git a/packages/api/src/controllers/timeseriesEngine.ts b/packages/api/src/controllers/timeseriesEngine.ts new file mode 100644 index 0000000000..ba088cc60c --- /dev/null +++ b/packages/api/src/controllers/timeseriesEngine.ts @@ -0,0 +1,137 @@ +import { + ChSql, + chSql, + concatChSql, +} from '@hyperdx/common-utils/dist/clickhouse'; +import { + Metadata, + MetadataCache, +} from '@hyperdx/common-utils/dist/core/metadata'; + +import { ClickhouseClient } from '@/clickhouse'; +import logger from '@/utils/logger'; + +export const PROMETHEUS_MAX_EXECUTION_SEC = 30; +export const PROMETHEUS_MAX_RESULT_ROWS = 100000; + +/** + * Indicates whether the tags inner table associated with the given TimeSeries table + * includes the optional `min_time` and `max_time` columns. + */ +async function timeSeriesTagsTableHasTimeBounds({ + metadata, + connectionId, + database, + table, +}: { + metadata: Metadata; + connectionId: string; + database: string; + table: string; +}): Promise { + try { + const columns = await metadata.getTimeSeriesTableColumns({ + connectionId, + databaseName: database, + tableName: table, + innerTableType: 'Tags', + }); + + const columnNames = new Set(columns.map(column => column.name)); + return columnNames.has('min_time') && columnNames.has('max_time'); + } catch (e) { + logger.warn( + { err: e, database, table }, + 'Failed to check if TimeSeries tags table has time bounds columns', + ); + return false; + } +} + +/** + * Queries distinct values for the given label name from the given TimeSeries table, + * optionally filtering by a time range (if the table has the optional time range columns). + */ +export async function queryLabelValues({ + client, + databaseName, + connectionId, + tableName, + labelName, + startMs, + endMs, + limit, +}: { + client: ClickhouseClient; + connectionId: string; + databaseName: string; + tableName: string; + labelName: string; + startMs?: number; + endMs?: number; + limit?: number; +}): Promise { + const isMetricName = labelName === '__name__'; + const value = isMetricName + ? chSql`${{ Identifier: 'metric_name' }} ` + : chSql`${{ Identifier: 'tags' }}[${{ String: labelName }}]`; + + const conditions: ChSql[] = []; + if (!isMetricName) + conditions.push( + chSql`mapContains(${{ Identifier: 'tags' }}, ${{ String: labelName }})`, + ); + + const metadata = new Metadata(client, new MetadataCache()); + const hasStartTime = startMs != null; + const hasEndTime = endMs != null; + + // min and max time columns are optional in the tags inner table, + // so we check if they exist before adding time conditions + const tableHasTimeBounds = + (hasStartTime || hasEndTime) && // Short-circuit if no time conditions are needed + (await timeSeriesTagsTableHasTimeBounds({ + metadata, + connectionId, + database: databaseName, + table: tableName, + })); + + if (tableHasTimeBounds && endMs != null) + conditions.push( + chSql`(min_time IS NULL OR min_time <= fromUnixTimestamp64Milli(${{ Int64: endMs }}))`, + ); + + if (tableHasTimeBounds && startMs != null) + conditions.push( + chSql`(max_time IS NULL OR max_time >= fromUnixTimestamp64Milli(${{ Int64: startMs }}))`, + ); + + const where = conditions.length + ? chSql`WHERE ${concatChSql(' AND ', ...conditions)}` + : chSql``; + const limitSql = limit + ? chSql`LIMIT ${{ Int32: Math.min(limit, PROMETHEUS_MAX_RESULT_ROWS) }}` + : chSql``; + const query = chSql` + SELECT DISTINCT ${value} AS val + FROM timeSeriesTags(${{ Identifier: databaseName }}, ${{ Identifier: tableName }}) + ${where} + ORDER BY val + ${limitSql} + `; + + const resp = await client.query({ + query: query.sql, + query_params: query.params, + format: 'JSON', + clickhouse_settings: { + allow_experimental_time_series_table: 1, + max_execution_time: PROMETHEUS_MAX_EXECUTION_SEC, + max_result_rows: String(PROMETHEUS_MAX_RESULT_ROWS), + }, + }); + + const json = await resp.json<{ val: string }>(); + return json.data.map(r => r.val); +} diff --git a/packages/api/src/fixtures.ts b/packages/api/src/fixtures.ts index 0cc0d35e33..26ad95229c 100644 --- a/packages/api/src/fixtures.ts +++ b/packages/api/src/fixtures.ts @@ -258,6 +258,96 @@ export const executeSqlCommand = async (sql: string) => { }); }; +// The TimeSeries engine is experimental, and its flag is a *query* setting — it +// cannot ride along in a CREATE's own SETTINGS clause, which only takes storage +// settings. Every statement therefore carries it, which is why these tables +// cannot go through `executeSqlCommand`. +export const executeTimeSeriesSqlCommand = async (sql: string) => { + const client = await getTestFixtureClickHouseClient(); + return await client.command({ + query: sql, + clickhouse_settings: { + allow_experimental_time_series_table: 1, + wait_end_of_query: 1, + }, + }); +}; + +export const dropTimeSeriesTable = async ({ + table, + database = DEFAULT_DATABASE, +}: { + table: string; + database?: string; +}) => executeTimeSeriesSqlCommand(`DROP TABLE IF EXISTS ${database}.${table}`); + +export type TimeSeriesFixtureSeries = { + metricName: string; + /** Labels other than `__name__`, which is derived from `metricName`. */ + tags: Record; + /** Series window in unix seconds; ignored when the table stores no bounds. */ + startSec: number; + endSec: number; +}; + +/** + * (Re)creates a TimeSeries table and writes `series` straight into its tags + * inner table — Prometheus remote-write is the only other way in, and label + * lookups never read the data inner table. + * + * `storeTimeBounds: false` creates the table with + * `store_min_time_and_max_time = 0`, which leaves the tags table without the + * min_time/max_time columns a time-bounded lookup reads. + */ +export const seedTimeSeriesTagsTable = async ({ + table, + series, + database = DEFAULT_DATABASE, + storeTimeBounds = true, +}: { + table: string; + series: TimeSeriesFixtureSeries[]; + database?: string; + storeTimeBounds?: boolean; +}) => { + await dropTimeSeriesTable({ table, database }); + await executeTimeSeriesSqlCommand( + `CREATE TABLE ${database}.${table} ENGINE = TimeSeries${ + storeTimeBounds ? '' : ' SETTINGS store_min_time_and_max_time = 0' + }`, + ); + + const quoted = (v: string) => `'${v.replace(/'/g, "\\'")}'`; + const mapLiteral = (tags: Record) => + `map(${Object.entries(tags) + .flatMap(([k, v]) => [quoted(k), quoted(v)]) + .join(', ')})`; + + const columns = storeTimeBounds + ? '(metric_name, tags, all_tags, min_time, max_time)' + : '(metric_name, tags, all_tags)'; + const values = series + .map(s => { + const row = [ + quoted(s.metricName), + mapLiteral(s.tags), + mapLiteral({ __name__: s.metricName, ...s.tags }), + ]; + if (storeTimeBounds) { + row.push( + `toDateTime64(${s.startSec}, 3)`, + `toDateTime64(${s.endSec}, 3)`, + ); + } + return `(${row.join(', ')})`; + }) + .join(', '); + + await executeTimeSeriesSqlCommand( + `INSERT INTO TABLE FUNCTION timeSeriesTags('${database}', '${table}') ${columns} VALUES ${values}`, + ); +}; + export const clearClickhouseTables = async () => { if (!config.IS_CI) { throw new Error('ONLY execute this in CI env 😈 !!!'); diff --git a/packages/api/src/routers/api/__tests__/prometheus.int.test.ts b/packages/api/src/routers/api/__tests__/prometheus.int.test.ts index e5703f8ca6..f25cc6cab8 100644 --- a/packages/api/src/routers/api/__tests__/prometheus.int.test.ts +++ b/packages/api/src/routers/api/__tests__/prometheus.int.test.ts @@ -1,11 +1,18 @@ import { Types } from 'mongoose'; import * as config from '@/config'; -import { getAgent, getLoggedInAgent, getServer } from '@/fixtures'; +import { + DEFAULT_DATABASE, + dropTimeSeriesTable, + getAgent, + getLoggedInAgent, + getServer, + seedTimeSeriesTagsTable, +} from '@/fixtures'; import Connection from '@/models/connection'; import { PROMETHEUS_MAX_EXEMPLAR_WINDOW_SEC } from '@/routers/api/prometheus'; -const mockFetch = global.fetch as jest.Mock; +const mockFetch = jest.mocked(global.fetch); // The proxy now streams the upstream response straight through (no // `await resp.json()`), so test mocks must expose the fields the pipeline @@ -289,7 +296,7 @@ describe('prometheus router', () => { expect(res.body).toEqual(promResponse); expect(mockFetch).toHaveBeenCalledTimes(1); - const calledUrl = mockFetch.mock.calls[0][0] as string; + const calledUrl = mockFetch.mock.calls[0][0]; expect(calledUrl).toContain('http://prom.example.com'); expect(calledUrl).toContain('/api/v1/query_range'); expect(calledUrl).toContain('query=up'); @@ -382,7 +389,7 @@ describe('prometheus router', () => { expect(res.body).toEqual(promResponse); expect(mockFetch).toHaveBeenCalledTimes(1); - const calledUrl = mockFetch.mock.calls[0][0] as string; + const calledUrl = mockFetch.mock.calls[0][0]; expect(calledUrl).toContain('/api/v1/query'); expect(calledUrl).not.toContain('/api/v1/query_range'); }); @@ -394,6 +401,26 @@ describe('prometheus router', () => { await agent.get('/v1/prometheus/label/__name__/values').expect(400); }); + // An empty or malformed id must fail in the schema. Reaching Mongoose + // instead turns a client error into a cast failure that the handler's + // catch-all reports as a backend fault. + it.each(['', 'not-an-object-id'])( + 'returns 400 for connectionId %p', + async connectionId => { + const { agent } = await getLoggedInAgent(server); + const res = await agent + .get('/v1/prometheus/label/__name__/values') + .query({ connectionId }) + .expect(400); + + expect(res.body).toMatchObject({ + status: 'error', + errorType: 'bad_data', + error: expect.stringContaining('connectionId'), + }); + }, + ); + it('returns 404 when connection does not exist', async () => { const { agent } = await getLoggedInAgent(server); await agent @@ -415,9 +442,360 @@ describe('prometheus router', () => { .expect(200); expect(res.body).toEqual(promResponse); - const calledUrl = mockFetch.mock.calls[0][0] as string; + const calledUrl = mockFetch.mock.calls[0][0]; expect(calledUrl).toContain('/api/v1/label/__name__/values'); }); + + it('forwards normalized bounds upstream', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ + connectionId: conn._id.toString(), + start: '2023-11-14T22:13:20Z', + end: '1700000060', + }) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.get('start')).toBe('1700000000'); + expect(requested.searchParams.get('end')).toBe('1700000060'); + }); + + it('sends no bounds upstream when the caller gives none', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString() }) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.has('start')).toBe(false); + expect(requested.searchParams.has('end')).toBe(false); + }); + + // `?start=` is what a client sends for "no bound"; it must not become the + // epoch, which is what parseTimestamp('') -> Number('') would make it. + it('treats an empty bound as absent rather than as the epoch', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), start: '', end: '' }) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.has('start')).toBe(false); + expect(requested.searchParams.has('end')).toBe(false); + }); + + // Query-string values arrive as strings; a schema that only accepts numbers + // would 400 every request carrying a limit. + it('forwards limit upstream', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), limit: '25' }) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.get('limit')).toBe('25'); + }); + + it('treats an empty limit as absent', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), limit: '' }) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.has('limit')).toBe(false); + }); + + // Prometheus reads limit=0 as unlimited, which is also what sending no + // limit means. Rejecting it would break a caller Prometheus itself accepts. + it('drops a zero limit and still proxies', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), limit: '0' }) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.has('limit')).toBe(false); + }); + + it('rejects a negative limit', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), limit: '-1' }) + .expect(400); + + expect(mockFetch).not.toHaveBeenCalled(); + }); + + it('rejects a non-numeric limit', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), limit: 'lots' }) + .expect(400); + + expect(mockFetch).not.toHaveBeenCalled(); + }); + + it('rejects a malformed bound before reaching upstream', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + const res = await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), start: 'not-a-time' }) + .expect(400); + + expect(res.body).toMatchObject({ + status: 'error', + errorType: 'bad_data', + error: 'start: invalid timestamp, expected RFC3339 or unix seconds', + }); + expect(mockFetch).not.toHaveBeenCalled(); + }); + + // Upstream 400s on an inverted range, so rejecting it here keeps the two + // backends answering the same request the same way. + it('rejects an inverted range before reaching upstream', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + const res = await agent + .get('/v1/prometheus/label/job/values') + .query({ + connectionId: conn._id.toString(), + start: '1700000000', + end: '1600000000', + }) + .expect(400); + + expect(res.body).toMatchObject({ + status: 'error', + errorType: 'bad_data', + error: 'end: end timestamp must not be before start time', + }); + expect(mockFetch).not.toHaveBeenCalled(); + }); + + // `match[]` is how a client scopes label values to one metric; dropping it + // silently widens the answer to every series in the store. Repeatable, and + // forwarded verbatim — a selector is PromQL, which only upstream parses. + it('forwards every match[] selector verbatim', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query( + `connectionId=${conn._id.toString()}&start=2023-11-14T22:13:20Z` + + `&match[]=${encodeURIComponent('up{env="prod"}')}` + + `&match[]=${encodeURIComponent('down')}`, + ) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.getAll('match[]')).toEqual([ + 'up{env="prod"}', + 'down', + ]); + expect(requested.searchParams.getAll('start')).toEqual(['1700000000']); + }); + + // qs strips the brackets, so the unbracketed spelling arrives on the same + // key. It still has to leave as `match[]`, which is the only name + // Prometheus reads. + it('restores the bracketed name for an unbracketed match', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), match: 'up' }) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.getAll('match[]')).toEqual(['up']); + expect(requested.searchParams.has('match')).toBe(false); + }); + + it('treats an empty match as absent', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedPrometheusConnection(team._id); + + await agent + .get('/v1/prometheus/label/job/values') + .query({ connectionId: conn._id.toString(), match: '' }) + .expect(200); + + const requested = new URL(String(mockFetch.mock.calls[0][0])); + expect(requested.searchParams.has('match[]')).toBe(false); + }); + + // queryLabelValues' own bounds/limit/fallback behaviour is covered in + // controllers/__tests__/timeseriesEngine.int.test.ts. What only the route + // can show is that query-string params reach it intact — bounds in unix + // seconds where the controller takes milliseconds, and a stringified limit. + describe('ClickHouse-backed', () => { + const TABLE = 'prom_label_values_bounded_test'; + const OLD_METRIC = 'label_values_old_metric'; + const RECENT_METRIC = 'label_values_recent_metric'; + const OLD_START = 1600000000; + const RECENT_START = 1700000000; + const SERIES_LENGTH_SEC = 3600; + + const labelValues = async (query: Record) => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedClickHouseConnection(team._id); + const res = await agent + .get('/v1/prometheus/label/__name__/values') + .query({ + connectionId: conn._id.toString(), + database: DEFAULT_DATABASE, + table: TABLE, + ...query, + }) + .expect(200); + return res.body; + }; + + beforeAll(async () => { + await seedTimeSeriesTagsTable({ + table: TABLE, + series: [ + { + metricName: OLD_METRIC, + tags: { job: 'batch' }, + startSec: OLD_START, + endSec: OLD_START + SERIES_LENGTH_SEC, + }, + { + metricName: RECENT_METRIC, + tags: { job: 'api' }, + startSec: RECENT_START, + endSec: RECENT_START + SERIES_LENGTH_SEC, + }, + ], + }); + }); + + afterAll(async () => { + await dropTimeSeriesTable({ table: TABLE }); + }); + + it('returns every value when no bounds are given', async () => { + expect(await labelValues({})).toEqual({ + status: 'success', + data: [OLD_METRIC, RECENT_METRIC], + }); + }); + + // Bounds arrive in unix seconds. Passed through unscaled they land in + // 1970, where the min_time predicate excludes every series instead. + it('reads bounds as unix seconds', async () => { + expect( + await labelValues({ + start: String(RECENT_START), + end: String(RECENT_START + SERIES_LENGTH_SEC), + }), + ).toEqual({ status: 'success', data: [RECENT_METRIC] }); + }); + + // Without the range check this answers 200 with whatever series span the + // inverted gap, while the proxy path 400s the identical request. + it('rejects an inverted range instead of answering', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedClickHouseConnection(team._id); + + const res = await agent + .get('/v1/prometheus/label/__name__/values') + .query({ + connectionId: conn._id.toString(), + database: DEFAULT_DATABASE, + table: TABLE, + start: String(RECENT_START), + end: String(OLD_START), + }) + .expect(400); + + expect(res.body).toMatchObject({ + status: 'error', + errorType: 'bad_data', + error: 'end: end timestamp must not be before start time', + }); + }); + + // A zero-width range is a point query, not an inversion. + it('accepts equal bounds', async () => { + expect( + await labelValues({ + start: String(RECENT_START), + end: String(RECENT_START), + }), + ).toEqual({ status: 'success', data: [RECENT_METRIC] }); + }); + + it('honours a limit given as a query string', async () => { + expect(await labelValues({ limit: '1' })).toEqual({ + status: 'success', + data: [OLD_METRIC], + }); + }); + + it('reads a zero limit as unlimited', async () => { + expect(await labelValues({ limit: '0' })).toEqual({ + status: 'success', + data: [OLD_METRIC, RECENT_METRIC], + }); + }); + + // Nothing here evaluates a PromQL selector, so answering 200 with + // unfiltered values would be a wrong answer the caller trusts. + it('rejects match[] instead of ignoring it', async () => { + const { agent, team } = await getLoggedInAgent(server); + const conn = await seedClickHouseConnection(team._id); + + const res = await agent + .get('/v1/prometheus/label/__name__/values') + .query({ + connectionId: conn._id.toString(), + database: DEFAULT_DATABASE, + table: TABLE, + 'match[]': 'up', + }) + .expect(400); + + expect(res.body).toMatchObject({ + status: 'error', + errorType: 'bad_data', + error: + 'match[] is not supported for ClickHouse-backed PromQL connections', + }); + }); + }); }); describe('GET /v1/prometheus/query_exemplars', () => { @@ -485,7 +863,7 @@ describe('prometheus router', () => { .expect(200); expect(res.body).toEqual(promResponse); - const calledUrl = mockFetch.mock.calls[0][0] as string; + const calledUrl = mockFetch.mock.calls[0][0]; expect(calledUrl).toContain('/api/v1/query_exemplars'); expect(calledUrl).toContain('query=up'); }); diff --git a/packages/api/src/routers/api/prometheus.ts b/packages/api/src/routers/api/prometheus.ts index e3831b4e95..f1bc3b3069 100644 --- a/packages/api/src/routers/api/prometheus.ts +++ b/packages/api/src/routers/api/prometheus.ts @@ -2,12 +2,19 @@ import express from 'express'; import { performance } from 'perf_hooks'; import { Readable } from 'stream'; import { pipeline } from 'stream/promises'; +import z from 'zod'; import { ClickhouseClient } from '@/clickhouse'; import { getConnectionById } from '@/controllers/connection'; +import { + PROMETHEUS_MAX_EXECUTION_SEC, + PROMETHEUS_MAX_RESULT_ROWS, + queryLabelValues, +} from '@/controllers/timeseriesEngine'; import { getNonNullUserWithTeam } from '@/middleware/auth'; import { getCounter, getHistogram } from '@/utils/instrumentation'; import logger from '@/utils/logger'; +import { objectIdSchema } from '@/utils/zod'; const router = express.Router(); @@ -152,8 +159,6 @@ export function formatVectorResponse( const PROMETHEUS_PROXY_TIMEOUT_MS = 90_000; const PROMETHEUS_CH_TIMEOUT_MS = 30_000; -const PROMETHEUS_MAX_EXECUTION_SEC = 30; -const PROMETHEUS_MAX_RESULT_ROWS = '100000'; const PROMETHEUS_MAX_RESOLUTION = 11_000; // Widest window /query_exemplars will proxy. Prometheus's exemplar store is a // small circular buffer, so a wider range mostly costs a bigger streamed body @@ -191,7 +196,7 @@ export function isClientDisconnect(err: unknown): boolean { async function proxyToPrometheus( upstreamHost: string, path: string, - params: Record, + params: Record, res: express.Response, ): Promise { let url: URL; @@ -207,7 +212,13 @@ async function proxyToPrometheus( } for (const [k, v] of Object.entries(params)) { if (['connectionId', 'database', 'table'].includes(k)) continue; - if (v != null) url.searchParams.set(k, v); + if (v == null) continue; + // A repeatable param (`match[]`) appends every value + if (Array.isArray(v)) { + for (const item of v) url.searchParams.append(k, item); + } else { + url.searchParams.set(k, v); + } } const target = url.toString(); @@ -423,7 +434,7 @@ const queryRangeHandler: express.RequestHandler = async (req, res) => { clickhouse_settings: { allow_experimental_time_series_table: 1, max_execution_time: PROMETHEUS_MAX_EXECUTION_SEC, - max_result_rows: PROMETHEUS_MAX_RESULT_ROWS, + max_result_rows: String(PROMETHEUS_MAX_RESULT_ROWS), }, }); @@ -537,7 +548,7 @@ const queryHandler: express.RequestHandler = async (req, res) => { clickhouse_settings: { allow_experimental_time_series_table: 1, max_execution_time: PROMETHEUS_MAX_EXECUTION_SEC, - max_result_rows: PROMETHEUS_MAX_RESULT_ROWS, + max_result_rows: String(PROMETHEUS_MAX_RESULT_ROWS), }, }); @@ -719,6 +730,82 @@ router.post('/query_exemplars', queryExemplarsHandler); // https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels const PROMETHEUS_LABEL_NAME = /^[a-zA-Z_:][a-zA-Z0-9_:]*$/; +/** + * Prometheus timestamps are either RFC3339 strings or unix seconds. + * This schema coerces them to unix seconds. + * + * `?start=` with no value means the same as no `start` at all. Without the + * empty-string arm it would not: `parseTimestamp('')` goes through `Number('')`, + * which is 0, pinning the bound to the epoch. + */ +const prometheusTimestampSchema = z + .union([z.string(), z.number()]) + .transform((v, ctx) => { + if (v === '') return undefined; + try { + return parseTimestamp(v); + } catch { + ctx.addIssue({ + code: 'custom', + message: `invalid timestamp, expected RFC3339 or unix seconds`, + }); + return z.NEVER; + } + }); + +/** + * Prometheus's series selector, spelled `match[]` and repeatable. Express's query + * parser drops the brackets, so both `match[]=up` and `match=up` land on a + * `match` key and a repeated one arrives as an array. The selectors themselves + * are never interpreted here — only Prometheus can — so an empty one is treated + * as absent rather than forwarded for upstream to reject. + */ +const prometheusMatchSchema = z + .union([z.string(), z.array(z.string())]) + .transform(v => { + const selectors = (Array.isArray(v) ? v : [v]).filter(m => m !== ''); + return selectors.length ? selectors : undefined; + }); + +const labelValuesRequestQuerySchema = z + .object({ + connectionId: objectIdSchema, + start: prometheusTimestampSchema.optional(), + end: prometheusTimestampSchema.optional(), + match: prometheusMatchSchema.optional(), + database: z.string().optional(), + table: z.string().optional(), + limit: z.preprocess( + v => (v === '' ? undefined : v), + z.coerce.number().int().nonnegative().optional(), + ), + }) + .superRefine((params, ctx) => { + if ( + params.start != null && + params.end != null && + params.end < params.start + ) { + ctx.addIssue({ + code: 'custom', + path: ['end'], + message: 'end timestamp must not be before start time', + }); + } + }); + +// A Prometheus error body carries one human-readable string, so zod's own +// `message` — a JSON dump of the issue list — is not usable as-is. +function formatQueryParamIssues(error: z.ZodError): string { + return error.issues + .map(issue => + issue.path.length + ? `${issue.path.join('.')}: ${issue.message}` + : issue.message, + ) + .join(', '); +} + router.get('/label/:name/values', async (req, res) => { const startedAt = performance.now(); let backend: PrometheusBackend = 'unknown'; @@ -732,17 +819,18 @@ router.get('/label/:name/values', async (req, res) => { error: 'Invalid label name', }); } - const params = req.query as Record; - const connectionId = params.connectionId; - if (!connectionId) { + const parseResult = labelValuesRequestQuerySchema.safeParse(req.query); + if (!parseResult.success) { return res.status(400).json({ status: 'error', errorType: 'bad_data', - error: 'missing required parameter: connectionId', + error: formatQueryParamIssues(parseResult.error), }); } + const params = parseResult.data; + const { connectionId, start, end, limit, match } = params; const connection = await getConnectionById( teamId.toString(), connectionId, @@ -762,7 +850,13 @@ router.get('/label/:name/values', async (req, res) => { const status = await proxyToPrometheus( connection.host, `/api/v1/label/${labelName}/values`, - params, + { + ...(start != null ? { start: String(start) } : {}), + ...(end != null ? { end: String(end) } : {}), + ...(limit ? { limit: String(limit) } : {}), + // Restored under the name Prometheus expects + ...(match != null ? { 'match[]': match } : {}), + }, res, ); recordProxyOutcome(status, 'label_values', backend); @@ -770,6 +864,17 @@ router.get('/label/:name/values', async (req, res) => { } backend = 'clickhouse'; + + // TODO: Support `match[]` for ClickHouse-Timeseries engine path + if (match != null) { + return res.status(400).json({ + status: 'error', + errorType: 'bad_data', + error: + 'match[] is not supported for ClickHouse-backed PromQL connections', + }); + } + const database = params.database ?? 'default'; const table = params.table; if (!table) { @@ -787,23 +892,18 @@ router.get('/label/:name/values', async (req, res) => { requestTimeout: PROMETHEUS_CH_TIMEOUT_MS, }); - const tagsQuery = - labelName === '__name__' - ? `SELECT DISTINCT metric_name AS val FROM timeSeriesTags({db:String}, {table:String}) ORDER BY val SETTINGS allow_experimental_time_series_table = 1` - : `SELECT DISTINCT all_tags[{label:String}] AS val FROM timeSeriesTags({db:String}, {table:String}) WHERE mapContains(all_tags, {label:String}) ORDER BY val SETTINGS allow_experimental_time_series_table = 1`; - - const resp = await client.query({ - query: tagsQuery, - query_params: { db: database, table, label: labelName }, - format: 'JSON', - clickhouse_settings: { - allow_experimental_time_series_table: 1, - max_execution_time: PROMETHEUS_MAX_EXECUTION_SEC, - max_result_rows: PROMETHEUS_MAX_RESULT_ROWS, - }, + const startMs = start != null ? Math.floor(start * 1000) : undefined; + const endMs = end != null ? Math.ceil(end * 1000) : undefined; + const values = await queryLabelValues({ + client, + labelName, + connectionId: connection.id, + databaseName: database, + tableName: table, + startMs, + endMs, + limit, }); - const json = await resp.json(); - const values: string[] = json.data.map((r: any) => r.val); return res.json({ status: 'success', data: values }); } catch (e) { diff --git a/packages/app/src/api.ts b/packages/app/src/api.ts index 0260eaf281..a9195c91b4 100644 --- a/packages/app/src/api.ts +++ b/packages/app/src/api.ts @@ -646,6 +646,8 @@ export const prometheusApi = { connectionId: string; database?: string; table?: string; + start?: number; + end?: number; }): Promise => server .get(`v1/prometheus/label/${params.label}/values`, { @@ -653,6 +655,8 @@ export const prometheusApi = { connectionId: params.connectionId, ...(params.database ? { database: params.database } : {}), ...(params.table ? { table: params.table } : {}), + ...(params.start != null ? { start: String(params.start) } : {}), + ...(params.end != null ? { end: String(params.end) } : {}), }, }) .json(), diff --git a/packages/common-utils/src/__tests__/metadata.int.test.ts b/packages/common-utils/src/__tests__/metadata.int.test.ts index 44b7e37585..478a9a84a1 100644 --- a/packages/common-utils/src/__tests__/metadata.int.test.ts +++ b/packages/common-utils/src/__tests__/metadata.int.test.ts @@ -1144,4 +1144,170 @@ describe('Metadata Integration Tests', () => { expect(result.names).toEqual([]); }); }); + + describe('getTimeSeriesTableColumns', () => { + const boundedTable = 'test_ts_bounded'; + const unboundedTable = 'test_ts_unbounded'; + let metadata: Metadata; + + // The TimeSeries engine is experimental, so both the DDL and every read of + // an inner table need the setting — it cannot be attached to the CREATE's + // own SETTINGS clause, which only takes storage settings. + const timeSeriesCommand = (query: string) => + client.command({ + query, + clickhouse_settings: { allow_experimental_time_series_table: 1 }, + }); + + beforeAll(async () => { + await timeSeriesCommand(`DROP TABLE IF EXISTS default.${boundedTable}`); + await timeSeriesCommand(`DROP TABLE IF EXISTS default.${unboundedTable}`); + await timeSeriesCommand( + `CREATE TABLE default.${boundedTable} ENGINE = TimeSeries`, + ); + await timeSeriesCommand( + `CREATE TABLE default.${unboundedTable} ENGINE = TimeSeries SETTINGS store_min_time_and_max_time = 0`, + ); + }); + + afterAll(async () => { + await timeSeriesCommand(`DROP TABLE IF EXISTS default.${boundedTable}`); + await timeSeriesCommand(`DROP TABLE IF EXISTS default.${unboundedTable}`); + }); + + beforeEach(() => { + metadata = new Metadata(hdxClient, new MetadataCache()); + }); + + it('describes the tags inner table, not the TimeSeries table itself', async () => { + const columns = await metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: boundedTable, + innerTableType: 'Tags', + }); + + const byName = new Map(columns.map(c => [c.name, c])); + expect([...byName.keys()].sort()).toEqual([ + 'all_tags', + 'id', + 'max_time', + 'metric_name', + 'min_time', + 'tags', + ]); + expect(byName.get('min_time')?.type).toBe( + 'SimpleAggregateFunction(min, Nullable(DateTime64(3)))', + ); + // Ephemeral, so it cannot be selected — the reason a caller has to + // inspect the columns rather than assume the documented shape. + expect(byName.get('all_tags')?.default_type).toBe('EPHEMERAL'); + }); + + // What `store_min_time_and_max_time = 0` costs: the time-bound columns are + // simply absent, and referencing them is a hard ClickHouse error. + it('omits min_time/max_time when the table does not store them', async () => { + const columns = await metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: unboundedTable, + innerTableType: 'Tags', + }); + + const names = columns.map(c => c.name); + expect(names).toEqual( + expect.arrayContaining(['metric_name', 'tags', 'id']), + ); + expect(names).not.toContain('min_time'); + expect(names).not.toContain('max_time'); + }); + + it.each([ + ['Metrics', ['metric_family_name', 'type', 'unit', 'help']], + ['Data', ['id', 'timestamp', 'value']], + ] as const)( + 'describes the %s inner table', + async (innerTableType, expected) => { + const columns = await metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: boundedTable, + innerTableType, + }); + + expect(columns.map(c => c.name)).toEqual(expected); + }, + ); + + // Matching on the message, not just "it threw": an unbound {db:String} + // placeholder also throws, and would let both of these pass while the + // database and table never reached ClickHouse at all. + it('rejects when the table does not exist', async () => { + await expect( + metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: 'test_ts_missing', + innerTableType: 'Tags', + }), + ).rejects.toThrow(/default\.test_ts_missing does not exist/); + }); + + it('rejects when the table is not a TimeSeries table', async () => { + await timeSeriesCommand( + `CREATE OR REPLACE TABLE default.test_ts_not_timeseries (ts DateTime) ENGINE = MergeTree ORDER BY ts`, + ); + try { + await expect( + metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: 'test_ts_not_timeseries', + innerTableType: 'Tags', + }), + ).rejects.toThrow(/TimeSeries table only/); + } finally { + await timeSeriesCommand( + 'DROP TABLE IF EXISTS default.test_ts_not_timeseries', + ); + } + }); + + it('caches per database, table and inner table type', async () => { + const querySpy = jest.spyOn(hdxClient, 'query'); + + const first = await metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: boundedTable, + innerTableType: 'Tags', + }); + const second = await metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: boundedTable, + innerTableType: 'Tags', + }); + expect(second).toBe(first); + expect(querySpy).toHaveBeenCalledTimes(1); + + await metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: unboundedTable, + innerTableType: 'Tags', + }); + expect(querySpy).toHaveBeenCalledTimes(2); + + await metadata.getTimeSeriesTableColumns({ + connectionId: 'test_connection', + databaseName: 'default', + tableName: boundedTable, + innerTableType: 'Metrics', + }); + expect(querySpy).toHaveBeenCalledTimes(3); + + querySpy.mockRestore(); + }); + }); }); diff --git a/packages/common-utils/src/core/metadata.ts b/packages/common-utils/src/core/metadata.ts index 37309be73b..c892345979 100644 --- a/packages/common-utils/src/core/metadata.ts +++ b/packages/common-utils/src/core/metadata.ts @@ -525,6 +525,39 @@ export class Metadata { ); } + /** Queries and returns the columns of a TimeSeries table's inner table */ + async getTimeSeriesTableColumns({ + connectionId, + databaseName, + tableName, + innerTableType, + }: { + connectionId: string; + databaseName: string; + tableName: string; + innerTableType: 'Tags' | 'Metrics' | 'Data'; + }) { + return this.cache.getOrFetch( + `${connectionId}.${databaseName}.${tableName}.${innerTableType}.getTimeSeriesTableColumns`, + async () => { + const sql = chSql`DESCRIBE TABLE timeSeries${innerTableType}(${{ String: databaseName }}, ${{ String: tableName }})`; + const columns = await this.clickhouseClient + .query<'JSON'>({ + query: sql.sql, + query_params: sql.params, + connectionId, + clickhouse_settings: { + ...this.getClickHouseSettings(), + allow_experimental_time_series_table: 1, + }, + }) + .then(res => res.json()) + .then(d => d.data); + return columns; + }, + ); + } + async getMaterializedColumnsLookupTable({ databaseName, tableName,