paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 43 Packages Projects Releases Wiki Activity

change to using a scope parameter instead

Browse Source 
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
2022-04-22 20:06:31 +02:00
parent cfd779a9bc
commit d63133e521
9 changed files with 211 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/swift-parrots-hammer.md
+5 -1
View File
@@ -2,4 +2,8 @@
'@backstage/backend-tasks': patch
---
The task scheduler now has two new methods: `scheduleLocalTask` and `createScheduledLocalTaskRunner`, which can be used to perform local-only recurring tasks. This can be used to replace usages of `runPeriodically` helpers.
Scheduled tasks now have an optional `scope` field. If unset, or having the
value `'global'`, the old behavior with cross-worker locking is retained. If
having the value `'local'`, there is no coordination across workers and the
behavior is more like `setInterval`. This can be used to replace usages of
`runPeriodically` helpers.
packages/backend-tasks/api-report.md
+1 -4
View File
@@ -11,11 +11,7 @@ import { Logger } from 'winston';
// @public
export interface PluginTaskScheduler {
createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner;
createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner;
scheduleLocalTask(
task: TaskScheduleDefinition & TaskInvocationDefinition,
): Promise<void>;
scheduleTask(
task: TaskScheduleDefinition & TaskInvocationDefinition,
): Promise<void>;
@@ -47,6 +43,7 @@ export interface TaskScheduleDefinition {
}
| Duration;
initialDelay?: Duration;
scope?: 'global' | 'local';
timeout: Duration;
}
packages/backend-tasks/src/tasks/LocalTaskWorker.ts
+18 -10
View File
@@ -22,6 +22,11 @@ import { Logger } from 'winston';
import { TaskFunction, TaskSettingsV2 } from './types';
import { delegateAbortController, sleep } from './util';
/**
* Implements tasks that run locally without cross-host collaboration.
*
* @private
*/
export class LocalTaskWorker {
private abortWait: AbortController | undefined;
@@ -31,10 +36,6 @@ export class LocalTaskWorker {
private readonly logger: Logger,
) {}
get id(): string {
return this.taskId;
}
start(settings: TaskSettingsV2, options?: { signal?: AbortSignal }) {
this.logger.info(
`Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`,
@@ -43,19 +44,19 @@ export class LocalTaskWorker {
(async () => {
try {
if (settings.initialDelayDuration) {
await sleep(
await this.sleep(
Duration.fromISO(settings.initialDelayDuration),
options?.signal,
);
}
while (!options?.signal?.aborted) {
const startTime = Date.now();
const startTime = process.hrtime();
await this.runOnce(settings, options?.signal);
const endTime = Date.now();
const timeTaken = process.hrtime(startTime);
await this.waitUntilNext(
settings,
endTime - startTime,
(timeTaken[0] + timeTaken[1] / 1e9) * 1000,
options?.signal,
);
}
@@ -130,8 +131,15 @@ export class LocalTaskWorker {
)}`,
);
this.abortWait = delegateAbortController(signal);
await sleep(Duration.fromMillis(dt), this.abortWait.signal);
await this.sleep(Duration.fromMillis(dt), signal);
}
private async sleep(
duration: Duration,
abortSignal?: AbortSignal,
): Promise<void> {
this.abortWait = delegateAbortController(abortSignal);
await sleep(duration, this.abortWait.signal);
this.abortWait.abort(); // cleans up resources
this.abortWait = undefined;
}
packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts
+110 -4
View File
@@ -16,11 +16,11 @@
import { getVoidLogger } from '@backstage/backend-common';
import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
import { ConflictError, NotFoundError } from '@backstage/errors';
import { Duration } from 'luxon';
import { AbortSignal } from 'node-abort-controller';
import { migrateBackendTasks } from '../database/migrateBackendTasks';
import { PluginTaskSchedulerImpl } from './PluginTaskSchedulerImpl';
import { ConflictError, NotFoundError } from '@backstage/errors';
import { AbortSignal } from 'node-abort-controller';
jest.useFakeTimers();
@@ -49,7 +49,7 @@ describe('PluginTaskManagerImpl', () => {
// This is just to test the wrapper code; most of the actual tests are in
// TaskWorker.test.ts
describe('scheduleTask', () => {
describe('scheduleTask with global scope', () => {
it.each(databases.eachSupportedId())(
'can run the v1 happy path, %p',
async databaseId => {
@@ -62,6 +62,7 @@ describe('PluginTaskManagerImpl', () => {
timeout: Duration.fromMillis(5000),
frequency: Duration.fromMillis(5000),
fn,
scope: 'global',
});
await promise;
@@ -82,6 +83,7 @@ describe('PluginTaskManagerImpl', () => {
timeout: Duration.fromMillis(5000),
frequency: { cron: '* * * * * *' },
fn,
scope: 'global',
});
await promise;
@@ -91,7 +93,7 @@ describe('PluginTaskManagerImpl', () => {
);
});
describe('triggerTask', () => {
describe('triggerTask with global scope', () => {
it.each(databases.eachSupportedId())(
'can manually trigger a task, %p',
async databaseId => {
@@ -105,6 +107,7 @@ describe('PluginTaskManagerImpl', () => {
frequency: Duration.fromObject({ years: 1 }),
initialDelay: Duration.fromObject({ years: 1 }),
fn,
scope: 'global',
});
await manager.triggerTask('task1');
@@ -127,6 +130,7 @@ describe('PluginTaskManagerImpl', () => {
timeout: Duration.fromMillis(5000),
frequency: Duration.fromObject({ years: 1 }),
fn,
scope: 'global',
});
await expect(() => manager.triggerTask('task2')).rejects.toThrow(
@@ -151,6 +155,7 @@ describe('PluginTaskManagerImpl', () => {
resolve();
await new Promise(r => setTimeout(r, 20000));
},
scope: 'global',
});
await promise;
@@ -162,6 +167,106 @@ describe('PluginTaskManagerImpl', () => {
);
});
// This is just to test the wrapper code; most of the actual tests are in
// TaskWorker.test.ts
describe('scheduleTask with local scope', () => {
it('can run the v1 happy path', async () => {
const { manager } = await init('SQLITE_3');
const fn = jest.fn();
const promise = new Promise(resolve => fn.mockImplementation(resolve));
await manager.scheduleTask({
id: 'task1',
timeout: Duration.fromMillis(5000),
frequency: Duration.fromMillis(5000),
fn,
scope: 'local',
});
await promise;
expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal));
}, 60_000);
it('can run the v2 happy path', async () => {
const { manager } = await init('SQLITE_3');
const fn = jest.fn();
const promise = new Promise(resolve => fn.mockImplementation(resolve));
await manager.scheduleTask({
id: 'task2',
timeout: Duration.fromMillis(5000),
frequency: { cron: '* * * * * *' },
fn,
scope: 'local',
});
await promise;
expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal));
}, 60_000);
});
describe('triggerTask with local scope', () => {
it('can manually trigger a task', async () => {
const { manager } = await init('SQLITE_3');
const fn = jest.fn();
const promise = new Promise(resolve => fn.mockImplementation(resolve));
await manager.scheduleTask({
id: 'task1',
timeout: Duration.fromMillis(5000),
frequency: Duration.fromObject({ years: 1 }),
initialDelay: Duration.fromObject({ years: 1 }),
fn,
scope: 'local',
});
await manager.triggerTask('task1');
jest.advanceTimersByTime(5000);
await promise;
expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal));
}, 60_000);
it('cant trigger a non-existent task', async () => {
const { manager } = await init('SQLITE_3');
const fn = jest.fn();
await manager.scheduleTask({
id: 'task1',
timeout: Duration.fromMillis(5000),
frequency: Duration.fromObject({ years: 1 }),
fn,
scope: 'local',
});
await expect(() => manager.triggerTask('task2')).rejects.toThrow(
NotFoundError,
);
}, 60_000);
it('cant trigger a running task', async () => {
const { manager } = await init('SQLITE_3');
const { promise, resolve } = defer();
await manager.scheduleTask({
id: 'task1',
timeout: Duration.fromMillis(5000),
frequency: Duration.fromObject({ years: 1 }),
fn: async () => {
resolve();
await new Promise(r => setTimeout(r, 20000));
},
scope: 'local',
});
await promise;
await expect(() => manager.triggerTask('task1')).rejects.toThrow(
ConflictError,
);
}, 60_000);
});
// This is just to test the wrapper code; most of the actual tests are in
// TaskWorker.test.ts
describe('createScheduledTaskRunner', () => {
@@ -176,6 +281,7 @@ describe('PluginTaskManagerImpl', () => {
.createScheduledTaskRunner({
timeout: Duration.fromMillis(5000),
frequency: Duration.fromMillis(5000),
scope: 'global',
})
.run({
id: 'task1',
packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts
+36 -47
View File
@@ -52,49 +52,46 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
task: TaskScheduleDefinition & TaskInvocationDefinition,
): Promise<void> {
validateId(task.id);
const scope = task.scope ?? 'global';
const knex = await this.databaseFactory();
const worker = new TaskWorker(task.id, task.fn, knex, this.logger);
if (scope === 'global') {
const knex = await this.databaseFactory();
const worker = new TaskWorker(task.id, task.fn, knex, this.logger);
await worker.start(
{
version: 2,
cadence:
'cron' in task.frequency
? task.frequency.cron
: task.frequency.toISO(),
initialDelayDuration: task.initialDelay?.toISO(),
timeoutAfterDuration: task.timeout.toISO(),
},
{
signal: task.signal,
},
);
}
await worker.start(
{
version: 2,
cadence:
'cron' in task.frequency
? task.frequency.cron
: task.frequency.toISO(),
initialDelayDuration: task.initialDelay?.toISO(),
timeoutAfterDuration: task.timeout.toISO(),
},
{
signal: task.signal,
},
);
} else {
const worker = new LocalTaskWorker(task.id, task.fn, this.logger);
async scheduleLocalTask(
task: TaskScheduleDefinition & TaskInvocationDefinition,
): Promise<void> {
validateId(task.id);
worker.start(
{
version: 2,
cadence:
'cron' in task.frequency
? task.frequency.cron
: task.frequency.toISO(),
initialDelayDuration: task.initialDelay?.toISO(),
timeoutAfterDuration: task.timeout.toISO(),
},
{
signal: task.signal,
},
);
const worker = new LocalTaskWorker(task.id, task.fn, this.logger);
worker.start(
{
version: 2,
cadence:
'cron' in task.frequency
? task.frequency.cron
: task.frequency.toISO(),
initialDelayDuration: task.initialDelay?.toISO(),
timeoutAfterDuration: task.timeout.toISO(),
},
{
signal: task.signal,
},
);
this.localTasksById.set(task.id, worker);
this.localTasksById.set(task.id, worker);
}
}
createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner {
@@ -104,12 +101,4 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
},
};
}
createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner {
return {
run: async task => {
await this.scheduleLocalTask({ ...task, ...schedule });
},
};
}
}
packages/backend-tasks/src/tasks/TaskScheduler.test.ts
+1 -1
View File
@@ -17,8 +17,8 @@
import { DatabaseManager, getVoidLogger } from '@backstage/backend-common';
import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
import { Duration } from 'luxon';
import { TaskScheduler } from './TaskScheduler';
import waitForExpect from 'wait-for-expect';
import { TaskScheduler } from './TaskScheduler';
describe('TaskScheduler', () => {
const logger = getVoidLogger();
packages/backend-tasks/src/tasks/TaskWorker.test.ts
+1 -1
View File
@@ -14,10 +14,10 @@
* limitations under the License.
*/
import { AbortController } from 'node-abort-controller';
import { getVoidLogger } from '@backstage/backend-common';
import { TestDatabases } from '@backstage/backend-test-utils';
import { Duration } from 'luxon';
import { AbortController } from 'node-abort-controller';
import waitForExpect from 'wait-for-expect';
import { migrateBackendTasks } from '../database/migrateBackendTasks';
import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables';
packages/backend-tasks/src/tasks/TaskWorker.ts
+1 -1
View File
@@ -28,7 +28,7 @@ import { delegateAbortController, nowPlus, sleep } from './util';
const DEFAULT_WORK_CHECK_FREQUENCY = Duration.fromObject({ seconds: 5 });
/**
* Performs the actual work of a task.
* Implements tasks that run across worker hosts, with collaborative locking.
*
* @private
*/
packages/backend-tasks/src/tasks/types.ts
+38 -46
View File
@@ -14,10 +14,10 @@
* limitations under the License.
*/
import { CronTime } from 'cron';
import { Duration } from 'luxon';
import { AbortSignal } from 'node-abort-controller';
import { z } from 'zod';
import { CronTime } from 'cron';
/**
* A function that can be called as a scheduled task.
@@ -93,14 +93,37 @@ export interface TaskScheduleDefinition {
* will happen as soon as possible according to the cadence.
*
* NOTE: This is a per-worker delay. If you have a cluster of workers all
* collaborating on a task, then you may still see the task being processed by
* other long-lived workers, while any given single worker is in its initial
* sleep delay time e.g. after a deployment. Therefore this parameter is not
* useful for "globally" pausing work; its main intended use is for individual
* machines to get a chance to reach some equilibrium at startup before
* triggering heavy batch workloads.
* collaborating on a task that has its `scope` field set to `'global'`, then
* you may still see the task being processed by other long-lived workers,
* while any given single worker is in its initial sleep delay time e.g. after
* a deployment. Therefore this parameter is not useful for "globally" pausing
* work; its main intended use is for individual machines to get a chance to
* reach some equilibrium at startup before triggering heavy batch workloads.
*/
initialDelay?: Duration;
/**
* Sets the scope of concurrency control / locking to apply for invocations of
* this task.
*
* @remarks
*
* When the scope is set to the default value `'global'`, the scheduler will
* attempt to ensure that only one worker machine runs the task at a time,
* according to the given cadence. This means that as the number of worker
* hosts increases, the invocation frequency of this task will not go up.
* Instead the load is spread randomly across hosts. This setting is useful
* for tasks that access shared resources, for example catalog ingestion tasks
* where you do not want many machines to repeatedly import the same data and
* trample over each other.
*
* When the scope is set to `'local'`, there is no concurrency control across
* hosts. Each host runs the task according to the given cadence similarly to
* `setInterval`, but the runtime ensures that there are no overlapping runs.
*
* @defaultValue 'global'
*/
scope?: 'global' | 'local';
}
/**
@@ -149,24 +172,23 @@ export interface PluginTaskScheduler {
/**
* Manually triggers a task by ID.
*
* If the task doesn't exist, a NotFoundError is thrown.
* If the task is currently running, a ConflictError is thrown.
* If the task doesn't exist, a NotFoundError is thrown. If the task is
* currently running, a ConflictError is thrown.
*
* @param id - The task ID
*
*/
triggerTask(id: string): Promise<void>;
/**
* Schedules a task function for coordinated exclusive invocation across
* workers. This convenience method performs both the scheduling and
* invocation in one go.
* Schedules a task function for recurring runs.
*
* @remarks
*
* If the task was already scheduled since before by us or by another party,
* its options are just overwritten with the given options, and things
* continue from there.
* The `scope` task field controls whether to use coordinated exclusive
* invocation across workers, or to just coordinate within the current worker.
*
* This convenience method performs both the scheduling and invocation in one
* go.
*
* @param task - The task definition
*/
@@ -174,21 +196,6 @@ export interface PluginTaskScheduler {
task: TaskScheduleDefinition & TaskInvocationDefinition,
): Promise<void>;
/**
* Schedules a task function for running only locally on this machine, without
* any form of coordination, locking or concurrency control across hosts. This
* convenience method performs both the scheduling and invocation in one go.
*
* @remarks
*
* If a task with this ID was already registered, a ConflictError is thrown.
*
* @param task - The task definition
*/
scheduleLocalTask(
task: TaskScheduleDefinition & TaskInvocationDefinition,
): Promise<void>;
/**
* Creates a scheduled but dormant recurring task, ready to be launched at a
* later time.
@@ -202,21 +209,6 @@ export interface PluginTaskScheduler {
* @param schedule - The task schedule
*/
createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner;
/**
* Creates a local (without any form of coordination, locking or concurrency
* control across hosts) scheduled but dormant recurring task, ready to be
* launched at a later time.
*
* @remarks
*
* This method is useful for pre-creating a schedule in outer code to be
* passed into an inner implementation, such that the outer code controls
* scheduling while inner code controls implementation.
*
* @param schedule - The task schedule
*/
createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner;
}
function isValidOptionalDurationString(d: string | undefined): boolean {