[Task Manager] Correctly handle running tasks when calling RunNow and reduce flakiness in related tests (#73244)

This PR addresses two issues which caused several tests to be flaky in TM.

When `runNow` was introduced to TM we added a pinned query which returned specific tasks by ID.
This query does not have the filter applied to it which causes task to return when they're already marked as `running` but we didn't address these correctly which caused flakyness in the tests.
This didn't cause a broken beahviour, but it did cause beahviour that was hard to reason about - we now address them correctly.

It seems that sometimes, especially if the ES queue is overworked, it can take some time for the update to the underlying task to be visible (we don't user `refresh:true` on purpose), so adding a wait for the index to refresh to make sure the task is updated in time for the next stage of the test.

Author: gmmorris

x-pack/plugins/alerts/server/alerts_client.ts

@@ -387,11 +387,18 @@ export class AlertsClient {
           updateResult.scheduledTaskId &&
           !isEqual(alertSavedObject.attributes.schedule, updateResult.schedule)
         ) {
-          this.taskManager.runNow(updateResult.scheduledTaskId).catch((err: Error) => {
-            this.logger.error(
-              `Alert update failed to run its underlying task. TaskManager runNow failed with Error: ${err.message}`
-            );
-          });
+          this.taskManager
+            .runNow(updateResult.scheduledTaskId)
+            .then(() => {
+              this.logger.debug(
+                `Alert update has rescheduled the underlying task: ${updateResult.scheduledTaskId}`
+              );
+            })
+            .catch((err: Error) => {
+              this.logger.error(
+                `Alert update failed to run its underlying task. TaskManager runNow failed with Error: ${err.message}`
+              );
+            });
         }
       })(),
     ]);

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

@@ -4,6 +4,8 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+import { Option } from 'fp-ts/lib/Option';
+
 import { ConcreteTaskInstance } from './task';
 
 import { Result, Err } from './lib/result_type';
@@ -22,7 +24,7 @@ export interface TaskEvent<T, E> {
 }
 export type TaskMarkRunning = TaskEvent<ConcreteTaskInstance, Error>;
 export type TaskRun = TaskEvent<ConcreteTaskInstance, Error>;
-export type TaskClaim = TaskEvent<ConcreteTaskInstance, Error>;
+export type TaskClaim = TaskEvent<ConcreteTaskInstance, Option<ConcreteTaskInstance>>;
 export type TaskRunRequest = TaskEvent<ConcreteTaskInstance, Error>;
 
 export function asTaskMarkRunningEvent(
@@ -46,7 +48,7 @@ export function asTaskRunEvent(id: string, event: Result<ConcreteTaskInstance, E
 
 export function asTaskClaimEvent(
   id: string,
-  event: Result<ConcreteTaskInstance, Error>
+  event: Result<ConcreteTaskInstance, Option<ConcreteTaskInstance>>
 ): TaskClaim {
   return {
     id,

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

@@ -7,6 +7,7 @@
 import _ from 'lodash';
 import sinon from 'sinon';
 import { Subject } from 'rxjs';
+import { none } from 'fp-ts/lib/Option';
 
 import {
   asTaskMarkRunningEvent,
@@ -297,7 +298,9 @@ describe('TaskManager', () => {
         events$.next(asTaskMarkRunningEvent(id, asOk(task)));
         events$.next(asTaskRunEvent(id, asErr(new Error('some thing gone wrong'))));
 
-        return expect(result).rejects.toEqual(new Error('some thing gone wrong'));
+        return expect(result).rejects.toMatchInlineSnapshot(
+          `[Error: Failed to run task "01ddff11-e88a-4d13-bc4e-256164e755e2": Error: some thing gone wrong]`
+        );
       });
 
       test('rejects when the task mark as running fails', () => {
@@ -311,7 +314,9 @@ describe('TaskManager', () => {
         events$.next(asTaskClaimEvent(id, asOk(task)));
         events$.next(asTaskMarkRunningEvent(id, asErr(new Error('some thing gone wrong'))));
 
-        return expect(result).rejects.toEqual(new Error('some thing gone wrong'));
+        return expect(result).rejects.toMatchInlineSnapshot(
+          `[Error: Failed to run task "01ddff11-e88a-4d13-bc4e-256164e755e2": Error: some thing gone wrong]`
+        );
       });
 
       test('when a task claim fails we ensure the task exists', async () => {
@@ -321,7 +326,7 @@ describe('TaskManager', () => {
 
         const result = awaitTaskRunResult(id, events$, getLifecycle);
 
-        events$.next(asTaskClaimEvent(id, asErr(new Error('failed to claim'))));
+        events$.next(asTaskClaimEvent(id, asErr(none)));
 
         await expect(result).rejects.toEqual(
           new Error(`Failed to run task "${id}" as it does not exist`)
@@ -337,7 +342,7 @@ describe('TaskManager', () => {
 
         const result = awaitTaskRunResult(id, events$, getLifecycle);
 
-        events$.next(asTaskClaimEvent(id, asErr(new Error('failed to claim'))));
+        events$.next(asTaskClaimEvent(id, asErr(none)));
 
         await expect(result).rejects.toEqual(
           new Error(`Failed to run task "${id}" as it is currently running`)
@@ -353,7 +358,7 @@ describe('TaskManager', () => {
 
         const result = awaitTaskRunResult(id, events$, getLifecycle);
 
-        events$.next(asTaskClaimEvent(id, asErr(new Error('failed to claim'))));
+        events$.next(asTaskClaimEvent(id, asErr(none)));
 
         await expect(result).rejects.toEqual(
           new Error(`Failed to run task "${id}" as it is currently running`)
@@ -386,9 +391,11 @@ describe('TaskManager', () => {
 
         const result = awaitTaskRunResult(id, events$, getLifecycle);
 
-        events$.next(asTaskClaimEvent(id, asErr(new Error('failed to claim'))));
+        events$.next(asTaskClaimEvent(id, asErr(none)));
 
-        await expect(result).rejects.toEqual(new Error('failed to claim'));
+        await expect(result).rejects.toMatchInlineSnapshot(
+          `[Error: Failed to run task "01ddff11-e88a-4d13-bc4e-256164e755e2" for unknown reason (Current Task Lifecycle is "idle")]`
+        );
 
         expect(getLifecycle).toHaveBeenCalledWith(id);
       });
@@ -400,9 +407,11 @@ describe('TaskManager', () => {
 
         const result = awaitTaskRunResult(id, events$, getLifecycle);
 
-        events$.next(asTaskClaimEvent(id, asErr(new Error('failed to claim'))));
+        events$.next(asTaskClaimEvent(id, asErr(none)));
 
-        await expect(result).rejects.toEqual(new Error('failed to claim'));
+        await expect(result).rejects.toMatchInlineSnapshot(
+          `[Error: Failed to run task "01ddff11-e88a-4d13-bc4e-256164e755e2" for unknown reason (Current Task Lifecycle is "failed")]`
+        );
 
         expect(getLifecycle).toHaveBeenCalledWith(id);
       });
@@ -424,7 +433,9 @@ describe('TaskManager', () => {
 
         events$.next(asTaskRunEvent(id, asErr(new Error('some thing gone wrong'))));
 
-        return expect(result).rejects.toEqual(new Error('some thing gone wrong'));
+        return expect(result).rejects.toMatchInlineSnapshot(
+          `[Error: Failed to run task "01ddff11-e88a-4d13-bc4e-256164e755e2": Error: some thing gone wrong]`
+        );
       });
     });
   });

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

@@ -9,13 +9,14 @@ import { filter } from 'rxjs/operators';
 import { performance } from 'perf_hooks';
 
 import { pipe } from 'fp-ts/lib/pipeable';
-import { Option, some, map as mapOptional } from 'fp-ts/lib/Option';
+import { Option, some, map as mapOptional, getOrElse } from 'fp-ts/lib/Option';
+
 import {
   SavedObjectsSerializer,
   ILegacyScopedClusterClient,
   ISavedObjectsRepository,
 } from '../../../../src/core/server';
-import { Result, asErr, either, map, mapErr, promiseResult } from './lib/result_type';
+import { Result, asOk, asErr, either, map, mapErr, promiseResult } from './lib/result_type';
 import { TaskManagerConfig } from './config';
 
 import { Logger } from './types';
@@ -405,7 +406,9 @@ export async function claimAvailableTasks(
 
       if (docs.length !== claimedTasks) {
         logger.warn(
-          `[Task Ownership error]: (${claimedTasks}) tasks were claimed by Kibana, but (${docs.length}) tasks were fetched`
+          `[Task Ownership error]: ${claimedTasks} tasks were claimed by Kibana, but ${
+            docs.length
+          } task(s) were fetched (${docs.map((doc) => doc.id).join(', ')})`
         );
       }
       return docs;
@@ -437,48 +440,65 @@ export async function awaitTaskRunResult(
       // listen for all events related to the current task
       .pipe(filter(({ id }: TaskLifecycleEvent) => id === taskId))
       .subscribe((taskEvent: TaskLifecycleEvent) => {
-        either(
-          taskEvent.event,
-          (taskInstance: ConcreteTaskInstance) => {
-            // resolve if the task has run sucessfully
-            if (isTaskRunEvent(taskEvent)) {
-              subscription.unsubscribe();
-              resolve({ id: taskInstance.id });
-            }
-          },
-          async (error: Error) => {
+        if (isTaskClaimEvent(taskEvent)) {
+          mapErr(async (error: Option<ConcreteTaskInstance>) => {
             // reject if any error event takes place for the requested task
             subscription.unsubscribe();
-            if (isTaskRunRequestEvent(taskEvent)) {
-              return reject(
-                new Error(
-                  `Failed to run task "${taskId}" as Task Manager is at capacity, please try again later`
-                )
-              );
-            } else if (isTaskClaimEvent(taskEvent)) {
-              reject(
-                map(
-                  // if the error happened in the Claim phase - we try to provide better insight
-                  // into why we failed to claim by getting the task's current lifecycle status
-                  await promiseResult<TaskLifecycle, Error>(getLifecycle(taskId)),
-                  (taskLifecycleStatus: TaskLifecycle) => {
-                    if (taskLifecycleStatus === TaskLifecycleResult.NotFound) {
-                      return new Error(`Failed to run task "${taskId}" as it does not exist`);
-                    } else if (
-                      taskLifecycleStatus === TaskStatus.Running ||
-                      taskLifecycleStatus === TaskStatus.Claiming
-                    ) {
-                      return new Error(`Failed to run task "${taskId}" as it is currently running`);
-                    }
-                    return error;
-                  },
-                  () => error
-                )
-              );
+            return reject(
+              map(
+                await pipe(
+                  error,
+                  mapOptional(async (taskReturnedBySweep) => asOk(taskReturnedBySweep.status)),
+                  getOrElse(() =>
+                    // if the error happened in the Claim phase - we try to provide better insight
+                    // into why we failed to claim by getting the task's current lifecycle status
+                    promiseResult<TaskLifecycle, Error>(getLifecycle(taskId))
+                  )
+                ),
+                (taskLifecycleStatus: TaskLifecycle) => {
+                  if (taskLifecycleStatus === TaskLifecycleResult.NotFound) {
+                    return new Error(`Failed to run task "${taskId}" as it does not exist`);
+                  } else if (
+                    taskLifecycleStatus === TaskStatus.Running ||
+                    taskLifecycleStatus === TaskStatus.Claiming
+                  ) {
+                    return new Error(`Failed to run task "${taskId}" as it is currently running`);
+                  }
+                  return new Error(
+                    `Failed to run task "${taskId}" for unknown reason (Current Task Lifecycle is "${taskLifecycleStatus}")`
+                  );
+                },
+                (getLifecycleError: Error) =>
+                  new Error(
+                    `Failed to run task "${taskId}" and failed to get current Status:${getLifecycleError}`
+                  )
+              )
+            );
+          }, taskEvent.event);
+        } else {
+          either<ConcreteTaskInstance, Error | Option<ConcreteTaskInstance>>(
+            taskEvent.event,
+            (taskInstance: ConcreteTaskInstance) => {
+              // resolve if the task has run sucessfully
+              if (isTaskRunEvent(taskEvent)) {
+                subscription.unsubscribe();
+                resolve({ id: taskInstance.id });
+              }
+            },
+            async (error: Error | Option<ConcreteTaskInstance>) => {
+              // reject if any error event takes place for the requested task
+              subscription.unsubscribe();
+              if (isTaskRunRequestEvent(taskEvent)) {
+                return reject(
+                  new Error(
+                    `Failed to run task "${taskId}" as Task Manager is at capacity, please try again later`
+                  )
+                );
+              }
+              return reject(new Error(`Failed to run task "${taskId}": ${error}`));
             }
-            return reject(error);
-          }
-        );
+          );
+        }
       });
   });
 }