github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 21:49:36 +02:00
Code Issues Actions 13 Packages Projects Releases Wiki Activity Graph

feat(Databricks Node): Add polling transport for job runs and pipeline events (no-changelog) (#38629)

Browse Source
This commit is contained in:
Stephen Wright 2026-09-16 10:00:42 +00:00 committed by GitHub
parent 533442071f
commit db55515f7a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 845 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
packages/nodes-base/nodes/Databricks/actions/helpers.ts
View File

@ -6,38 +6,34 @@ import type {
IExecuteFunctions,
IHttpRequestOptions,
ILoadOptionsFunctions,
IPollFunctions,
} from 'n8n-workflow';
import { DATABRICKS_PARTNER_USER_AGENT } from '../constants';
import type { DatabricksCredentials, OpenAPISchema } from './interfaces';
export type DatabricksContext = IExecuteFunctions | ILoadOptionsFunctions | IPollFunctions;
export type DatabricksCredentialType = 'databricksApi' | 'databricksOAuth2Api';
const RATE_LIMIT_MAX_RETRIES = 3;
const RATE_LIMIT_FALLBACK_DELAY_MS = 1_000;
const RATE_LIMIT_MAX_DELAY_MS = 30_000;
/**
* Single egress point for the Databricks API, so every request carries the
* partner User-Agent. Enforced by eslint-user-agent-restriction.mjs.
*
* Takes `context` explicitly rather than the house `this`-binding style because
* some callers (e.g. `fetchResourcesInSchema` in methods/listSearch.ts) are plain
* functions with no `this`.
*
* Setting a User-Agent deliberately opts these calls out of the instance-wide
* outbound UA, including `N8N_GLOBAL_USER_AGENT_VALUE` — partner attribution
* requires a single predictable token.
* Single egress point for the Databricks API, enforced by eslint-user-agent-restriction.mjs.
* Setting a User-Agent opts these calls out of N8N_GLOBAL_USER_AGENT_VALUE on purpose:
* partner attribution needs one predictable token.
*/
export async function databricksApiRequest(
context: IExecuteFunctions | ILoadOptionsFunctions,
credentialType: 'databricksApi' | 'databricksOAuth2Api',
context: DatabricksContext,
credentialType: DatabricksCredentialType,
options: IHttpRequestOptions,
): ReturnType<IExecuteFunctions['helpers']['httpRequestWithAuthentication']> {
const requestOptions: IHttpRequestOptions = {
...options,
headers: {
...options.headers,
// Last, so a caller cannot override it
'User-Agent': DATABRICKS_PARTNER_USER_AGENT,
},
};
@ -63,44 +59,37 @@ export async function databricksApiRequest(
}
export function getActiveCredentialType(
context: IExecuteFunctions | ILoadOptionsFunctions,
context: DatabricksContext,
itemIndex = 0,
): 'databricksApi' | 'databricksOAuth2Api' {
const authentication = context.getNodeParameter(
'authentication',
itemIndex,
'accessToken',
) as string;
): DatabricksCredentialType {
const authentication =
'getInputData' in context
? context.getNodeParameter('authentication', itemIndex, 'accessToken')
: context.getNodeParameter('authentication', 'accessToken');
return authentication === 'oAuth2' ? 'databricksOAuth2Api' : 'databricksApi';
}
export async function getHost(
context: IExecuteFunctions | ILoadOptionsFunctions,
credentialType: 'databricksApi' | 'databricksOAuth2Api',
context: DatabricksContext,
credentialType: DatabricksCredentialType,
): Promise<string> {
const credentials = await context.getCredentials<DatabricksCredentials>(credentialType);
return credentials.host.replace(/\/$/, '');
}
// Body text comes from whatever server `host` points at — truncate and strip
// control chars before promoting it to a visible error message
// Body text comes from whatever server `host` points at
export function sanitizeApiMessage(message: string): string {
// eslint-disable-next-line no-control-regex
return message.replace(/[\x00-\x1f\x7f]+/g, ' ').slice(0, 500);
}
// Must be called at every request entry point (router catch, listSearch wrapper):
// databricksApiRequest() only attaches the User-Agent and deliberately does not
// wrap errors, so callers still own their catch. Keyed on PERMISSION_DENIED only; widen
// the key if other Databricks error_codes with legible messages show up. Keyed on
// the error_code, not HTTP 403, so expired-token 403s (which core retries via
// refresh) aren't mislabeled if they leak through. Mutates rather than re-wraps:
// `new NodeApiError(node, existingNodeApiError)` returns the original untouched.
// Called at every request entry point (router catch, listSearch wrapper) because
// databricksApiRequest() does not wrap errors. Keyed on error_code, not HTTP 403, so
// expired-token 403s are not mislabeled. Mutates: re-wrapping a NodeApiError returns the same instance.
export function makePermissionErrorLegible(error: unknown): void {
if (!(error instanceof NodeApiError)) return;
// Requests with encoding: 'arraybuffer' (file downloads) receive their 403
// JSON body as raw bytes, so parse Buffer/string bodies before reading it
// File downloads receive the 403 body as raw bytes
let data = error.context.data;
if (Buffer.isBuffer(data) || typeof data === 'string') {
try {
@ -292,9 +281,7 @@ export function generateExampleFromSchema(schema: unknown, format: string): stri
if (Object.keys(exampleObj).length > 0) {
return JSON.stringify(exampleObj, null, 2);
}
} catch (e) {
// Fall through to default examples
}
} catch {}
}
const examples: Record<string, string> = {

1
packages/nodes-base/nodes/Databricks/actions/interfaces.ts
View File

@ -74,6 +74,7 @@ export type DatabricksJobRunTask = {
start_time?: number;
end_time?: number;
status?: DatabricksJobRunStatus;
/** @deprecated Jobs API 2.2 reports `status` instead */
state?: DatabricksJobRunLegacyState;
};

51
packages/nodes-base/nodes/Databricks/eslint-user-agent-restriction.mjs
View File

@ -1,11 +1,8 @@
// Forces every Databricks API call through databricksApiRequest() in
// actions/helpers.ts, which is what attaches the partner User-Agent. Spread into
// packages/nodes-base's eslint.config.mjs, which is where the `files`/`ignores`
// globs below are resolved from (this file's own location doesn't affect path
// resolution).
// Every Databricks API call must go through databricksApiRequest() in actions/helpers.ts.
// Spread into packages/nodes-base/eslint.config.mjs; the globs below resolve from there.
// Flat-config `rules` blocks replace, not merge, for any file matched by more
// than one block — so the scoped `no-restricted-syntax` below must repeat the
// than one block, so the scoped `no-restricted-syntax` below must repeat the
// base config's raw-enum rule, or it silently stops applying there.
const NO_RAW_ENUM_SYNTAX_RULE = {
selector: 'TSEnumDeclaration:not([const=true])',
@ -13,38 +10,42 @@ const NO_RAW_ENUM_SYNTAX_RULE = {
'Do not declare raw enums as it leads to runtime overhead. Use const enum instead. See https://www.typescriptlang.org/docs/handbook/enums.html#const-enums',
};
// Raw request helpers that would skip the partner User-Agent, shared by both
// selectors below so the list can't drift between them.
// `request` also names unrelated members (error.request), so it stays anchored to `helpers`.
const RESTRICTED_HELPER_METHODS =
'/^(httpRequest|httpRequestWithAuthentication|request|requestWithAuthentication|requestWithAuthenticationPaginated)$/';
'/^(httpRequest|httpRequestWithAuthentication|request|requestWithAuthentication|requestWithAuthenticationPaginated|requestOAuth1|requestOAuth2)$/';
const DISTINCTIVE_HELPER_METHODS =
'/^(httpRequest|httpRequestWithAuthentication|requestWithAuthentication|requestWithAuthenticationPaginated|requestOAuth1|requestOAuth2)$/';
const USE_DATABRICKS_API_REQUEST =
'Use databricksApiRequest() from actions/helpers.ts so the partner User-Agent is sent.';
export const databricksUserAgentRestriction = [
{
files: ['./nodes/Databricks/**/*.ts'],
// helpers.ts is the one file allowed to call the raw helper; tests stub it directly.
// helpers.ts is the one file allowed to call the raw helpers; tests stub them directly.
ignores: ['./nodes/Databricks/actions/helpers.ts', './nodes/Databricks/test/**'],
rules: {
'no-restricted-syntax': [
'error',
NO_RAW_ENUM_SYNTAX_RULE,
{
// Siblings of httpRequestWithAuthentication are restricted too: an
// operation written with helpers.httpRequest would otherwise skip
// the User-Agent silently. Anchored to `helpers` (as a member or a bare
// binding from `const { helpers } = this`) so unrelated members like
// `error.request` don't trip the guard.
selector: `MemberExpression:matches([object.property.name="helpers"], [object.name="helpers"])[property.name=${RESTRICTED_HELPER_METHODS}]`,
message:
'Use databricksApiRequest() from actions/helpers.ts so the partner User-Agent is sent.',
selector: `MemberExpression[computed=false][property.name=${DISTINCTIVE_HELPER_METHODS}]`,
message: USE_DATABRICKS_API_REQUEST,
},
{
// The same helpers reached by destructuring rather than member access.
// Anchored to a `*.helpers` or bare `helpers` initialiser so an unrelated
// `const { request } = response` doesn't trip the guard with a confusing
// message.
selector: `VariableDeclarator:matches([init.name="helpers"], [init.property.name="helpers"]) > ObjectPattern > Property[key.name=${RESTRICTED_HELPER_METHODS}]`,
message:
'Use databricksApiRequest() from actions/helpers.ts so the partner User-Agent is sent.',
selector: `MemberExpression[computed=true] > Literal.property[value=${RESTRICTED_HELPER_METHODS}]`,
message: USE_DATABRICKS_API_REQUEST,
},
{
selector: `MemberExpression:matches([object.property.name="helpers"], [object.name="helpers"])[property.name="request"]`,
message: USE_DATABRICKS_API_REQUEST,
},
{
selector: `ObjectPattern > Property[key.name=${DISTINCTIVE_HELPER_METHODS}]`,
message: USE_DATABRICKS_API_REQUEST,
},
{
selector: `VariableDeclarator:matches([init.name="helpers"], [init.property.name="helpers"]) > ObjectPattern > Property[key.name="request"]`,
message: USE_DATABRICKS_API_REQUEST,
},
],
},

7
packages/nodes-base/nodes/Databricks/methods/listSearch.ts
View File

@ -12,6 +12,7 @@ import {
getHost,
makePermissionErrorLegible,
sanitizeApiMessage,
type DatabricksCredentialType,
} from '../actions/helpers';
import type { DatabricksJobRun } from '../actions/interfaces';
@ -19,7 +20,7 @@ import type { DatabricksJobRun } from '../actions/interfaces';
// doesn't cover them — apply it here for every listSearch call site instead
async function listRequest<T>(
context: ILoadOptionsFunctions,
credentialType: 'databricksApi' | 'databricksOAuth2Api',
credentialType: DatabricksCredentialType,
options: IHttpRequestOptions,
): Promise<T> {
try {
@ -208,7 +209,7 @@ export async function getSchemas(
async function fetchResourcesInSchema<T extends { name: string }>(
context: ILoadOptionsFunctions,
credentialType: 'databricksApi' | 'databricksOAuth2Api',
credentialType: DatabricksCredentialType,
host: string,
apiPath: string,
catalogName: string,
@ -434,7 +435,7 @@ type JobsListPage = { jobs?: JobSummary[]; next_page_token?: string };
async function fetchListPage<T>(
context: ILoadOptionsFunctions,
credentialType: 'databricksApi' | 'databricksOAuth2Api',
credentialType: DatabricksCredentialType,
host: string,
path: string,
limit: number,

12
packages/nodes-base/nodes/Databricks/test/Databricks.node.test.ts
View File

@ -56,6 +56,12 @@ const apiErrorFromBody = (status: number, data: unknown) =>
}) as unknown as JsonObject,
);
const createExecuteContext = () => {
const context = mockDeep<IExecuteFunctions>();
context.getInputData.mockReturnValue([]);
return context;
};
describe('Databricks', () => {
const credentials = {
databricksApi: {
@ -799,7 +805,7 @@ describe('listSearch -> PERMISSION_DENIED surfaces the Databricks message', () =
describe('Databricks SQL -> Execute Query (FAILED/CANCELED statement)', () => {
const setupContext = (status: unknown) => {
const context = mockDeep<IExecuteFunctions>();
const context = createExecuteContext();
context.getNode.mockReturnValue(node);
context.getNodeParameter.mockImplementation((name) => {
if (name === 'warehouseId') return 'warehouse123';
@ -882,7 +888,7 @@ describe('Databricks SQL -> Execute Query (async polling)', () => {
});
it('should poll until the statement reaches SUCCEEDED and map rows to items', async () => {
const context = mockDeep<IExecuteFunctions>();
const context = createExecuteContext();
context.getNode.mockReturnValue(node);
context.getNodeParameter.mockImplementation((name) => {
if (name === 'warehouseId') return 'warehouse123';
@ -962,7 +968,7 @@ describe('Job -> Run (wait for completion)', () => {
options: { timeout: 10 },
...overrides,
};
const context = mockDeep<IExecuteFunctions>();
const context = createExecuteContext();
context.getNode.mockReturnValue(node);
context.getExecutionCancelSignal.mockReturnValue(cancelSignal);
context.getNodeParameter.mockImplementation((name, index) =>

36
packages/nodes-base/nodes/Databricks/test/helpers.test.ts
View File

@ -5,13 +5,13 @@ import type {
IHttpRequestOptions,
ILoadOptionsFunctions,
INode,
IPollFunctions,
JsonObject,
} from 'n8n-workflow';
import type { Mock } from 'vitest';
import { mock, mockDeep } from 'vitest-mock-extended';
import { databricksApiRequest, readIdParameter } from '../actions/helpers';
import { DATABRICKS_PARTNER_USER_AGENT } from '../constants';
import { databricksApiRequest, getActiveCredentialType, readIdParameter } from '../actions/helpers';
vi.mock('@n8n/utils/sleep', () => ({
sleep: vi.fn().mockResolvedValue(undefined),
@ -86,8 +86,6 @@ describe('databricksApiRequest', () => {
});
it('should forward the credential type and bind the call to the passed context', async () => {
// `fetchResourcesInSchema` in methods/listSearch.ts passes a context object
// rather than `this`, so the receiver must come from the argument.
const loadOptionsContext = mock<ILoadOptionsFunctions>({
getNode: () => mock<INode>({ typeVersion: 1 }),
helpers: { httpRequestWithAuthentication },
@ -101,27 +99,23 @@ describe('databricksApiRequest', () => {
expect(httpRequestWithAuthentication.mock.calls[0][0]).toBe('databricksOAuth2Api');
expect(httpRequestWithAuthentication.mock.instances[0]).toBe(loadOptionsContext);
});
});
it('should track the integration version rather than the node typeVersion', async () => {
// A node instance pinned to an older typeVersion must still report the version
// of the integration that is actually running.
const staleContext = mock<IExecuteFunctions>({
getNode: () => mock<INode>({ typeVersion: 0.1 }),
helpers: { httpRequestWithAuthentication },
});
describe('getActiveCredentialType', () => {
it('should read the authentication parameter of the given item on an execute context', () => {
const context = mock<IExecuteFunctions>({ getInputData: vi.fn() });
context.getNodeParameter.mockReturnValue('oAuth2');
await databricksApiRequest(staleContext, 'databricksApi', {
method: 'GET',
url: 'https://example.databricks.com/api/2.1/unity-catalog/catalogs',
});
expect(capturedOptions().headers).toEqual({
'User-Agent': DATABRICKS_PARTNER_USER_AGENT,
});
expect(getActiveCredentialType(context, 2)).toBe('databricksOAuth2Api');
expect(context.getNodeParameter).toHaveBeenCalledWith('authentication', 2, 'accessToken');
});
it('should send the unversioned partner User-Agent', () => {
expect(DATABRICKS_PARTNER_USER_AGENT).toBe('n8n_DatabricksNode');
it('should read the authentication parameter without an item index on a polling context', () => {
const context = mock<IPollFunctions>();
context.getNodeParameter.mockReturnValue('accessToken');
expect(getActiveCredentialType(context)).toBe('databricksApi');
expect(context.getNodeParameter).toHaveBeenCalledWith('authentication', 'accessToken');
});
});

444
packages/nodes-base/nodes/Databricks/test/transport.test.ts Normal file
View File

@ -0,0 +1,444 @@
import type { INode, IPollFunctions } from 'n8n-workflow';
import { NodeOperationError, UnexpectedError } from 'n8n-workflow';
import { mock, mockDeep } from 'vitest-mock-extended';
import { DATABRICKS_PARTNER_USER_AGENT } from '../constants';
import {
DEFAULT_MAX_PAGES,
JOB_RUNS_MAX_PAGE_SIZE,
listAllJobRuns,
listAllPipelineEvents,
listJobRuns,
listPipelineEvents,
PIPELINE_EVENT_LEVELS,
PIPELINE_EVENTS_MAX_PAGE_SIZE,
type ListJobRunsParams,
type ListPipelineEventsParams,
type PageLimits,
type PipelineEventLevel,
} from '../transport';
import { clampPageSize } from '../transport/pagination';
import { buildPipelineEventsFilter } from '../transport/pipelineEvents';
const HOST = 'https://adb-1234567890.1.azuredatabricks.net';
const JOB_ID = 281874479417551;
const PIPELINE_ID = '8199cd89-e2f5-4169-a6aa-656a24c8886d';
const CURSOR = '2026-09-01T14:20:50Z';
const NO_DEADLINE: PageLimits = { deadlineEpochMs: Number.POSITIVE_INFINITY };
const run = (runId: number) => ({ job_id: JOB_ID, run_id: runId, status: { state: 'TERMINATED' } });
const event = (id: string) => ({ id, event_type: 'update_progress', timestamp: CURSOR });
const createPollContext = () => {
const context = mockDeep<IPollFunctions>();
context.getNode.mockReturnValue(mock<INode>({ name: 'Databricks Trigger', typeVersion: 1 }));
context.getCredentials.mockResolvedValue({ host: HOST });
return context;
};
const apiMock = (context: ReturnType<typeof createPollContext>) =>
context.helpers.httpRequestWithAuthentication;
const requestQuery = (context: ReturnType<typeof createPollContext>, call = 0) =>
apiMock(context).mock.calls[call][1].qs;
describe('clampPageSize', () => {
it.each([
['an in-range size', 7, 7],
['a size below the range', 0, 1],
['a fractional size', 7.9, 7],
['no size', undefined, undefined],
['a non-numeric size', Number.NaN, undefined],
])('clamps %s', (_label, value, expected) => {
expect(clampPageSize(value, 25)).toBe(expected);
});
});
describe('listJobRuns', () => {
it('requests one page with the given filters and the partner User-Agent', async () => {
const context = createPollContext();
apiMock(context).mockResolvedValue({ runs: [run(1)], next_page_token: 'page-2' });
const page = await listJobRuns(context, 'databricksApi', {
jobId: JOB_ID,
startTimeFromMs: 1788260000000,
state: 'completed',
expandTasks: true,
pageSize: 25,
});
expect(apiMock(context)).toHaveBeenCalledWith(
'databricksApi',
expect.objectContaining({
method: 'GET',
url: `${HOST}/api/2.2/jobs/runs/list`,
qs: {
job_id: JOB_ID,
start_time_from: 1788260000000,
completed_only: true,
expand_tasks: true,
limit: 25,
},
headers: expect.objectContaining({ 'User-Agent': DATABRICKS_PARTNER_USER_AGENT }),
}),
);
expect(page).toEqual({ items: [run(1)], nextPageToken: 'page-2' });
});
const queryCases: Array<[string, ListJobRunsParams, Record<string, unknown>]> = [
['no parameters', {}, {}],
['a false expandTasks flag', { expandTasks: false }, {}],
['active runs only', { state: 'active' }, { active_only: true }],
[
'a time window',
{ startTimeFromMs: 1, startTimeToMs: 2 },
{ start_time_from: 1, start_time_to: 2 },
],
['a page token', { pageToken: 'p2' }, { page_token: 'p2' }],
['an empty page token', { pageToken: '' }, {}],
['a page size above the range', { pageSize: 100 }, { limit: JOB_RUNS_MAX_PAGE_SIZE }],
];
it.each(queryCases)('sends the query for %s', async (_label, params, qs) => {
const context = createPollContext();
apiMock(context).mockResolvedValue({ runs: [] });
await listJobRuns(context, 'databricksOAuth2Api', params);
expect(apiMock(context)).toHaveBeenCalledWith(
'databricksOAuth2Api',
expect.objectContaining({ qs }),
);
});
it.each([
['a page without runs', {}, { items: [] }],
['an empty token', { runs: [run(1)], next_page_token: '' }, { items: [run(1)] }],
])('normalises %s', async (_label, response, expected) => {
const context = createPollContext();
apiMock(context).mockResolvedValue(response);
await expect(listJobRuns(context, 'databricksApi')).resolves.toEqual(expected);
});
it.each([
['an HTML body', '<html>sign in</html>'],
['a runs field that is not a list', { runs: 'none' }],
])('rejects %s instead of returning an empty page', async (_label, response) => {
const context = createPollContext();
apiMock(context).mockResolvedValue(response);
await expect(listJobRuns(context, 'databricksApi')).rejects.toThrow(NodeOperationError);
});
});
describe('listAllJobRuns', () => {
let nowMs: number;
beforeEach(() => {
nowMs = 1_000;
vi.spyOn(Date, 'now').mockImplementation(() => nowMs);
});
afterEach(() => {
vi.restoreAllMocks();
});
it('follows next_page_token from the start token, keeping the filters, until the last page', async () => {
const context = createPollContext();
apiMock(context)
.mockResolvedValueOnce({ runs: [run(1), run(2)], next_page_token: 'p2' })
.mockResolvedValueOnce({ runs: [run(3)] });
const result = await listAllJobRuns(
context,
'databricksApi',
{ jobId: JOB_ID, pageSize: 2, pageToken: 'p1' },
NO_DEADLINE,
);
expect(result).toEqual({ items: [run(1), run(2), run(3)] });
expect(requestQuery(context, 0)).toEqual({ job_id: JOB_ID, limit: 2, page_token: 'p1' });
expect(requestQuery(context, 1)).toEqual({ job_id: JOB_ID, limit: 2, page_token: 'p2' });
expect(context.getCredentials).toHaveBeenCalledTimes(1);
});
const caps: Array<[string, PageLimits, number]> = [
['the given page cap', { ...NO_DEADLINE, maxPages: 3 }, 3],
['the default page cap', NO_DEADLINE, DEFAULT_MAX_PAGES],
[
'the page cap when the deadline is not a number',
{ maxPages: 3, deadlineEpochMs: Number.NaN },
3,
],
];
it.each(caps)('stops at %s and hands back the next token', async (_label, limits, pages) => {
const context = createPollContext();
apiMock(context).mockResolvedValue({ runs: [run(1)], next_page_token: 'more' });
const result = await listAllJobRuns(context, 'databricksApi', {}, limits);
expect(apiMock(context)).toHaveBeenCalledTimes(pages);
expect(result).toEqual({ items: Array<object>(pages).fill(run(1)), nextPageToken: 'more' });
});
const spentDeadlines: Array<[string, PageLimits]> = [
['the deadline equals the clock', { deadlineEpochMs: 1_000 }],
['the deadline has passed', { deadlineEpochMs: 0 }],
['the deadline has passed before the page cap', { maxPages: 3, deadlineEpochMs: 0 }],
];
it.each(spentDeadlines)(
'fetches one page and hands back the token when %s',
async (_label, limits) => {
const context = createPollContext();
apiMock(context).mockResolvedValue({ runs: [run(1)], next_page_token: 'more' });
const result = await listAllJobRuns(context, 'databricksApi', {}, limits);
expect(apiMock(context)).toHaveBeenCalledTimes(1);
expect(result).toEqual({ items: [run(1)], nextPageToken: 'more' });
},
);
it('stops between pages once the clock passes the deadline', async () => {
const context = createPollContext();
const pages = [
{ runs: [run(1)], next_page_token: 'p2' },
{ runs: [run(2)], next_page_token: 'p3' },
{ runs: [run(3)] },
];
apiMock(context).mockImplementation(async () => {
nowMs += 600;
return pages.shift();
});
const result = await listAllJobRuns(
context,
'databricksApi',
{ jobId: JOB_ID },
{ deadlineEpochMs: 2_100 },
);
expect(apiMock(context)).toHaveBeenCalledTimes(2);
expect(requestQuery(context, 1)).toEqual({ job_id: JOB_ID, page_token: 'p2' });
expect(result).toEqual({ items: [run(1), run(2)], nextPageToken: 'p3' });
expect(context.getCredentials).toHaveBeenCalledTimes(1);
});
it('returns no token when the last page lands after the deadline', async () => {
const context = createPollContext();
apiMock(context).mockResolvedValue({ runs: [run(1)] });
await expect(
listAllJobRuns(context, 'databricksApi', {}, { deadlineEpochMs: 0 }),
).resolves.toEqual({ items: [run(1)] });
});
});
describe('buildPipelineEventsFilter', () => {
const cases: Array<
[string, Pick<ListPipelineEventsParams, 'after' | 'levels'>, string | undefined]
> = [
['nothing', {}, undefined],
['no levels', { levels: [] }, undefined],
['a cursor', { after: CURSOR }, `timestamp > '${CURSOR}'`],
[
'a cursor with fractional seconds',
{ after: '2026-09-01T14:20:31.066Z' },
"timestamp > '2026-09-01T14:20:31.066Z'",
],
['levels', { levels: ['ERROR', 'WARN'] }, "level in ('ERROR', 'WARN')"],
[
'every level',
{ levels: PIPELINE_EVENT_LEVELS },
"level in ('INFO', 'WARN', 'ERROR', 'METRICS')",
],
[
'levels and a cursor',
{ after: CURSOR, levels: ['ERROR'] },
`level in ('ERROR') AND timestamp > '${CURSOR}'`,
],
];
it.each(cases)('builds the filter for %s', (_label, params, expected) => {
expect(buildPipelineEventsFilter(params)).toBe(expected);
});
it.each([
'2026-09-01',
'2026-09-01T14:20:50+02:00',
"2026-09-01T14:20:50Z' OR level='INFO",
'2026-02-30T14:20:50Z',
'2026-09-01T25:00:00Z',
])('rejects the cursor %s', (after) => {
expect(() => buildPipelineEventsFilter({ after })).toThrow(UnexpectedError);
});
it.each(['DEBUG', 'error'])('rejects the level %s', (level) => {
const levels = ['ERROR', level] as PipelineEventLevel[];
expect(() => buildPipelineEventsFilter({ levels })).toThrow(UnexpectedError);
});
});
describe('listPipelineEvents', () => {
it('requests events after the cursor in ascending order with the partner User-Agent', async () => {
const context = createPollContext();
apiMock(context).mockResolvedValue({ events: [event('e1')], next_page_token: 'p2' });
const page = await listPipelineEvents(context, 'databricksApi', {
pipelineId: PIPELINE_ID,
after: CURSOR,
levels: ['ERROR', 'WARN'],
pageSize: 25,
});
expect(apiMock(context)).toHaveBeenCalledWith(
'databricksApi',
expect.objectContaining({
method: 'GET',
url: `${HOST}/api/2.0/pipelines/${PIPELINE_ID}/events`,
qs: {
max_results: 25,
filter: `level in ('ERROR', 'WARN') AND timestamp > '${CURSOR}'`,
order_by: 'timestamp asc',
},
headers: expect.objectContaining({ 'User-Agent': DATABRICKS_PARTNER_USER_AGENT }),
}),
);
expect(page).toEqual({ items: [event('e1')], nextPageToken: 'p2' });
});
const queryCases: Array<
[string, Omit<ListPipelineEventsParams, 'pipelineId'>, Record<string, unknown>]
> = [
['ascending order by default', {}, { order_by: 'timestamp asc' }],
['descending order', { order: 'desc' }, { order_by: 'timestamp desc' }],
[
'a page size above the range',
{ pageSize: 5000 },
{ max_results: PIPELINE_EVENTS_MAX_PAGE_SIZE, order_by: 'timestamp asc' },
],
[
'a continuation, without the filter',
{ after: CURSOR, levels: ['ERROR'], pageSize: 25, pageToken: 'p2' },
{ max_results: 25, page_token: 'p2' },
],
[
'an empty page token',
{ after: CURSOR, pageToken: '' },
{ filter: `timestamp > '${CURSOR}'`, order_by: 'timestamp asc' },
],
];
it.each(queryCases)('sends the query for %s', async (_label, params, qs) => {
const context = createPollContext();
apiMock(context).mockResolvedValue({});
await listPipelineEvents(context, 'databricksOAuth2Api', {
pipelineId: PIPELINE_ID,
...params,
});
expect(apiMock(context)).toHaveBeenCalledWith(
'databricksOAuth2Api',
expect.objectContaining({ qs }),
);
});
it.each(['pipelines', '../updates', `${PIPELINE_ID}/events`, ''])(
'rejects the pipeline ID %j before any request',
async (pipelineId) => {
const context = createPollContext();
await expect(listPipelineEvents(context, 'databricksApi', { pipelineId })).rejects.toThrow(
NodeOperationError,
);
expect(apiMock(context)).not.toHaveBeenCalled();
},
);
it('rejects an invalid cursor before any request', async () => {
const context = createPollContext();
await expect(
listPipelineEvents(context, 'databricksApi', {
pipelineId: PIPELINE_ID,
after: '2026-09-01',
}),
).rejects.toThrow(UnexpectedError);
expect(apiMock(context)).not.toHaveBeenCalled();
});
it('normalises a page without events', async () => {
const context = createPollContext();
apiMock(context).mockResolvedValue({});
await expect(
listPipelineEvents(context, 'databricksApi', { pipelineId: PIPELINE_ID }),
).resolves.toEqual({ items: [] });
});
it.each([
['an HTML body', '<html>sign in</html>'],
['an events field that is not a list', { events: 'none' }],
])('rejects %s instead of returning an empty page', async (_label, response) => {
const context = createPollContext();
apiMock(context).mockResolvedValue(response);
await expect(
listPipelineEvents(context, 'databricksApi', { pipelineId: PIPELINE_ID }),
).rejects.toThrow(NodeOperationError);
});
});
describe('listAllPipelineEvents', () => {
it('follows next_page_token until the last page', async () => {
const context = createPollContext();
apiMock(context)
.mockResolvedValueOnce({ events: [event('e1')], next_page_token: 'p2' })
.mockResolvedValueOnce({ events: [event('e2')] });
const result = await listAllPipelineEvents(
context,
'databricksApi',
{ pipelineId: PIPELINE_ID, after: CURSOR, pageSize: 25 },
NO_DEADLINE,
);
expect(result).toEqual({ items: [event('e1'), event('e2')] });
expect(requestQuery(context, 0)).toEqual({
max_results: 25,
filter: `timestamp > '${CURSOR}'`,
order_by: 'timestamp asc',
});
expect(requestQuery(context, 1)).toEqual({ max_results: 25, page_token: 'p2' });
expect(context.getCredentials).toHaveBeenCalledTimes(1);
});
const stops: Array<[string, PageLimits, number]> = [
['the page cap', { ...NO_DEADLINE, maxPages: 2 }, 2],
['a spent deadline', { deadlineEpochMs: 0 }, 1],
];
it.each(stops)(
'stops at %s, resuming from the given token, and hands back the next token',
async (_label, limits, pages) => {
const context = createPollContext();
apiMock(context).mockResolvedValue({ events: [event('e1')], next_page_token: 'more' });
const result = await listAllPipelineEvents(
context,
'databricksApi',
{ pipelineId: PIPELINE_ID, pageToken: 'p1' },
limits,
);
expect(requestQuery(context, 0)).toEqual({ page_token: 'p1' });
expect(apiMock(context)).toHaveBeenCalledTimes(pages);
expect(result).toEqual({
items: Array<object>(pages).fill(event('e1')),
nextPageToken: 'more',
});
},
);
});

17
packages/nodes-base/nodes/Databricks/transport/index.ts Normal file
View File

@ -0,0 +1,17 @@
export {
JOB_RUNS_MAX_PAGE_SIZE,
listAllJobRuns,
listJobRuns,
type ListJobRunsParams,
} from './jobRuns';
export { DEFAULT_MAX_PAGES, type Page, type PageLimits } from './pagination';
export {
isPipelineEventLevel,
listAllPipelineEvents,
listPipelineEvents,
PIPELINE_EVENT_LEVELS,
PIPELINE_EVENTS_MAX_PAGE_SIZE,
type ListPipelineEventsParams,
type PipelineEvent,
type PipelineEventLevel,
} from './pipelineEvents';

95
packages/nodes-base/nodes/Databricks/transport/jobRuns.ts Normal file
View File

@ -0,0 +1,95 @@
import { isRecord } from '@n8n/utils/is-record';
import type { IDataObject } from 'n8n-workflow';
import { NodeOperationError } from 'n8n-workflow';
import {
databricksApiRequest,
getHost,
type DatabricksContext,
type DatabricksCredentialType,
} from '../actions/helpers';
import type { DatabricksJobRun } from '../actions/interfaces';
import { clampPageSize, collectPages, toPage, type Page, type PageLimits } from './pagination';
export const JOB_RUNS_MAX_PAGE_SIZE = 25;
export interface ListJobRunsParams {
jobId?: number;
startTimeFromMs?: number;
startTimeToMs?: number;
state?: 'active' | 'completed';
expandTasks?: boolean;
pageSize?: number;
pageToken?: string;
}
type JobRunsListResponse = { runs?: DatabricksJobRun[]; next_page_token?: string };
function isJobRunsListResponse(value: unknown): value is JobRunsListResponse {
return isRecord(value) && (value.runs === undefined || Array.isArray(value.runs));
}
function toQuery(params: ListJobRunsParams): IDataObject {
const qs: IDataObject = {};
if (params.jobId !== undefined) qs.job_id = params.jobId;
if (params.startTimeFromMs !== undefined) qs.start_time_from = params.startTimeFromMs;
if (params.startTimeToMs !== undefined) qs.start_time_to = params.startTimeToMs;
if (params.state === 'active') qs.active_only = true;
if (params.state === 'completed') qs.completed_only = true;
if (params.expandTasks) qs.expand_tasks = true;
const limit = clampPageSize(params.pageSize, JOB_RUNS_MAX_PAGE_SIZE);
if (limit !== undefined) qs.limit = limit;
if (params.pageToken) qs.page_token = params.pageToken;
return qs;
}
async function fetchJobRunsPage(
context: DatabricksContext,
credentialType: DatabricksCredentialType,
host: string,
params: ListJobRunsParams,
): Promise<Page<DatabricksJobRun>> {
const response: unknown = await databricksApiRequest(context, credentialType, {
method: 'GET',
url: `${host}/api/2.2/jobs/runs/list`,
qs: toQuery(params),
headers: { Accept: 'application/json' },
json: true,
});
if (!isJobRunsListResponse(response)) {
throw new NodeOperationError(
context.getNode(),
'Databricks did not return a JSON list of job runs',
{ description: `Check that ${host} is the URL of a Databricks workspace.` },
);
}
return toPage(response.runs, response.next_page_token);
}
export async function listJobRuns(
context: DatabricksContext,
credentialType: DatabricksCredentialType,
params: ListJobRunsParams = {},
): Promise<Page<DatabricksJobRun>> {
return await fetchJobRunsPage(
context,
credentialType,
await getHost(context, credentialType),
params,
);
}
export async function listAllJobRuns(
context: DatabricksContext,
credentialType: DatabricksCredentialType,
params: ListJobRunsParams,
limits: PageLimits,
): Promise<Page<DatabricksJobRun>> {
const host = await getHost(context, credentialType);
return await collectPages(
async (pageToken) =>
await fetchJobRunsPage(context, credentialType, host, { ...params, pageToken }),
limits,
params.pageToken,
);
}

36
packages/nodes-base/nodes/Databricks/transport/pagination.ts Normal file
View File

@ -0,0 +1,36 @@
export const DEFAULT_MAX_PAGES = 40;
export interface Page<T> {
items: T[];
nextPageToken?: string;
}
export interface PageLimits {
deadlineEpochMs: number;
maxPages?: number;
}
export function toPage<T>(items: T[] | undefined, nextPageToken: string | undefined): Page<T> {
return { items: items ?? [], nextPageToken: nextPageToken || undefined };
}
export function clampPageSize(value: number | undefined, max: number): number | undefined {
if (value === undefined || !Number.isFinite(value)) return undefined;
return Math.min(Math.max(Math.trunc(value), 1), max);
}
export async function collectPages<T>(
fetchPage: (pageToken?: string) => Promise<Page<T>>,
{ deadlineEpochMs, maxPages = DEFAULT_MAX_PAGES }: PageLimits,
startToken?: string,
): Promise<Page<T>> {
const items: T[] = [];
let pageToken = startToken;
for (let page = 0; page < maxPages; page++) {
const result = await fetchPage(pageToken);
items.push(...result.items);
pageToken = result.nextPageToken;
if (pageToken === undefined || Date.now() >= deadlineEpochMs) break;
}
return { items, nextPageToken: pageToken };
}

175
packages/nodes-base/nodes/Databricks/transport/pipelineEvents.ts Normal file
View File

@ -0,0 +1,175 @@
import { isRecord } from '@n8n/utils/is-record';
import type { IDataObject } from 'n8n-workflow';
import { NodeOperationError, UnexpectedError } from 'n8n-workflow';
import {
databricksApiRequest,
getHost,
type DatabricksContext,
type DatabricksCredentialType,
} from '../actions/helpers';
import { clampPageSize, collectPages, toPage, type Page, type PageLimits } from './pagination';
export const PIPELINE_EVENTS_MAX_PAGE_SIZE = 1000;
const PIPELINE_ID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
const ISO_TIMESTAMP_PATTERN = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{1,9})?Z$/;
export const PIPELINE_EVENT_LEVELS = ['INFO', 'WARN', 'ERROR', 'METRICS'] as const;
export type PipelineEventLevel = (typeof PIPELINE_EVENT_LEVELS)[number];
export type IsoUtcTimestamp = string;
export type PipelineEvent = {
id?: string;
sequence?: {
control_plane_seq_no?: number;
data_plane_id?: { instance?: string; seq_no?: number };
};
event_type?: string;
level?: PipelineEventLevel;
message?: string;
timestamp?: IsoUtcTimestamp;
maturity_level?: string;
origin?: {
pipeline_id?: string;
pipeline_name?: string;
update_id?: string;
flow_id?: string;
flow_name?: string;
[key: string]: unknown;
};
details?: { update_progress?: { state?: string }; [key: string]: unknown };
error?: {
fatal?: boolean;
exceptions?: Array<{
class_name?: string;
error_class?: string;
sql_state?: string;
message?: string;
stack?: Array<{
declaring_class?: string;
method_name?: string;
file_name?: string;
line_number?: number;
}>;
}>;
};
truncation?: { truncated_fields?: Array<{ field_name?: string }> };
};
export interface ListPipelineEventsParams {
pipelineId: string;
after?: IsoUtcTimestamp;
levels?: readonly PipelineEventLevel[];
order?: 'asc' | 'desc';
pageSize?: number;
pageToken?: string;
}
type PipelineEventsResponse = { events?: PipelineEvent[]; next_page_token?: string };
function isPipelineEventsResponse(value: unknown): value is PipelineEventsResponse {
return isRecord(value) && (value.events === undefined || Array.isArray(value.events));
}
export function isPipelineEventLevel(level: string): level is PipelineEventLevel {
return PIPELINE_EVENT_LEVELS.some((known) => known === level);
}
function isIsoUtcTimestamp(value: string): boolean {
if (!ISO_TIMESTAMP_PATTERN.test(value)) return false;
const parsed = new Date(value);
return !Number.isNaN(parsed.getTime()) && parsed.toISOString().startsWith(value.slice(0, 19));
}
export function buildPipelineEventsFilter(
params: Pick<ListPipelineEventsParams, 'after' | 'levels'>,
): string | undefined {
const clauses: string[] = [];
if (params.levels?.length) {
const invalid = params.levels.filter((level) => !isPipelineEventLevel(level));
if (invalid.length > 0) {
throw new UnexpectedError('Pipeline event level is not one of INFO, WARN, ERROR, METRICS', {
extra: { levels: invalid },
});
}
clauses.push(`level in (${params.levels.map((level) => `'${level}'`).join(', ')})`);
}
if (params.after !== undefined) {
if (!isIsoUtcTimestamp(params.after)) {
throw new UnexpectedError('Pipeline events cursor must be an ISO 8601 UTC timestamp', {
extra: { after: params.after },
});
}
clauses.push(`timestamp > '${params.after}'`);
}
return clauses.length > 0 ? clauses.join(' AND ') : undefined;
}
// The API rejects any field other than max_results next to page_token
function toQuery(params: ListPipelineEventsParams): IDataObject {
const qs: IDataObject = {};
const pageSize = clampPageSize(params.pageSize, PIPELINE_EVENTS_MAX_PAGE_SIZE);
if (pageSize !== undefined) qs.max_results = pageSize;
if (params.pageToken) {
qs.page_token = params.pageToken;
return qs;
}
const filter = buildPipelineEventsFilter(params);
if (filter !== undefined) qs.filter = filter;
qs.order_by = `timestamp ${params.order ?? 'asc'}`;
return qs;
}
async function fetchPipelineEventsPage(
context: DatabricksContext,
credentialType: DatabricksCredentialType,
host: string,
params: ListPipelineEventsParams,
): Promise<Page<PipelineEvent>> {
if (!PIPELINE_ID_PATTERN.test(params.pipelineId)) {
throw new NodeOperationError(context.getNode(), 'Pipeline ID must be a UUID');
}
const response: unknown = await databricksApiRequest(context, credentialType, {
method: 'GET',
url: `${host}/api/2.0/pipelines/${params.pipelineId}/events`,
qs: toQuery(params),
headers: { Accept: 'application/json' },
json: true,
});
if (!isPipelineEventsResponse(response)) {
throw new NodeOperationError(
context.getNode(),
'Databricks did not return a JSON list of pipeline events',
{ description: `Check that ${host} is the URL of a Databricks workspace.` },
);
}
return toPage(response.events, response.next_page_token);
}
export async function listPipelineEvents(
context: DatabricksContext,
credentialType: DatabricksCredentialType,
params: ListPipelineEventsParams,
): Promise<Page<PipelineEvent>> {
return await fetchPipelineEventsPage(
context,
credentialType,
await getHost(context, credentialType),
params,
);
}
export async function listAllPipelineEvents(
context: DatabricksContext,
credentialType: DatabricksCredentialType,
params: ListPipelineEventsParams,
limits: PageLimits,
): Promise<Page<PipelineEvent>> {
const host = await getHost(context, credentialType);
return await collectPages(
async (pageToken) =>
await fetchPipelineEventsPage(context, credentialType, host, { ...params, pageToken }),
limits,
params.pageToken,
);
}