add the scheduler definitions to backend-plugin-api/scheduler

Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
2024-04-24 15:15:10 +02:00
parent d31ddb5cd2
commit abc983847f
15 changed files with 681 additions and 14 deletions
@@ -1,5 +1,5 @@
/*
* Copyright 2022 The Backstage Authors
* Copyright 2024 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.
@@ -14,7 +14,10 @@
* limitations under the License.
*/
import { PluginTaskScheduler } from '@backstage/backend-tasks';
import { type SchedulerService as SchedulerService_ } from '@backstage/backend-plugin-api/scheduler';
/** @public */
export interface SchedulerService extends PluginTaskScheduler {}
/**
* @public
* @deprecated import from `@backstage/backend-plugin-api/scheduler` instead
*/
export type SchedulerService = SchedulerService_;
@@ -0,0 +1,26 @@
/*
* Copyright 2024 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.
*/
export { readTaskScheduleDefinitionFromConfig } from './readTaskScheduleDefinitionFromConfig';
export type {
SchedulerService,
TaskDescriptor,
TaskFunction,
TaskInvocationDefinition,
TaskRunner,
TaskScheduleDefinition,
TaskScheduleDefinitionConfig,
} from './types';
@@ -0,0 +1,132 @@
/*
* 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 { ConfigReader } from '@backstage/config';
import { HumanDuration } from '@backstage/types';
import { Duration } from 'luxon';
import { readTaskScheduleDefinitionFromConfig } from './readTaskScheduleDefinitionFromConfig';
describe('readTaskScheduleDefinitionFromConfig', () => {
it('all valid values', () => {
const config = new ConfigReader({
frequency: {
cron: '0 30 * * * *',
},
timeout: 'PT3M',
initialDelay: {
minutes: 20,
},
scope: 'global',
});
const result = readTaskScheduleDefinitionFromConfig(config);
expect((result.frequency as { cron: string }).cron).toBe('0 30 * * * *');
expect(result.timeout).toEqual(Duration.fromISO('PT3M'));
expect((result.initialDelay as HumanDuration).minutes).toEqual(20);
expect(result.scope).toBe('global');
});
it('all valid required values', () => {
const config = new ConfigReader({
frequency: {
cron: '0 30 * * * *',
},
timeout: 'PT3M',
});
const result = readTaskScheduleDefinitionFromConfig(config);
expect((result.frequency as { cron: string }).cron).toBe('0 30 * * * *');
expect(result.timeout).toEqual(Duration.fromISO('PT3M'));
expect(result.initialDelay).toBeUndefined();
expect(result.scope).toBeUndefined();
});
it('fail without required frequency', () => {
const config = new ConfigReader({
timeout: 'PT3M',
});
expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow(
"Missing required config value at 'frequency'",
);
});
it('fail without required timeout', () => {
const config = new ConfigReader({
frequency: 'PT30M',
});
expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow(
"Missing required config value at 'timeout'",
);
});
it('invalid frequency key', () => {
const config = new ConfigReader({
frequency: {
invalid: 'value',
},
timeout: 'PT3M',
});
expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow(
"Failed to read duration from config at 'frequency', Error: Needs one or more of 'years', 'months', 'weeks', 'days', 'hours', 'minutes', 'seconds', 'milliseconds'",
);
});
it('invalid frequency value', () => {
const config = new ConfigReader({
frequency: {
minutes: 'value',
},
timeout: 'PT3M',
});
expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow(
"Failed to read duration from config, Error: Unable to convert config value for key 'frequency.minutes' in 'mock-config' to a number",
);
});
it('frequency value with additional invalid prop', () => {
const config = new ConfigReader({
frequency: {
minutes: 20,
invalid: 'value',
},
timeout: 'PT3M',
});
expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow(
"Failed to read duration from config at 'frequency', Error: Unknown property 'invalid'; expected one or more of 'years', 'months', 'weeks', 'days', 'hours', 'minutes', 'seconds', 'milliseconds'",
);
});
it('invalid scope value', () => {
const config = new ConfigReader({
frequency: {
years: 2,
},
timeout: 'PT3M',
scope: 'invalid',
});
expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow(
'Only "global" or "local" are allowed for TaskScheduleDefinition.scope, but got: invalid',
);
});
});
@@ -0,0 +1,78 @@
/*
* 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 { Config, readDurationFromConfig } from '@backstage/config';
import { HumanDuration } from '@backstage/types';
import { TaskScheduleDefinition } from './types';
import { Duration } from 'luxon';
function readDuration(config: Config, key: string): Duration | HumanDuration {
if (typeof config.get(key) === 'string') {
const value = config.getString(key);
const duration = Duration.fromISO(value);
if (!duration.isValid) {
throw new Error(`Invalid duration: ${value}`);
}
return duration;
}
return readDurationFromConfig(config, { key });
}
function readCronOrDuration(
config: Config,
key: string,
): { cron: string } | Duration | HumanDuration {
const value = config.get(key);
if (typeof value === 'object' && (value as { cron?: string }).cron) {
return value as { cron: string };
}
return readDuration(config, key);
}
/**
* Reads a TaskScheduleDefinition from a Config.
* Expects the config not to be the root config,
* but the config for the definition.
*
* @param config - config for a TaskScheduleDefinition.
* @public
*/
export function readTaskScheduleDefinitionFromConfig(
config: Config,
): TaskScheduleDefinition {
const frequency = readCronOrDuration(config, 'frequency');
const timeout = readDuration(config, 'timeout');
const initialDelay = config.has('initialDelay')
? readDuration(config, 'initialDelay')
: undefined;
const scope = config.getOptionalString('scope');
if (scope && !['global', 'local'].includes(scope)) {
throw new Error(
`Only "global" or "local" are allowed for TaskScheduleDefinition.scope, but got: ${scope}`,
);
}
return {
frequency,
timeout,
initialDelay,
scope: scope as 'global' | 'local' | undefined,
};
}
@@ -0,0 +1,343 @@
/*
* Copyright 2021 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 { HumanDuration, JsonObject } from '@backstage/types';
import { Duration } from 'luxon';
/**
* A function that can be called as a scheduled task.
*
* It may optionally accept an abort signal argument. When the signal triggers,
* processing should abort and return as quickly as possible.
*
* @public
*/
export type TaskFunction =
| ((abortSignal: AbortSignal) => void | Promise<void>)
| (() => void | Promise<void>);
/**
* A semi-opaque type to describe an actively scheduled task.
*
* @public
*/
export type TaskDescriptor = {
/**
* The unique identifier of the task.
*/
id: string;
/**
* The scope of the task.
*/
scope: 'global' | 'local';
/**
* The settings that control the task flow. This is a semi-opaque structure
* that is mainly there for debugging purposes. Do not make any assumptions
* about the contents of this field.
*/
settings: { version: number } & JsonObject;
};
/**
* Options that control the scheduling of a task.
*
* @public
*/
export interface TaskScheduleDefinition {
/**
* How often you want the task to run. The system does its best to avoid
* overlapping invocations.
*
* @remarks
*
* This is the 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
* invocation of this task will be delayed until after the previous one
* finishes.
*
* This is a required field.
*/
frequency:
| {
/**
* A crontab style string.
*
* @remarks
*
* Overview:
*
* ```
* ┌────────────── second (optional)
* │ ┌──────────── minute
* │ │ ┌────────── hour
* │ │ │ ┌──────── day of month
* │ │ │ │ ┌────── month
* │ │ │ │ │ ┌──── day of week
* │ │ │ │ │ │
* │ │ │ │ │ │
* * * * * * *
* ```
*/
cron: string;
}
| Duration
| HumanDuration;
/**
* The maximum amount of time that a single task invocation can take, before
* it's considered timed out and gets "released" such that a new invocation
* is permitted to take place (possibly, then, on a different worker).
*/
timeout: Duration | HumanDuration;
/**
* The amount of time that should pass before the first invocation happens.
*
* @remarks
*
* 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 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 | HumanDuration;
/**
* 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';
}
/**
* Config options for {@link TaskScheduleDefinition}
* that control the scheduling of a task.
*
* @public
*/
export interface TaskScheduleDefinitionConfig {
/**
* How often you want the task to run. The system does its best to avoid
* overlapping invocations.
*
* @remarks
*
* This is the 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
* invocation of this task will be delayed until after the previous one
* finishes.
*
* This is a required field.
*/
frequency:
| {
/**
* A crontab style string.
*
* @remarks
*
* Overview:
*
* ```
* ┌────────────── second (optional)
* │ ┌──────────── minute
* │ │ ┌────────── hour
* │ │ │ ┌──────── day of month
* │ │ │ │ ┌────── month
* │ │ │ │ │ ┌──── day of week
* │ │ │ │ │ │
* │ │ │ │ │ │
* * * * * * *
* ```
*/
cron: string;
}
| string
| HumanDuration;
/**
* The maximum amount of time that a single task invocation can take, before
* it's considered timed out and gets "released" such that a new invocation
* is permitted to take place (possibly, then, on a different worker).
*/
timeout: string | HumanDuration;
/**
* The amount of time that should pass before the first invocation happens.
*
* @remarks
*
* 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 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?: string | HumanDuration;
/**
* 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';
}
/**
* Options that apply to the invocation of a given task.
*
* @public
*/
export interface TaskInvocationDefinition {
/**
* A unique ID (within the scope of the plugin) for the task.
*/
id: string;
/**
* The actual task function to be invoked regularly.
*/
fn: TaskFunction;
/**
* An abort signal that, when triggered, will stop the recurring execution of
* the task.
*/
signal?: AbortSignal;
}
/**
* A previously prepared task schedule, ready to be invoked.
*
* @public
*/
export interface TaskRunner {
/**
* Takes the schedule and executes an actual task using it.
*
* @param task - The actual runtime properties of the task
*/
run(task: TaskInvocationDefinition): Promise<void>;
}
/**
* Deals with the scheduling of distributed tasks, for a given plugin.
*
* @public
*/
export interface SchedulerService {
/**
* 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.
*
* @param id - The task ID
*/
triggerTask(id: string): Promise<void>;
/**
* Schedules a task function for recurring runs.
*
* @remarks
*
* 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
*/
scheduleTask(
task: TaskScheduleDefinition & TaskInvocationDefinition,
): Promise<void>;
/**
* Creates a 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
*/
createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner;
/**
* Returns all scheduled tasks registered to this scheduler.
*
* @remarks
*
* This method is useful for triggering tasks manually using the triggerTask
* functionality. Note that the returned tasks contain only tasks that have
* been initialized in this instance of the scheduler.
*
* @returns Scheduled tasks
*/
getScheduledTasks(): Promise<TaskDescriptor[]>;
}
+1
View File
@@ -24,3 +24,4 @@ export * from './services';
export type { BackendFeature } from './types';
export * from './paths';
export * from './wiring';
export * from './deprecated';
@@ -165,7 +165,7 @@ export namespace coreServices {
* @public
*/
export const scheduler = createServiceRef<
import('./SchedulerService').SchedulerService
import('@backstage/backend-plugin-api/scheduler').SchedulerService
>({ id: 'core.scheduler' });
/**
@@ -52,7 +52,6 @@ export type { PluginMetadataService } from './PluginMetadataService';
export type { RootHttpRouterService } from './RootHttpRouterService';
export type { RootLifecycleService } from './RootLifecycleService';
export type { RootLoggerService } from './RootLoggerService';
export type { SchedulerService } from './SchedulerService';
export type { TokenManagerService } from './TokenManagerService';
export type {
ReadTreeOptions,