diff --git a/src/cache.ts b/src/cache.ts index 2150b5de9be..87a7902bc84 100644 --- a/src/cache.ts +++ b/src/cache.ts @@ -197,8 +197,31 @@ export class ListWatch implements ObjectCache, In return ms + this.randFn() * ListWatch.BACKOFF_JITTER * ms; } + private errorStatusCode(err: any): number | undefined { + const statusCode = err?.statusCode ?? err?.code; + return typeof statusCode === 'number' ? statusCode : undefined; + } + + private isRetryableError(err: any): boolean { + const statusCode = this.errorStatusCode(err); + return statusCode === undefined || statusCode === 408 || statusCode === 429 || statusCode >= 500; + } + + private async waitForRetry(): Promise { + if (this.reconnectDelayMs === 0) { + this.reconnectDelayMs = this.nextBackoffLevelMs(); + } + await this.delayFn(this.withJitter(this.reconnectDelayMs)); + if (this.stopped) { + return false; + } + this.reconnectDelayMs = this.nextBackoffLevelMs(); + return true; + } + private async doneHandler(err: any): Promise { this._stop(); + let hasBackedOff = false; if ( err && ((err as { statusCode?: number }).statusCode === 410 || (err as { code?: number }).code === 410) @@ -212,7 +235,10 @@ export class ListWatch implements ObjectCache, In this.reconnectDelayMs = 0; } else if (err) { this.callbackCache[ERROR].forEach((elt: ErrorCallback) => elt(err)); - return; + if (!this.isRetryableError(err) || this.stopped || !(await this.waitForRetry())) { + return; + } + hasBackedOff = true; } if (this.stopped) { // do not auto-restart @@ -220,17 +246,25 @@ export class ListWatch implements ObjectCache, In } this.callbackCache[CONNECT].forEach((elt: ErrorCallback) => elt(undefined)); if (!this.resourceVersion) { - let list: KubernetesListObject; - try { - const promise = this.listFn(); - list = await promise; - } catch (err) { - this.callbackCache[ERROR].forEach((elt: ErrorCallback) => elt(err)); - return; + let listed = false; + while (!listed) { + let list: KubernetesListObject; + try { + const promise = this.listFn(); + list = await promise; + } catch (err) { + this.callbackCache[ERROR].forEach((elt: ErrorCallback) => elt(err)); + if (!this.isRetryableError(err) || this.stopped || !(await this.waitForRetry())) { + return; + } + hasBackedOff = true; + continue; + } + this.objects = deleteItems(this.objects, list.items, this.callbackCache[DELETE].slice()); + this.addOrUpdateItems(list.items); + this.resourceVersion = list.metadata ? list.metadata!.resourceVersion || '' : ''; + listed = true; } - this.objects = deleteItems(this.objects, list.items, this.callbackCache[DELETE].slice()); - this.addOrUpdateItems(list.items); - this.resourceVersion = list.metadata ? list.metadata!.resourceVersion || '' : ''; } const queryParams = { resourceVersion: this.resourceVersion, @@ -245,10 +279,13 @@ export class ListWatch implements ObjectCache, In if (this.fieldSelector !== undefined) { queryParams.fieldSelector = ObjectSerializer.serialize(this.fieldSelector, 'string'); } - if (this.hasConnected) { + if (this.hasConnected && !hasBackedOff) { if (this.reconnectDelayMs > 0) { await this.delayFn(this.withJitter(this.reconnectDelayMs)); } + if (this.stopped) { + return; + } this.reconnectDelayMs = this.nextBackoffLevelMs(); } this.hasConnected = true; diff --git a/src/cache_test.ts b/src/cache_test.ts index 550e9905597..da689afcc50 100644 --- a/src/cache_test.ts +++ b/src/cache_test.ts @@ -1106,7 +1106,7 @@ describe('ListWatchCache', () => { ).twice(); }); - it('does not auto-restart after an error', async () => { + it('does not auto-restart after a non-retryable error', async () => { const fakeWatch = mock.mock(Watch); const list: V1Pod[] = [ { @@ -1150,7 +1150,7 @@ describe('ListWatchCache', () => { const [, , , doneHandler] = mock.capture(fakeWatch.watch).last(); - const error = new Error('testing'); + const error = Object.assign(new Error('Bad Request'), { statusCode: 400 }); await doneHandler(error); mock.verify( @@ -1523,6 +1523,7 @@ describe('ListWatchCache', () => { }, }); + await informer.stop(); mockAgent.assertNoPendingInterceptors(); }); @@ -1686,6 +1687,98 @@ describe('ListWatchCache', () => { strictEqual(errorEmitted, false); }); + it('should reconnect with backoff after a retryable watch error', async () => { + const fakeWatch = mock.mock(Watch); + const listObj = { + metadata: { resourceVersion: '12345' } as V1ListMeta, + items: [] as V1Namespace[], + } as V1NamespaceList; + const listFn: ListPromise = () => Promise.resolve(listObj); + + let watchCalls = 0; + const delays: number[] = []; + mock.when( + fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()), + ).thenCall(() => { + watchCalls++; + return Promise.resolve(new AbortController()); + }); + + const informer = new ListWatch( + '/some/path', + mock.instance(fakeWatch), + listFn, + false, + undefined, + undefined, + { + delayFn: (ms: number) => { + delays.push(ms); + return Promise.resolve(); + }, + randFn: () => 0, + }, + ); + const errors: Error[] = []; + informer.on('error', (err) => errors.push(err)); + await informer.start(); + + const [, , , doneHandler] = mock.capture(fakeWatch.watch).last(); + const error = Object.assign(new Error('Service Unavailable'), { statusCode: 503 }); + await doneHandler(error); + + strictEqual(watchCalls, 2); + deepStrictEqual(delays, [800]); + deepStrictEqual(errors, [error]); + }); + + it('should retry the initial list with backoff after a retryable error', async () => { + const fakeWatch = mock.mock(Watch); + const listObj = { + metadata: { resourceVersion: '12345' } as V1ListMeta, + items: [] as V1Namespace[], + } as V1NamespaceList; + const error = Object.assign(new Error('Too Many Requests'), { code: 429 }); + let listCalls = 0; + const listFn: ListPromise = () => { + listCalls++; + return listCalls === 1 ? Promise.reject(error) : Promise.resolve(listObj); + }; + const delays: number[] = []; + let watchCalls = 0; + mock.when( + fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()), + ).thenCall(() => { + watchCalls++; + return Promise.resolve(new AbortController()); + }); + + const informer = new ListWatch( + '/some/path', + mock.instance(fakeWatch), + listFn, + false, + undefined, + undefined, + { + delayFn: (ms: number) => { + delays.push(ms); + return Promise.resolve(); + }, + randFn: () => 0, + }, + ); + const errors: Error[] = []; + informer.on('error', (err) => errors.push(err)); + + await informer.start(); + + strictEqual(listCalls, 2); + strictEqual(watchCalls, 1); + deepStrictEqual(delays, [800]); + deepStrictEqual(errors, [error]); + }); + it('should not back off between repeated TimeoutErrors', async () => { const fakeWatch = mock.mock(Watch); const listObj = { @@ -1896,7 +1989,7 @@ describe('delete items', () => { strictEqual(await connectPromise, true); }); - it('does calls connect after a restart after an error', async () => { + it('calls connect after manually restarting from a non-retryable error', async () => { const fakeWatch = mock.mock(Watch); const list: V1Pod[] = [ { @@ -1940,7 +2033,7 @@ describe('delete items', () => { const [, , , doneHandler] = mock.capture(fakeWatch.watch).last(); - const error = new Error('testing'); + const error = Object.assign(new Error('Bad Request'), { statusCode: 400 }); await doneHandler(error); mock.verify( @@ -1960,7 +2053,7 @@ describe('delete items', () => { it('should correctly handle errors in the initial list', async () => { const fake = mock.mock(Watch); - const requestErr = Error('request failed'); + const requestErr = Object.assign(Error('request failed'), { statusCode: 400 }); const listFn: ListPromise = function (): Promise { return new Promise((resolve, reject) => { reject(requestErr);