Skip to content

Commit

Permalink
[Response Ops][Task Manager] Propagate msearch error status code so…
Browse files Browse the repository at this point in the history
… backpressure mechanism responds correctly (elastic#197501)

Resolves elastic/response-ops-team#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 <elasticmachine@users.noreply.github.com>
(cherry picked from commit 043e18b)
  • Loading branch information
ymao1 committed Oct 28, 2024
1 parent 15aa7b4 commit 3fd8212
Show file tree
Hide file tree
Showing 5 changed files with 60 additions and 14 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand All @@ -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).'
Expand All @@ -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);
Expand All @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand Down
25 changes: 25 additions & 0 deletions x-pack/plugins/task_manager/server/lib/msearch_error.ts
Original file line number Diff line number Diff line change
@@ -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;
}
}
12 changes: 9 additions & 3 deletions x-pack/plugins/task_manager/server/task_store.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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]`
);
Expand Down
3 changes: 2 additions & 1 deletion x-pack/plugins/task_manager/server/task_store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}
Expand Down

0 comments on commit 3fd8212

Please sign in to comment.