[RAM] adds bulkUpdatesSchedules method to Task Manager API

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

@@ -12,6 +12,10 @@ import { ILicenseState, RuleTypeDisabledError } from '../lib';
 import { verifyAccessAndContext, rewriteRule, handleDisabledApiKeysError } from './lib';
 import { AlertingRequestHandlerContext, INTERNAL_BASE_ALERTING_API_PATH } from '../types';
 
+const scheduleSchema = schema.object({
+  interval: schema.string(),
+});
+
 const ruleActionSchema = schema.object({
   group: schema.string(),
   id: schema.string(),
@@ -34,6 +38,11 @@ const operationsSchema = schema.arrayOf(
       field: schema.literal('actions'),
       value: schema.arrayOf(ruleActionSchema),
     }),
+    schema.object({
+      operation: schema.literal('set'),
+      field: schema.literal('schedule'),
+      value: scheduleSchema,
+    }),
   ]),
   { minSize: 1 }
 );

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

@@ -211,7 +211,7 @@ export interface FindOptions extends IndexType {
   filter?: string;
 }
 
-export type BulkEditFields = keyof Pick<Rule, 'actions' | 'tags'>;
+export type BulkEditFields = keyof Pick<Rule, 'actions' | 'tags' | 'schedule'>;
 
 export type BulkEditOperation =
   | {
@@ -223,6 +223,11 @@ export type BulkEditOperation =
       operation: 'add' | 'set';
       field: Extract<BulkEditFields, 'actions'>;
       value: NormalizedAlertAction[];
+    }
+  | {
+      operation: 'set';
+      field: Extract<BulkEditFields, 'schedule'>;
+      value: Rule['schedule'];
     };
 
 // schedule, throttle, notifyWhen is commented out before https://github.com/elastic/kibana/issues/124850 will be implemented
@@ -1494,6 +1499,32 @@ export class RulesClient {
       );
     });
 
+    // update schedules only if schedule operation is present
+    const scheduleOperation = options.operations.find(
+      (op): op is Extract<BulkEditOperation, { field: Extract<BulkEditFields, 'schedule'> }> =>
+        op.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);
+      } catch (error) {
+        this.auditLogger?.log(
+          ruleAuditEvent({
+            action: RuleAuditAction.BULK_EDIT,
+            error,
+          })
+        );
+      }
+    }
+
     return { rules: updatedRules, errors, total };
   }
 

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,
     };

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

@@ -6,15 +6,16 @@
  */
 
 import { filter, take } from 'rxjs/operators';
-
+import pMap from 'p-map';
 import { pipe } from 'fp-ts/lib/pipeable';
 import { Option, map as mapOptional, getOrElse, isSome } from 'fp-ts/lib/Option';
 
 import uuid from 'uuid';
-import { pick } from 'lodash';
+import { pick, chunk } from 'lodash';
 import { merge, Subject } from 'rxjs';
 import agent from 'elastic-apm-node';
 import { Logger } from '@kbn/core/server';
+import { mustBeAllOf } from './queries/query_clauses';
 import { asOk, either, map, mapErr, promiseResult, isErr } from './lib/result_type';
 import {
   isTaskRunEvent,
@@ -28,6 +29,7 @@ import {
   TaskClaimErrorType,
 } from './task_events';
 import { Middleware } from './lib/middleware';
+import { parseIntervalAsMillisecond } from './lib/intervals';
 import {
   ConcreteTaskInstance,
   TaskInstanceWithId,
@@ -36,8 +38,9 @@ import {
   TaskLifecycleResult,
   TaskStatus,
   EphemeralTask,
+  IntervalSchedule,
 } from './task';
-import { TaskStore } from './task_store';
+import { TaskStore, BulkUpdateResult } from './task_store';
 import { ensureDeprecatedFieldsAreCorrected } from './lib/correct_deprecated_fields';
 import { TaskLifecycleEvent, TaskPollingLifecycle } from './polling_lifecycle';
 import { TaskTypeDictionary } from './task_type_dictionary';
@@ -111,6 +114,51 @@ export class TaskScheduling {
     });
   }
 
+  /**
+   * Bulk updates schedules for tasks by ids.
+   *
+   * @param taskIs - list of task ids
+   * @param schedule - new schedule
+   * @returns {Promise<ConcreteTaskInstance[]>}
+   */
+  public async bulkUpdateSchedules(
+    taskIds: string[],
+    schedule: IntervalSchedule
+  ): Promise<BulkUpdateResult[]> {
+    const tasks = await pMap(chunk(taskIds, 100), async (taskIdsChunk) =>
+      this.store.fetch({
+        query: mustBeAllOf(
+          {
+            terms: {
+              _id: taskIdsChunk.map((taskId) => `task:${taskId}`),
+            },
+          },
+          {
+            term: {
+              'task.status': 'idle',
+            },
+          }
+        ),
+        size: 100,
+      })
+    );
+
+    const updatedTasks = tasks
+      .flatMap(({ docs }) => docs)
+      .map((task) => {
+        const oldInterval = parseIntervalAsMillisecond(task.schedule?.interval ?? '0s');
+
+        const newRunAtInMs = Math.max(
+          Date.now(),
+          task.runAt.getTime() - oldInterval + parseIntervalAsMillisecond(schedule.interval)
+        );
+
+        return { ...task, schedule, runAt: new Date(newRunAtInMs) };
+      });
+
+    return this.store.bulkUpdate(updatedTasks);
+  }
+
   /**
    * Run  task.
    *