From a83babdd6366c054372b2584716d2731869ccec9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Wed, 30 Mar 2022 10:31:24 +0200 Subject: [PATCH] fix initialDelay for tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/ninety-radios-protect.md | 14 +++++ .../src/tasks/TaskWorker.test.ts | 55 +++++++++++++++++++ .../backend-tasks/src/tasks/TaskWorker.ts | 9 +++ packages/backend-tasks/src/tasks/types.ts | 20 +++++-- 4 files changed, 94 insertions(+), 4 deletions(-) create mode 100644 .changeset/ninety-radios-protect.md diff --git a/.changeset/ninety-radios-protect.md b/.changeset/ninety-radios-protect.md new file mode 100644 index 0000000000..fa3fae281f --- /dev/null +++ b/.changeset/ninety-radios-protect.md @@ -0,0 +1,14 @@ +--- +'@backstage/backend-tasks': patch +--- + +Fixed the `initialDelay` parameter of tasks to properly make task workers +_always_ wait before the first invocations on startup, not just the very first +time that the task is ever created. This behavior is more in line with +expectations. Callers to not need to update their code. + +Also clarified in the doc comment for the field that this wait applies only on +an individual worker level. That is, if you have a cluster of workers then each +individual machine may postpone its first task invocation by the given amount of +time to leave room for the service to settle, but _other_ workers may still +continue to invoke the task on the regular cadence in the meantime. diff --git a/packages/backend-tasks/src/tasks/TaskWorker.test.ts b/packages/backend-tasks/src/tasks/TaskWorker.test.ts index f235c58973..0c64d06287 100644 --- a/packages/backend-tasks/src/tasks/TaskWorker.test.ts +++ b/packages/backend-tasks/src/tasks/TaskWorker.test.ts @@ -14,6 +14,7 @@ * 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'; @@ -257,4 +258,58 @@ describe('TaskWorker', () => { }, 60_000, ); + + it.each(databases.eachSupportedId())( + 'respects initialDelayDuration per worker, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const abortFirst = new AbortController(); + const settings: TaskSettingsV2 = { + version: 2, + initialDelayDuration: 'PT0.3S', + cadence: 'PT0.1S', + timeoutAfterDuration: 'PT10S', + }; + + // Start a single worker and make sure it waits and then goes to work + const fn1 = jest.fn(async () => {}); + const worker1 = new TaskWorker( + 'task1', + fn1, + knex, + logger, + Duration.fromMillis(10), + ); + await worker1.start(settings, { signal: abortFirst.signal }); + + expect(fn1).toBeCalledTimes(0); + await new Promise(resolve => setTimeout(resolve, 250)); + expect(fn1).toBeCalledTimes(0); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(fn1.mock.calls.length).toBeGreaterThan(0); + + // Start a second worker and make sure it waits but the first worker still works along + const fn2 = jest.fn(); + const promise2 = new Promise(resolve => fn2.mockImplementation(resolve)); + const worker2 = new TaskWorker( + 'task1', + fn2, + knex, + logger, + Duration.fromMillis(10), + ); + await worker2.start(settings); + + // We eventually abort the first worker just to make sure that the second + // one for sure will get a go at running the task + setTimeout(() => abortFirst.abort(), 1000); + + const before = fn1.mock.calls.length; + await promise2; + expect(fn1.mock.calls.length).toBeGreaterThan(before); + }, + 60_000, + ); }); diff --git a/packages/backend-tasks/src/tasks/TaskWorker.ts b/packages/backend-tasks/src/tasks/TaskWorker.ts index edec453a69..02bc115ddd 100644 --- a/packages/backend-tasks/src/tasks/TaskWorker.ts +++ b/packages/backend-tasks/src/tasks/TaskWorker.ts @@ -53,6 +53,13 @@ export class TaskWorker { (async () => { try { + if (settings.initialDelayDuration) { + await sleep( + Duration.fromISO(settings.initialDelayDuration), + options?.signal, + ); + } + while (!options?.signal?.aborted) { const runResult = await this.runOnce(options?.signal); if (runResult.result === 'abort') { @@ -61,6 +68,7 @@ export class TaskWorker { await sleep(this.workCheckFrequency, options?.signal); } + this.logger.info(`Task worker finished: ${this.taskId}`); } catch (e) { this.logger.warn(`Task worker failed unexpectedly, ${e}`); @@ -106,6 +114,7 @@ export class TaskWorker { try { await this.fn(taskAbortController.signal); + taskAbortController.abort(); // releases resources } catch (e) { await this.tryReleaseTask(ticket, taskSettings); return { result: 'failed' }; diff --git a/packages/backend-tasks/src/tasks/types.ts b/packages/backend-tasks/src/tasks/types.ts index 513e0dbae4..1e09e52cd0 100644 --- a/packages/backend-tasks/src/tasks/types.ts +++ b/packages/backend-tasks/src/tasks/types.ts @@ -41,6 +41,8 @@ export interface TaskScheduleDefinition { * How often you want the task to run. The system does its best to avoid * overlapping invocations. * + * @remarks + * * This is a best effort value; under some circumstances there can be * deviations. For example, if the task runtime is longer than the frequency * and the timeout has not been given or not been exceeded yet, the next @@ -54,6 +56,8 @@ export interface TaskScheduleDefinition { /** * A crontab style string. * + * @remarks + * * Overview: * * ``` @@ -82,11 +86,19 @@ export interface TaskScheduleDefinition { /** * The amount of time that should pass before the first invocation happens. * - * This can be useful in cold start scenarios to stagger or delay some heavy - * compute jobs. + * @remarks * - * If no value is given for this field then the first invocation will happen - * as soon as possible according to the cadence. + * This can be useful in cold start scenarios to stagger or delay some heavy + * compute jobs. If no value is given for this field then the first invocation + * 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. */ initialDelay?: Duration; }