[8.x] Add more logs to Task Manager poller (#194741) (#194865)

# Backport This will backport the following commits from `main` to `8.x`: - [Add more logs to Task Manager poller (#194741)](https://github.com/elastic/kibana/pull/194741)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  Co-authored-by: Mike Côté <mikecote@users.noreply.github.com>
2025-04-24 17:59:23 -04:00 · 2024-10-04 04:55:31 +10:00 · 2024-10-04 04:55:31 +10:00 · e422533e3d
commit e422533e3d
parent 6a782989f8
4 changed files with 36 additions and 13 deletions
--- a/x-pack/plugins/task_manager/server/polling/task_poller.test.ts
+++ b/x-pack/plugins/task_manager/server/polling/task_poller.test.ts
@ -239,13 +239,14 @@ describe('TaskPoller', () => {
    const pollInterval = 100;

    const handler = jest.fn();
+    const workError = new Error('failed to work');
    const poller = createTaskPoller<string, string[]>({
      initialPollInterval: pollInterval,
      logger: loggingSystemMock.create().get(),
      pollInterval$: of(pollInterval),
      pollIntervalDelay$: of(0),
      work: async (...args) => {
-        throw new Error('failed to work');
+        throw workError;
      },
      getCapacity: () => 5,
    });
@ -256,12 +257,13 @@ describe('TaskPoller', () => {
    await new Promise((resolve) => setImmediate(resolve));

    const expectedError = new PollingError<string>(
-      'Failed to poll for work: Error: failed to work',
+      'Failed to poll for work: failed to work',
      PollingErrorType.WorkError,
      none
    );
    expect(handler).toHaveBeenCalledWith(asErr(expectedError));
    expect(handler.mock.calls[0][0].error.type).toEqual(PollingErrorType.WorkError);
+    expect(handler.mock.calls[0][0].error.stack).toContain(workError.stack);
  });

  test('continues polling after work fails', async () => {
@ -269,10 +271,11 @@ describe('TaskPoller', () => {

    const handler = jest.fn();
    let callCount = 0;
+    const workError = new Error('failed to work');
    const work = jest.fn(async () => {
      callCount++;
      if (callCount === 2) {
-        throw new Error('failed to work');
+        throw workError;
      }
      return callCount;
    });
@ -296,12 +299,13 @@ describe('TaskPoller', () => {
    await new Promise((resolve) => setImmediate(resolve));

    const expectedError = new PollingError<string>(
-      'Failed to poll for work: Error: failed to work',
+      'Failed to poll for work: failed to work',
      PollingErrorType.WorkError,
      none
    );
    expect(handler).toHaveBeenCalledWith(asErr(expectedError));
    expect(handler.mock.calls[1][0].error.type).toEqual(PollingErrorType.WorkError);
+    expect(handler.mock.calls[1][0].error.stack).toContain(workError.stack);
    expect(handler).not.toHaveBeenCalledWith(asOk(2));

    clock.tick(pollInterval);
@ -342,7 +346,7 @@ describe('TaskPoller', () => {
    await new Promise((resolve) => setImmediate(resolve));

    const expectedError = new PollingError<string>(
-      'Failed to poll for work: Error: error getting capacity',
+      'Failed to poll for work: error getting capacity',
      PollingErrorType.WorkError,
      none
    );
--- a/x-pack/plugins/task_manager/server/polling/task_poller.ts
+++ b/x-pack/plugins/task_manager/server/polling/task_poller.ts
@ -65,6 +65,8 @@ export function createTaskPoller<T, H>({
      if (hasCapacity()) {
        const result = await work();
        subject.next(asOk(result));
+      } else {
+        logger.debug('Skipping polling cycle because there is no capacity available');
      }
    } catch (e) {
      subject.next(asPollingError<T>(e, PollingErrorType.WorkError));
@ -73,11 +75,16 @@ export function createTaskPoller<T, H>({
    if (running) {
      // Set the next runCycle call
      timeoutId = setTimeout(
-        () => runCycle().catch(() => {}),
+        () =>
+          runCycle().catch((e) => {
+            subject.next(asPollingError(e, PollingErrorType.PollerError));
+          }),
        Math.max(pollInterval - (Date.now() - start) + (pollIntervalDelay % pollInterval), 0)
      );
      // Reset delay, it's designed to shuffle only once
      pollIntervalDelay = 0;
+    } else {
+      logger.info('Task poller finished running its last cycle');
    }
  }

@ -113,14 +120,18 @@ export function createTaskPoller<T, H>({
    events$: subject,
    start: () => {
      if (!running) {
+        logger.info('Starting the task poller');
        running = true;
-        runCycle().catch(() => {});
+        runCycle().catch((e) => {
+          subject.next(asPollingError(e, PollingErrorType.PollerError));
+        });
        // We need to subscribe shortly after start. Otherwise, the observables start emiting events
        // too soon for the task run statistics module to capture.
        setTimeout(() => subscribe(), 0);
      }
    },
    stop: () => {
+      logger.info('Stopping the task poller');
      if (timeoutId) {
        clearTimeout(timeoutId);
        timeoutId = null;
@ -134,21 +145,25 @@ export enum PollingErrorType {
  WorkError,
  WorkTimeout,
  RequestCapacityReached,
+  PollerError,
 }

-function asPollingError<T>(err: string | Error, type: PollingErrorType, data: Option<T> = none) {
-  return asErr(new PollingError<T>(`Failed to poll for work: ${err}`, type, data));
+function asPollingError<T>(err: Error, type: PollingErrorType, data: Option<T> = none) {
+  return asErr(new PollingError<T>(`Failed to poll for work: ${err.message}`, type, data, err));
 }

 export class PollingError<T> extends Error {
  public readonly type: PollingErrorType;
  public readonly data: Option<T>;
  public readonly source: TaskErrorSource;
-  constructor(message: string, type: PollingErrorType, data: Option<T>) {
-    super(message);
+  constructor(message: string, type: PollingErrorType, data: Option<T>, cause?: Error) {
+    super(message, { cause });
    Object.setPrototypeOf(this, new.target.prototype);
    this.type = type;
    this.data = data;
    this.source = TaskErrorSource.FRAMEWORK;
+    if (cause) {
+      this.stack = `${this.stack}\nCaused by:\n${cause.stack}`;
+    }
  }
 }
--- a/x-pack/plugins/task_manager/server/polling_lifecycle.test.ts
+++ b/x-pack/plugins/task_manager/server/polling_lifecycle.test.ts
@ -515,9 +515,10 @@ describe('TaskPollingLifecycle', () => {
        (event: TaskLifecycleEvent) => event.type === TaskEventType.TASK_POLLING_CYCLE
      );

+      expect(pollingCycleEvent!.event.tag).toEqual('err');
      expect(pollingCycleEvent!.event).toEqual({
        tag: 'err',
-        error: new Error(`Failed to poll for work: Error: booo`),
+        error: new Error(`Failed to poll for work: booo`),
      });
    });

--- a/x-pack/plugins/task_manager/server/polling_lifecycle.ts
+++ b/x-pack/plugins/task_manager/server/polling_lifecycle.ts
@ -199,6 +199,9 @@ export class TaskPollingLifecycle implements ITaskEventEmitter<TaskLifecycleEven
        // start polling for work
        poller.start();
      } else if (!areESAndSOAvailable) {
+        this.logger.info(
+          `Stopping the task poller because Elasticsearch and/or saved-objects service became unavailable`
+        );
        poller.stop();
        this.pool.cancelRunningTasks();
      }
@ -285,7 +288,7 @@ export class TaskPollingLifecycle implements ITaskEventEmitter<TaskLifecycleEven
                mapOptional((id) => this.emitEvent(asTaskRunRequestEvent(id, asErr(error))))
              );
            }
-            this.logger.error(error.message);
+            this.logger.error(error.message, { error: { stack_trace: error.stack } });

            // Emit event indicating task manager utilization % at the end of a polling cycle
            // Because there was a polling error, no tasks were claimed so this represents the number of workers busy