mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
feat(Databricks Node): Add the Get Run Output operation to the hidden Job resource (no-changelog) (#38709)
This commit is contained in:
Stephen Wright
GitHub
parent
e56b2a7654
commit
13c8a3bbd4
|
|
@ -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<IExecuteFunctions['helpers']['httpRequestWithAuthentication']> {
|
||||
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 || '';
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>;
|
||||
dbt_output?: Record<string, unknown>;
|
||||
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;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1 +1,2 @@
|
|||
export * as getRunOutput from './getRunOutput.operation';
|
||||
export * as run from './run.operation';
|
||||
|
|
|
|||
|
|
@ -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<DatabricksJobRun> {
|
||||
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<INodeExecutionData[]> {
|
||||
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;
|
||||
}
|
||||
|
|
@ -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<INodeExecutionData[]> {
|
||||
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<INode
|
|||
method: 'POST',
|
||||
url: `${host}/api/2.2/jobs/run-now`,
|
||||
body: {
|
||||
job_id: Number(jobId),
|
||||
job_id: jobId,
|
||||
...(Object.keys(jobParameters).length > 0 && { job_parameters: jobParameters }),
|
||||
},
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
|
|
|
|||
|
|
@ -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<T>(
|
||||
context: ILoadOptionsFunctions,
|
||||
credentialType: 'databricksApi' | 'databricksOAuth2Api',
|
||||
host: string,
|
||||
path: string,
|
||||
limit: number,
|
||||
pageToken?: string,
|
||||
): Promise<JobsListPage> {
|
||||
const qs: IDataObject = { limit: JOBS_PAGE_SIZE };
|
||||
): Promise<T> {
|
||||
const qs: IDataObject = { limit };
|
||||
if (pageToken) qs.page_token = pageToken;
|
||||
return await listRequest<JobsListPage>(context, credentialType, {
|
||||
return await listRequest<T>(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<JobsListPage>(
|
||||
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<INodeListSearchResult> {
|
||||
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<RunsListPage>(
|
||||
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 };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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<ILoadOptionsFunctions>();
|
||||
context.getNodeParameter.mockReturnValue('accessToken');
|
||||
|
|
|
|||
441
packages/nodes-base/nodes/Databricks/test/getRunOutput.test.ts
Normal file
441
packages/nodes-base/nodes/Databricks/test/getRunOutput.test.ts
Normal file
|
|
@ -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<string, NodeParameterValueType | object> = {},
|
||||
itemIndex = 0,
|
||||
) => {
|
||||
const parameters: Record<string, NodeParameterValueType | object> = {
|
||||
authentication: 'accessToken',
|
||||
runId: String(RUN_ID),
|
||||
...overrides,
|
||||
};
|
||||
const context = mockDeep<IExecuteFunctions>();
|
||||
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<typeof setupContext>) =>
|
||||
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<ILoadOptionsFunctions>();
|
||||
context.getNodeParameter.mockReturnValue('accessToken');
|
||||
context.getCredentials.mockResolvedValue({ host: HOST });
|
||||
return context;
|
||||
};
|
||||
const apiMock = (context: ReturnType<typeof setupContext>) =>
|
||||
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' });
|
||||
});
|
||||
});
|
||||
|
|
@ -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<string, string> },
|
||||
) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
|
||||
const apiError = (status: number, headers?: Record<string, string>) =>
|
||||
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<IExecuteFunctions>({
|
||||
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<IExecuteFunctions>({
|
||||
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<ILoadOptionsFunctions>({
|
||||
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<IExecuteFunctions>();
|
||||
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 } });
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user