-
Notifications
You must be signed in to change notification settings - Fork 8.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Shameless copy of the retryTransientEsErrors from fleet
- Loading branch information
Showing
4 changed files
with
189 additions
and
13 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| /* | ||
| * 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. | ||
| */ | ||
|
|
||
| jest.mock('timers/promises'); | ||
| import { setTimeout } from 'timers/promises'; | ||
| import { loggerMock } from '@kbn/logging-mocks'; | ||
| import { errors as EsErrors } from '@elastic/elasticsearch'; | ||
|
|
||
| import { retryTransientEsErrors } from './retry'; | ||
|
|
||
| const setTimeoutMock = setTimeout as jest.Mock< | ||
| ReturnType<typeof setTimeout>, | ||
| Parameters<typeof setTimeout> | ||
| >; | ||
|
|
||
| describe('retryTransientErrors', () => { | ||
| beforeEach(() => { | ||
| setTimeoutMock.mockClear(); | ||
| }); | ||
|
|
||
| it("doesn't retry if operation is successful", async () => { | ||
| const esCallMock = jest.fn().mockResolvedValue('success'); | ||
| expect(await retryTransientEsErrors(esCallMock)).toEqual('success'); | ||
| expect(esCallMock).toHaveBeenCalledTimes(1); | ||
| }); | ||
|
|
||
| it('logs an warning message on retry', async () => { | ||
| const logger = loggerMock.create(); | ||
| const esCallMock = jest | ||
| .fn() | ||
| .mockRejectedValueOnce(new EsErrors.ConnectionError('foo')) | ||
| .mockResolvedValue('success'); | ||
|
|
||
| await retryTransientEsErrors(esCallMock, { logger }); | ||
| expect(logger.warn).toHaveBeenCalledTimes(1); | ||
| expect(logger.warn.mock.calls[0][0]).toMatch( | ||
| `Retrying Elasticsearch operation after [2s] due to error: ConnectionError: foo ConnectionError: foo` | ||
| ); | ||
| }); | ||
|
|
||
| it('retries with an exponential backoff', async () => { | ||
| let attempt = 0; | ||
| const esCallMock = jest.fn(async () => { | ||
| attempt++; | ||
| if (attempt < 5) { | ||
| throw new EsErrors.ConnectionError('foo'); | ||
| } else { | ||
| return 'success'; | ||
| } | ||
| }); | ||
|
|
||
| expect(await retryTransientEsErrors(esCallMock)).toEqual('success'); | ||
| expect(setTimeoutMock.mock.calls).toEqual([[2000], [4000], [8000], [16000]]); | ||
| expect(esCallMock).toHaveBeenCalledTimes(5); | ||
| }); | ||
|
|
||
| it('retries each supported error type', async () => { | ||
| const errors = [ | ||
| new EsErrors.NoLivingConnectionsError('no living connection', { | ||
| warnings: [], | ||
| meta: {} as any, | ||
| }), | ||
| new EsErrors.ConnectionError('no connection'), | ||
| new EsErrors.TimeoutError('timeout'), | ||
| new EsErrors.ResponseError({ statusCode: 503, meta: {} as any, warnings: [] }), | ||
| new EsErrors.ResponseError({ statusCode: 408, meta: {} as any, warnings: [] }), | ||
| new EsErrors.ResponseError({ statusCode: 410, meta: {} as any, warnings: [] }), | ||
| ]; | ||
|
|
||
| for (const error of errors) { | ||
| const esCallMock = jest.fn().mockRejectedValueOnce(error).mockResolvedValue('success'); | ||
| expect(await retryTransientEsErrors(esCallMock)).toEqual('success'); | ||
| expect(esCallMock).toHaveBeenCalledTimes(2); | ||
| } | ||
| }); | ||
|
|
||
| it('does not retry unsupported errors', async () => { | ||
| const error = new Error('foo!'); | ||
| const esCallMock = jest.fn().mockRejectedValueOnce(error).mockResolvedValue('success'); | ||
| await expect(retryTransientEsErrors(esCallMock)).rejects.toThrow(error); | ||
| expect(esCallMock).toHaveBeenCalledTimes(1); | ||
| }); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,53 @@ | ||
| /* | ||
| * 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. | ||
| */ | ||
|
|
||
| import { setTimeout } from 'timers/promises'; | ||
| import { errors as EsErrors } from '@elastic/elasticsearch'; | ||
| import type { Logger } from '@kbn/logging'; | ||
|
|
||
| const MAX_ATTEMPTS = 5; | ||
|
|
||
| const retryResponseStatuses = [ | ||
| 503, // ServiceUnavailable | ||
| 408, // RequestTimeout | ||
| 410, // Gone | ||
| ]; | ||
|
|
||
| const isRetryableError = (e: any) => | ||
| e instanceof EsErrors.NoLivingConnectionsError || | ||
| e instanceof EsErrors.ConnectionError || | ||
| e instanceof EsErrors.TimeoutError || | ||
| (e instanceof EsErrors.ResponseError && retryResponseStatuses.includes(e?.statusCode!)); | ||
|
|
||
| /** | ||
| * Retries any transient network or configuration issues encountered from Elasticsearch with an exponential backoff. | ||
| * Should only be used to wrap operations that are idempotent and can be safely executed more than once. | ||
| */ | ||
| export const retryTransientEsErrors = async <T>( | ||
| esCall: () => Promise<T>, | ||
| { logger, attempt = 0 }: { logger?: Logger; attempt?: number } = {} | ||
| ): Promise<T> => { | ||
| try { | ||
| return await esCall(); | ||
| } catch (e) { | ||
| if (attempt < MAX_ATTEMPTS && isRetryableError(e)) { | ||
| const retryCount = attempt + 1; | ||
| const retryDelaySec = Math.min(Math.pow(2, retryCount), 64); // 2s, 4s, 8s, 16s, 32s, 64s, 64s, 64s ... | ||
|
|
||
| logger?.warn( | ||
| `Retrying Elasticsearch operation after [${retryDelaySec}s] due to error: ${e.toString()} ${ | ||
| e.stack | ||
| }` | ||
| ); | ||
|
|
||
| await setTimeout(retryDelaySec * 1000); | ||
| return retryTransientEsErrors(esCall, { logger, attempt: retryCount }); | ||
| } | ||
|
|
||
| throw e; | ||
| } | ||
| }; |