diff --git a/.changeset/tender-cloths-switch.md b/.changeset/tender-cloths-switch.md new file mode 100644 index 0000000000..591e53fefc --- /dev/null +++ b/.changeset/tender-cloths-switch.md @@ -0,0 +1,5 @@ +--- +'@hyperdx/api': minor +--- + +feat: Add POST /api/v2/search endpoint for querying raw log and trace rows programmatically diff --git a/packages/api/openapi.json b/packages/api/openapi.json index c5a894b830..13050ca52e 100644 --- a/packages/api/openapi.json +++ b/packages/api/openapi.json @@ -31,6 +31,10 @@ { "name": "Webhooks", "description": "Endpoints for managing webhooks" + }, + { + "name": "Search", + "description": "Endpoints for querying raw data from log and trace sources" } ], "components": { @@ -2608,6 +2612,96 @@ } } }, + "SearchRequest": { + "type": "object", + "required": [ + "sourceId" + ], + "properties": { + "sourceId": { + "type": "string", + "description": "Source ID to query. Call GET /api/v2/sources to list available sources.\nThe source determines the underlying ClickHouse table (e.g. otel.otel_logs,\notel.otel_traces) and its column schema.\n", + "example": "69b46cb0d964ce2d0b9506a8" + }, + "startTime": { + "type": "string", + "format": "date-time", + "description": "Start of the query window (ISO 8601). Defaults to 15 minutes before endTime. Must be before endTime.\n", + "example": "2026-05-10T00:00:00Z" + }, + "endTime": { + "type": "string", + "format": "date-time", + "description": "End of the query window (ISO 8601). Defaults to now.", + "example": "2026-05-10T01:00:00Z" + }, + "where": { + "type": "string", + "maxLength": 8192, + "default": "", + "description": "Row filter expression. The language is controlled by whereLanguage.\n\nLucene examples (default):\n SeverityText:ERROR\n pipedream.pipeline_name:my-pipeline AND SeverityText:ERROR\n Body:timeout\n\nSQL examples (whereLanguage: \"sql\"):\n SeverityText = 'ERROR'\n `pipedream.pipeline_name` = 'my-pipeline'\n", + "example": "SeverityText:ERROR" + }, + "whereLanguage": { + "type": "string", + "enum": [ + "lucene", + "sql" + ], + "example": "lucene", + "default": "lucene", + "description": "Language used for the where filter. Default is lucene." + }, + "select": { + "type": "string", + "maxLength": 4096, + "default": "", + "description": "Comma-separated list of ClickHouse column expressions to include in\neach result row. When omitted the source's default select expression\nis used.\n\nEach entry is a ClickHouse SQL expression executed under the team's\ndatabase user. Semicolons and subqueries (SELECT keyword) are\nrejected; use column references, map lookups, or function calls only.\n\nHyperDX rewrites known attribute column names to their materialized\nequivalents automatically; you can still pass the logical name.\n", + "example": "Timestamp,SeverityText,Body,pipedream.pipeline_name" + }, + "orderBy": { + "type": "string", + "maxLength": 1024, + "description": "ClickHouse ORDER BY expression. When omitted the source's default\nordering (typically timestamp DESC) is used.\n", + "example": "Timestamp DESC" + }, + "maxResults": { + "type": "integer", + "minimum": 1, + "maximum": 2000, + "default": 100, + "description": "Maximum number of rows to return. Default is 100, max is 2000." + }, + "offset": { + "type": "integer", + "minimum": 0, + "maximum": 10000, + "default": 0, + "description": "Number of rows to skip (best-effort offset pagination). Default is\n0, max is 10000. Offset pagination is non-deterministic when\nmultiple rows share the same timestamp; for reliable deep paging\nfilter by the last Timestamp value returned in the previous page\ninstead of using a large offset.\n" + } + } + }, + "SearchRow": { + "type": "object", + "description": "A single result row. Keys correspond to the requested columns (or the source default columns when columns was omitted). Values are strings or numbers as returned by ClickHouse.\n", + "additionalProperties": true + }, + "SearchResponse": { + "type": "object", + "properties": { + "data": { + "type": "array", + "description": "Array of result rows. Each row is an object with keys corresponding to the requested columns.", + "items": { + "$ref": "#/components/schemas/SearchRow" + } + }, + "rows": { + "type": "integer", + "description": "Number of rows in this response (not total matching rows)." + } + } + }, "QuerySetting": { "type": "object", "required": [ @@ -4884,6 +4978,142 @@ } } }, + "/api/v2/search": { + "post": { + "summary": "Search Raw Logs and Traces", + "description": "Fetch individual log or trace rows from a HyperDX source.\n\nThis endpoint mirrors the \"search\" panel mode in the HyperDX UI.\nHyperDX applies the same query optimizations used in the UI:\n - Named attribute columns (e.g. \"pipedream.pipeline_name\") are\n rewritten to their indexed materialized equivalents when the\n source schema exposes them, avoiding slow Map lookups.\n - Rows are ordered by timestamp descending (most recent first).\n - The source's built-in PREWHERE / partition pruning is applied.\n\nAuthentication: Bearer token (personal API key from Team Settings).\n", + "operationId": "searchEvents", + "tags": [ + "Search" + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SearchRequest" + }, + "examples": { + "recentErrors": { + "summary": "Recent errors for a service", + "value": { + "sourceId": "69b46cb0d964ce2d0b9506a8", + "startTime": "2026-05-10T00:00:00Z", + "endTime": "2026-05-10T01:00:00Z", + "where": "SeverityText:ERROR", + "select": "Timestamp,SeverityText,Body,ServiceName", + "maxResults": 50 + } + }, + "pipedreamTaskLogs": { + "summary": "Pipedream task logs with materialized column filter", + "value": { + "sourceId": "69b46cb0d964ce2d0b9506a8", + "startTime": "2026-05-10T00:00:00Z", + "endTime": "2026-05-10T01:00:00Z", + "where": "pipedream.pipeline_name:my-pipeline", + "select": "Timestamp,SeverityText,Body,pipedream.pipeline_name,pipedream.stage_name,pipedream.task_index", + "maxResults": 200 + } + }, + "traceSearch": { + "summary": "Slow spans for a service", + "value": { + "sourceId": "69b46cb0d964ce2d0b9508b2", + "startTime": "2026-05-10T00:00:00Z", + "endTime": "2026-05-10T01:00:00Z", + "where": "ServiceName:my-service AND Duration:>1000000000", + "select": "Timestamp,TraceId,SpanId,SpanName,ServiceName,Duration,StatusCode", + "maxResults": 100 + } + }, + "paginated": { + "summary": "Paginated fetch", + "value": { + "sourceId": "69b46cb0d964ce2d0b9506a8", + "startTime": "2026-05-10T00:00:00Z", + "endTime": "2026-05-10T01:00:00Z", + "maxResults": 100, + "offset": 100 + } + } + } + } + } + }, + "responses": { + "200": { + "description": "Matching rows returned successfully", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SearchResponse" + }, + "example": { + "data": [ + { + "Timestamp": "2026-05-10T00:01:23.456789000Z", + "SeverityText": "ERROR", + "Body": "connection refused: redis:6379", + "ServiceName": "api-service" + } + ], + "rows": 1 + } + } + } + }, + "400": { + "description": "Invalid request parameters or query error", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "message": { + "type": "string" + } + } + } + } + } + }, + "401": { + "description": "Missing or invalid API key" + }, + "404": { + "description": "Source or connection not found", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "message": { + "type": "string" + } + } + } + } + } + }, + "500": { + "description": "Server error", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "message": { + "type": "string" + } + } + } + } + } + } + } + } + }, "/api/v2/sources": { "get": { "summary": "List Sources", diff --git a/packages/api/src/mcp/tools/query/helpers.ts b/packages/api/src/mcp/tools/query/helpers.ts index 79d34eaa79..e4b5413605 100644 --- a/packages/api/src/mcp/tools/query/helpers.ts +++ b/packages/api/src/mcp/tools/query/helpers.ts @@ -33,6 +33,9 @@ export function parseTimeRange( error: 'Invalid startTime or endTime: must be valid ISO 8601 strings', }; } + if (startDate > endDate) { + return { error: 'startTime must not be after endTime' }; + } return { startDate, endDate }; } diff --git a/packages/api/src/routers/external-api/__tests__/search.test.ts b/packages/api/src/routers/external-api/__tests__/search.test.ts new file mode 100644 index 0000000000..36f8f6ff8d --- /dev/null +++ b/packages/api/src/routers/external-api/__tests__/search.test.ts @@ -0,0 +1,560 @@ +import { SourceKind } from '@hyperdx/common-utils/dist/types'; +import mongoose from 'mongoose'; +import request from 'supertest'; + +import * as config from '../../../config'; +import { + bulkInsertLogs, + DEFAULT_DATABASE, + DEFAULT_LOGS_TABLE, + getLoggedInAgent, + getServer, +} from '../../../fixtures'; +import Connection from '../../../models/connection'; +import type { ISource } from '../../../models/source'; +import { Source } from '../../../models/source'; +import type { IUser } from '../../../models/user'; + +const DEFAULT_END_TIME = Math.floor(Date.now() / 60000) * 60000; +const DEFAULT_START_TIME = DEFAULT_END_TIME - 3600 * 1000; + +const iso = (ms: number) => new Date(ms).toISOString(); + +describe('External API v2 Search', () => { + const server = getServer(); + let agent: request.SuperTest; + let user: IUser; + let logSource: ISource; + + beforeAll(async () => { + await server.start(); + }); + + beforeEach(async () => { + const result = await getLoggedInAgent(server); + agent = result.agent; + user = result.user; + const team = result.team; + + const connection = await Connection.create({ + team: team._id, + name: 'Default', + host: config.CLICKHOUSE_HOST, + username: config.CLICKHOUSE_USER, + password: config.CLICKHOUSE_PASSWORD, + }); + + logSource = await Source.create({ + kind: SourceKind.Log, + team: team._id, + from: { + databaseName: DEFAULT_DATABASE, + tableName: DEFAULT_LOGS_TABLE, + }, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression: 'Timestamp', + connection: connection._id, + name: 'Logs', + }); + + const now = new Date(DEFAULT_END_TIME - 10000); + await bulkInsertLogs([ + { + ServiceName: 'search-svc', + Timestamp: now, + SeverityText: 'INFO', + Body: 'hello from search test', + }, + { + ServiceName: 'search-svc', + Timestamp: new Date(now.getTime() + 1000), + SeverityText: 'ERROR', + Body: 'error in search test', + }, + { + ServiceName: 'other-svc', + Timestamp: new Date(now.getTime() + 2000), + SeverityText: 'INFO', + Body: 'other service log', + }, + ]); + }); + + afterEach(async () => { + await server.clearDBs(); + }); + + afterAll(async () => { + await server.stop(); + }); + + // Helper: authenticated POST to /api/v2/search + const search = (body: Record) => + agent + .post('/api/v2/search') + .set('Authorization', `Bearer ${user.accessKey}`) + .send(body); + + // ------------------------------------------------------------------------- + // Validation + // ------------------------------------------------------------------------- + + it('returns 400 when sourceId is missing', async () => { + const res = await search({ + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when sourceId is not a valid ObjectId', async () => { + const res = await search({ + sourceId: 'not-an-objectid', + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when startTime is not a valid ISO string', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: 'not-a-date', + endTime: iso(DEFAULT_END_TIME), + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when startTime is after endTime', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_END_TIME), + endTime: iso(DEFAULT_START_TIME), + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when where exceeds 8192 characters', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + where: 'x'.repeat(8 * 1024 + 1), + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when select exceeds 4096 characters', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + select: 'x'.repeat(4 * 1024 + 1), + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when offset exceeds 10000', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + offset: 10_001, + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when maxResults exceeds 2000', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + maxResults: 9999, + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 401 when no auth token is provided', async () => { + const res = await agent.post('/api/v2/search').send({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }); + expect(res.status).toBe(401); + }); + + it('returns 404 when sourceId does not exist', async () => { + const res = await search({ + sourceId: '000000000000000000000000', + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }); + expect(res.status).toBe(404); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 404 when the connection for a source no longer exists', async () => { + const team = logSource.team; + const deadConn = await Connection.create({ + team, + name: 'Dead Connection', + host: config.CLICKHOUSE_HOST, + username: config.CLICKHOUSE_USER, + password: config.CLICKHOUSE_PASSWORD, + }); + const orphanSource = await Source.create({ + kind: SourceKind.Log, + team, + from: { databaseName: DEFAULT_DATABASE, tableName: DEFAULT_LOGS_TABLE }, + timestampValueExpression: 'Timestamp', + connection: deadConn._id, + name: 'Orphan', + }); + await Connection.deleteOne({ _id: deadConn._id }); + + const res = await search({ + sourceId: orphanSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }); + expect(res.status).toBe(404); + expect(res.body).toHaveProperty('message'); + }); + + // ------------------------------------------------------------------------- + // Basic fetch + response shape + // ------------------------------------------------------------------------- + + it('returns rows from the source with default settings', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }); + + expect(res.status).toBe(200); + expect(res.body).toHaveProperty('data'); + expect(res.body).toHaveProperty('rows'); + expect(Array.isArray(res.body.data)).toBe(true); + expect(res.body.data.length).toBeGreaterThan(0); + expect(res.body.rows).toBe(res.body.data.length); + }); + + it('returns flat row objects with column names as top-level keys (not a nested envelope)', async () => { + // Guards against the ResponseJSON-envelope bug where result.data was the + // entire { data: [...], meta, rows, statistics } object instead of the + // row array, making every response look like { data: [], rows: 1 }. + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + select: 'Timestamp,SeverityText,Body', + }); + + expect(res.status).toBe(200); + expect(res.body.data.length).toBeGreaterThan(0); + for (const row of res.body.data) { + // Each element must be a flat object with string column keys, not a + // nested { data: [...], meta, rows } envelope. + expect(typeof row.Timestamp).toBe('string'); + expect(typeof row.SeverityText).toBe('string'); + expect(typeof row.Body).toBe('string'); + expect(row).not.toHaveProperty('meta'); + expect(row).not.toHaveProperty('statistics'); + } + }); + + // ------------------------------------------------------------------------- + // Column selection + // ------------------------------------------------------------------------- + + it('returns exactly the requested select columns and no others', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + select: 'Timestamp,SeverityText,Body', + }); + + expect(res.status).toBe(200); + expect(res.body.data.length).toBeGreaterThan(0); + for (const row of res.body.data) { + expect(Object.keys(row).sort()).toEqual([ + 'Body', + 'SeverityText', + 'Timestamp', + ]); + } + }); + + // ------------------------------------------------------------------------- + // Lucene filter + // ------------------------------------------------------------------------- + + it('filters rows by Lucene where clause', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + where: 'SeverityText:ERROR', + select: 'SeverityText,Body', + }); + + expect(res.status).toBe(200); + expect(res.body.data.length).toBeGreaterThan(0); + for (const row of res.body.data) { + expect(row.SeverityText).toBe('ERROR'); + } + }); + + // ------------------------------------------------------------------------- + // Pagination + // ------------------------------------------------------------------------- + + it('respects maxResults', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + maxResults: 1, + }); + + expect(res.status).toBe(200); + expect(res.body.data.length).toBeLessThanOrEqual(1); + }); + + it('respects offset for pagination with deterministic DESC timestamp ordering', async () => { + // beforeEach inserts 3 rows: now, now+1s, now+2s. + // ORDER BY Timestamp DESC => row order: [now+2s, now+1s, now]. + // Fetching 2 rows then 1 with offset=1 must yield an exact continuation. + const page1 = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + maxResults: 2, + offset: 0, + select: 'Timestamp,SeverityText,Body', + }); + + const page2 = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + maxResults: 1, + offset: 1, + select: 'Timestamp,SeverityText,Body', + }); + + expect(page1.status).toBe(200); + expect(page2.status).toBe(200); + expect(page1.body.data.length).toBe(2); + expect(page2.body.data.length).toBe(1); + // The single row on page 2 must equal the second row on page 1 + expect(page2.body.data[0].Body).toBe(page1.body.data[1].Body); + expect(page2.body.data[0].Timestamp).toBe(page1.body.data[1].Timestamp); + }); + + // ------------------------------------------------------------------------- + // SQL where language + // ------------------------------------------------------------------------- + + it('supports SQL where language', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + where: "SeverityText = 'ERROR'", + whereLanguage: 'sql', + select: 'SeverityText,Body', + }); + + expect(res.status).toBe(200); + expect(res.body.data.length).toBeGreaterThan(0); + for (const row of res.body.data) { + expect(row.SeverityText).toBe('ERROR'); + } + }); + + it('defaults whereLanguage to lucene when omitted', async () => { + // Verifies processRequestWithEnhancedErrors applies the Zod default so + // the Lucene filter is active even without an explicit whereLanguage field. + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + where: 'SeverityText:ERROR', + // whereLanguage intentionally omitted + select: 'SeverityText,Body', + }); + expect(res.status).toBe(200); + expect(res.body.data.length).toBeGreaterThan(0); + for (const row of res.body.data) { + expect(row.SeverityText).toBe('ERROR'); + } + }); + + // ------------------------------------------------------------------------- + // Empty time range + // ------------------------------------------------------------------------- + + it('returns empty data when time range has no matching rows', async () => { + const farPast = new Date('2000-01-01T00:00:00Z').getTime(); + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(farPast), + endTime: iso(farPast + 60_000), + }); + + expect(res.status).toBe(200); + expect(res.body.data).toEqual([]); + expect(res.body.rows).toBe(0); + }); + + // ------------------------------------------------------------------------- + // Error response shape + // ------------------------------------------------------------------------- + + it('uses { message } envelope for 4xx errors', async () => { + const res = await search({ sourceId: 'not-an-objectid' }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + expect(typeof res.body.message).toBe('string'); + expect(res.body.message.length).toBeGreaterThan(0); + expect(res.body).not.toHaveProperty('error'); + }); + + it('uses { message } envelope for 404 errors', async () => { + const res = await search({ + sourceId: '000000000000000000000000', + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }); + expect(res.status).toBe(404); + expect(res.body).toHaveProperty('message'); + expect(typeof res.body.message).toBe('string'); + expect(res.body).not.toHaveProperty('error'); + }); + + it('returns 400 when whereLanguage is not lucene or sql', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + whereLanguage: 'graphql', + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when select contains a semicolon', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + select: 'Timestamp; DROP TABLE otel_logs', + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + it('returns 400 when select contains a SELECT subquery', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + select: '(SELECT max(Timestamp) FROM system.tables)', + }); + expect(res.status).toBe(400); + expect(res.body).toHaveProperty('message'); + }); + + // ------------------------------------------------------------------------- + // Boundary values + // ------------------------------------------------------------------------- + + it('accepts maxResults at the upper boundary of 2000', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + maxResults: 2000, + }); + expect(res.status).toBe(200); + expect(res.body).toHaveProperty('data'); + }); + + it('accepts offset at the upper boundary of 10000', async () => { + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + offset: 10000, + }); + expect(res.status).toBe(200); + expect(res.body.data).toEqual([]); + }); + + it('applies maxResults default of 100 when omitted', async () => { + // Insert 101 rows to ensure the default cap of 100 is observable. + await bulkInsertLogs( + Array.from({ length: 101 }, (_, i) => ({ + ServiceName: 'cap-svc', + Timestamp: new Date(DEFAULT_END_TIME - 100_000 + i * 10), + SeverityText: 'DEBUG', + Body: `row ${i}`, + })), + ); + + const res = await search({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + // maxResults intentionally omitted + }); + expect(res.status).toBe(200); + expect(res.body.rows).toBeLessThanOrEqual(100); + }); + + // ------------------------------------------------------------------------- + // Cross-team isolation + // ------------------------------------------------------------------------- + + it('returns 404 when sourceId belongs to a different team', async () => { + // The /register/password endpoint only allows one team to register, so + // create a synthetic "other team" source directly in Mongo and verify + // the authenticated user (team 1) cannot access it. + const otherTeamId = new mongoose.Types.ObjectId(); + const otherConnection = await Connection.create({ + team: otherTeamId, + name: 'Other Team Connection', + host: config.CLICKHOUSE_HOST, + username: config.CLICKHOUSE_USER, + password: config.CLICKHOUSE_PASSWORD, + }); + const otherSource = await Source.create({ + kind: SourceKind.Log, + team: otherTeamId, + from: { databaseName: DEFAULT_DATABASE, tableName: DEFAULT_LOGS_TABLE }, + timestampValueExpression: 'Timestamp', + connection: otherConnection._id, + name: 'Other Team Source', + }); + + const res = await search({ + sourceId: otherSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }); + + expect(res.status).toBe(404); + expect(res.body).toHaveProperty('message'); + }); +}); diff --git a/packages/api/src/routers/external-api/v2/index.ts b/packages/api/src/routers/external-api/v2/index.ts index f2f1b09811..cb4baafc00 100644 --- a/packages/api/src/routers/external-api/v2/index.ts +++ b/packages/api/src/routers/external-api/v2/index.ts @@ -4,6 +4,7 @@ import { validateUserAccessKey } from '@/middleware/auth'; import alertsRouter from '@/routers/external-api/v2/alerts'; import chartsRouter from '@/routers/external-api/v2/charts'; import dashboardRouter from '@/routers/external-api/v2/dashboards'; +import searchRouter from '@/routers/external-api/v2/search'; import sourcesRouter from '@/routers/external-api/v2/sources'; import webhooksRouter from '@/routers/external-api/v2/webhooks'; import rateLimiter, { rateLimiterKeyGenerator } from '@/utils/rateLimiter'; @@ -43,6 +44,8 @@ router.use( sourcesRouter, ); +router.use('/search', defaultRateLimiter, validateUserAccessKey, searchRouter); + router.use( '/webhooks', defaultRateLimiter, diff --git a/packages/api/src/routers/external-api/v2/search.ts b/packages/api/src/routers/external-api/v2/search.ts new file mode 100644 index 0000000000..37f951a6b9 --- /dev/null +++ b/packages/api/src/routers/external-api/v2/search.ts @@ -0,0 +1,432 @@ +import { ClickHouseQueryError } from '@hyperdx/common-utils/dist/clickhouse'; +import express from 'express'; +import { ObjectId } from 'mongodb'; +import { z } from 'zod'; + +import { parseTimeRange } from '@/mcp/tools/query/helpers'; +import { getNonNullUserWithTeam } from '@/middleware/auth'; +import { processRequestWithEnhancedErrors as validateRequest } from '@/utils/enhancedErrors'; +import logger from '@/utils/logger'; +import { externalDashboardSearchRequestSchema } from '@/utils/zod'; + +import { runSearchConfig, type SearchErrorCode } from './utils/search'; + +// CH error types caused by user-supplied query content — map to 400. +const CH_USER_INPUT_ERRORS = new Set([ + 'SYNTAX_ERROR', + 'UNKNOWN_IDENTIFIER', + 'UNKNOWN_TABLE', + 'UNKNOWN_FUNCTION', + 'UNKNOWN_AGGREGATE_FUNCTION', + 'UNKNOWN_TYPE', + 'BAD_ARGUMENTS', + 'CANNOT_CONVERT_TYPE', + 'THERE_IS_NO_COLUMN', + 'NO_COMMON_TYPE', + 'ILLEGAL_TYPE_OF_ARGUMENT', +]); + +/** + * @openapi + * components: + * schemas: + * SearchRequest: + * type: object + * required: + * - sourceId + * properties: + * sourceId: + * type: string + * description: | + * Source ID to query. Call GET /api/v2/sources to list available sources. + * The source determines the underlying ClickHouse table (e.g. otel.otel_logs, + * otel.otel_traces) and its column schema. + * example: "69b46cb0d964ce2d0b9506a8" + * startTime: + * type: string + * format: date-time + * description: > + * Start of the query window (ISO 8601). Defaults to 15 minutes before + * endTime. Must be before endTime. + * example: "2026-05-10T00:00:00Z" + * endTime: + * type: string + * format: date-time + * description: End of the query window (ISO 8601). Defaults to now. + * example: "2026-05-10T01:00:00Z" + * where: + * type: string + * maxLength: 8192 + * default: "" + * description: | + * Row filter expression. The language is controlled by whereLanguage. + * + * Lucene examples (default): + * SeverityText:ERROR + * pipedream.pipeline_name:my-pipeline AND SeverityText:ERROR + * Body:timeout + * + * SQL examples (whereLanguage: "sql"): + * SeverityText = 'ERROR' + * `pipedream.pipeline_name` = 'my-pipeline' + * example: "SeverityText:ERROR" + * whereLanguage: + * type: string + * enum: [lucene, sql] + * example: lucene + * default: lucene + * description: Language used for the where filter. Default is lucene. + * select: + * type: string + * maxLength: 4096 + * default: "" + * description: | + * Comma-separated list of ClickHouse column expressions to include in + * each result row. When omitted the source's default select expression + * is used. + * + * Each entry is a ClickHouse SQL expression executed under the team's + * database user. Semicolons and subqueries (SELECT keyword) are + * rejected; use column references, map lookups, or function calls only. + * + * HyperDX rewrites known attribute column names to their materialized + * equivalents automatically; you can still pass the logical name. + * example: "Timestamp,SeverityText,Body,pipedream.pipeline_name" + * orderBy: + * type: string + * maxLength: 1024 + * description: | + * ClickHouse ORDER BY expression. When omitted the source's default + * ordering (typically timestamp DESC) is used. + * example: "Timestamp DESC" + * maxResults: + * type: integer + * minimum: 1 + * maximum: 2000 + * default: 100 + * description: Maximum number of rows to return. Default is 100, max is 2000. + * offset: + * type: integer + * minimum: 0 + * maximum: 10000 + * default: 0 + * description: | + * Number of rows to skip (best-effort offset pagination). Default is + * 0, max is 10000. Offset pagination is non-deterministic when + * multiple rows share the same timestamp; for reliable deep paging + * filter by the last Timestamp value returned in the previous page + * instead of using a large offset. + * + * SearchRow: + * type: object + * description: > + * A single result row. Keys correspond to the requested columns (or the + * source default columns when columns was omitted). Values are strings + * or numbers as returned by ClickHouse. + * additionalProperties: true + * + * SearchResponse: + * type: object + * properties: + * data: + * type: array + * description: Array of result rows. Each row is an object with keys corresponding to the requested columns. + * items: + * $ref: '#/components/schemas/SearchRow' + * rows: + * type: integer + * description: Number of rows in this response (not total matching rows). + */ + +// Rejects semicolons and SELECT subqueries in column expressions. +// Word-boundary anchor prevents blocking identifiers like "selectId". +const DISALLOWED_COLUMNS_PATTERN = /;|(? ObjectId.isValid(val), { + message: 'sourceId must be a valid ObjectId', + }) + .describe( + 'Source ID to query. Call GET /api/v2/sources to list available sources.', + ), + startTime: z + .string() + .optional() + .describe( + 'Start of the query window as ISO 8601. Default: 15 minutes ago.', + ), + endTime: z + .string() + .optional() + .describe('End of the query window as ISO 8601. Default: now.'), + where: z + .string() + .max(8 * 1024) + .optional() + .default('') + .describe( + 'Row filter in Lucene syntax (default) or SQL (set whereLanguage: "sql"). ' + + 'Examples: "SeverityText:ERROR", "pipedream.pipeline_name:my-pipeline AND SeverityText:ERROR"', + ), + whereLanguage: z + .enum(['lucene', 'sql']) + .optional() + .default('lucene') + .describe('Language for the where filter. Default: lucene'), + select: z + .string() + .max(4 * 1024) + .optional() + .default('') + .refine(validateColumnsExpression, { + message: + 'select must not contain semicolons or SELECT subqueries; ' + + 'use column references, map lookups, or scalar functions only', + }) + .describe( + 'Comma-separated list of column expressions to return. ' + + 'When omitted the source default columns are used. ' + + 'Named attribute columns such as "pipedream.pipeline_name" or "k8s.pod.name" ' + + 'are automatically rewritten to their materialized equivalents when available.', + ), + orderBy: z + .string() + .max(1024) + .optional() + .describe( + 'ORDER BY expression. Defaults to the source timestamp expression DESC.', + ), + maxResults: z + .number() + .int() + .min(1) + .max(2000) + .optional() + .default(100) + .describe('Maximum rows to return (1-2000). Default: 100.'), + offset: z + .number() + .int() + .min(0) + .max(10_000) + .optional() + .default(0) + .describe( + 'Number of rows to skip for pagination (0-10000). Default: 0. ' + + 'Prefer timestamp-cursor pagination for large datasets.', + ), +}); + +/** + * @openapi + * /api/v2/search: + * post: + * summary: Search Raw Logs and Traces + * description: | + * Fetch individual log or trace rows from a HyperDX source. + * + * This endpoint mirrors the "search" panel mode in the HyperDX UI. + * HyperDX applies the same query optimizations used in the UI: + * - Named attribute columns (e.g. "pipedream.pipeline_name") are + * rewritten to their indexed materialized equivalents when the + * source schema exposes them, avoiding slow Map lookups. + * - Rows are ordered by timestamp descending (most recent first). + * - The source's built-in PREWHERE / partition pruning is applied. + * + * Authentication: Bearer token (personal API key from Team Settings). + * operationId: searchEvents + * tags: [Search] + * requestBody: + * required: true + * content: + * application/json: + * schema: + * $ref: '#/components/schemas/SearchRequest' + * examples: + * recentErrors: + * summary: Recent errors for a service + * value: + * sourceId: "69b46cb0d964ce2d0b9506a8" + * startTime: "2026-05-10T00:00:00Z" + * endTime: "2026-05-10T01:00:00Z" + * where: "SeverityText:ERROR" + * select: "Timestamp,SeverityText,Body,ServiceName" + * maxResults: 50 + * pipedreamTaskLogs: + * summary: Pipedream task logs with materialized column filter + * value: + * sourceId: "69b46cb0d964ce2d0b9506a8" + * startTime: "2026-05-10T00:00:00Z" + * endTime: "2026-05-10T01:00:00Z" + * where: "pipedream.pipeline_name:my-pipeline" + * select: "Timestamp,SeverityText,Body,pipedream.pipeline_name,pipedream.stage_name,pipedream.task_index" + * maxResults: 200 + * traceSearch: + * summary: Slow spans for a service + * value: + * sourceId: "69b46cb0d964ce2d0b9508b2" + * startTime: "2026-05-10T00:00:00Z" + * endTime: "2026-05-10T01:00:00Z" + * where: "ServiceName:my-service AND Duration:>1000000000" + * select: "Timestamp,TraceId,SpanId,SpanName,ServiceName,Duration,StatusCode" + * maxResults: 100 + * paginated: + * summary: Paginated fetch + * value: + * sourceId: "69b46cb0d964ce2d0b9506a8" + * startTime: "2026-05-10T00:00:00Z" + * endTime: "2026-05-10T01:00:00Z" + * maxResults: 100 + * offset: 100 + * responses: + * '200': + * description: Matching rows returned successfully + * content: + * application/json: + * schema: + * $ref: '#/components/schemas/SearchResponse' + * example: + * data: + * - Timestamp: "2026-05-10T00:01:23.456789000Z" + * SeverityText: "ERROR" + * Body: "connection refused: redis:6379" + * ServiceName: "api-service" + * rows: 1 + * '400': + * description: Invalid request parameters or query error + * content: + * application/json: + * schema: + * type: object + * properties: + * message: + * type: string + * '401': + * description: Missing or invalid API key + * '404': + * description: Source or connection not found + * content: + * application/json: + * schema: + * type: object + * properties: + * message: + * type: string + * '500': + * description: Server error + * content: + * application/json: + * schema: + * type: object + * properties: + * message: + * type: string + */ + +function codeToStatus(code: SearchErrorCode): number { + switch (code) { + case 'SOURCE_NOT_FOUND': + case 'CONNECTION_NOT_FOUND': + return 404; + default: { + const _exhaustive: never = code; + void _exhaustive; + return 500; + } + } +} + +const router = express.Router(); + +router.post( + '/', + validateRequest({ body: searchRequestSchema }), + async (req, res) => { + try { + const { teamId } = getNonNullUserWithTeam(req); + const { + sourceId, + startTime, + endTime, + where, + whereLanguage, + select, + orderBy, + maxResults, + offset, + } = req.body; + + const timeRange = parseTimeRange(startTime, endTime); + if ('error' in timeRange) { + return res.status(400).json({ message: timeRange.error }); + } + const { startDate, endDate } = timeRange; + + const config = externalDashboardSearchRequestSchema.parse({ + displayType: 'search' as const, + sourceId, + select, + where, + whereLanguage, + orderBy, + }); + + const result = await runSearchConfig({ + teamId: teamId.toString(), + config, + startDate, + endDate, + maxResults, + offset, + }); + + if (result.isError) { + return res + .status(codeToStatus(result.code)) + .json({ message: result.message }); + } + + return res.json({ data: result.data, rows: result.data.length }); + } catch (err) { + if (err instanceof ClickHouseQueryError) { + const chType = ((err.cause as Record | undefined) + ?.type ?? 'UNKNOWN') as string; + const safeMsg = + (err.message.split('\n')[0] ?? '').slice(0, 300) || 'Query error'; + logger.error({ chType, safeMsg }, '[search] ClickHouse query error'); + + if (CH_USER_INPUT_ERRORS.has(chType)) { + return res.status(400).json({ message: `${chType}: ${safeMsg}` }); + } + return res.status(500).json({ message: 'Query execution failed' }); + } + + const rawCode = + err != null && typeof err === 'object' + ? (err as Record).statusCode + : undefined; + const statusCode = + typeof rawCode === 'number' && rawCode >= 400 && rawCode <= 599 + ? rawCode + : 500; + if (statusCode >= 500) { + logger.error({ err }, '[search] unexpected error'); + return res + .status(statusCode) + .json({ message: 'Internal server error' }); + } + const msg = err instanceof Error ? err.message : 'Request failed'; + return res.status(statusCode).json({ message: msg }); + } + }, +); + +export default router; diff --git a/packages/api/src/routers/external-api/v2/utils/search.ts b/packages/api/src/routers/external-api/v2/utils/search.ts new file mode 100644 index 0000000000..2fc750dd3d --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/search.ts @@ -0,0 +1,172 @@ +import { ClickhouseClient } from '@hyperdx/common-utils/dist/clickhouse/node'; +import { getMetadata } from '@hyperdx/common-utils/dist/core/metadata'; +import { buildSearchChartConfig } from '@hyperdx/common-utils/dist/core/searchChartConfig'; +import { + getFirstTimestampValueExpression, + splitAndTrimWithBracket, +} from '@hyperdx/common-utils/dist/core/utils'; +import type { + ChartConfigWithDateRange, + TSource, +} from '@hyperdx/common-utils/dist/types'; +import { + DisplayType, + isLogSource, + isTraceSource, +} from '@hyperdx/common-utils/dist/types'; + +import { getConnectionById } from '@/controllers/connection'; +import { getSource } from '@/controllers/sources'; +import type { ExternalDashboardSearchRequestConfig } from '@/utils/zod'; + +export type SearchErrorCode = 'SOURCE_NOT_FOUND' | 'CONNECTION_NOT_FOUND'; + +type SearchError = { + isError: true; + code: SearchErrorCode; + message: string; +}; + +type SearchResults = { + isError: false; + data: Record[]; +}; + +// Mirrors `optimizeDefaultOrderBy` + `useDefaultOrderBy` from DBSearchPage.tsx. +// Uses `source.orderByExpression` when set, otherwise derives an ORDER BY string +// from the source's timestamp expressions. +// +// The UI version also folds in `tableMetadata.sorting_key` (fetched from CH) to +// pick up extra timestamp-like columns from the table's sort key. We skip that +// step here to avoid an extra CH round-trip on the critical path. Sources that +// need the full sorting-key-aware behaviour should set `orderByExpression`. +function resolveSearchOrderBy(source: TSource): string { + const explicit = + isLogSource(source) || isTraceSource(source) + ? source.orderByExpression?.trim() + : undefined; + if (explicit) return explicit; + + const timestampExpr = source.timestampValueExpression ?? ''; + const displayedExpr = + isLogSource(source) || isTraceSource(source) + ? source.displayedTimestampValueExpression?.trim() + : undefined; + + const timestampParts = splitAndTrimWithBracket(timestampExpr); + const candidates = displayedExpr + ? [...timestampParts, displayedExpr] + : [...timestampParts]; + + const seen = new Set(); + const orderByParts: string[] = []; + for (const key of candidates) { + if (!seen.has(key)) { + seen.add(key); + orderByParts.push(key); + } + } + + if (orderByParts.length === 0) { + orderByParts.push( + getFirstTimestampValueExpression(timestampExpr) ?? 'Timestamp', + ); + } + + return orderByParts.length > 1 + ? `(${orderByParts.join(', ')}) DESC` + : `${orderByParts[0]} DESC`; +} + +export async function runSearchConfig({ + teamId, + config, + startDate, + endDate, + maxResults, + offset, +}: { + teamId: string; + config: ExternalDashboardSearchRequestConfig; + startDate: Date; + endDate: Date; + maxResults: number; + offset: number; +}): Promise { + const source = await getSource(teamId, config.sourceId); + if (!source) { + return { + isError: true, + code: 'SOURCE_NOT_FOUND', + message: `Source not found: ${config.sourceId}`, + }; + } + + const connection = await getConnectionById( + teamId, + source.connection.toString(), + true, + ); + if (!connection) { + return { + isError: true, + code: 'CONNECTION_NOT_FOUND', + message: `Connection not found for source: ${config.sourceId}`, + }; + } + + // Set client-side HTTP timeout slightly above the source's max_execution_time + // so CH can return a clean error first. value=0 means no server limit. + const maxExecSetting = source.querySettings?.find( + s => s.setting === 'max_execution_time', + ); + const maxExecSeconds = maxExecSetting ? Number(maxExecSetting.value) : NaN; + const requestTimeout = + maxExecSeconds > 0 && isFinite(maxExecSeconds) + ? maxExecSeconds * 1000 + 2_000 + : undefined; + + const clickhouseClient = new ClickhouseClient({ + host: connection.host, + username: connection.username, + password: connection.password, + ...(requestTimeout != null ? { requestTimeout } : {}), + }); + + const searchBase = buildSearchChartConfig(source, { + where: typeof config.where === 'string' ? config.where : '', + whereLanguage: config.whereLanguage ?? 'lucene', + select: config.select ?? null, + displayType: DisplayType.Search, + orderBy: config.orderBy?.trim() || resolveSearchOrderBy(source), + dateRange: [startDate, endDate], + }); + + // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion + const chartConfig: ChartConfigWithDateRange = { + ...searchBase, + connection: source.connection.toString(), + limit: { limit: maxResults, offset }, + } as ChartConfigWithDateRange; + + const metadata = getMetadata(clickhouseClient); + const result = await clickhouseClient.queryChartConfig({ + config: chartConfig, + metadata, + querySettings: source.querySettings, + }); + + if ( + result == null || + typeof result !== 'object' || + !('data' in result) || + !Array.isArray((result as { data: unknown }).data) + ) { + throw new Error('Unexpected ClickHouse response shape: missing data array'); + } + + return { + isError: false, + data: (result as { data: unknown[] }).data as Record[], + }; +} diff --git a/packages/api/src/utils/swagger.ts b/packages/api/src/utils/swagger.ts index c81756b2ef..6cf9192176 100644 --- a/packages/api/src/utils/swagger.ts +++ b/packages/api/src/utils/swagger.ts @@ -40,6 +40,11 @@ export const swaggerOptions = { name: 'Webhooks', description: 'Endpoints for managing webhooks', }, + { + name: 'Search', + description: + 'Endpoints for querying raw data from log and trace sources', + }, ], components: { securitySchemes: { diff --git a/packages/api/src/utils/zod.ts b/packages/api/src/utils/zod.ts index 6cc1c7abc0..b2d3e53ebb 100644 --- a/packages/api/src/utils/zod.ts +++ b/packages/api/src/utils/zod.ts @@ -367,6 +367,17 @@ const externalDashboardSearchChartConfigSchema = z.object({ whereLanguage: whereLanguageSchema, }); +// Extended schema for the /api/v2/search endpoint — adds orderBy which is not +// applicable to dashboard tiles. +export const externalDashboardSearchRequestSchema = + externalDashboardSearchChartConfigSchema.extend({ + orderBy: z.string().max(1024).optional(), + }); + +export type ExternalDashboardSearchRequestConfig = z.infer< + typeof externalDashboardSearchRequestSchema +>; + const externalDashboardMarkdownChartConfigSchema = z.object({ displayType: z.literal('markdown'), markdown: z.string().max(50000).optional(),