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

Add local task running support to TaskScheduler

Browse Source 
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
2022-04-21 14:18:15 +02:00
parent 6aea66b400
commit cfd779a9bc
7 changed files with 348 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/swift-parrots-hammer.md
+5
View File
@@ -0,0 +1,5 @@
---
'@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.
packages/backend-tasks/api-report.md
+4
View File
@@ -11,7 +11,11 @@ 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>;
packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts
+105
View File
@@ -0,0 +1,105 @@
/*
* Copyright 2022 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { getVoidLogger } from '@backstage/backend-common';
import { AbortController } from 'node-abort-controller';
import { LocalTaskWorker } from './LocalTaskWorker';
describe('LocalTaskWorker', () => {
const logger = getVoidLogger();
it('runs the happy path (with iso duration) and handles cancellation', async () => {
const fn = jest.fn();
const controller = new AbortController();
const worker = new LocalTaskWorker('a', fn, logger);
worker.start(
{
version: 2,
initialDelayDuration: 'PT0.2S',
cadence: 'PT0.2S',
timeoutAfterDuration: 'PT1S',
},
{ signal: controller.signal },
);
// TODO(freben): Rewrite to fake timers - tried, but it wouldn't work
expect(fn).toBeCalledTimes(0);
await new Promise(r => setTimeout(r, 100));
expect(fn).toBeCalledTimes(0);
await new Promise(r => setTimeout(r, 200));
expect(fn).toBeCalledTimes(1);
await new Promise(r => setTimeout(r, 200));
expect(fn).toBeCalledTimes(2);
controller.abort();
await new Promise(r => setTimeout(r, 200));
expect(fn).toBeCalledTimes(2);
});
it('runs the happy path (with a cron expression) and handles cancellation', async () => {
const fn = jest.fn();
const controller = new AbortController();
// Await until system time is just past a second boundary (since cron is
// wall clock based)
await new Promise(r => setTimeout(r, 1000 - (Date.now() % 1000) + 10));
const worker = new LocalTaskWorker('a', fn, logger);
worker.start(
{
version: 2,
initialDelayDuration: 'PT0.2S',
cadence: '* * * * * *',
timeoutAfterDuration: 'PT1S',
},
{ signal: controller.signal },
);
// TODO(freben): Rewrite to fake timers - tried, but it wouldn't work
expect(fn).toBeCalledTimes(0);
await new Promise(r => setTimeout(r, 100));
expect(fn).toBeCalledTimes(0);
await new Promise(r => setTimeout(r, 200));
expect(fn).toBeCalledTimes(1);
await new Promise(r => setTimeout(r, 1000));
expect(fn).toBeCalledTimes(2);
controller.abort();
await new Promise(r => setTimeout(r, 1000));
expect(fn).toBeCalledTimes(2);
});
it('can trigger to abort wait', async () => {
const fn = jest.fn();
const worker = new LocalTaskWorker('a', fn, logger);
worker.start({
version: 2,
initialDelayDuration: 'PT0.2S',
cadence: 'PT0.2S',
timeoutAfterDuration: 'PT1S',
});
// TODO(freben): Rewrite to fake timers - tried, but it wouldn't work
expect(fn).toBeCalledTimes(0);
await new Promise(r => setTimeout(r, 100));
expect(fn).toBeCalledTimes(0);
await new Promise(r => setTimeout(r, 200));
expect(fn).toBeCalledTimes(1);
worker.trigger();
await new Promise(r => setTimeout(r, 10));
expect(fn).toBeCalledTimes(2);
});
});
packages/backend-tasks/src/tasks/LocalTaskWorker.ts
+138
View File
@@ -0,0 +1,138 @@
/*
* Copyright 2022 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { ConflictError } from '@backstage/errors';
import { CronTime } from 'cron';
import { DateTime, Duration } from 'luxon';
import { AbortController, AbortSignal } from 'node-abort-controller';
import { Logger } from 'winston';
import { TaskFunction, TaskSettingsV2 } from './types';
import { delegateAbortController, sleep } from './util';
export class LocalTaskWorker {
private abortWait: AbortController | undefined;
constructor(
private readonly taskId: string,
private readonly fn: TaskFunction,
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)}`,
);
(async () => {
try {
if (settings.initialDelayDuration) {
await sleep(
Duration.fromISO(settings.initialDelayDuration),
options?.signal,
);
}
while (!options?.signal?.aborted) {
const startTime = Date.now();
await this.runOnce(settings, options?.signal);
const endTime = Date.now();
await this.waitUntilNext(
settings,
endTime - startTime,
options?.signal,
);
}
this.logger.info(`Task worker finished: ${this.taskId}`);
} catch (e) {
this.logger.warn(`Task worker failed unexpectedly, ${e}`);
}
})();
}
trigger(): void {
if (!this.abortWait) {
throw new ConflictError(`Task ${this.taskId} is currently running`);
}
this.abortWait.abort();
}
/**
* Makes a single attempt at running the task to completion.
*/
private async runOnce(
settings: TaskSettingsV2,
signal?: AbortSignal,
): Promise<void> {
// Abort the task execution either if the worker is stopped, or if the
// task timeout is hit
const taskAbortController = delegateAbortController(signal);
const timeoutHandle = setTimeout(() => {
taskAbortController.abort();
}, Duration.fromISO(settings.timeoutAfterDuration).as('milliseconds'));
try {
await this.fn(taskAbortController.signal);
} catch (e) {
// ignore intentionally
}
// release resources
clearTimeout(timeoutHandle);
taskAbortController.abort();
}
/**
* Sleeps until it's time to run the task again.
*/
private async waitUntilNext(
settings: TaskSettingsV2,
lastRunMillis: number,
signal?: AbortSignal,
) {
if (signal?.aborted) {
return;
}
const isCron = !settings.cadence.startsWith('P');
let dt: number;
if (isCron) {
const nextRun = +new CronTime(settings.cadence).sendAt().toDate();
dt = nextRun - Date.now();
} else {
dt =
Duration.fromISO(settings.cadence).as('milliseconds') - lastRunMillis;
}
dt = Math.max(dt, 0);
this.logger.debug(
`task: ${this.taskId} will next occur around ${DateTime.now().plus(
Duration.fromMillis(dt),
)}`,
);
this.abortWait = delegateAbortController(signal);
await sleep(Duration.fromMillis(dt), this.abortWait.signal);
this.abortWait.abort(); // cleans up resources
this.abortWait = undefined;
}
}
packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts
+43 -20
View File
@@ -16,6 +16,7 @@
import { Knex } from 'knex';
import { Logger } from 'winston';
import { LocalTaskWorker } from './LocalTaskWorker';
import { TaskWorker } from './TaskWorker';
import {
PluginTaskScheduler,
@@ -24,38 +25,27 @@ import {
TaskScheduleDefinition,
} from './types';
import { validateId } from './util';
import { DB_TASKS_TABLE, DbTasksRow } from '../database/tables';
import { ConflictError, NotFoundError } from '@backstage/errors';
/**
* Implements the actual task management.
*/
export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
private readonly localTasksById = new Map<string, LocalTaskWorker>();
constructor(
private readonly databaseFactory: () => Promise<Knex>,
private readonly logger: Logger,
) {}
async triggerTask(id: string): Promise<void> {
const localTask = this.localTasksById.get(id);
if (localTask) {
localTask.trigger();
return;
}
const knex = await this.databaseFactory();
// check if task exists
const rows = await knex<DbTasksRow>(DB_TASKS_TABLE)
.select(knex.raw(1))
.where('id', '=', id);
if (rows.length !== 1) {
throw new NotFoundError(`Task ${id} does not exist`);
}
const updatedRows = await knex<DbTasksRow>(DB_TASKS_TABLE)
.where('id', '=', id)
.whereNull('current_run_ticket')
.update({
next_run_start_at: knex.fn.now(),
});
if (updatedRows < 1) {
throw new ConflictError(`Task ${id} is currently running`);
}
await TaskWorker.trigger(knex, id);
}
async scheduleTask(
@@ -82,6 +72,31 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
);
}
async scheduleLocalTask(
task: TaskScheduleDefinition & TaskInvocationDefinition,
): Promise<void> {
validateId(task.id);
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);
}
createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner {
return {
run: async task => {
@@ -89,4 +104,12 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
},
};
}
createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner {
return {
run: async task => {
await this.scheduleLocalTask({ ...task, ...schedule });
},
};
}
}
packages/backend-tasks/src/tasks/TaskWorker.ts
+23 -2
View File
@@ -14,6 +14,8 @@
* limitations under the License.
*/
import { ConflictError, NotFoundError } from '@backstage/errors';
import { CronTime } from 'cron';
import { Knex } from 'knex';
import { DateTime, Duration } from 'luxon';
import { AbortSignal } from 'node-abort-controller';
@@ -22,7 +24,6 @@ import { Logger } from 'winston';
import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables';
import { TaskFunction, TaskSettingsV2, taskSettingsV2Schema } from './types';
import { delegateAbortController, nowPlus, sleep } from './util';
import { CronTime } from 'cron';
const DEFAULT_WORK_CHECK_FREQUENCY = Duration.fromObject({ seconds: 5 });
@@ -76,12 +77,32 @@ export class TaskWorker {
})();
}
static async trigger(knex: Knex, taskId: string): Promise<void> {
// check if task exists
const rows = await knex<DbTasksRow>(DB_TASKS_TABLE)
.select(knex.raw(1))
.where('id', '=', taskId);
if (rows.length !== 1) {
throw new NotFoundError(`Task ${taskId} does not exist`);
}
const updatedRows = await knex<DbTasksRow>(DB_TASKS_TABLE)
.where('id', '=', taskId)
.whereNull('current_run_ticket')
.update({
next_run_start_at: knex.fn.now(),
});
if (updatedRows < 1) {
throw new ConflictError(`Task ${taskId} is currently running`);
}
}
/**
* Makes a single attempt at running the task to completion, if ready.
*
* @returns The outcome of the attempt
*/
async runOnce(
private async runOnce(
signal?: AbortSignal,
): Promise<
| { result: 'not-ready-yet' }
packages/backend-tasks/src/tasks/types.ts
+30
View File
@@ -174,6 +174,21 @@ 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.
@@ -187,6 +202,21 @@ 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 {