diff --git a/src/cache.ts b/src/cache.ts index d7fcbc9dd6a..2150b5de9be 100644 --- a/src/cache.ts +++ b/src/cache.ts @@ -26,11 +26,22 @@ export type CacheMap = Map>; export interface ListWatchOptions { delayFn?: (ms: number) => Promise; + // Clock source, injectable for testing. + nowFn?: () => number; + // Randomness source in [0, 1), injectable for testing. + randFn?: () => number; } export class ListWatch implements ObjectCache, Informer { - private static readonly BASE_RECONNECT_DELAY_MS = 1000; - private static readonly MAX_RECONNECT_DELAY_MS = 30000; + // Mirrors k8s.io/client-go's defaultBackoff{Init,Max,Factor,Jitter,Reset}, + // used by the reflector via NewExponentialBackoffManager. Jitter spreads a + // fleet out so that everyone does not retry in lockstep, and the delay + // resets once it has gone unused for BACKOFF_RESET_MS. + private static readonly BACKOFF_INIT_MS = 800; + private static readonly BACKOFF_MAX_MS = 30000; + private static readonly BACKOFF_FACTOR = 2; + private static readonly BACKOFF_JITTER = 1.0; + private static readonly BACKOFF_RESET_MS = 120000; private objects: CacheMap = new Map(); private resourceVersion: string; @@ -39,8 +50,11 @@ export class ListWatch implements ObjectCache, In private request: AbortController | undefined; private stopped: boolean = false; private reconnectDelayMs: number = 0; + private lastBackoffAt: number | undefined; private hasConnected: boolean = false; private readonly delayFn: (ms: number) => Promise; + private readonly nowFn: () => number; + private readonly randFn: () => number; private readonly path: string; private readonly watch: Watch; private readonly listFn: ListPromise; @@ -63,6 +77,8 @@ export class ListWatch implements ObjectCache, In this.labelSelector = labelSelector; this.fieldSelector = fieldSelector; this.delayFn = options?.delayFn ?? setTimeout; + this.nowFn = options?.nowFn ?? Date.now; + this.randFn = options?.randFn ?? Math.random; this.callbackCache[ADD] = []; this.callbackCache[UPDATE] = []; @@ -78,6 +94,7 @@ export class ListWatch implements ObjectCache, In public async start(): Promise { this.stopped = false; this.reconnectDelayMs = 0; + this.lastBackoffAt = undefined; this.hasConnected = false; await this.doneHandler(null); } @@ -160,6 +177,26 @@ export class ListWatch implements ObjectCache, In } } + // The next backoff level, before jitter. Grows exponentially up to a cap, + // and starts over once the backoff has gone unused for BACKOFF_RESET_MS, + // which is how a connection that recovered stops paying for old failures. + private nextBackoffLevelMs(): number { + const now = this.nowFn(); + const idle = + this.lastBackoffAt !== undefined && now - this.lastBackoffAt >= ListWatch.BACKOFF_RESET_MS; + this.lastBackoffAt = now; + if (this.reconnectDelayMs === 0 || idle) { + return ListWatch.BACKOFF_INIT_MS; + } + return Math.min(this.reconnectDelayMs * ListWatch.BACKOFF_FACTOR, ListWatch.BACKOFF_MAX_MS); + } + + // Spreads the delay over [ms, ms * (1 + BACKOFF_JITTER)) so that many + // clients disconnected by the same event do not all retry together. + private withJitter(ms: number): number { + return ms + this.randFn() * ListWatch.BACKOFF_JITTER * ms; + } + private async doneHandler(err: any): Promise { this._stop(); if ( @@ -208,14 +245,11 @@ export class ListWatch implements ObjectCache, In if (this.fieldSelector !== undefined) { queryParams.fieldSelector = ObjectSerializer.serialize(this.fieldSelector, 'string'); } - if (this.reconnectDelayMs > 0 && this.hasConnected) { - await this.delayFn(this.reconnectDelayMs); - } if (this.hasConnected) { - this.reconnectDelayMs = Math.min( - this.reconnectDelayMs > 0 ? this.reconnectDelayMs * 2 : ListWatch.BASE_RECONNECT_DELAY_MS, - ListWatch.MAX_RECONNECT_DELAY_MS, - ); + if (this.reconnectDelayMs > 0) { + await this.delayFn(this.withJitter(this.reconnectDelayMs)); + } + this.reconnectDelayMs = this.nextBackoffLevelMs(); } this.hasConnected = true; this.request = await this.watch.watch( diff --git a/src/cache_test.ts b/src/cache_test.ts index 20fb08f78cd..550e9905597 100644 --- a/src/cache_test.ts +++ b/src/cache_test.ts @@ -1,5 +1,5 @@ import { describe, it } from 'node:test'; -import { deepStrictEqual, notStrictEqual, strictEqual, throws } from 'node:assert'; +import { deepStrictEqual, notStrictEqual, ok, strictEqual, throws } from 'node:assert'; import mock from 'ts-mockito'; import { V1Namespace, V1NamespaceList, V1ObjectMeta, V1Pod, V1PodList, V1ListMeta } from './api.js'; @@ -1561,6 +1561,8 @@ describe('ListWatchCache', () => { delayValues.push(ms); return Promise.resolve(); }, + // Pin the jitter off so the growth itself can be asserted. + randFn: () => 0, }, ); await promise; @@ -1574,15 +1576,15 @@ describe('ListWatchCache', () => { await doneHandler(null); strictEqual(watchCalls, 3); - deepStrictEqual(delayValues, [1000]); + deepStrictEqual(delayValues, [800]); await doneHandler(null); strictEqual(watchCalls, 4); - deepStrictEqual(delayValues, [1000, 2000]); + deepStrictEqual(delayValues, [800, 1600]); await doneHandler(null); strictEqual(watchCalls, 5); - deepStrictEqual(delayValues, [1000, 2000, 4000]); + deepStrictEqual(delayValues, [800, 1600, 3200]); }); it('should reset backoff after receiving a watch event', async () => { @@ -1617,6 +1619,7 @@ describe('ListWatchCache', () => { delayValues.push(ms); return Promise.resolve(); }, + randFn: () => 0, }, ); await promise; @@ -1625,7 +1628,7 @@ describe('ListWatchCache', () => { await doneHandler(null); await doneHandler(null); - deepStrictEqual(delayValues, [1000]); + deepStrictEqual(delayValues, [800]); watchHandler('ADDED', { metadata: { name: 'reset', namespace: 'default', resourceVersion: '99' } as V1ObjectMeta, @@ -1636,7 +1639,7 @@ describe('ListWatchCache', () => { deepStrictEqual(delayValues, []); await doneHandler(null); - deepStrictEqual(delayValues, [1000]); + deepStrictEqual(delayValues, [800]); }); it('should reconnect on TimeoutError', async () => { @@ -1718,6 +1721,7 @@ describe('ListWatchCache', () => { delayValues.push(ms); return Promise.resolve(); }, + randFn: () => 0, }, ); await promise; @@ -1737,7 +1741,84 @@ describe('ListWatchCache', () => { // Backoff is still applied to non-timeout reconnects. await doneHandler(null); - deepStrictEqual(delayValues, [1000]); + deepStrictEqual(delayValues, [800]); + }); + + // Builds a cache whose delay, clock and randomness are all observable, so + // the backoff can be driven deterministically. + async function setupBackoffCache(opts: { delays: number[]; now?: () => number; rand?: () => number }) { + const fakeWatch = mock.mock(Watch); + const listObj = { + metadata: { resourceVersion: '12345' } as V1ListMeta, + items: [] as V1Namespace[], + } as V1NamespaceList; + const listFn: ListPromise = () => Promise.resolve(listObj); + + const promise = new Promise((resolve) => { + mock.when( + fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()), + ).thenCall(() => { + resolve(new AbortController()); + return Promise.resolve(new AbortController()); + }); + }); + + // eslint-disable-next-line @typescript-eslint/no-unused-vars + const cache = new ListWatch( + '/some/path', + mock.instance(fakeWatch), + listFn, + true, + undefined, + undefined, + { + delayFn: (ms: number) => { + opts.delays.push(ms); + return Promise.resolve(); + }, + nowFn: opts.now, + randFn: opts.rand, + }, + ); + await promise; + const [, , , doneHandler] = mock.capture(fakeWatch.watch).last(); + return { done: doneHandler }; + } + + it('should spread the backoff over a jittered range', async () => { + // Full jitter: each delay lands in [level, 2*level). Two clients that + // failed together therefore do not retry together. + const low: number[] = []; + const high: number[] = []; + const lowCache = await setupBackoffCache({ delays: low, rand: () => 0 }); + const highCache = await setupBackoffCache({ delays: high, rand: () => 0.999 }); + + for (let i = 0; i < 3; i++) { + await lowCache.done(null); + await highCache.done(null); + } + + deepStrictEqual(low, [800, 1600]); + ok(high[0] > 800 && high[0] < 1600, `expected [800, 1600), got ${high[0]}`); + ok(high[1] > 1600 && high[1] < 3200, `expected [1600, 3200), got ${high[1]}`); + }); + + it('should cap the backoff and reset it once idle', async () => { + const delays: number[] = []; + let now = 0; + const h = await setupBackoffCache({ delays, now: () => now, rand: () => 0 }); + + for (let i = 0; i < 12; i++) { + await h.done(null); + } + strictEqual(delays[delays.length - 1], 30000, 'expected the delay to cap at 30s'); + + // Two idle minutes means the previous failures no longer count. + now += 120000; + await h.done(null); + strictEqual(delays[delays.length - 1], 30000, 'the pending delay is still the capped one'); + await h.done(null); + strictEqual(delays[delays.length - 1], 800, 'expected the backoff to start over'); }); });