[8.16] [Response Ops][Task Manager] Propagate `msearch` error status code so backpressure mechanism responds correctly (#197501) (#198034)

# Backport This will backport the following commits from `main` to `8.16`: - [[Response Ops][Task Manager] Propagate `msearch` error status code so backpressure mechanism responds correctly (#197501)](https://github.com/elastic/kibana/pull/197501)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  Co-authored-by: Ying Mao <ying.mao@elastic.co>
2025-04-24 01:38:56 -04:00 · 2024-10-29 04:31:24 +11:00 · 2024-10-29 04:31:24 +11:00 · 9bdaef80cd
commit 9bdaef80cd
parent ac44c8bbca
5 changed files with 60 additions and 14 deletions
--- a/x-pack/plugins/task_manager/server/lib/create_managed_configuration.test.ts
+++ b/x-pack/plugins/task_manager/server/lib/create_managed_configuration.test.ts
@ -14,6 +14,7 @@ import {
 } from './create_managed_configuration';
 import { mockLogger } from '../test_utils';
 import { CLAIM_STRATEGY_UPDATE_BY_QUERY, CLAIM_STRATEGY_MGET, TaskManagerConfig } from '../config';
+import { MsearchError } from './msearch_error';

 describe('createManagedConfiguration()', () => {
  let clock: sinon.SinonFakeTimers;
@ -235,9 +236,9 @@ describe('createManagedConfiguration()', () => {
    });

    describe('mget claim strategy', () => {
-      test('should decrease configuration at the next interval when an error is emitted', async () => {
-        const { subscription, errors$ } = setupScenario(10, CLAIM_STRATEGY_MGET);
-        errors$.next(SavedObjectsErrorHelpers.createTooManyRequestsError('a', 'b'));
+      test('should decrease configuration at the next interval when an msearch 429 error is emitted', async () => {
+        const { subscription, errors$ } = setupScenario(10);
+        errors$.next(new MsearchError(429));
        clock.tick(ADJUST_THROUGHPUT_INTERVAL - 1);
        expect(subscription).toHaveBeenCalledTimes(1);
        expect(subscription).toHaveBeenNthCalledWith(1, 10);
@ -246,9 +247,9 @@ describe('createManagedConfiguration()', () => {
        expect(subscription).toHaveBeenNthCalledWith(2, 8);
      });

-      test('should decrease configuration at the next interval when a 503 error is emitted', async () => {
-        const { subscription, errors$ } = setupScenario(10, CLAIM_STRATEGY_MGET);
-        errors$.next(SavedObjectsErrorHelpers.createGenericNotFoundEsUnavailableError('a', 'b'));
+      test('should decrease configuration at the next interval when an msearch 503 error is emitted', async () => {
+        const { subscription, errors$ } = setupScenario(10);
+        errors$.next(new MsearchError(503));
        clock.tick(ADJUST_THROUGHPUT_INTERVAL - 1);
        expect(subscription).toHaveBeenCalledTimes(1);
        expect(subscription).toHaveBeenNthCalledWith(1, 10);
@ -257,9 +258,19 @@ describe('createManagedConfiguration()', () => {
        expect(subscription).toHaveBeenNthCalledWith(2, 8);
      });

+      test('should not change configuration at the next interval when other msearch error is emitted', async () => {
+        const { subscription, errors$ } = setupScenario(10);
+        errors$.next(new MsearchError(404));
+        clock.tick(ADJUST_THROUGHPUT_INTERVAL - 1);
+        expect(subscription).toHaveBeenCalledTimes(1);
+        expect(subscription).toHaveBeenNthCalledWith(1, 10);
+        clock.tick(1);
+        expect(subscription).toHaveBeenCalledTimes(1);
+      });
+
      test('should log a warning when the configuration changes from the starting value', async () => {
        const { errors$ } = setupScenario(10, CLAIM_STRATEGY_MGET);
-        errors$.next(SavedObjectsErrorHelpers.createTooManyRequestsError('a', 'b'));
+        errors$.next(new MsearchError(429));
        clock.tick(ADJUST_THROUGHPUT_INTERVAL);
        expect(logger.warn).toHaveBeenCalledWith(
          'Capacity configuration is temporarily reduced after Elasticsearch returned 1 "too many request" and/or "execute [inline] script" error(s).'
@ -268,7 +279,7 @@ describe('createManagedConfiguration()', () => {

      test('should increase configuration back to normal incrementally after an error is emitted', async () => {
        const { subscription, errors$ } = setupScenario(10, CLAIM_STRATEGY_MGET);
-        errors$.next(SavedObjectsErrorHelpers.createTooManyRequestsError('a', 'b'));
+        errors$.next(new MsearchError(429));
        clock.tick(ADJUST_THROUGHPUT_INTERVAL * 10);
        expect(subscription).toHaveBeenNthCalledWith(1, 10);
        expect(subscription).toHaveBeenNthCalledWith(2, 8);
@ -281,7 +292,7 @@ describe('createManagedConfiguration()', () => {
      test('should keep reducing configuration when errors keep emitting until it reaches minimum', async () => {
        const { subscription, errors$ } = setupScenario(10, CLAIM_STRATEGY_MGET);
        for (let i = 0; i < 20; i++) {
-          errors$.next(SavedObjectsErrorHelpers.createTooManyRequestsError('a', 'b'));
+          errors$.next(new MsearchError(429));
          clock.tick(ADJUST_THROUGHPUT_INTERVAL);
        }
        expect(subscription).toHaveBeenNthCalledWith(1, 10);
--- a/x-pack/plugins/task_manager/server/lib/create_managed_configuration.ts
+++ b/x-pack/plugins/task_manager/server/lib/create_managed_configuration.ts
@ -12,6 +12,7 @@ import { Logger } from '@kbn/core/server';
 import { isEsCannotExecuteScriptError } from './identify_es_error';
 import { CLAIM_STRATEGY_MGET, DEFAULT_CAPACITY, MAX_CAPACITY, TaskManagerConfig } from '../config';
 import { TaskCost } from '../task';
+import { getMsearchStatusCode } from './msearch_error';

 const FLUSH_MARKER = Symbol('flush');
 export const ADJUST_THROUGHPUT_INTERVAL = 10 * 1000;
@ -164,7 +165,9 @@ function countErrors(errors$: Observable<Error>, countInterval: number): Observa
        (e) =>
          SavedObjectsErrorHelpers.isTooManyRequestsError(e) ||
          SavedObjectsErrorHelpers.isEsUnavailableError(e) ||
-          isEsCannotExecuteScriptError(e)
+          isEsCannotExecuteScriptError(e) ||
+          getMsearchStatusCode(e) === 429 ||
+          getMsearchStatusCode(e) === 503
      )
    )
  ).pipe(
--- a/x-pack/plugins/task_manager/server/lib/msearch_error.ts
+++ b/x-pack/plugins/task_manager/server/lib/msearch_error.ts
@ -0,0 +1,25 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+export class MsearchError extends Error {
+  private _statusCode?: number;
+
+  constructor(statusCode?: number) {
+    super(`Unexpected status code from taskStore::msearch: ${statusCode ?? 'unknown'}`);
+    this._statusCode = statusCode;
+  }
+
+  public get statusCode() {
+    return this._statusCode;
+  }
+}
+
+export function getMsearchStatusCode(error: Error | MsearchError): number | undefined {
+  if (Boolean(error && error instanceof MsearchError)) {
+    return (error as MsearchError).statusCode;
+  }
+}
--- a/x-pack/plugins/task_manager/server/task_store.test.ts
+++ b/x-pack/plugins/task_manager/server/task_store.test.ts
@ -30,6 +30,7 @@ import { mockLogger } from './test_utils';
 import { AdHocTaskCounter } from './lib/adhoc_task_counter';
 import { asErr, asOk } from './lib/result_type';
 import { UpdateByQueryResponse } from '@elastic/elasticsearch/lib/api/types';
+import { MsearchError } from './lib/msearch_error';

 const mockGetValidatedTaskInstanceFromReading = jest.fn();
 const mockGetValidatedTaskInstanceForUpdating = jest.fn();
@ -490,9 +491,14 @@ describe('TaskStore', () => {
          },
        ],
      } as estypes.MsearchResponse);
-      await expect(store.msearch([{}])).rejects.toThrowErrorMatchingInlineSnapshot(
-        `"Unexpected status code from taskStore::msearch: 429"`
-      );
+
+      try {
+        await store.msearch([{}]);
+        throw new Error('should have thrown');
+      } catch (err) {
+        expect(err instanceof MsearchError).toBe(true);
+        expect(err.statusCode).toEqual(429);
+      }
      expect(await firstErrorPromise).toMatchInlineSnapshot(
        `[Error: Unexpected status code from taskStore::msearch: 429]`
      );
--- a/x-pack/plugins/task_manager/server/task_store.ts
+++ b/x-pack/plugins/task_manager/server/task_store.ts
@ -47,6 +47,7 @@ import { TaskValidator } from './task_validator';
 import { claimSort } from './queries/mark_available_tasks_as_claimed';
 import { MAX_PARTITIONS } from './lib/task_partitioner';
 import { ErrorOutput } from './lib/bulk_operation_buffer';
+import { MsearchError } from './lib/msearch_error';

 export interface StoreOpts {
  esClient: ElasticsearchClient;
@ -575,7 +576,7 @@ export class TaskStore {

    for (const response of responses) {
      if (response.status !== 200) {
-        const err = new Error(`Unexpected status code from taskStore::msearch: ${response.status}`);
+        const err = new MsearchError(response.status);
        this.errors$.next(err);
        throw err;
      }