From 3fd8212feaae9656f19e3e4071fcb77c66717b5f Mon Sep 17 00:00:00 2001 From: Ying Mao Date: Mon, 28 Oct 2024 11:43:51 -0400 Subject: [PATCH] [Response Ops][Task Manager] Propagate `msearch` error status code so backpressure mechanism responds correctly (#197501) Resolves https://github.com/elastic/response-ops-team/issues/240 ## Summary Creating an `MsearchError` class that preserves the status code from any msearch errors. These errors are already piped to the managed configuration observable that watches for and responds to ES errors from the update by query claim strategy so I updated that filter to filter for msearch 429 and 503 errors as well. ## To Verify 1. Make sure you're using the mget claim strategy (`xpack.task_manager.claim_strategy: 'mget'`) and start ES and Kibana. 2. Inject a 429 error into an msearch response. ``` --- a/x-pack/plugins/task_manager/server/task_store.ts +++ b/x-pack/plugins/task_manager/server/task_store.ts @@ -571,6 +571,8 @@ export class TaskStore { }); const { responses } = result; + responses[0].status = 429; + const versionMap = this.createVersionMap([]); ``` 3. See task manager log the msearch errors and eventually reduce polling capacity ``` [2024-10-23T15:35:59.255-04:00][ERROR][plugins.taskManager] Failed to poll for work: Unexpected status code from taskStore::msearch: 429 [2024-10-23T15:35:59.756-04:00][ERROR][plugins.taskManager] Failed to poll for work: Unexpected status code from taskStore::msearch: 429 [2024-10-23T15:36:00.257-04:00][ERROR][plugins.taskManager] Failed to poll for work: Unexpected status code from taskStore::msearch: 429 [2024-10-23T15:36:00.757-04:00][ERROR][plugins.taskManager] Failed to poll for work: Unexpected status code from taskStore::msearch: 429 ... [2024-10-23T15:36:06.267-04:00][WARN ][plugins.taskManager] Poll interval configuration is temporarily increased after Elasticsearch returned 19 "too many request" and/or "execute [inline] script" error(s). [2024-10-23T15:36:06.268-04:00][WARN ][plugins.taskManager] Capacity configuration is temporarily reduced after Elasticsearch returned 19 "too many request" and/or "execute [inline] script" error(s). ``` --------- Co-authored-by: Elastic Machine (cherry picked from commit 043e18b6a097f4405ff37a99396c0c8c92db6b44) --- .../lib/create_managed_configuration.test.ts | 29 +++++++++++++------ .../lib/create_managed_configuration.ts | 5 +++- .../task_manager/server/lib/msearch_error.ts | 25 ++++++++++++++++ .../task_manager/server/task_store.test.ts | 12 ++++++-- .../plugins/task_manager/server/task_store.ts | 3 +- 5 files changed, 60 insertions(+), 14 deletions(-) create mode 100644 x-pack/plugins/task_manager/server/lib/msearch_error.ts 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 index d863fb9b300fd..5e0a5ed4f2e67 100644 --- 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 index a0adee6d05818..8a76029efb8eb 100644 --- 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, 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 new file mode 100644 index 0000000000000..8cd61a53dd2a7 --- /dev/null +++ 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 index 18dc3fa3c44ce..f1374f6d27b76 100644 --- 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 index 83c69ebdb2d88..2b3440e87c0f8 100644 --- 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; }