From db55515f7a50e413e81bfa4e519aa0dbe96a01bf Mon Sep 17 00:00:00 2001 From: Stephen Wright Date: Wed, 16 Sep 2026 10:00:42 +0000 Subject: [PATCH] feat(Databricks Node): Add polling transport for job runs and pipeline events (no-changelog) (#38629) --- .../nodes/Databricks/actions/helpers.ts | 59 +-- .../nodes/Databricks/actions/interfaces.ts | 1 + .../eslint-user-agent-restriction.mjs | 51 +- .../nodes/Databricks/methods/listSearch.ts | 7 +- .../Databricks/test/Databricks.node.test.ts | 12 +- .../nodes/Databricks/test/helpers.test.ts | 36 +- .../nodes/Databricks/test/transport.test.ts | 444 ++++++++++++++++++ .../nodes/Databricks/transport/index.ts | 17 + .../nodes/Databricks/transport/jobRuns.ts | 95 ++++ .../nodes/Databricks/transport/pagination.ts | 36 ++ .../Databricks/transport/pipelineEvents.ts | 175 +++++++ 11 files changed, 845 insertions(+), 88 deletions(-) create mode 100644 packages/nodes-base/nodes/Databricks/test/transport.test.ts create mode 100644 packages/nodes-base/nodes/Databricks/transport/index.ts create mode 100644 packages/nodes-base/nodes/Databricks/transport/jobRuns.ts create mode 100644 packages/nodes-base/nodes/Databricks/transport/pagination.ts create mode 100644 packages/nodes-base/nodes/Databricks/transport/pipelineEvents.ts diff --git a/packages/nodes-base/nodes/Databricks/actions/helpers.ts b/packages/nodes-base/nodes/Databricks/actions/helpers.ts index 33a0ab8eece..cd2af816c34 100644 --- a/packages/nodes-base/nodes/Databricks/actions/helpers.ts +++ b/packages/nodes-base/nodes/Databricks/actions/helpers.ts @@ -6,38 +6,34 @@ import type { IExecuteFunctions, IHttpRequestOptions, ILoadOptionsFunctions, + IPollFunctions, } from 'n8n-workflow'; import { DATABRICKS_PARTNER_USER_AGENT } from '../constants'; import type { DatabricksCredentials, OpenAPISchema } from './interfaces'; +export type DatabricksContext = IExecuteFunctions | ILoadOptionsFunctions | IPollFunctions; +export type DatabricksCredentialType = 'databricksApi' | 'databricksOAuth2Api'; + const RATE_LIMIT_MAX_RETRIES = 3; const RATE_LIMIT_FALLBACK_DELAY_MS = 1_000; const RATE_LIMIT_MAX_DELAY_MS = 30_000; /** - * Single egress point for the Databricks API, so every request carries the - * partner User-Agent. Enforced by eslint-user-agent-restriction.mjs. - * - * Takes `context` explicitly rather than the house `this`-binding style because - * some callers (e.g. `fetchResourcesInSchema` in methods/listSearch.ts) are plain - * functions with no `this`. - * - * Setting a User-Agent deliberately opts these calls out of the instance-wide - * outbound UA, including `N8N_GLOBAL_USER_AGENT_VALUE` — partner attribution - * requires a single predictable token. + * Single egress point for the Databricks API, enforced by eslint-user-agent-restriction.mjs. + * Setting a User-Agent opts these calls out of N8N_GLOBAL_USER_AGENT_VALUE on purpose: + * partner attribution needs one predictable token. */ export async function databricksApiRequest( - context: IExecuteFunctions | ILoadOptionsFunctions, - credentialType: 'databricksApi' | 'databricksOAuth2Api', + context: DatabricksContext, + credentialType: DatabricksCredentialType, options: IHttpRequestOptions, ): ReturnType { const requestOptions: IHttpRequestOptions = { ...options, headers: { ...options.headers, - // Last, so a caller cannot override it 'User-Agent': DATABRICKS_PARTNER_USER_AGENT, }, }; @@ -63,44 +59,37 @@ export async function databricksApiRequest( } export function getActiveCredentialType( - context: IExecuteFunctions | ILoadOptionsFunctions, + context: DatabricksContext, itemIndex = 0, -): 'databricksApi' | 'databricksOAuth2Api' { - const authentication = context.getNodeParameter( - 'authentication', - itemIndex, - 'accessToken', - ) as string; +): DatabricksCredentialType { + const authentication = + 'getInputData' in context + ? context.getNodeParameter('authentication', itemIndex, 'accessToken') + : context.getNodeParameter('authentication', 'accessToken'); return authentication === 'oAuth2' ? 'databricksOAuth2Api' : 'databricksApi'; } export async function getHost( - context: IExecuteFunctions | ILoadOptionsFunctions, - credentialType: 'databricksApi' | 'databricksOAuth2Api', + context: DatabricksContext, + credentialType: DatabricksCredentialType, ): Promise { const credentials = await context.getCredentials(credentialType); return credentials.host.replace(/\/$/, ''); } -// Body text comes from whatever server `host` points at — truncate and strip -// control chars before promoting it to a visible error message +// Body text comes from whatever server `host` points at export function sanitizeApiMessage(message: string): string { // eslint-disable-next-line no-control-regex return message.replace(/[\x00-\x1f\x7f]+/g, ' ').slice(0, 500); } -// Must be called at every request entry point (router catch, listSearch wrapper): -// databricksApiRequest() only attaches the User-Agent and deliberately does not -// wrap errors, so callers still own their catch. Keyed on PERMISSION_DENIED only; widen -// the key if other Databricks error_codes with legible messages show up. Keyed on -// the error_code, not HTTP 403, so expired-token 403s (which core retries via -// refresh) aren't mislabeled if they leak through. Mutates rather than re-wraps: -// `new NodeApiError(node, existingNodeApiError)` returns the original untouched. +// Called at every request entry point (router catch, listSearch wrapper) because +// databricksApiRequest() does not wrap errors. Keyed on error_code, not HTTP 403, so +// expired-token 403s are not mislabeled. Mutates: re-wrapping a NodeApiError returns the same instance. export function makePermissionErrorLegible(error: unknown): void { if (!(error instanceof NodeApiError)) return; - // Requests with encoding: 'arraybuffer' (file downloads) receive their 403 - // JSON body as raw bytes, so parse Buffer/string bodies before reading it + // File downloads receive the 403 body as raw bytes let data = error.context.data; if (Buffer.isBuffer(data) || typeof data === 'string') { try { @@ -292,9 +281,7 @@ export function generateExampleFromSchema(schema: unknown, format: string): stri if (Object.keys(exampleObj).length > 0) { return JSON.stringify(exampleObj, null, 2); } - } catch (e) { - // Fall through to default examples - } + } catch {} } const examples: Record = { diff --git a/packages/nodes-base/nodes/Databricks/actions/interfaces.ts b/packages/nodes-base/nodes/Databricks/actions/interfaces.ts index 3e355b91204..2581b6e3939 100644 --- a/packages/nodes-base/nodes/Databricks/actions/interfaces.ts +++ b/packages/nodes-base/nodes/Databricks/actions/interfaces.ts @@ -74,6 +74,7 @@ export type DatabricksJobRunTask = { start_time?: number; end_time?: number; status?: DatabricksJobRunStatus; + /** @deprecated Jobs API 2.2 reports `status` instead */ state?: DatabricksJobRunLegacyState; }; diff --git a/packages/nodes-base/nodes/Databricks/eslint-user-agent-restriction.mjs b/packages/nodes-base/nodes/Databricks/eslint-user-agent-restriction.mjs index 1e31685e11f..6d1e2b4ff45 100644 --- a/packages/nodes-base/nodes/Databricks/eslint-user-agent-restriction.mjs +++ b/packages/nodes-base/nodes/Databricks/eslint-user-agent-restriction.mjs @@ -1,11 +1,8 @@ -// Forces every Databricks API call through databricksApiRequest() in -// actions/helpers.ts, which is what attaches the partner User-Agent. Spread into -// packages/nodes-base's eslint.config.mjs, which is where the `files`/`ignores` -// globs below are resolved from (this file's own location doesn't affect path -// resolution). +// Every Databricks API call must go through databricksApiRequest() in actions/helpers.ts. +// Spread into packages/nodes-base/eslint.config.mjs; the globs below resolve from there. // Flat-config `rules` blocks replace, not merge, for any file matched by more -// than one block — so the scoped `no-restricted-syntax` below must repeat the +// than one block, so the scoped `no-restricted-syntax` below must repeat the // base config's raw-enum rule, or it silently stops applying there. const NO_RAW_ENUM_SYNTAX_RULE = { selector: 'TSEnumDeclaration:not([const=true])', @@ -13,38 +10,42 @@ const NO_RAW_ENUM_SYNTAX_RULE = { 'Do not declare raw enums as it leads to runtime overhead. Use const enum instead. See https://www.typescriptlang.org/docs/handbook/enums.html#const-enums', }; -// Raw request helpers that would skip the partner User-Agent, shared by both -// selectors below so the list can't drift between them. +// `request` also names unrelated members (error.request), so it stays anchored to `helpers`. const RESTRICTED_HELPER_METHODS = - '/^(httpRequest|httpRequestWithAuthentication|request|requestWithAuthentication|requestWithAuthenticationPaginated)$/'; + '/^(httpRequest|httpRequestWithAuthentication|request|requestWithAuthentication|requestWithAuthenticationPaginated|requestOAuth1|requestOAuth2)$/'; +const DISTINCTIVE_HELPER_METHODS = + '/^(httpRequest|httpRequestWithAuthentication|requestWithAuthentication|requestWithAuthenticationPaginated|requestOAuth1|requestOAuth2)$/'; +const USE_DATABRICKS_API_REQUEST = + 'Use databricksApiRequest() from actions/helpers.ts so the partner User-Agent is sent.'; export const databricksUserAgentRestriction = [ { files: ['./nodes/Databricks/**/*.ts'], - // helpers.ts is the one file allowed to call the raw helper; tests stub it directly. + // helpers.ts is the one file allowed to call the raw helpers; tests stub them directly. ignores: ['./nodes/Databricks/actions/helpers.ts', './nodes/Databricks/test/**'], rules: { 'no-restricted-syntax': [ 'error', NO_RAW_ENUM_SYNTAX_RULE, { - // Siblings of httpRequestWithAuthentication are restricted too: an - // operation written with helpers.httpRequest would otherwise skip - // the User-Agent silently. Anchored to `helpers` (as a member or a bare - // binding from `const { helpers } = this`) so unrelated members like - // `error.request` don't trip the guard. - selector: `MemberExpression:matches([object.property.name="helpers"], [object.name="helpers"])[property.name=${RESTRICTED_HELPER_METHODS}]`, - message: - 'Use databricksApiRequest() from actions/helpers.ts so the partner User-Agent is sent.', + selector: `MemberExpression[computed=false][property.name=${DISTINCTIVE_HELPER_METHODS}]`, + message: USE_DATABRICKS_API_REQUEST, }, { - // The same helpers reached by destructuring rather than member access. - // Anchored to a `*.helpers` or bare `helpers` initialiser so an unrelated - // `const { request } = response` doesn't trip the guard with a confusing - // message. - selector: `VariableDeclarator:matches([init.name="helpers"], [init.property.name="helpers"]) > ObjectPattern > Property[key.name=${RESTRICTED_HELPER_METHODS}]`, - message: - 'Use databricksApiRequest() from actions/helpers.ts so the partner User-Agent is sent.', + selector: `MemberExpression[computed=true] > Literal.property[value=${RESTRICTED_HELPER_METHODS}]`, + message: USE_DATABRICKS_API_REQUEST, + }, + { + selector: `MemberExpression:matches([object.property.name="helpers"], [object.name="helpers"])[property.name="request"]`, + message: USE_DATABRICKS_API_REQUEST, + }, + { + selector: `ObjectPattern > Property[key.name=${DISTINCTIVE_HELPER_METHODS}]`, + message: USE_DATABRICKS_API_REQUEST, + }, + { + selector: `VariableDeclarator:matches([init.name="helpers"], [init.property.name="helpers"]) > ObjectPattern > Property[key.name="request"]`, + message: USE_DATABRICKS_API_REQUEST, }, ], }, diff --git a/packages/nodes-base/nodes/Databricks/methods/listSearch.ts b/packages/nodes-base/nodes/Databricks/methods/listSearch.ts index dfe9cb0b43e..de4d730c67c 100644 --- a/packages/nodes-base/nodes/Databricks/methods/listSearch.ts +++ b/packages/nodes-base/nodes/Databricks/methods/listSearch.ts @@ -12,6 +12,7 @@ import { getHost, makePermissionErrorLegible, sanitizeApiMessage, + type DatabricksCredentialType, } from '../actions/helpers'; import type { DatabricksJobRun } from '../actions/interfaces'; @@ -19,7 +20,7 @@ import type { DatabricksJobRun } from '../actions/interfaces'; // doesn't cover them — apply it here for every listSearch call site instead async function listRequest( context: ILoadOptionsFunctions, - credentialType: 'databricksApi' | 'databricksOAuth2Api', + credentialType: DatabricksCredentialType, options: IHttpRequestOptions, ): Promise { try { @@ -208,7 +209,7 @@ export async function getSchemas( async function fetchResourcesInSchema( context: ILoadOptionsFunctions, - credentialType: 'databricksApi' | 'databricksOAuth2Api', + credentialType: DatabricksCredentialType, host: string, apiPath: string, catalogName: string, @@ -434,7 +435,7 @@ type JobsListPage = { jobs?: JobSummary[]; next_page_token?: string }; async function fetchListPage( context: ILoadOptionsFunctions, - credentialType: 'databricksApi' | 'databricksOAuth2Api', + credentialType: DatabricksCredentialType, host: string, path: string, limit: number, diff --git a/packages/nodes-base/nodes/Databricks/test/Databricks.node.test.ts b/packages/nodes-base/nodes/Databricks/test/Databricks.node.test.ts index 11d301b484b..20435b646a3 100644 --- a/packages/nodes-base/nodes/Databricks/test/Databricks.node.test.ts +++ b/packages/nodes-base/nodes/Databricks/test/Databricks.node.test.ts @@ -56,6 +56,12 @@ const apiErrorFromBody = (status: number, data: unknown) => }) as unknown as JsonObject, ); +const createExecuteContext = () => { + const context = mockDeep(); + context.getInputData.mockReturnValue([]); + return context; +}; + describe('Databricks', () => { const credentials = { databricksApi: { @@ -799,7 +805,7 @@ describe('listSearch -> PERMISSION_DENIED surfaces the Databricks message', () = describe('Databricks SQL -> Execute Query (FAILED/CANCELED statement)', () => { const setupContext = (status: unknown) => { - const context = mockDeep(); + const context = createExecuteContext(); context.getNode.mockReturnValue(node); context.getNodeParameter.mockImplementation((name) => { if (name === 'warehouseId') return 'warehouse123'; @@ -882,7 +888,7 @@ describe('Databricks SQL -> Execute Query (async polling)', () => { }); it('should poll until the statement reaches SUCCEEDED and map rows to items', async () => { - const context = mockDeep(); + const context = createExecuteContext(); context.getNode.mockReturnValue(node); context.getNodeParameter.mockImplementation((name) => { if (name === 'warehouseId') return 'warehouse123'; @@ -962,7 +968,7 @@ describe('Job -> Run (wait for completion)', () => { options: { timeout: 10 }, ...overrides, }; - const context = mockDeep(); + const context = createExecuteContext(); context.getNode.mockReturnValue(node); context.getExecutionCancelSignal.mockReturnValue(cancelSignal); context.getNodeParameter.mockImplementation((name, index) => diff --git a/packages/nodes-base/nodes/Databricks/test/helpers.test.ts b/packages/nodes-base/nodes/Databricks/test/helpers.test.ts index a0c586735fb..35728b5be9b 100644 --- a/packages/nodes-base/nodes/Databricks/test/helpers.test.ts +++ b/packages/nodes-base/nodes/Databricks/test/helpers.test.ts @@ -5,13 +5,13 @@ import type { IHttpRequestOptions, ILoadOptionsFunctions, INode, + IPollFunctions, JsonObject, } from 'n8n-workflow'; import type { Mock } from 'vitest'; import { mock, mockDeep } from 'vitest-mock-extended'; -import { databricksApiRequest, readIdParameter } from '../actions/helpers'; -import { DATABRICKS_PARTNER_USER_AGENT } from '../constants'; +import { databricksApiRequest, getActiveCredentialType, readIdParameter } from '../actions/helpers'; vi.mock('@n8n/utils/sleep', () => ({ sleep: vi.fn().mockResolvedValue(undefined), @@ -86,8 +86,6 @@ describe('databricksApiRequest', () => { }); it('should forward the credential type and bind the call to the passed context', async () => { - // `fetchResourcesInSchema` in methods/listSearch.ts passes a context object - // rather than `this`, so the receiver must come from the argument. const loadOptionsContext = mock({ getNode: () => mock({ typeVersion: 1 }), helpers: { httpRequestWithAuthentication }, @@ -101,27 +99,23 @@ describe('databricksApiRequest', () => { expect(httpRequestWithAuthentication.mock.calls[0][0]).toBe('databricksOAuth2Api'); expect(httpRequestWithAuthentication.mock.instances[0]).toBe(loadOptionsContext); }); +}); - it('should track the integration version rather than the node typeVersion', async () => { - // A node instance pinned to an older typeVersion must still report the version - // of the integration that is actually running. - const staleContext = mock({ - getNode: () => mock({ typeVersion: 0.1 }), - helpers: { httpRequestWithAuthentication }, - }); +describe('getActiveCredentialType', () => { + it('should read the authentication parameter of the given item on an execute context', () => { + const context = mock({ getInputData: vi.fn() }); + context.getNodeParameter.mockReturnValue('oAuth2'); - await databricksApiRequest(staleContext, 'databricksApi', { - method: 'GET', - url: 'https://example.databricks.com/api/2.1/unity-catalog/catalogs', - }); - - expect(capturedOptions().headers).toEqual({ - 'User-Agent': DATABRICKS_PARTNER_USER_AGENT, - }); + expect(getActiveCredentialType(context, 2)).toBe('databricksOAuth2Api'); + expect(context.getNodeParameter).toHaveBeenCalledWith('authentication', 2, 'accessToken'); }); - it('should send the unversioned partner User-Agent', () => { - expect(DATABRICKS_PARTNER_USER_AGENT).toBe('n8n_DatabricksNode'); + it('should read the authentication parameter without an item index on a polling context', () => { + const context = mock(); + context.getNodeParameter.mockReturnValue('accessToken'); + + expect(getActiveCredentialType(context)).toBe('databricksApi'); + expect(context.getNodeParameter).toHaveBeenCalledWith('authentication', 'accessToken'); }); }); diff --git a/packages/nodes-base/nodes/Databricks/test/transport.test.ts b/packages/nodes-base/nodes/Databricks/test/transport.test.ts new file mode 100644 index 00000000000..833c38c8fd7 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/test/transport.test.ts @@ -0,0 +1,444 @@ +import type { INode, IPollFunctions } from 'n8n-workflow'; +import { NodeOperationError, UnexpectedError } from 'n8n-workflow'; +import { mock, mockDeep } from 'vitest-mock-extended'; + +import { DATABRICKS_PARTNER_USER_AGENT } from '../constants'; +import { + DEFAULT_MAX_PAGES, + JOB_RUNS_MAX_PAGE_SIZE, + listAllJobRuns, + listAllPipelineEvents, + listJobRuns, + listPipelineEvents, + PIPELINE_EVENT_LEVELS, + PIPELINE_EVENTS_MAX_PAGE_SIZE, + type ListJobRunsParams, + type ListPipelineEventsParams, + type PageLimits, + type PipelineEventLevel, +} from '../transport'; +import { clampPageSize } from '../transport/pagination'; +import { buildPipelineEventsFilter } from '../transport/pipelineEvents'; + +const HOST = 'https://adb-1234567890.1.azuredatabricks.net'; +const JOB_ID = 281874479417551; +const PIPELINE_ID = '8199cd89-e2f5-4169-a6aa-656a24c8886d'; +const CURSOR = '2026-09-01T14:20:50Z'; +const NO_DEADLINE: PageLimits = { deadlineEpochMs: Number.POSITIVE_INFINITY }; + +const run = (runId: number) => ({ job_id: JOB_ID, run_id: runId, status: { state: 'TERMINATED' } }); +const event = (id: string) => ({ id, event_type: 'update_progress', timestamp: CURSOR }); + +const createPollContext = () => { + const context = mockDeep(); + context.getNode.mockReturnValue(mock({ name: 'Databricks Trigger', typeVersion: 1 })); + context.getCredentials.mockResolvedValue({ host: HOST }); + return context; +}; +const apiMock = (context: ReturnType) => + context.helpers.httpRequestWithAuthentication; +const requestQuery = (context: ReturnType, call = 0) => + apiMock(context).mock.calls[call][1].qs; + +describe('clampPageSize', () => { + it.each([ + ['an in-range size', 7, 7], + ['a size below the range', 0, 1], + ['a fractional size', 7.9, 7], + ['no size', undefined, undefined], + ['a non-numeric size', Number.NaN, undefined], + ])('clamps %s', (_label, value, expected) => { + expect(clampPageSize(value, 25)).toBe(expected); + }); +}); + +describe('listJobRuns', () => { + it('requests one page with the given filters and the partner User-Agent', async () => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({ runs: [run(1)], next_page_token: 'page-2' }); + + const page = await listJobRuns(context, 'databricksApi', { + jobId: JOB_ID, + startTimeFromMs: 1788260000000, + state: 'completed', + expandTasks: true, + pageSize: 25, + }); + + expect(apiMock(context)).toHaveBeenCalledWith( + 'databricksApi', + expect.objectContaining({ + method: 'GET', + url: `${HOST}/api/2.2/jobs/runs/list`, + qs: { + job_id: JOB_ID, + start_time_from: 1788260000000, + completed_only: true, + expand_tasks: true, + limit: 25, + }, + headers: expect.objectContaining({ 'User-Agent': DATABRICKS_PARTNER_USER_AGENT }), + }), + ); + expect(page).toEqual({ items: [run(1)], nextPageToken: 'page-2' }); + }); + + const queryCases: Array<[string, ListJobRunsParams, Record]> = [ + ['no parameters', {}, {}], + ['a false expandTasks flag', { expandTasks: false }, {}], + ['active runs only', { state: 'active' }, { active_only: true }], + [ + 'a time window', + { startTimeFromMs: 1, startTimeToMs: 2 }, + { start_time_from: 1, start_time_to: 2 }, + ], + ['a page token', { pageToken: 'p2' }, { page_token: 'p2' }], + ['an empty page token', { pageToken: '' }, {}], + ['a page size above the range', { pageSize: 100 }, { limit: JOB_RUNS_MAX_PAGE_SIZE }], + ]; + + it.each(queryCases)('sends the query for %s', async (_label, params, qs) => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({ runs: [] }); + + await listJobRuns(context, 'databricksOAuth2Api', params); + + expect(apiMock(context)).toHaveBeenCalledWith( + 'databricksOAuth2Api', + expect.objectContaining({ qs }), + ); + }); + + it.each([ + ['a page without runs', {}, { items: [] }], + ['an empty token', { runs: [run(1)], next_page_token: '' }, { items: [run(1)] }], + ])('normalises %s', async (_label, response, expected) => { + const context = createPollContext(); + apiMock(context).mockResolvedValue(response); + + await expect(listJobRuns(context, 'databricksApi')).resolves.toEqual(expected); + }); + + it.each([ + ['an HTML body', 'sign in'], + ['a runs field that is not a list', { runs: 'none' }], + ])('rejects %s instead of returning an empty page', async (_label, response) => { + const context = createPollContext(); + apiMock(context).mockResolvedValue(response); + + await expect(listJobRuns(context, 'databricksApi')).rejects.toThrow(NodeOperationError); + }); +}); + +describe('listAllJobRuns', () => { + let nowMs: number; + + beforeEach(() => { + nowMs = 1_000; + vi.spyOn(Date, 'now').mockImplementation(() => nowMs); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('follows next_page_token from the start token, keeping the filters, until the last page', async () => { + const context = createPollContext(); + apiMock(context) + .mockResolvedValueOnce({ runs: [run(1), run(2)], next_page_token: 'p2' }) + .mockResolvedValueOnce({ runs: [run(3)] }); + + const result = await listAllJobRuns( + context, + 'databricksApi', + { jobId: JOB_ID, pageSize: 2, pageToken: 'p1' }, + NO_DEADLINE, + ); + + expect(result).toEqual({ items: [run(1), run(2), run(3)] }); + expect(requestQuery(context, 0)).toEqual({ job_id: JOB_ID, limit: 2, page_token: 'p1' }); + expect(requestQuery(context, 1)).toEqual({ job_id: JOB_ID, limit: 2, page_token: 'p2' }); + expect(context.getCredentials).toHaveBeenCalledTimes(1); + }); + + const caps: Array<[string, PageLimits, number]> = [ + ['the given page cap', { ...NO_DEADLINE, maxPages: 3 }, 3], + ['the default page cap', NO_DEADLINE, DEFAULT_MAX_PAGES], + [ + 'the page cap when the deadline is not a number', + { maxPages: 3, deadlineEpochMs: Number.NaN }, + 3, + ], + ]; + + it.each(caps)('stops at %s and hands back the next token', async (_label, limits, pages) => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({ runs: [run(1)], next_page_token: 'more' }); + + const result = await listAllJobRuns(context, 'databricksApi', {}, limits); + + expect(apiMock(context)).toHaveBeenCalledTimes(pages); + expect(result).toEqual({ items: Array(pages).fill(run(1)), nextPageToken: 'more' }); + }); + + const spentDeadlines: Array<[string, PageLimits]> = [ + ['the deadline equals the clock', { deadlineEpochMs: 1_000 }], + ['the deadline has passed', { deadlineEpochMs: 0 }], + ['the deadline has passed before the page cap', { maxPages: 3, deadlineEpochMs: 0 }], + ]; + + it.each(spentDeadlines)( + 'fetches one page and hands back the token when %s', + async (_label, limits) => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({ runs: [run(1)], next_page_token: 'more' }); + + const result = await listAllJobRuns(context, 'databricksApi', {}, limits); + + expect(apiMock(context)).toHaveBeenCalledTimes(1); + expect(result).toEqual({ items: [run(1)], nextPageToken: 'more' }); + }, + ); + + it('stops between pages once the clock passes the deadline', async () => { + const context = createPollContext(); + const pages = [ + { runs: [run(1)], next_page_token: 'p2' }, + { runs: [run(2)], next_page_token: 'p3' }, + { runs: [run(3)] }, + ]; + apiMock(context).mockImplementation(async () => { + nowMs += 600; + return pages.shift(); + }); + + const result = await listAllJobRuns( + context, + 'databricksApi', + { jobId: JOB_ID }, + { deadlineEpochMs: 2_100 }, + ); + + expect(apiMock(context)).toHaveBeenCalledTimes(2); + expect(requestQuery(context, 1)).toEqual({ job_id: JOB_ID, page_token: 'p2' }); + expect(result).toEqual({ items: [run(1), run(2)], nextPageToken: 'p3' }); + expect(context.getCredentials).toHaveBeenCalledTimes(1); + }); + + it('returns no token when the last page lands after the deadline', async () => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({ runs: [run(1)] }); + + await expect( + listAllJobRuns(context, 'databricksApi', {}, { deadlineEpochMs: 0 }), + ).resolves.toEqual({ items: [run(1)] }); + }); +}); + +describe('buildPipelineEventsFilter', () => { + const cases: Array< + [string, Pick, string | undefined] + > = [ + ['nothing', {}, undefined], + ['no levels', { levels: [] }, undefined], + ['a cursor', { after: CURSOR }, `timestamp > '${CURSOR}'`], + [ + 'a cursor with fractional seconds', + { after: '2026-09-01T14:20:31.066Z' }, + "timestamp > '2026-09-01T14:20:31.066Z'", + ], + ['levels', { levels: ['ERROR', 'WARN'] }, "level in ('ERROR', 'WARN')"], + [ + 'every level', + { levels: PIPELINE_EVENT_LEVELS }, + "level in ('INFO', 'WARN', 'ERROR', 'METRICS')", + ], + [ + 'levels and a cursor', + { after: CURSOR, levels: ['ERROR'] }, + `level in ('ERROR') AND timestamp > '${CURSOR}'`, + ], + ]; + + it.each(cases)('builds the filter for %s', (_label, params, expected) => { + expect(buildPipelineEventsFilter(params)).toBe(expected); + }); + + it.each([ + '2026-09-01', + '2026-09-01T14:20:50+02:00', + "2026-09-01T14:20:50Z' OR level='INFO", + '2026-02-30T14:20:50Z', + '2026-09-01T25:00:00Z', + ])('rejects the cursor %s', (after) => { + expect(() => buildPipelineEventsFilter({ after })).toThrow(UnexpectedError); + }); + + it.each(['DEBUG', 'error'])('rejects the level %s', (level) => { + const levels = ['ERROR', level] as PipelineEventLevel[]; + expect(() => buildPipelineEventsFilter({ levels })).toThrow(UnexpectedError); + }); +}); + +describe('listPipelineEvents', () => { + it('requests events after the cursor in ascending order with the partner User-Agent', async () => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({ events: [event('e1')], next_page_token: 'p2' }); + + const page = await listPipelineEvents(context, 'databricksApi', { + pipelineId: PIPELINE_ID, + after: CURSOR, + levels: ['ERROR', 'WARN'], + pageSize: 25, + }); + + expect(apiMock(context)).toHaveBeenCalledWith( + 'databricksApi', + expect.objectContaining({ + method: 'GET', + url: `${HOST}/api/2.0/pipelines/${PIPELINE_ID}/events`, + qs: { + max_results: 25, + filter: `level in ('ERROR', 'WARN') AND timestamp > '${CURSOR}'`, + order_by: 'timestamp asc', + }, + headers: expect.objectContaining({ 'User-Agent': DATABRICKS_PARTNER_USER_AGENT }), + }), + ); + expect(page).toEqual({ items: [event('e1')], nextPageToken: 'p2' }); + }); + + const queryCases: Array< + [string, Omit, Record] + > = [ + ['ascending order by default', {}, { order_by: 'timestamp asc' }], + ['descending order', { order: 'desc' }, { order_by: 'timestamp desc' }], + [ + 'a page size above the range', + { pageSize: 5000 }, + { max_results: PIPELINE_EVENTS_MAX_PAGE_SIZE, order_by: 'timestamp asc' }, + ], + [ + 'a continuation, without the filter', + { after: CURSOR, levels: ['ERROR'], pageSize: 25, pageToken: 'p2' }, + { max_results: 25, page_token: 'p2' }, + ], + [ + 'an empty page token', + { after: CURSOR, pageToken: '' }, + { filter: `timestamp > '${CURSOR}'`, order_by: 'timestamp asc' }, + ], + ]; + + it.each(queryCases)('sends the query for %s', async (_label, params, qs) => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({}); + + await listPipelineEvents(context, 'databricksOAuth2Api', { + pipelineId: PIPELINE_ID, + ...params, + }); + + expect(apiMock(context)).toHaveBeenCalledWith( + 'databricksOAuth2Api', + expect.objectContaining({ qs }), + ); + }); + + it.each(['pipelines', '../updates', `${PIPELINE_ID}/events`, ''])( + 'rejects the pipeline ID %j before any request', + async (pipelineId) => { + const context = createPollContext(); + + await expect(listPipelineEvents(context, 'databricksApi', { pipelineId })).rejects.toThrow( + NodeOperationError, + ); + expect(apiMock(context)).not.toHaveBeenCalled(); + }, + ); + + it('rejects an invalid cursor before any request', async () => { + const context = createPollContext(); + + await expect( + listPipelineEvents(context, 'databricksApi', { + pipelineId: PIPELINE_ID, + after: '2026-09-01', + }), + ).rejects.toThrow(UnexpectedError); + expect(apiMock(context)).not.toHaveBeenCalled(); + }); + + it('normalises a page without events', async () => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({}); + + await expect( + listPipelineEvents(context, 'databricksApi', { pipelineId: PIPELINE_ID }), + ).resolves.toEqual({ items: [] }); + }); + + it.each([ + ['an HTML body', 'sign in'], + ['an events field that is not a list', { events: 'none' }], + ])('rejects %s instead of returning an empty page', async (_label, response) => { + const context = createPollContext(); + apiMock(context).mockResolvedValue(response); + + await expect( + listPipelineEvents(context, 'databricksApi', { pipelineId: PIPELINE_ID }), + ).rejects.toThrow(NodeOperationError); + }); +}); + +describe('listAllPipelineEvents', () => { + it('follows next_page_token until the last page', async () => { + const context = createPollContext(); + apiMock(context) + .mockResolvedValueOnce({ events: [event('e1')], next_page_token: 'p2' }) + .mockResolvedValueOnce({ events: [event('e2')] }); + + const result = await listAllPipelineEvents( + context, + 'databricksApi', + { pipelineId: PIPELINE_ID, after: CURSOR, pageSize: 25 }, + NO_DEADLINE, + ); + + expect(result).toEqual({ items: [event('e1'), event('e2')] }); + expect(requestQuery(context, 0)).toEqual({ + max_results: 25, + filter: `timestamp > '${CURSOR}'`, + order_by: 'timestamp asc', + }); + expect(requestQuery(context, 1)).toEqual({ max_results: 25, page_token: 'p2' }); + expect(context.getCredentials).toHaveBeenCalledTimes(1); + }); + + const stops: Array<[string, PageLimits, number]> = [ + ['the page cap', { ...NO_DEADLINE, maxPages: 2 }, 2], + ['a spent deadline', { deadlineEpochMs: 0 }, 1], + ]; + + it.each(stops)( + 'stops at %s, resuming from the given token, and hands back the next token', + async (_label, limits, pages) => { + const context = createPollContext(); + apiMock(context).mockResolvedValue({ events: [event('e1')], next_page_token: 'more' }); + + const result = await listAllPipelineEvents( + context, + 'databricksApi', + { pipelineId: PIPELINE_ID, pageToken: 'p1' }, + limits, + ); + + expect(requestQuery(context, 0)).toEqual({ page_token: 'p1' }); + expect(apiMock(context)).toHaveBeenCalledTimes(pages); + expect(result).toEqual({ + items: Array(pages).fill(event('e1')), + nextPageToken: 'more', + }); + }, + ); +}); diff --git a/packages/nodes-base/nodes/Databricks/transport/index.ts b/packages/nodes-base/nodes/Databricks/transport/index.ts new file mode 100644 index 00000000000..17a404ed270 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/transport/index.ts @@ -0,0 +1,17 @@ +export { + JOB_RUNS_MAX_PAGE_SIZE, + listAllJobRuns, + listJobRuns, + type ListJobRunsParams, +} from './jobRuns'; +export { DEFAULT_MAX_PAGES, type Page, type PageLimits } from './pagination'; +export { + isPipelineEventLevel, + listAllPipelineEvents, + listPipelineEvents, + PIPELINE_EVENT_LEVELS, + PIPELINE_EVENTS_MAX_PAGE_SIZE, + type ListPipelineEventsParams, + type PipelineEvent, + type PipelineEventLevel, +} from './pipelineEvents'; diff --git a/packages/nodes-base/nodes/Databricks/transport/jobRuns.ts b/packages/nodes-base/nodes/Databricks/transport/jobRuns.ts new file mode 100644 index 00000000000..f62cf0a3967 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/transport/jobRuns.ts @@ -0,0 +1,95 @@ +import { isRecord } from '@n8n/utils/is-record'; +import type { IDataObject } from 'n8n-workflow'; +import { NodeOperationError } from 'n8n-workflow'; + +import { + databricksApiRequest, + getHost, + type DatabricksContext, + type DatabricksCredentialType, +} from '../actions/helpers'; +import type { DatabricksJobRun } from '../actions/interfaces'; +import { clampPageSize, collectPages, toPage, type Page, type PageLimits } from './pagination'; + +export const JOB_RUNS_MAX_PAGE_SIZE = 25; + +export interface ListJobRunsParams { + jobId?: number; + startTimeFromMs?: number; + startTimeToMs?: number; + state?: 'active' | 'completed'; + expandTasks?: boolean; + pageSize?: number; + pageToken?: string; +} + +type JobRunsListResponse = { runs?: DatabricksJobRun[]; next_page_token?: string }; + +function isJobRunsListResponse(value: unknown): value is JobRunsListResponse { + return isRecord(value) && (value.runs === undefined || Array.isArray(value.runs)); +} + +function toQuery(params: ListJobRunsParams): IDataObject { + const qs: IDataObject = {}; + if (params.jobId !== undefined) qs.job_id = params.jobId; + if (params.startTimeFromMs !== undefined) qs.start_time_from = params.startTimeFromMs; + if (params.startTimeToMs !== undefined) qs.start_time_to = params.startTimeToMs; + if (params.state === 'active') qs.active_only = true; + if (params.state === 'completed') qs.completed_only = true; + if (params.expandTasks) qs.expand_tasks = true; + const limit = clampPageSize(params.pageSize, JOB_RUNS_MAX_PAGE_SIZE); + if (limit !== undefined) qs.limit = limit; + if (params.pageToken) qs.page_token = params.pageToken; + return qs; +} + +async function fetchJobRunsPage( + context: DatabricksContext, + credentialType: DatabricksCredentialType, + host: string, + params: ListJobRunsParams, +): Promise> { + const response: unknown = await databricksApiRequest(context, credentialType, { + method: 'GET', + url: `${host}/api/2.2/jobs/runs/list`, + qs: toQuery(params), + headers: { Accept: 'application/json' }, + json: true, + }); + if (!isJobRunsListResponse(response)) { + throw new NodeOperationError( + context.getNode(), + 'Databricks did not return a JSON list of job runs', + { description: `Check that ${host} is the URL of a Databricks workspace.` }, + ); + } + return toPage(response.runs, response.next_page_token); +} + +export async function listJobRuns( + context: DatabricksContext, + credentialType: DatabricksCredentialType, + params: ListJobRunsParams = {}, +): Promise> { + return await fetchJobRunsPage( + context, + credentialType, + await getHost(context, credentialType), + params, + ); +} + +export async function listAllJobRuns( + context: DatabricksContext, + credentialType: DatabricksCredentialType, + params: ListJobRunsParams, + limits: PageLimits, +): Promise> { + const host = await getHost(context, credentialType); + return await collectPages( + async (pageToken) => + await fetchJobRunsPage(context, credentialType, host, { ...params, pageToken }), + limits, + params.pageToken, + ); +} diff --git a/packages/nodes-base/nodes/Databricks/transport/pagination.ts b/packages/nodes-base/nodes/Databricks/transport/pagination.ts new file mode 100644 index 00000000000..108ee61125c --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/transport/pagination.ts @@ -0,0 +1,36 @@ +export const DEFAULT_MAX_PAGES = 40; + +export interface Page { + items: T[]; + nextPageToken?: string; +} + +export interface PageLimits { + deadlineEpochMs: number; + maxPages?: number; +} + +export function toPage(items: T[] | undefined, nextPageToken: string | undefined): Page { + return { items: items ?? [], nextPageToken: nextPageToken || undefined }; +} + +export function clampPageSize(value: number | undefined, max: number): number | undefined { + if (value === undefined || !Number.isFinite(value)) return undefined; + return Math.min(Math.max(Math.trunc(value), 1), max); +} + +export async function collectPages( + fetchPage: (pageToken?: string) => Promise>, + { deadlineEpochMs, maxPages = DEFAULT_MAX_PAGES }: PageLimits, + startToken?: string, +): Promise> { + const items: T[] = []; + let pageToken = startToken; + for (let page = 0; page < maxPages; page++) { + const result = await fetchPage(pageToken); + items.push(...result.items); + pageToken = result.nextPageToken; + if (pageToken === undefined || Date.now() >= deadlineEpochMs) break; + } + return { items, nextPageToken: pageToken }; +} diff --git a/packages/nodes-base/nodes/Databricks/transport/pipelineEvents.ts b/packages/nodes-base/nodes/Databricks/transport/pipelineEvents.ts new file mode 100644 index 00000000000..f9dec800af1 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/transport/pipelineEvents.ts @@ -0,0 +1,175 @@ +import { isRecord } from '@n8n/utils/is-record'; +import type { IDataObject } from 'n8n-workflow'; +import { NodeOperationError, UnexpectedError } from 'n8n-workflow'; + +import { + databricksApiRequest, + getHost, + type DatabricksContext, + type DatabricksCredentialType, +} from '../actions/helpers'; +import { clampPageSize, collectPages, toPage, type Page, type PageLimits } from './pagination'; + +export const PIPELINE_EVENTS_MAX_PAGE_SIZE = 1000; +const PIPELINE_ID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +const ISO_TIMESTAMP_PATTERN = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{1,9})?Z$/; + +export const PIPELINE_EVENT_LEVELS = ['INFO', 'WARN', 'ERROR', 'METRICS'] as const; +export type PipelineEventLevel = (typeof PIPELINE_EVENT_LEVELS)[number]; +export type IsoUtcTimestamp = string; + +export type PipelineEvent = { + id?: string; + sequence?: { + control_plane_seq_no?: number; + data_plane_id?: { instance?: string; seq_no?: number }; + }; + event_type?: string; + level?: PipelineEventLevel; + message?: string; + timestamp?: IsoUtcTimestamp; + maturity_level?: string; + origin?: { + pipeline_id?: string; + pipeline_name?: string; + update_id?: string; + flow_id?: string; + flow_name?: string; + [key: string]: unknown; + }; + details?: { update_progress?: { state?: string }; [key: string]: unknown }; + error?: { + fatal?: boolean; + exceptions?: Array<{ + class_name?: string; + error_class?: string; + sql_state?: string; + message?: string; + stack?: Array<{ + declaring_class?: string; + method_name?: string; + file_name?: string; + line_number?: number; + }>; + }>; + }; + truncation?: { truncated_fields?: Array<{ field_name?: string }> }; +}; + +export interface ListPipelineEventsParams { + pipelineId: string; + after?: IsoUtcTimestamp; + levels?: readonly PipelineEventLevel[]; + order?: 'asc' | 'desc'; + pageSize?: number; + pageToken?: string; +} + +type PipelineEventsResponse = { events?: PipelineEvent[]; next_page_token?: string }; + +function isPipelineEventsResponse(value: unknown): value is PipelineEventsResponse { + return isRecord(value) && (value.events === undefined || Array.isArray(value.events)); +} + +export function isPipelineEventLevel(level: string): level is PipelineEventLevel { + return PIPELINE_EVENT_LEVELS.some((known) => known === level); +} + +function isIsoUtcTimestamp(value: string): boolean { + if (!ISO_TIMESTAMP_PATTERN.test(value)) return false; + const parsed = new Date(value); + return !Number.isNaN(parsed.getTime()) && parsed.toISOString().startsWith(value.slice(0, 19)); +} + +export function buildPipelineEventsFilter( + params: Pick, +): string | undefined { + const clauses: string[] = []; + if (params.levels?.length) { + const invalid = params.levels.filter((level) => !isPipelineEventLevel(level)); + if (invalid.length > 0) { + throw new UnexpectedError('Pipeline event level is not one of INFO, WARN, ERROR, METRICS', { + extra: { levels: invalid }, + }); + } + clauses.push(`level in (${params.levels.map((level) => `'${level}'`).join(', ')})`); + } + if (params.after !== undefined) { + if (!isIsoUtcTimestamp(params.after)) { + throw new UnexpectedError('Pipeline events cursor must be an ISO 8601 UTC timestamp', { + extra: { after: params.after }, + }); + } + clauses.push(`timestamp > '${params.after}'`); + } + return clauses.length > 0 ? clauses.join(' AND ') : undefined; +} + +// The API rejects any field other than max_results next to page_token +function toQuery(params: ListPipelineEventsParams): IDataObject { + const qs: IDataObject = {}; + const pageSize = clampPageSize(params.pageSize, PIPELINE_EVENTS_MAX_PAGE_SIZE); + if (pageSize !== undefined) qs.max_results = pageSize; + if (params.pageToken) { + qs.page_token = params.pageToken; + return qs; + } + const filter = buildPipelineEventsFilter(params); + if (filter !== undefined) qs.filter = filter; + qs.order_by = `timestamp ${params.order ?? 'asc'}`; + return qs; +} + +async function fetchPipelineEventsPage( + context: DatabricksContext, + credentialType: DatabricksCredentialType, + host: string, + params: ListPipelineEventsParams, +): Promise> { + if (!PIPELINE_ID_PATTERN.test(params.pipelineId)) { + throw new NodeOperationError(context.getNode(), 'Pipeline ID must be a UUID'); + } + const response: unknown = await databricksApiRequest(context, credentialType, { + method: 'GET', + url: `${host}/api/2.0/pipelines/${params.pipelineId}/events`, + qs: toQuery(params), + headers: { Accept: 'application/json' }, + json: true, + }); + if (!isPipelineEventsResponse(response)) { + throw new NodeOperationError( + context.getNode(), + 'Databricks did not return a JSON list of pipeline events', + { description: `Check that ${host} is the URL of a Databricks workspace.` }, + ); + } + return toPage(response.events, response.next_page_token); +} + +export async function listPipelineEvents( + context: DatabricksContext, + credentialType: DatabricksCredentialType, + params: ListPipelineEventsParams, +): Promise> { + return await fetchPipelineEventsPage( + context, + credentialType, + await getHost(context, credentialType), + params, + ); +} + +export async function listAllPipelineEvents( + context: DatabricksContext, + credentialType: DatabricksCredentialType, + params: ListPipelineEventsParams, + limits: PageLimits, +): Promise> { + const host = await getHost(context, credentialType); + return await collectPages( + async (pageToken) => + await fetchPipelineEventsPage(context, credentialType, host, { ...params, pageToken }), + limits, + params.pageToken, + ); +}