[RAM] adds bulkUpdatesSchedules method to Task Manager API

x-pack/plugins/alerting/server/routes/bulk_edit_rules.ts

@@ -8,7 +8,7 @@
 import { schema } from '@kbn/config-schema';
 import { IRouter } from '@kbn/core/server';
 
-import { ILicenseState, RuleTypeDisabledError } from '../lib';
+import { ILicenseState, RuleTypeDisabledError, validateDurationSchema } from '../lib';
 import { verifyAccessAndContext, rewriteRule, handleDisabledApiKeysError } from './lib';
 import { AlertingRequestHandlerContext, INTERNAL_BASE_ALERTING_API_PATH } from '../types';
 
@@ -34,6 +34,27 @@ const operationsSchema = schema.arrayOf(
       field: schema.literal('actions'),
       value: schema.arrayOf(ruleActionSchema),
     }),
+    schema.object({
+      operation: schema.literal('set'),
+      field: schema.literal('schedule'),
+      value: schema.object({ interval: schema.string({ validate: validateDurationSchema }) }),
+    }),
+    schema.object({
+      operation: schema.literal('set'),
+      field: schema.literal('throttle'),
+      value: schema.nullable(schema.string()),
+    }),
+    schema.object({
+      operation: schema.literal('set'),
+      field: schema.literal('notifyWhen'),
+      value: schema.nullable(
+        schema.oneOf([
+          schema.literal('onActionGroupChange'),
+          schema.literal('onActiveAlert'),
+          schema.literal('onThrottleInterval'),
+        ])
+      ),
+    }),
   ]),
   { minSize: 1 }
 );

x-pack/plugins/alerting/server/rules_client/rules_client.ts

@@ -211,7 +211,10 @@ export interface FindOptions extends IndexType {
   filter?: string;
 }
 
-export type BulkEditFields = keyof Pick<Rule, 'actions' | 'tags'>;
+export type BulkEditFields = keyof Pick<
+  Rule,
+  'actions' | 'tags' | 'schedule' | 'throttle' | 'notifyWhen'
+>;
 
 export type BulkEditOperation =
   | {
@@ -223,25 +226,23 @@ export type BulkEditOperation =
       operation: 'add' | 'set';
       field: Extract<BulkEditFields, 'actions'>;
       value: NormalizedAlertAction[];
+    }
+  | {
+      operation: 'set';
+      field: Extract<BulkEditFields, 'schedule'>;
+      value: Rule['schedule'];
+    }
+  | {
+      operation: 'set';
+      field: Extract<BulkEditFields, 'throttle'>;
+      value: Rule['throttle'];
+    }
+  | {
+      operation: 'set';
+      field: Extract<BulkEditFields, 'notifyWhen'>;
+      value: Rule['notifyWhen'];
     };
 
-// schedule, throttle, notifyWhen is commented out before https://github.com/elastic/kibana/issues/124850 will be implemented
-// | {
-//     operation: 'set';
-//     field: Extract<BulkEditFields, 'schedule'>;
-//     value: Rule['schedule'];
-//   }
-// | {
-//     operation: 'set';
-//     field: Extract<BulkEditFields, 'throttle'>;
-//     value: Rule['throttle'];
-//   }
-// | {
-//     operation: 'set';
-//     field: Extract<BulkEditFields, 'notifyWhen'>;
-//     value: Rule['notifyWhen'];
-//   };
-
 type RuleParamsModifier<Params extends RuleTypeParams> = (params: Params) => Promise<Params>;
 
 export interface BulkEditOptionsFilter<Params extends RuleTypeParams> {
@@ -1494,6 +1495,36 @@ export class RulesClient {
       );
     });
 
+    // update schedules only if schedule operation is present
+    const scheduleOperation = options.operations.find(
+      (
+        operation
+      ): operation is Extract<BulkEditOperation, { field: Extract<BulkEditFields, 'schedule'> }> =>
+        operation.field === 'schedule'
+    );
+
+    if (scheduleOperation?.value) {
+      const taskIds = updatedRules.reduce<string[]>((acc, rule) => {
+        if (rule.scheduledTaskId) {
+          acc.push(rule.scheduledTaskId);
+        }
+        return acc;
+      }, []);
+
+      try {
+        await this.taskManager.bulkUpdateSchedules(taskIds, scheduleOperation.value);
+        this.logger.debug(
+          `Successfully updated schedules for underlying tasks: ${taskIds.join(', ')}`
+        );
+      } catch (error) {
+        this.logger.error(
+          `Failure to update schedules for underlying tasks: ${taskIds.join(
+            ', '
+          )}. TaskManager bulkUpdateSchedules failed with Error: ${error.message}`
+        );
+      }
+    }
+
     return { rules: updatedRules, errors, total };
   }
 

x-pack/plugins/alerting/server/rules_client/tests/bulk_edit.test.ts

@@ -899,4 +899,81 @@ describe('bulkEdit()', () => {
       );
     });
   });
+
+  describe('task manager', () => {
+    test('should call task manager method bulkUpdateSchedules if operation set new schedules', async () => {
+      unsecuredSavedObjectsClient.bulkUpdate.mockResolvedValue({
+        saved_objects: [
+          {
+            id: '1',
+            type: 'alert',
+            attributes: {
+              enabled: true,
+              tags: ['foo'],
+              alertTypeId: 'myType',
+              schedule: { interval: '1m' },
+              consumer: 'myApp',
+              scheduledTaskId: 'task-123',
+              params: { index: ['test-index-*'] },
+              throttle: null,
+              notifyWhen: null,
+              actions: [],
+            },
+            references: [],
+            version: '123',
+          },
+        ],
+      });
+
+      await rulesClient.bulkEdit({
+        operations: [
+          {
+            field: 'schedule',
+            operation: 'set',
+            value: { interval: '10m' },
+          },
+        ],
+      });
+
+      expect(taskManager.bulkUpdateSchedules).toHaveBeenCalledWith(['task-123'], {
+        interval: '10m',
+      });
+    });
+
+    test('should not call task manager method bulkUpdateSchedules if operation is not set schedule', async () => {
+      unsecuredSavedObjectsClient.bulkUpdate.mockResolvedValue({
+        saved_objects: [
+          {
+            id: '1',
+            type: 'alert',
+            attributes: {
+              enabled: true,
+              tags: ['foo'],
+              alertTypeId: 'myType',
+              schedule: { interval: '1m' },
+              consumer: 'myApp',
+              params: { index: ['test-index-*'] },
+              throttle: null,
+              notifyWhen: null,
+              actions: [],
+            },
+            references: [],
+            version: '123',
+          },
+        ],
+      });
+
+      await rulesClient.bulkEdit({
+        operations: [
+          {
+            field: 'tags',
+            operation: 'set',
+            value: ['test-tag'],
+          },
+        ],
+      });
+
+      expect(taskManager.bulkUpdateSchedules).not.toHaveBeenCalled();
+    });
+  });
 });

x-pack/plugins/task_manager/README.md

@@ -328,6 +328,9 @@ The _Start_ Plugin api allow you to use Task Manager to facilitate your Plugin's
   runNow: (taskId: string) =>  {
     // ...
   },
+  bulkUpdateSchedules: (taskIds: string[], schedule: IntervalSchedule) =>  {
+    // ...
+  },
   ensureScheduled: (taskInstance: TaskInstanceWithId, options?: any) => {
     // ...
   },
@@ -415,6 +418,38 @@ export class Plugin {
 }
 ```
 
+#### bulkUpdateSchedules
+Using `bulkUpdatesSchedules` you can instruct TaskManger to update interval of tasks that are in `idle` status
+(for the tasks which have `running` status,  `schedule` and `runAt` will be recalculated after task run finishes).
+When interval updated, new `runAt` will be computed and task will be updated with that value, using formula
+```
+newRunAt = oldRunAt - oldInterval + newInterval
+```
+
+Example:
+```js
+export class Plugin {
+  constructor() {
+  }
+
+  public setup(core: CoreSetup, plugins: { taskManager }) {
+  }
+
+  public start(core: CoreStart, plugins: { taskManager }) {
+    try {
+      const bulkUpdateResults = await taskManager.bulkUpdateSchedule(
+        ['97c2c4e7-d850-11ec-bf95-895ffd19f959', 'a5ee24d1-dce2-11ec-ab8d-cf74da82133d'],
+        { interval: '10m' },
+      );
+      // If no error is thrown, the bulkUpdateSchedule has completed successfully.
+      // But some updates of some tasks can be failed, due to OCC 409 conflict for example
+    } catch(err: Error) {
+      // if error is caught, means the whole method requested has failed and tasks weren't updated
+    }    
+  }
+}
+```
+
 #### more options
 
 More custom access to the tasks can be done directly via Elasticsearch, though that won't be officially supported, as we can change the document structure at any time.

x-pack/plugins/task_manager/server/index.ts

@@ -30,7 +30,7 @@ export {
   throwUnrecoverableError,
   isEphemeralTaskRejectedDueToCapacityError,
 } from './task_running';
-export type { RunNowResult } from './task_scheduling';
+export type { RunNowResult, BulkUpdateSchedulesResult } from './task_scheduling';
 export { getOldestIdleActionTask } from './queries/oldest_idle_action_task';
 export {
   IdleTaskWithExpiredRunAt,

x-pack/plugins/task_manager/server/mocks.ts

@@ -27,6 +27,7 @@ const createStartMock = () => {
     ensureScheduled: jest.fn(),
     removeIfExists: jest.fn(),
     supportsEphemeralTasks: jest.fn(),
+    bulkUpdateSchedules: jest.fn(),
   };
   return mock;
 };

x-pack/plugins/task_manager/server/plugin.ts

@@ -48,7 +48,7 @@ export interface TaskManagerSetupContract {
 
 export type TaskManagerStartContract = Pick<
   TaskScheduling,
-  'schedule' | 'runNow' | 'ephemeralRunNow' | 'ensureScheduled'
+  'schedule' | 'runNow' | 'ephemeralRunNow' | 'ensureScheduled' | 'bulkUpdateSchedules'
 > &
   Pick<TaskStore, 'fetch' | 'get' | 'remove'> & {
     removeIfExists: TaskStore['remove'];
@@ -238,6 +238,7 @@ export class TaskManagerPlugin
       schedule: (...args) => taskScheduling.schedule(...args),
       ensureScheduled: (...args) => taskScheduling.ensureScheduled(...args),
       runNow: (...args) => taskScheduling.runNow(...args),
+      bulkUpdateSchedules: (...args) => taskScheduling.bulkUpdateSchedules(...args),
       ephemeralRunNow: (task: EphemeralTask) => taskScheduling.ephemeralRunNow(task),
       supportsEphemeralTasks: () => this.config.ephemeral_tasks.enabled,
     };