[8.16] [Response Ops][Task Manager] Propagate `msearch` err…

…or 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)](#197501)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)  Co-authored-by: Ying Mao <[email protected]>
elastic · Oct 28, 2024 · 9bdaef8 · 9bdaef8
1 parent ac44c8b
commit 9bdaef8
Show file tree

Hide file tree

Showing 5 changed files with 60 additions and 14 deletions.
diff --git 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);

diff --git 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(

diff --git 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;
+  }
+}
diff --git 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]`
       );

diff --git 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;
       }