From 13c8a3bbd472fb17a18e8514827f4b016b64fe06 Mon Sep 17 00:00:00 2001 From: Stephen Wright Date: Tue, 15 Sep 2026 14:20:37 +0000 Subject: [PATCH] feat(Databricks Node): Add the Get Run Output operation to the hidden Job resource (no-changelog) (#38709) --- .../nodes/Databricks/actions/helpers.ts | 56 ++- .../nodes/Databricks/actions/interfaces.ts | 65 ++- .../Databricks/actions/job/Job.resource.ts | 1 + .../actions/job/getRunOutput.operation.ts | 103 ++++ .../Databricks/actions/job/run.operation.ts | 19 +- .../nodes/Databricks/methods/listSearch.ts | 87 +++- .../Databricks/resources/job/operations.ts | 6 + .../Databricks/resources/job/parameters.ts | 50 ++ .../Databricks/test/Databricks.node.test.ts | 37 +- .../Databricks/test/getRunOutput.test.ts | 441 ++++++++++++++++++ .../nodes/Databricks/test/helpers.test.ts | 207 +++++++- .../test/job-get-run-output.workflow.json | 78 ++++ 12 files changed, 1107 insertions(+), 43 deletions(-) create mode 100644 packages/nodes-base/nodes/Databricks/actions/job/getRunOutput.operation.ts create mode 100644 packages/nodes-base/nodes/Databricks/test/getRunOutput.test.ts create mode 100644 packages/nodes-base/nodes/Databricks/test/job-get-run-output.workflow.json diff --git a/packages/nodes-base/nodes/Databricks/actions/helpers.ts b/packages/nodes-base/nodes/Databricks/actions/helpers.ts index 2739d3f1ef3..33a0ab8eece 100644 --- a/packages/nodes-base/nodes/Databricks/actions/helpers.ts +++ b/packages/nodes-base/nodes/Databricks/actions/helpers.ts @@ -1,4 +1,6 @@ -import { NodeApiError, UserError } from 'n8n-workflow'; +import { retryabilityFromError } from '@n8n/backend-network'; +import { sleep } from '@n8n/utils/sleep'; +import { NodeApiError, NodeOperationError, UserError } from 'n8n-workflow'; import type { IDataObject, IExecuteFunctions, @@ -10,6 +12,10 @@ import { DATABRICKS_PARTNER_USER_AGENT } from '../constants'; import type { DatabricksCredentials, OpenAPISchema } from './interfaces'; +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. @@ -27,14 +33,33 @@ export async function databricksApiRequest( credentialType: 'databricksApi' | 'databricksOAuth2Api', options: IHttpRequestOptions, ): ReturnType { - return await context.helpers.httpRequestWithAuthentication.call(context, credentialType, { + const requestOptions: IHttpRequestOptions = { ...options, headers: { ...options.headers, // Last, so a caller cannot override it 'User-Agent': DATABRICKS_PARTNER_USER_AGENT, }, - }); + }; + const abortSignal = + 'getExecutionCancelSignal' in context ? context.getExecutionCancelSignal() : undefined; + + for (let attempt = 0; ; attempt++) { + try { + return await context.helpers.httpRequestWithAuthentication.call( + context, + credentialType, + requestOptions, + ); + } catch (error) { + const { status, retryAfterMs } = retryabilityFromError(error); + if (status !== 429 || attempt >= RATE_LIMIT_MAX_RETRIES) throw error; + await sleep( + Math.min(retryAfterMs || RATE_LIMIT_FALLBACK_DELAY_MS, RATE_LIMIT_MAX_DELAY_MS), + abortSignal, + ); + } + } } export function getActiveCredentialType( @@ -96,6 +121,31 @@ export function makePermissionErrorLegible(error: unknown): void { } } +export function readIdParameter( + context: IExecuteFunctions, + itemIndex: number, + parameterName: string, + noun: string, +): number { + const label = `${noun.charAt(0).toUpperCase()}${noun.slice(1)} ID`; + const value = String( + context.getNodeParameter(parameterName, itemIndex, '', { extractValue: true }), + ); + if (!/^[0-9]+$/.test(value)) { + throw new NodeOperationError(context.getNode(), `${label} must be a whole number`, { + itemIndex, + description: `Use the numeric ID shown in the ${noun} URL in Databricks.`, + }); + } + if (!Number.isSafeInteger(Number(value))) { + throw new NodeOperationError(context.getNode(), `${label} is too large to send exactly`, { + itemIndex, + description: `IDs above ${Number.MAX_SAFE_INTEGER} lose precision in JavaScript, so the node cannot send this ${noun} ID.`, + }); + } + return Number(value); +} + export function extractResourceLocatorValue(param: unknown): string { if (typeof param === 'object' && param !== null) { return (param as { value?: string }).value || ''; diff --git a/packages/nodes-base/nodes/Databricks/actions/interfaces.ts b/packages/nodes-base/nodes/Databricks/actions/interfaces.ts index 911a1307989..3e355b91204 100644 --- a/packages/nodes-base/nodes/Databricks/actions/interfaces.ts +++ b/packages/nodes-base/nodes/Databricks/actions/interfaces.ts @@ -52,25 +52,64 @@ export type DatabricksRunNowResponse = { number_in_job?: number; }; +export type DatabricksJobRunStatus = { + state?: string; + termination_details?: { + code?: string; + type?: string; + message?: string; + }; +}; + +export type DatabricksJobRunLegacyState = { + life_cycle_state?: string; + result_state?: string; + state_message?: string; +}; + +export type DatabricksJobRunTask = { + task_key?: string; + run_id?: number; + run_page_url?: string; + start_time?: number; + end_time?: number; + status?: DatabricksJobRunStatus; + state?: DatabricksJobRunLegacyState; +}; + export type DatabricksJobRun = { job_id?: number; run_id?: number; run_name?: string; + run_type?: string; run_page_url?: string; + trigger?: string; + creator_user_name?: string; start_time?: number; end_time?: number; - status?: { - state?: string; - termination_details?: { - code?: string; - type?: string; - message?: string; - }; - }; + queue_duration?: number; + run_duration?: number; + status?: DatabricksJobRunStatus; /** @deprecated Jobs API 2.2 reports `status` instead */ - state?: { - life_cycle_state?: string; - result_state?: string; - state_message?: string; - }; + state?: DatabricksJobRunLegacyState; + tasks?: DatabricksJobRunTask[]; + has_more?: boolean; + next_page_token?: string; + job_parameters?: Array<{ name?: string; default?: string; value?: string }>; +}; + +export type DatabricksNotebookOutput = { result?: string; truncated?: boolean }; + +export type DatabricksRunOutput = { + metadata?: DatabricksJobRun; + notebook_output?: DatabricksNotebookOutput; + sql_output?: Record; + dbt_output?: Record; + run_job_output?: { run_id?: number }; + clean_rooms_notebook_output?: { notebook_output?: DatabricksNotebookOutput }; + logs?: string; + logs_truncated?: boolean; + error?: string; + error_trace?: string; + info?: string; }; diff --git a/packages/nodes-base/nodes/Databricks/actions/job/Job.resource.ts b/packages/nodes-base/nodes/Databricks/actions/job/Job.resource.ts index d14badbdd71..ced7f3a82d5 100644 --- a/packages/nodes-base/nodes/Databricks/actions/job/Job.resource.ts +++ b/packages/nodes-base/nodes/Databricks/actions/job/Job.resource.ts @@ -1 +1,2 @@ +export * as getRunOutput from './getRunOutput.operation'; export * as run from './run.operation'; diff --git a/packages/nodes-base/nodes/Databricks/actions/job/getRunOutput.operation.ts b/packages/nodes-base/nodes/Databricks/actions/job/getRunOutput.operation.ts new file mode 100644 index 00000000000..2d0b1b410b5 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/actions/job/getRunOutput.operation.ts @@ -0,0 +1,103 @@ +import { NodeOperationError } from 'n8n-workflow'; +import type { IDataObject, IExecuteFunctions, INodeExecutionData } from 'n8n-workflow'; + +import { + databricksApiRequest, + getActiveCredentialType, + getHost, + readIdParameter, +} from '../helpers'; +import type { DatabricksJobRun, DatabricksJobRunTask, DatabricksRunOutput } from '../interfaces'; + +const TASK_PAGES_MAX = 20; + +type RunTarget = { runId: number; taskKey?: string }; + +async function fetchRunPage( + context: IExecuteFunctions, + credentialType: 'databricksApi' | 'databricksOAuth2Api', + host: string, + runId: number, + pageToken?: string, +): Promise { + const qs: IDataObject = { run_id: runId }; + if (pageToken) qs.page_token = pageToken; + return await databricksApiRequest(context, credentialType, { + method: 'GET', + url: `${host}/api/2.2/jobs/runs/get`, + qs, + headers: { Accept: 'application/json' }, + json: true, + }); +} + +async function resolveTargets( + context: IExecuteFunctions, + credentialType: 'databricksApi' | 'databricksOAuth2Api', + host: string, + runId: number, + itemIndex: number, +): Promise<{ run: DatabricksJobRun; targets: RunTarget[] }> { + const run = await fetchRunPage(context, credentialType, host, runId); + const tasks: DatabricksJobRunTask[] = [...(run.tasks ?? [])]; + let pageToken = run.next_page_token; + for (let page = 1; pageToken && page < TASK_PAGES_MAX; page++) { + const next = await fetchRunPage(context, credentialType, host, runId, pageToken); + tasks.push(...(next.tasks ?? [])); + pageToken = next.next_page_token; + } + if (pageToken) { + throw new NodeOperationError( + context.getNode(), + `Run ${runId} has more tasks than the node can read`, + { + itemIndex, + description: `The node reads at most ${TASK_PAGES_MAX * 100} tasks of one run. Read the remaining outputs by their task run IDs.`, + }, + ); + } + + const targets = tasks + .filter((task): task is DatabricksJobRunTask & { run_id: number } => task.run_id !== undefined) + .map((task) => ({ runId: task.run_id, taskKey: task.task_key })); + return { run, targets: targets.length > 0 ? targets : [{ runId }] }; +} + +function isTruncated(output: DatabricksRunOutput): boolean { + return ( + output.notebook_output?.truncated === true || + output.clean_rooms_notebook_output?.notebook_output?.truncated === true || + output.logs_truncated === true + ); +} + +export async function execute(this: IExecuteFunctions, i: number): Promise { + const credentialType = getActiveCredentialType(this, i); + const host = await getHost(this, credentialType); + const runId = readIdParameter(this, i, 'runId', 'run'); + + const { run, targets } = await resolveTargets(this, credentialType, host, runId, i); + + const items: INodeExecutionData[] = []; + for (const target of targets) { + const output: DatabricksRunOutput = await databricksApiRequest(this, credentialType, { + method: 'GET', + url: `${host}/api/2.2/jobs/runs/get-output`, + qs: { run_id: target.runId }, + headers: { Accept: 'application/json' }, + json: true, + }); + items.push({ + json: { + run_id: runId, + job_id: run.job_id, + task_key: target.taskKey, + task_run_id: target.runId, + truncated: isTruncated(output), + ...output, + }, + pairedItem: { item: i }, + }); + } + return items; +} diff --git a/packages/nodes-base/nodes/Databricks/actions/job/run.operation.ts b/packages/nodes-base/nodes/Databricks/actions/job/run.operation.ts index 8acb16eda56..78f994426c0 100644 --- a/packages/nodes-base/nodes/Databricks/actions/job/run.operation.ts +++ b/packages/nodes-base/nodes/Databricks/actions/job/run.operation.ts @@ -7,6 +7,7 @@ import { databricksApiRequest, getActiveCredentialType, getHost, + readIdParameter, sanitizeApiMessage, } from '../helpers'; import type { DatabricksJobRun, DatabricksRunNowResponse } from '../interfaces'; @@ -56,21 +57,7 @@ function describeRunPage(run: DatabricksJobRun): string | undefined { export async function execute(this: IExecuteFunctions, i: number): Promise { const credentialType = getActiveCredentialType(this, i); const host = await getHost(this, credentialType); - const jobId = String(this.getNodeParameter('jobId', i, '', { extractValue: true })); - - if (!/^[0-9]+$/.test(jobId)) { - throw new NodeOperationError(this.getNode(), 'Job ID must be a whole number', { - itemIndex: i, - description: 'Use the numeric ID shown in the job URL in Databricks.', - }); - } - if (!Number.isSafeInteger(Number(jobId))) { - throw new NodeOperationError(this.getNode(), 'Job ID is too large to send exactly', { - itemIndex: i, - description: - 'IDs above 9007199254740991 lose precision in JavaScript, so the node cannot run this job.', - }); - } + const jobId = readIdParameter(this, i, 'jobId', 'job'); const jobParameters = readJobParameters(this, i); const waitForCompletion = this.getNodeParameter('waitForCompletion', i, false) === true; @@ -88,7 +75,7 @@ export async function execute(this: IExecuteFunctions, i: number): Promise 0 && { job_parameters: jobParameters }), }, headers: { 'Content-Type': 'application/json' }, diff --git a/packages/nodes-base/nodes/Databricks/methods/listSearch.ts b/packages/nodes-base/nodes/Databricks/methods/listSearch.ts index 083d197e57c..dfe9cb0b43e 100644 --- a/packages/nodes-base/nodes/Databricks/methods/listSearch.ts +++ b/packages/nodes-base/nodes/Databricks/methods/listSearch.ts @@ -13,6 +13,7 @@ import { makePermissionErrorLegible, sanitizeApiMessage, } from '../actions/helpers'; +import type { DatabricksJobRun } from '../actions/interfaces'; // Dropdown requests never pass through the router, so its permission-error hook // doesn't cover them — apply it here for every listSearch call site instead @@ -431,17 +432,19 @@ const JOBS_SEARCH_MAX_PAGES = 10; type JobSummary = { job_id: number; settings?: { name?: string } }; type JobsListPage = { jobs?: JobSummary[]; next_page_token?: string }; -async function fetchJobsPage( +async function fetchListPage( context: ILoadOptionsFunctions, credentialType: 'databricksApi' | 'databricksOAuth2Api', host: string, + path: string, + limit: number, pageToken?: string, -): Promise { - const qs: IDataObject = { limit: JOBS_PAGE_SIZE }; +): Promise { + const qs: IDataObject = { limit }; if (pageToken) qs.page_token = pageToken; - return await listRequest(context, credentialType, { + return await listRequest(context, credentialType, { method: 'GET', - url: `${host}/api/2.2/jobs/list`, + url: `${host}${path}`, qs, headers: { Accept: 'application/json' }, json: true, @@ -461,8 +464,18 @@ export async function getJobs( url: `${host}/jobs/${job.job_id}`, }); + const fetchPage = async (pageToken?: string) => + await fetchListPage( + this, + credentialType, + host, + '/api/2.2/jobs/list', + JOBS_PAGE_SIZE, + pageToken, + ); + if (!filter) { - const page = await fetchJobsPage(this, credentialType, host, paginationToken); + const page = await fetchPage(paginationToken); return { results: (page.jobs ?? []).map(toListItem), paginationToken: page.next_page_token }; } @@ -471,7 +484,7 @@ export async function getJobs( const results: INodeListSearchResult['results'] = []; let pageToken = paginationToken; for (let page = 0; page < JOBS_SEARCH_MAX_PAGES && (page === 0 || pageToken); page++) { - const response = await fetchJobsPage(this, credentialType, host, pageToken); + const response = await fetchPage(pageToken); results.push( ...(response.jobs ?? []) .filter((job) => (job.settings?.name ?? '').toLowerCase().includes(filterLower)) @@ -482,3 +495,63 @@ export async function getJobs( return { results, paginationToken: pageToken }; } + +const RUNS_PAGE_SIZE = 25; +const RUNS_SEARCH_MAX_PAGES = 10; + +type RunsListPage = { runs?: DatabricksJobRun[]; next_page_token?: string }; + +function describeRun(run: DatabricksJobRun): string { + const state = run.status?.state ?? run.state?.life_cycle_state; + const outcome = run.status?.termination_details?.code ?? run.state?.result_state; + const startedAt = run.start_time + ? `${new Date(run.start_time).toISOString().replace('T', ' ').slice(0, 19)} UTC` + : undefined; + return [run.run_name || `Job ${run.job_id}`, outcome ?? state, startedAt, `Run ${run.run_id}`] + .filter(Boolean) + .join(' · '); +} + +export async function getRuns( + this: ILoadOptionsFunctions, + filter?: string, + paginationToken?: string, +): Promise { + const credentialType = getActiveCredentialType(this); + const host = await getHost(this, credentialType); + const toListItem = (run: DatabricksJobRun) => ({ + name: describeRun(run), + value: String(run.run_id), + url: run.run_page_url, + }); + + const fetchPage = async (pageToken?: string) => + await fetchListPage( + this, + credentialType, + host, + '/api/2.2/jobs/runs/list', + RUNS_PAGE_SIZE, + pageToken, + ); + + if (!filter) { + const page = await fetchPage(paginationToken); + return { results: (page.runs ?? []).map(toListItem), paginationToken: page.next_page_token }; + } + + const filterLower = filter.toLowerCase(); + const results: INodeListSearchResult['results'] = []; + let pageToken = paginationToken; + for (let page = 0; page < RUNS_SEARCH_MAX_PAGES && (page === 0 || pageToken); page++) { + const response = await fetchPage(pageToken); + results.push( + ...(response.runs ?? []) + .map(toListItem) + .filter((item) => item.name.toLowerCase().includes(filterLower)), + ); + pageToken = response.next_page_token; + } + + return { results, paginationToken: pageToken }; +} diff --git a/packages/nodes-base/nodes/Databricks/resources/job/operations.ts b/packages/nodes-base/nodes/Databricks/resources/job/operations.ts index fdc2d24abba..1c59b246357 100644 --- a/packages/nodes-base/nodes/Databricks/resources/job/operations.ts +++ b/packages/nodes-base/nodes/Databricks/resources/job/operations.ts @@ -11,6 +11,12 @@ export const jobOperations: INodeProperties = { }, }, options: [ + { + name: 'Get Run Output', + value: 'getRunOutput', + description: 'Get what the tasks of a job run produced', + action: 'Get run output', + }, { name: 'Run', value: 'run', diff --git a/packages/nodes-base/nodes/Databricks/resources/job/parameters.ts b/packages/nodes-base/nodes/Databricks/resources/job/parameters.ts index 2da858cb8a8..6f29fa1f994 100644 --- a/packages/nodes-base/nodes/Databricks/resources/job/parameters.ts +++ b/packages/nodes-base/nodes/Databricks/resources/job/parameters.ts @@ -3,8 +3,58 @@ import type { INodeProperties } from 'n8n-workflow'; import { JOB_RUN_DEFAULT_TIMEOUT_SECONDS } from '../../constants'; const showForRun = { resource: ['job'], operation: ['run'] }; +const showForGetRunOutput = { resource: ['job'], operation: ['getRunOutput'] }; export const jobParameters: INodeProperties[] = [ + { + displayName: 'Run', + name: 'runId', + type: 'resourceLocator', + default: { mode: 'list', value: '' }, + required: true, + description: 'The job run to read the output of', + displayOptions: { + show: showForGetRunOutput, + }, + modes: [ + { + displayName: 'From List', + name: 'list', + type: 'list', + typeOptions: { + searchListMethod: 'getRuns', + searchable: true, + }, + }, + { + displayName: 'By ID', + name: 'id', + type: 'string', + placeholder: 'e.g. 41847992357943', + validation: [ + { + type: 'regex', + properties: { + regex: '^[0-9]+$', + errorMessage: 'Must be a numeric run ID', + }, + }, + ], + }, + { + displayName: 'By URL', + name: 'url', + type: 'string', + placeholder: + 'e.g. https://adb-xxx.azuredatabricks.net/jobs/281874479417551/runs/41847992357943', + extractValue: { + type: 'regex', + regex: + 'https://[^/]+/(?:jobs/[0-9]+/runs|\\?o=[0-9]+#job/[0-9]+/run|#job/[0-9]+/run)/([0-9]+)', + }, + }, + ], + }, { displayName: 'Job', name: 'jobId', 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 6c5b5ecfa28..11d301b484b 100644 --- a/packages/nodes-base/nodes/Databricks/test/Databricks.node.test.ts +++ b/packages/nodes-base/nodes/Databricks/test/Databricks.node.test.ts @@ -15,7 +15,7 @@ import { mockDeep } from 'vitest-mock-extended'; import { execute as executeQuery } from '../actions/databricksSql/executeQuery.operation'; import { makePermissionErrorLegible } from '../actions/helpers'; import { execute as runJob } from '../actions/job/run.operation'; -import { getCatalogs, getJobs, getSchemas } from '../methods/listSearch'; +import { getCatalogs, getJobs, getRuns, getSchemas } from '../methods/listSearch'; import { jobParameters } from '../resources/job/parameters'; // The operation is imported from source, so this mock replaces the real poll delay @@ -632,6 +632,40 @@ describe('Databricks', () => { }); }); + describe('Job -> Get Run Output', () => { + beforeAll(() => { + const databricksNock = nock(HOST); + databricksNock + .get('/api/2.2/jobs/runs/get') + .query({ run_id: '41847992357943' }) + .matchHeader('user-agent', 'n8n_DatabricksNode') + .reply(200, { + job_id: 281874479417551, + run_id: 41847992357943, + tasks: [{ task_key: 'main', run_id: 41847992357944 }], + }); + databricksNock + .get('/api/2.2/jobs/runs/get-output') + .query({ run_id: '41847992357944' }) + .matchHeader('user-agent', 'n8n_DatabricksNode') + .reply(200, { + metadata: { + run_id: 41847992357944, + task_key: 'main', + status: { state: 'TERMINATED', termination_details: { code: 'SUCCESS' } }, + }, + notebook_output: { result: 'done', truncated: false }, + }); + }); + + afterAll(() => nock.cleanAll()); + + new NodeTestHarness().setupTests({ + credentials, + workflowFiles: ['job-get-run-output.workflow.json'], + }); + }); + describe('Router -> PERMISSION_DENIED surfaces the Databricks message', () => { // A 403 PERMISSION_DENIED body must surface its legible Databricks message // instead of the generic "Forbidden - perhaps check your credentials?" — @@ -734,6 +768,7 @@ describe('listSearch -> PERMISSION_DENIED surfaces the Databricks message', () = it.each([ ['getCatalogs', getCatalogs], ['getJobs', getJobs], + ['getRuns', getRuns], ])('should reject with the legible message from %s', async (_name, method) => { const context = mockDeep(); context.getNodeParameter.mockReturnValue('accessToken'); diff --git a/packages/nodes-base/nodes/Databricks/test/getRunOutput.test.ts b/packages/nodes-base/nodes/Databricks/test/getRunOutput.test.ts new file mode 100644 index 00000000000..d2d84e74ced --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/test/getRunOutput.test.ts @@ -0,0 +1,441 @@ +import { NodeOperationError } from 'n8n-workflow'; +import type { + IExecuteFunctions, + ILoadOptionsFunctions, + INode, + NodeParameterValueType, +} from 'n8n-workflow'; +import { mockDeep } from 'vitest-mock-extended'; + +import { execute as getRunOutput } from '../actions/job/getRunOutput.operation'; +import { getRuns } from '../methods/listSearch'; +import { jobParameters } from '../resources/job/parameters'; + +const HOST = 'https://adb-1234567890.1.azuredatabricks.net'; +const JOB_ID = 281874479417551; +const RUN_ID = 41847992357943; + +const node: INode = { + id: '1', + name: 'Databricks', + type: 'n8n-nodes-base.databricks', + typeVersion: 1, + position: [0, 0], + parameters: {}, +}; + +describe('Job -> Get Run Output', () => { + const task = (taskKey: string, runId?: number) => ({ + task_key: taskKey, + ...(runId === undefined ? {} : { run_id: runId }), + status: { state: 'TERMINATED' }, + }); + const notebookOutput = { + metadata: { run_id: 1001, task_key: 'extract' }, + notebook_output: { result: '{"rows":42}', truncated: false }, + }; + const sqlOutput = { + metadata: { run_id: 1002, task_key: 'load' }, + sql_output: { query_output: { output_link: `${HOST}/sql/history` } }, + logs: 'done', + logs_truncated: false, + }; + + const setupContext = ( + overrides: Record = {}, + itemIndex = 0, + ) => { + const parameters: Record = { + authentication: 'accessToken', + runId: String(RUN_ID), + ...overrides, + }; + const context = mockDeep(); + context.getNode.mockReturnValue(node); + context.getInputData.mockReturnValue([]); + context.getExecutionCancelSignal.mockReturnValue(undefined); + context.getNodeParameter.mockImplementation((name, index) => + index === itemIndex ? parameters[name] : undefined, + ); + context.getCredentials.mockResolvedValue({ host: HOST }); + return context; + }; + const apiMock = (context: ReturnType) => + context.helpers.httpRequestWithAuthentication; + const getRequest = (path: string, qs: object) => + expect.objectContaining({ method: 'GET', url: `${HOST}${path}`, qs }); + + it('should return one item per task, tagged with the task and run IDs', async () => { + const context = setupContext({}, 2); + apiMock(context) + .mockResolvedValueOnce({ + job_id: JOB_ID, + run_id: RUN_ID, + tasks: [task('extract', 1001), task('load', 1002)], + }) + .mockResolvedValueOnce(notebookOutput) + .mockResolvedValueOnce(sqlOutput); + + const result = await getRunOutput.call(context, 2); + + expect(apiMock(context)).toHaveBeenCalledTimes(3); + expect(apiMock(context)).toHaveBeenNthCalledWith( + 1, + 'databricksApi', + getRequest('/api/2.2/jobs/runs/get', { run_id: RUN_ID }), + ); + expect(apiMock(context)).toHaveBeenNthCalledWith( + 2, + 'databricksApi', + getRequest('/api/2.2/jobs/runs/get-output', { run_id: 1001 }), + ); + expect(apiMock(context)).toHaveBeenNthCalledWith( + 3, + 'databricksApi', + getRequest('/api/2.2/jobs/runs/get-output', { run_id: 1002 }), + ); + expect(result).toEqual([ + { + json: { + run_id: RUN_ID, + job_id: JOB_ID, + task_key: 'extract', + task_run_id: 1001, + truncated: false, + ...notebookOutput, + }, + pairedItem: { item: 2 }, + }, + { + json: { + run_id: RUN_ID, + job_id: JOB_ID, + task_key: 'load', + task_run_id: 1002, + truncated: false, + ...sqlOutput, + }, + pairedItem: { item: 2 }, + }, + ]); + }); + + it.each([ + ['lists no tasks', { job_id: JOB_ID, run_id: RUN_ID }], + ['lists tasks without run IDs', { job_id: JOB_ID, run_id: RUN_ID, tasks: [task('main')] }], + ])('should read the output of the run itself when it %s', async (_label, run) => { + const context = setupContext(); + apiMock(context).mockResolvedValueOnce(run).mockResolvedValueOnce(notebookOutput); + + const result = await getRunOutput.call(context, 0); + + expect(apiMock(context)).toHaveBeenCalledTimes(2); + expect(apiMock(context)).toHaveBeenNthCalledWith( + 2, + 'databricksApi', + getRequest('/api/2.2/jobs/runs/get-output', { run_id: RUN_ID }), + ); + expect(result).toEqual([ + { + json: { + run_id: RUN_ID, + job_id: JOB_ID, + task_key: undefined, + task_run_id: RUN_ID, + truncated: false, + ...notebookOutput, + }, + pairedItem: { item: 0 }, + }, + ]); + }); + + it('should follow the task pages of a large run before reading outputs', async () => { + const context = setupContext(); + apiMock(context) + .mockResolvedValueOnce({ + job_id: JOB_ID, + run_id: RUN_ID, + tasks: [task('extract', 1001)], + next_page_token: 'tasks-2', + }) + .mockResolvedValueOnce({ job_id: JOB_ID, run_id: RUN_ID, tasks: [task('load', 1002)] }) + .mockResolvedValueOnce(notebookOutput) + .mockResolvedValueOnce(sqlOutput); + + const result = await getRunOutput.call(context, 0); + + expect(apiMock(context)).toHaveBeenCalledTimes(4); + expect(apiMock(context)).toHaveBeenNthCalledWith( + 2, + 'databricksApi', + getRequest('/api/2.2/jobs/runs/get', { run_id: RUN_ID, page_token: 'tasks-2' }), + ); + expect(result.map((item) => item.json.task_key)).toEqual(['extract', 'load']); + }); + + it('should fail instead of returning a partial task list when the run has more task pages than the node reads', async () => { + const context = setupContext({}, 2); + apiMock(context).mockResolvedValue({ + job_id: JOB_ID, + run_id: RUN_ID, + tasks: [task('extract', 1001)], + next_page_token: 'more', + }); + + const result = getRunOutput.call(context, 2); + + await expect(result).rejects.toBeInstanceOf(NodeOperationError); + await expect(result).rejects.toMatchObject({ + message: `Run ${RUN_ID} has more tasks than the node can read`, + context: { itemIndex: 2 }, + }); + expect(apiMock(context)).toHaveBeenCalledTimes(20); + expect(apiMock(context).mock.calls.map(([, options]) => options.url)).not.toContain( + `${HOST}/api/2.2/jobs/runs/get-output`, + ); + }); + + it.each([ + [ + 'the notebook output is truncated', + { notebook_output: { result: 'x', truncated: true } }, + true, + ], + ['the logs are truncated', { logs: 'y', logs_truncated: true }, true], + [ + 'the clean room notebook output is truncated', + { clean_rooms_notebook_output: { notebook_output: { result: 'x', truncated: true } } }, + true, + ], + ['nothing is truncated', { notebook_output: { result: 'x' }, logs: 'y' }, false], + ['the run has not produced output yet', { metadata: { status: { state: 'RUNNING' } } }, false], + ])('should flag truncation when %s', async (_label, output, truncated) => { + const context = setupContext(); + apiMock(context) + .mockResolvedValueOnce({ job_id: JOB_ID, run_id: RUN_ID, tasks: [task('main', 1001)] }) + .mockResolvedValueOnce(output); + + const [item] = await getRunOutput.call(context, 0); + + expect(item.json.truncated).toBe(truncated); + }); + + it.each([ + ['', 'Run ID must be a whole number'], + ['not-a-number', 'Run ID must be a whole number'], + ['9007199254740993', 'Run ID is too large to send exactly'], + ])('should reject the run ID %j before calling the API', async (runId, message) => { + const context = setupContext({ runId }, 2); + + const result = getRunOutput.call(context, 2); + + await expect(result).rejects.toBeInstanceOf(NodeOperationError); + await expect(result).rejects.toMatchObject({ message, context: { itemIndex: 2 } }); + expect(apiMock(context)).not.toHaveBeenCalled(); + }); +}); + +describe('Job -> Get Run Output (run locator URL mode)', () => { + const urlMode = jobParameters + .find((property) => property.name === 'runId') + ?.modes?.find((mode) => mode.name === 'url'); + const regexSource = urlMode?.extractValue?.type === 'regex' ? urlMode.extractValue.regex : ''; + const regex = new RegExp(String(regexSource)); + + it('should define a regex for the URL mode', () => { + expect(regexSource).not.toBe(''); + }); + + it.each([ + 'https://adb-1234567890.1.azuredatabricks.net/jobs/281874479417551/runs/41847992357943', + 'https://dbc-5a643033-7dd4.cloud.databricks.com/?o=7474656527543353#job/281874479417551/run/41847992357943', + 'https://dbc-5a643033-7dd4.cloud.databricks.com/#job/281874479417551/run/41847992357943', + ])('should extract the run ID from %s', (url) => { + const match = regex.exec(url); + expect(match).toHaveLength(2); + expect(match?.[1]).toBe('41847992357943'); + }); + + it.each([ + 'https://adb-1234567890.1.azuredatabricks.net/jobs/281874479417551', + 'https://dbc-5a643033-7dd4.cloud.databricks.com/#job/281874479417551', + 'http://adb-1234567890.1.azuredatabricks.net/jobs/281874479417551/runs/41847992357943', + 'https://adb-1234567890.1.azuredatabricks.net/jobs/281874479417551/runs/latest', + ])('should not match %s', (url) => { + expect(regex.exec(url)).toBeNull(); + }); +}); + +describe('listSearch -> getRuns', () => { + const run = (runId: number, fields: object = {}) => ({ + run_id: runId, + job_id: JOB_ID, + run_page_url: `${HOST}/jobs/${JOB_ID}/runs/${runId}`, + ...fields, + }); + const listItem = (runId: number, name: string) => ({ + name, + value: String(runId), + url: `${HOST}/jobs/${JOB_ID}/runs/${runId}`, + }); + const succeeded = { state: 'TERMINATED', termination_details: { code: 'SUCCESS' } }; + const failed = { state: 'TERMINATED', termination_details: { code: 'FAILED' } }; + + const setupContext = () => { + const context = mockDeep(); + context.getNodeParameter.mockReturnValue('accessToken'); + context.getCredentials.mockResolvedValue({ host: HOST }); + return context; + }; + const apiMock = (context: ReturnType) => + context.helpers.httpRequestWithAuthentication; + + it('should name each run by job, outcome, start time and run ID, and return the next token', async () => { + const context = setupContext(); + apiMock(context).mockResolvedValue({ + runs: [ + run(1, { run_name: 'Nightly ETL', status: succeeded, start_time: 1757923200000 }), + run(2, { status: { state: 'RUNNING' }, start_time: 1757926800000 }), + ], + has_more: true, + next_page_token: 'next-token', + }); + + const result = await getRuns.call(context, undefined, 'prev-token'); + + expect(apiMock(context)).toHaveBeenCalledTimes(1); + expect(apiMock(context)).toHaveBeenCalledWith( + 'databricksApi', + expect.objectContaining({ + method: 'GET', + url: `${HOST}/api/2.2/jobs/runs/list`, + qs: { limit: 25, page_token: 'prev-token' }, + }), + ); + expect(result).toEqual({ + results: [ + listItem(1, 'Nightly ETL · SUCCESS · 2025-09-15 08:00:00 UTC · Run 1'), + listItem(2, `Job ${JOB_ID} · RUNNING · 2025-09-15 09:00:00 UTC · Run 2`), + ], + paginationToken: 'next-token', + }); + }); + + it('should give two runs of the same job different names', async () => { + const context = setupContext(); + apiMock(context).mockResolvedValue({ + runs: [ + run(1, { run_name: 'Nightly ETL', status: succeeded, start_time: 1757923200000 }), + run(2, { run_name: 'Nightly ETL', status: failed, start_time: 1757836800000 }), + ], + }); + + const { results } = await getRuns.call(context); + + expect(results.map((item) => item.name)).toEqual([ + 'Nightly ETL · SUCCESS · 2025-09-15 08:00:00 UTC · Run 1', + 'Nightly ETL · FAILED · 2025-09-14 08:00:00 UTC · Run 2', + ]); + }); + + it('should describe a run reported only through the deprecated state object', async () => { + const context = setupContext(); + apiMock(context).mockResolvedValue({ + runs: [ + run(3, { + run_name: 'Backfill', + state: { life_cycle_state: 'TERMINATED', result_state: 'CANCELED' }, + }), + ], + }); + + const result = await getRuns.call(context); + + expect(result).toEqual({ results: [listItem(3, 'Backfill · CANCELED · Run 3')] }); + }); + + it('should handle a workspace without runs, whose page carries no runs key', async () => { + const context = setupContext(); + apiMock(context).mockResolvedValue({ has_more: false }); + + const result = await getRuns.call(context); + + expect(apiMock(context)).toHaveBeenCalledWith( + 'databricksApi', + expect.objectContaining({ qs: { limit: 25 } }), + ); + expect(result).toEqual({ results: [] }); + }); + + it('should scan pages and match the filter anywhere in the name, ignoring case', async () => { + const context = setupContext(); + apiMock(context) + .mockResolvedValueOnce({ + runs: [ + run(1, { run_name: 'Nightly ETL', status: failed }), + run(2, { run_name: 'Daily Load', status: succeeded }), + ], + has_more: true, + next_page_token: 'page-2', + }) + .mockResolvedValueOnce({ + runs: [ + run(3, { run_name: 'Backfill', status: failed }), + run(4, { run_name: 'nightly cleanup', status: succeeded }), + ], + has_more: false, + }); + + const result = await getRuns.call(context, 'Nightly'); + + expect(apiMock(context)).toHaveBeenCalledTimes(2); + expect(apiMock(context)).toHaveBeenNthCalledWith( + 2, + 'databricksApi', + expect.objectContaining({ qs: { limit: 25, page_token: 'page-2' } }), + ); + expect(result).toEqual({ + results: [ + listItem(1, 'Nightly ETL · FAILED · Run 1'), + listItem(4, 'nightly cleanup · SUCCESS · Run 4'), + ], + }); + }); + + it('should match the filter against the run outcome', async () => { + const context = setupContext(); + apiMock(context).mockResolvedValue({ + runs: [ + run(1, { run_name: 'Nightly ETL', status: failed }), + run(2, { run_name: 'Daily Load', status: succeeded }), + ], + }); + + const result = await getRuns.call(context, 'failed'); + + expect(result).toEqual({ results: [listItem(1, 'Nightly ETL · FAILED · Run 1')] }); + }); + + it('should resume a filtered search from the incoming token', async () => { + const context = setupContext(); + apiMock(context).mockResolvedValue({ runs: [run(9, { run_name: 'Nightly ETL' })] }); + + await getRuns.call(context, 'Nightly', 'prev-token'); + + expect(apiMock(context)).toHaveBeenCalledWith( + 'databricksApi', + expect.objectContaining({ qs: { limit: 25, page_token: 'prev-token' } }), + ); + }); + + it('should stop scanning after ten pages and hand back the continuation token', async () => { + const context = setupContext(); + apiMock(context).mockResolvedValue({ has_more: true, next_page_token: 'more' }); + + const result = await getRuns.call(context, 'missing'); + + expect(apiMock(context)).toHaveBeenCalledTimes(10); + expect(result).toEqual({ results: [], paginationToken: 'more' }); + }); +}); diff --git a/packages/nodes-base/nodes/Databricks/test/helpers.test.ts b/packages/nodes-base/nodes/Databricks/test/helpers.test.ts index 9ac405a6534..a0c586735fb 100644 --- a/packages/nodes-base/nodes/Databricks/test/helpers.test.ts +++ b/packages/nodes-base/nodes/Databricks/test/helpers.test.ts @@ -1,10 +1,22 @@ -import type { IExecuteFunctions, ILoadOptionsFunctions, INode } from 'n8n-workflow'; +import { sleep } from '@n8n/utils/sleep'; +import { NodeApiError, NodeOperationError } from 'n8n-workflow'; +import type { + IExecuteFunctions, + IHttpRequestOptions, + ILoadOptionsFunctions, + INode, + JsonObject, +} from 'n8n-workflow'; import type { Mock } from 'vitest'; -import { mock } from 'vitest-mock-extended'; +import { mock, mockDeep } from 'vitest-mock-extended'; -import { databricksApiRequest } from '../actions/helpers'; +import { databricksApiRequest, readIdParameter } from '../actions/helpers'; import { DATABRICKS_PARTNER_USER_AGENT } from '../constants'; +vi.mock('@n8n/utils/sleep', () => ({ + sleep: vi.fn().mockResolvedValue(undefined), +})); + describe('databricksApiRequest', () => { let httpRequestWithAuthentication: Mock; let context: IExecuteFunctions; @@ -112,3 +124,192 @@ describe('databricksApiRequest', () => { expect(DATABRICKS_PARTNER_USER_AGENT).toBe('n8n_DatabricksNode'); }); }); + +describe('databricksApiRequest rate limiting', () => { + const node: INode = { + id: '1', + name: 'Databricks', + type: 'n8n-nodes-base.databricks', + typeVersion: 1, + position: [0, 0], + parameters: {}, + }; + + class AxiosError extends Error { + constructor( + message: string, + readonly response: { status: number; data: unknown; headers?: Record }, + ) { + super(message); + } + } + + const apiError = (status: number, headers?: Record) => + new NodeApiError( + node, + new AxiosError(`Request failed with status code ${status}`, { + status, + data: { error_code: 'RESOURCE_EXHAUSTED', message: 'Too many requests' }, + headers, + }) as unknown as JsonObject, + ); + + const request: IHttpRequestOptions = { + method: 'GET', + url: 'https://example.databricks.com/api/2.2/jobs/runs/get', + }; + let httpRequestWithAuthentication: Mock; + let context: IExecuteFunctions; + + beforeEach(() => { + vi.mocked(sleep).mockClear(); + httpRequestWithAuthentication = vi.fn(); + context = mock({ + getNode: () => node, + getExecutionCancelSignal: () => undefined, + helpers: { httpRequestWithAuthentication }, + }); + }); + + const sleepDurations = () => vi.mocked(sleep).mock.calls.map(([ms]) => ms); + + it('should wait for the Retry-After seconds after a 429 and return the next response', async () => { + httpRequestWithAuthentication + .mockRejectedValueOnce(apiError(429, { 'retry-after': '2' })) + .mockResolvedValueOnce({ run_id: 1 }); + + await expect(databricksApiRequest(context, 'databricksApi', request)).resolves.toEqual({ + run_id: 1, + }); + + expect(httpRequestWithAuthentication).toHaveBeenCalledTimes(2); + expect(sleepDurations()).toEqual([2000]); + }); + + it.each([ + ['no header', undefined], + ['a non-numeric header', { 'retry-after': 'soon' }], + ['a zero header', { 'retry-after': '0' }], + ])('should wait one second when the 429 carries %s', async (_label, headers) => { + httpRequestWithAuthentication + .mockRejectedValueOnce(apiError(429, headers)) + .mockResolvedValueOnce({}); + + await databricksApiRequest(context, 'databricksApi', request); + + expect(sleepDurations()).toEqual([1000]); + }); + + it('should cap the wait at 30 seconds', async () => { + httpRequestWithAuthentication + .mockRejectedValueOnce(apiError(429, { 'retry-after': '120' })) + .mockResolvedValueOnce({}); + + await databricksApiRequest(context, 'databricksApi', request); + + expect(sleepDurations()).toEqual([30000]); + }); + + it('should give up after three retries and rethrow the 429', async () => { + const error = apiError(429, { 'retry-after': '1' }); + httpRequestWithAuthentication.mockRejectedValue(error); + + await expect(databricksApiRequest(context, 'databricksApi', request)).rejects.toBe(error); + + expect(httpRequestWithAuthentication).toHaveBeenCalledTimes(4); + expect(sleepDurations()).toEqual([1000, 1000, 1000]); + }); + + it('should not retry other API errors', async () => { + const error = apiError(503); + httpRequestWithAuthentication.mockRejectedValue(error); + + await expect(databricksApiRequest(context, 'databricksApi', request)).rejects.toBe(error); + + expect(httpRequestWithAuthentication).toHaveBeenCalledTimes(1); + expect(sleep).not.toHaveBeenCalled(); + }); + + it('should hand the execution cancel signal to the wait', async () => { + const signal = new AbortController().signal; + context = mock({ + getNode: () => node, + getExecutionCancelSignal: () => signal, + helpers: { httpRequestWithAuthentication }, + }); + httpRequestWithAuthentication + .mockRejectedValueOnce(apiError(429, { 'retry-after': '1' })) + .mockResolvedValueOnce({}); + + await databricksApiRequest(context, 'databricksApi', request); + + expect(sleep).toHaveBeenCalledWith(1000, signal); + }); + + it('should retry dropdown requests without a cancel signal', async () => { + const loadOptionsContext = mock({ + getNode: () => node, + helpers: { httpRequestWithAuthentication }, + }); + httpRequestWithAuthentication + .mockRejectedValueOnce(apiError(429, { 'retry-after': '1' })) + .mockResolvedValueOnce({ jobs: [] }); + + await expect( + databricksApiRequest(loadOptionsContext, 'databricksApi', request), + ).resolves.toEqual({ jobs: [] }); + + expect(sleep).toHaveBeenCalledWith(1000, undefined); + }); +}); + +describe('readIdParameter', () => { + const node: INode = { + id: '1', + name: 'Databricks', + type: 'n8n-nodes-base.databricks', + typeVersion: 1, + position: [0, 0], + parameters: {}, + }; + const setupContext = (value: string) => { + const context = mockDeep(); + context.getNode.mockReturnValue(node); + context.getInputData.mockReturnValue([]); + context.getNodeParameter.mockReturnValue(value); + return context; + }; + + it('should read the resource locator value and return it as a number', () => { + const context = setupContext('41847992357943'); + + expect(readIdParameter(context, 1, 'runId', 'run')).toBe(41847992357943); + expect(context.getNodeParameter).toHaveBeenCalledWith('runId', 1, '', { extractValue: true }); + }); + + it.each([ + ['', 'Job ID must be a whole number', 'Use the numeric ID shown in the job URL in Databricks.'], + [ + '12a', + 'Job ID must be a whole number', + 'Use the numeric ID shown in the job URL in Databricks.', + ], + [ + '9007199254740993', + 'Job ID is too large to send exactly', + 'IDs above 9007199254740991 lose precision in JavaScript, so the node cannot send this job ID.', + ], + ])('should reject %j with a user-facing error for the item', (value, message, description) => { + const context = setupContext(value); + + let thrown: unknown; + try { + readIdParameter(context, 3, 'jobId', 'job'); + } catch (error) { + thrown = error; + } + + expect(thrown).toBeInstanceOf(NodeOperationError); + expect(thrown).toMatchObject({ message, description, context: { itemIndex: 3 } }); + }); +}); diff --git a/packages/nodes-base/nodes/Databricks/test/job-get-run-output.workflow.json b/packages/nodes-base/nodes/Databricks/test/job-get-run-output.workflow.json new file mode 100644 index 00000000000..09beb7556d5 --- /dev/null +++ b/packages/nodes-base/nodes/Databricks/test/job-get-run-output.workflow.json @@ -0,0 +1,78 @@ +{ + "name": "Databricks Job Get Run Output Test Workflow", + "nodes": [ + { + "parameters": {}, + "type": "n8n-nodes-base.manualTrigger", + "typeVersion": 1, + "position": [0, 0], + "id": "trigger-001", + "name": "When clicking 'Execute workflow'" + }, + { + "parameters": { + "resource": "job", + "operation": "getRunOutput", + "runId": { + "mode": "id", + "value": "41847992357943" + } + }, + "type": "n8n-nodes-base.databricks", + "typeVersion": 1, + "position": [220, 0], + "id": "job-get-run-output-001", + "name": "Get Run Output", + "credentials": { + "databricksApi": { + "id": "cred-001", + "name": "Databricks account" + } + } + } + ], + "pinData": { + "Get Run Output": [ + { + "json": { + "run_id": 41847992357943, + "job_id": 281874479417551, + "task_key": "main", + "task_run_id": 41847992357944, + "truncated": false, + "metadata": { + "run_id": 41847992357944, + "task_key": "main", + "status": { + "state": "TERMINATED", + "termination_details": { + "code": "SUCCESS" + } + } + }, + "notebook_output": { + "result": "done", + "truncated": false + } + } + } + ] + }, + "connections": { + "When clicking 'Execute workflow'": { + "main": [ + [ + { + "node": "Get Run Output", + "type": "main", + "index": 0 + } + ] + ] + } + }, + "active": false, + "settings": { + "executionOrder": "v1" + } +}