Merge pull request #16527 from drodil/task_scheduler_get_tasks

feat: add functionality to get scheduled tasks

.changeset/fresh-hairs-switch.md

@@ -0,0 +1,5 @@
+---
+'@backstage/backend-tasks': minor
+---
+
+add functionality to get descriptions from the scheduler for triggering

packages/backend-tasks/api-report.md

@@ -7,6 +7,7 @@ import { Config } from '@backstage/config';
 import { DatabaseManager } from '@backstage/backend-common';
 import { Duration } from 'luxon';
 import { HumanDuration as HumanDuration_2 } from '@backstage/types';
+import { JsonObject } from '@backstage/types';
 import { Logger } from 'winston';
 import { PluginDatabaseManager } from '@backstage/backend-common';
 
@@ -16,6 +17,7 @@ export type HumanDuration = HumanDuration_2;
 // @public
 export interface PluginTaskScheduler {
   createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner;
+  getScheduledTasks(): Promise<TaskDescriptor[]>;
   scheduleTask(
     task: TaskScheduleDefinition & TaskInvocationDefinition,
   ): Promise<void>;
@@ -27,6 +29,15 @@ export function readTaskScheduleDefinitionFromConfig(
   config: Config,
 ): TaskScheduleDefinition;
 
+// @public
+export type TaskDescriptor = {
+  id: string;
+  scope: 'global' | 'local';
+  settings: {
+    version: number;
+  } & JsonObject;
+};
+
 // @public
 export type TaskFunction =
   | ((abortSignal: AbortSignal) => void | Promise<void>)

packages/backend-tasks/src/migrations.test.ts

@@ -0,0 +1,82 @@
+/*
+ * 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 { Knex } from 'knex';
+import { TestDatabases } from '@backstage/backend-test-utils';
+import fs from 'fs';
+
+const migrationsDir = `${__dirname}/../migrations`;
+const migrationsFiles = fs.readdirSync(migrationsDir).sort();
+
+async function migrateUpOnce(knex: Knex): Promise<void> {
+  await knex.migrate.up({ directory: migrationsDir });
+}
+
+async function migrateDownOnce(knex: Knex): Promise<void> {
+  await knex.migrate.down({ directory: migrationsDir });
+}
+
+async function migrateUntilBefore(knex: Knex, target: string): Promise<void> {
+  const index = migrationsFiles.indexOf(target);
+  if (index === -1) {
+    throw new Error(`Migration ${target} not found`);
+  }
+  for (let i = 0; i < index; i++) {
+    await migrateUpOnce(knex);
+  }
+}
+
+describe('migrations', () => {
+  const databases = TestDatabases.create({
+    ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
+  });
+
+  it.each(databases.eachSupportedId())(
+    '20210928160613_init.js, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+
+      await migrateUntilBefore(knex, '20210928160613_init.js');
+      await migrateUpOnce(knex);
+
+      await knex('backstage_backend_tasks__tasks').insert({
+        id: 'test',
+        settings_json: '{}',
+        next_run_start_at: knex.fn.now(),
+      });
+
+      await expect(knex('backstage_backend_tasks__tasks')).resolves.toEqual([
+        {
+          id: 'test',
+          settings_json: '{}',
+          next_run_start_at: expect.anything(),
+          current_run_ticket: null,
+          current_run_started_at: null,
+          current_run_expires_at: null,
+        },
+      ]);
+
+      await migrateDownOnce(knex);
+
+      await expect(knex('backstage_backend_tasks__tasks')).rejects.toThrow(
+        /backstage_backend_tasks__tasks/,
+      );
+
+      await knex.destroy();
+    },
+    60_000,
+  );
+});

packages/backend-tasks/src/tasks/LocalTaskWorker.ts

@@ -39,8 +39,9 @@ export class LocalTaskWorker {
     this.logger.info(
       `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`,
     );
-    let attemptNum = 1;
+
     (async () => {
+      let attemptNum = 1;
       for (;;) {
         try {
           if (settings.initialDelayDuration) {
@@ -60,6 +61,7 @@ export class LocalTaskWorker {
               options?.signal,
             );
           }
+
           this.logger.info(`Task worker finished: ${this.taskId}`);
           attemptNum = 0;
           break;

packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts

@@ -304,6 +304,46 @@ describe('PluginTaskManagerImpl', () => {
     );
   });
 
+  describe('can fetch task ids', () => {
+    it.each(databases.eachSupportedId())(
+      'can fetch both global and local task ids, %p',
+      async databaseId => {
+        const { manager } = await init(databaseId);
+        const fn = jest.fn();
+
+        await manager.scheduleTask({
+          id: 'task1',
+          timeout: Duration.fromMillis(5000),
+          frequency: Duration.fromMillis(5000),
+          fn,
+          scope: 'global',
+        });
+
+        await manager.scheduleTask({
+          id: 'task2',
+          timeout: Duration.fromMillis(5000),
+          frequency: Duration.fromMillis(5000),
+          fn,
+          scope: 'local',
+        });
+
+        await expect(manager.getScheduledTasks()).resolves.toEqual([
+          {
+            id: 'task1',
+            scope: 'global',
+            settings: expect.objectContaining({ cadence: 'PT5S' }),
+          },
+          {
+            id: 'task2',
+            scope: 'local',
+            settings: expect.objectContaining({ cadence: 'PT5S' }),
+          },
+        ]);
+      },
+      60_000,
+    );
+  });
+
   describe('parseDuration', () => {
     it('should parse durations', () => {
       expect(parseDuration({ milliseconds: 5000 })).toEqual('PT5S');

packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts

@@ -21,9 +21,11 @@ import { LocalTaskWorker } from './LocalTaskWorker';
 import { TaskWorker } from './TaskWorker';
 import {
   PluginTaskScheduler,
+  TaskDescriptor,
   TaskInvocationDefinition,
   TaskRunner,
   TaskScheduleDefinition,
+  TaskSettingsV2,
 } from './types';
 import { validateId } from './util';
 
@@ -32,6 +34,7 @@ import { validateId } from './util';
  */
 export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
   private readonly localTasksById = new Map<string, LocalTaskWorker>();
+  private readonly allScheduledTasks: TaskDescriptor[] = [];
 
   constructor(
     private readonly databaseFactory: () => Promise<Knex>,
@@ -55,6 +58,14 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
     validateId(task.id);
     const scope = task.scope ?? 'global';
 
+    const settings: TaskSettingsV2 = {
+      version: 2,
+      cadence: parseDuration(task.frequency),
+      initialDelayDuration:
+        task.initialDelay && parseDuration(task.initialDelay),
+      timeoutAfterDuration: parseDuration(task.timeout),
+    };
+
     if (scope === 'global') {
       const knex = await this.databaseFactory();
       const worker = new TaskWorker(
@@ -63,37 +74,22 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
         knex,
         this.logger.child({ task: task.id }),
       );
-
-      await worker.start(
-        {
-          version: 2,
-          cadence: parseDuration(task.frequency),
-          initialDelayDuration:
-            task.initialDelay && parseDuration(task.initialDelay),
-          timeoutAfterDuration: parseDuration(task.timeout),
-        },
-        {
-          signal: task.signal,
-        },
-      );
+      await worker.start(settings, { signal: task.signal });
     } else {
-      const worker = new LocalTaskWorker(task.id, task.fn, this.logger);
-
-      worker.start(
-        {
-          version: 2,
-          cadence: parseDuration(task.frequency),
-          initialDelayDuration:
-            task.initialDelay && parseDuration(task.initialDelay),
-          timeoutAfterDuration: parseDuration(task.timeout),
-        },
-        {
-          signal: task.signal,
-        },
+      const worker = new LocalTaskWorker(
+        task.id,
+        task.fn,
+        this.logger.child({ task: task.id }),
       );
-
+      worker.start(settings, { signal: task.signal });
       this.localTasksById.set(task.id, worker);
     }
+
+    this.allScheduledTasks.push({
+      id: task.id,
+      scope: scope,
+      settings: settings,
+    });
   }
 
   createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner {
@@ -103,6 +99,10 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
       },
     };
   }
+
+  async getScheduledTasks(): Promise<TaskDescriptor[]> {
+    return this.allScheduledTasks;
+  }
 }
 
 export function parseDuration(

packages/backend-tasks/src/tasks/index.ts

@@ -19,6 +19,7 @@ export { TaskScheduler } from './TaskScheduler';
 export type {
   PluginTaskScheduler,
   TaskFunction,
+  TaskDescriptor,
   TaskInvocationDefinition,
   TaskRunner,
   TaskScheduleDefinition,

packages/backend-tasks/src/tasks/types.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { HumanDuration } from '@backstage/types';
+import { HumanDuration, JsonObject } from '@backstage/types';
 import { CronTime } from 'cron';
 import { Duration } from 'luxon';
 import { z } from 'zod';
@@ -31,6 +31,28 @@ 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.
  *
@@ -307,6 +329,19 @@ export interface PluginTaskScheduler {
    * @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[]>;
 }
 
 function isValidOptionalDurationString(d: string | undefined): boolean {