Skip to content

Commit 8f23ee7

Browse files
davidgamerodavidgamero
committed
fix(cache): match client-go's reconnect backoff algorithm
The reconnect delay doubled from a 1s base with no jitter, so every client disconnected by the same event (an apiserver rollout, a load balancer cycling) retried at the same instants and kept colliding. Adopt client-go's defaultBackoff* values from the reflector's ExponentialBackoffManager: an 800ms base, a factor of 2, a 30s cap, full jitter, and a reset once the backoff has gone unused for two minutes. Full jitter spreads each retry over [level, 2*level), which is what breaks up the lockstep, and the idle reset means a client that recovered stops paying for failures it has already outlived. Only the delay computation changes. When a reconnect backs off at all is untouched: the first reconnect is still free, and receiving a watch event or a TimeoutError still clears the delay. nowFn and randFn join delayFn as injectable options so the backoff can be asserted deterministically instead of by sleeping.
1 parent 0e873f1 commit 8f23ee7

2 files changed

Lines changed: 131 additions & 16 deletions

File tree

‎src/cache.ts‎

Lines changed: 43 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -26,11 +26,22 @@ export type CacheMap<T extends KubernetesObject> = Map<string, Map<string, T>>;
2626

2727
export interface ListWatchOptions {
2828
delayFn?: (ms: number) => Promise<void>;
29+
// Clock source, injectable for testing.
30+
nowFn?: () => number;
31+
// Randomness source in [0, 1), injectable for testing.
32+
randFn?: () => number;
2933
}
3034

3135
export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, Informer<T> {
32-
private static readonly BASE_RECONNECT_DELAY_MS = 1000;
33-
private static readonly MAX_RECONNECT_DELAY_MS = 30000;
36+
// Mirrors k8s.io/client-go's defaultBackoff{Init,Max,Factor,Jitter,Reset},
37+
// used by the reflector via NewExponentialBackoffManager. Jitter spreads a
38+
// fleet out so that everyone does not retry in lockstep, and the delay
39+
// resets once it has gone unused for BACKOFF_RESET_MS.
40+
private static readonly BACKOFF_INIT_MS = 800;
41+
private static readonly BACKOFF_MAX_MS = 30000;
42+
private static readonly BACKOFF_FACTOR = 2;
43+
private static readonly BACKOFF_JITTER = 1.0;
44+
private static readonly BACKOFF_RESET_MS = 120000;
3445

3546
private objects: CacheMap<T> = new Map();
3647
private resourceVersion: string;
@@ -39,8 +50,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
3950
private request: AbortController | undefined;
4051
private stopped: boolean = false;
4152
private reconnectDelayMs: number = 0;
53+
private lastBackoffAt: number | undefined;
4254
private hasConnected: boolean = false;
4355
private readonly delayFn: (ms: number) => Promise<void>;
56+
private readonly nowFn: () => number;
57+
private readonly randFn: () => number;
4458
private readonly path: string;
4559
private readonly watch: Watch;
4660
private readonly listFn: ListPromise<T>;
@@ -63,6 +77,8 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
6377
this.labelSelector = labelSelector;
6478
this.fieldSelector = fieldSelector;
6579
this.delayFn = options?.delayFn ?? setTimeout;
80+
this.nowFn = options?.nowFn ?? Date.now;
81+
this.randFn = options?.randFn ?? Math.random;
6682

6783
this.callbackCache[ADD] = [];
6884
this.callbackCache[UPDATE] = [];
@@ -78,6 +94,7 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
7894
public async start(): Promise<void> {
7995
this.stopped = false;
8096
this.reconnectDelayMs = 0;
97+
this.lastBackoffAt = undefined;
8198
this.hasConnected = false;
8299
await this.doneHandler(null);
83100
}
@@ -160,6 +177,26 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
160177
}
161178
}
162179

180+
// The next backoff level, before jitter. Grows exponentially up to a cap,
181+
// and starts over once the backoff has gone unused for BACKOFF_RESET_MS,
182+
// which is how a connection that recovered stops paying for old failures.
183+
private nextBackoffLevelMs(): number {
184+
const now = this.nowFn();
185+
const idle =
186+
this.lastBackoffAt !== undefined && now - this.lastBackoffAt >= ListWatch.BACKOFF_RESET_MS;
187+
this.lastBackoffAt = now;
188+
if (this.reconnectDelayMs === 0 || idle) {
189+
return ListWatch.BACKOFF_INIT_MS;
190+
}
191+
return Math.min(this.reconnectDelayMs * ListWatch.BACKOFF_FACTOR, ListWatch.BACKOFF_MAX_MS);
192+
}
193+
194+
// Spreads the delay over [ms, ms * (1 + BACKOFF_JITTER)) so that many
195+
// clients disconnected by the same event do not all retry together.
196+
private withJitter(ms: number): number {
197+
return ms + this.randFn() * ListWatch.BACKOFF_JITTER * ms;
198+
}
199+
163200
private async doneHandler(err: any): Promise<void> {
164201
this._stop();
165202
if (
@@ -208,14 +245,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
208245
if (this.fieldSelector !== undefined) {
209246
queryParams.fieldSelector = ObjectSerializer.serialize(this.fieldSelector, 'string');
210247
}
211-
if (this.reconnectDelayMs > 0 && this.hasConnected) {
212-
await this.delayFn(this.reconnectDelayMs);
213-
}
214248
if (this.hasConnected) {
215-
this.reconnectDelayMs = Math.min(
216-
this.reconnectDelayMs > 0 ? this.reconnectDelayMs * 2 : ListWatch.BASE_RECONNECT_DELAY_MS,
217-
ListWatch.MAX_RECONNECT_DELAY_MS,
218-
);
249+
if (this.reconnectDelayMs > 0) {
250+
await this.delayFn(this.withJitter(this.reconnectDelayMs));
251+
}
252+
this.reconnectDelayMs = this.nextBackoffLevelMs();
219253
}
220254
this.hasConnected = true;
221255
this.request = await this.watch.watch(

‎src/cache_test.ts‎

Lines changed: 88 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { describe, it } from 'node:test';
2-
import { deepStrictEqual, notStrictEqual, strictEqual, throws } from 'node:assert';
2+
import { deepStrictEqual, notStrictEqual, ok, strictEqual, throws } from 'node:assert';
33
import mock from 'ts-mockito';
44

55
import { V1Namespace, V1NamespaceList, V1ObjectMeta, V1Pod, V1PodList, V1ListMeta } from './api.js';
@@ -1561,6 +1561,8 @@ describe('ListWatchCache', () => {
15611561
delayValues.push(ms);
15621562
return Promise.resolve();
15631563
},
1564+
// Pin the jitter off so the growth itself can be asserted.
1565+
randFn: () => 0,
15641566
},
15651567
);
15661568
await promise;
@@ -1574,15 +1576,15 @@ describe('ListWatchCache', () => {
15741576

15751577
await doneHandler(null);
15761578
strictEqual(watchCalls, 3);
1577-
deepStrictEqual(delayValues, [1000]);
1579+
deepStrictEqual(delayValues, [800]);
15781580

15791581
await doneHandler(null);
15801582
strictEqual(watchCalls, 4);
1581-
deepStrictEqual(delayValues, [1000, 2000]);
1583+
deepStrictEqual(delayValues, [800, 1600]);
15821584

15831585
await doneHandler(null);
15841586
strictEqual(watchCalls, 5);
1585-
deepStrictEqual(delayValues, [1000, 2000, 4000]);
1587+
deepStrictEqual(delayValues, [800, 1600, 3200]);
15861588
});
15871589

15881590
it('should reset backoff after receiving a watch event', async () => {
@@ -1617,6 +1619,7 @@ describe('ListWatchCache', () => {
16171619
delayValues.push(ms);
16181620
return Promise.resolve();
16191621
},
1622+
randFn: () => 0,
16201623
},
16211624
);
16221625
await promise;
@@ -1625,7 +1628,7 @@ describe('ListWatchCache', () => {
16251628

16261629
await doneHandler(null);
16271630
await doneHandler(null);
1628-
deepStrictEqual(delayValues, [1000]);
1631+
deepStrictEqual(delayValues, [800]);
16291632

16301633
watchHandler('ADDED', {
16311634
metadata: { name: 'reset', namespace: 'default', resourceVersion: '99' } as V1ObjectMeta,
@@ -1636,7 +1639,7 @@ describe('ListWatchCache', () => {
16361639
deepStrictEqual(delayValues, []);
16371640

16381641
await doneHandler(null);
1639-
deepStrictEqual(delayValues, [1000]);
1642+
deepStrictEqual(delayValues, [800]);
16401643
});
16411644

16421645
it('should reconnect on TimeoutError', async () => {
@@ -1718,6 +1721,7 @@ describe('ListWatchCache', () => {
17181721
delayValues.push(ms);
17191722
return Promise.resolve();
17201723
},
1724+
randFn: () => 0,
17211725
},
17221726
);
17231727
await promise;
@@ -1737,7 +1741,84 @@ describe('ListWatchCache', () => {
17371741

17381742
// Backoff is still applied to non-timeout reconnects.
17391743
await doneHandler(null);
1740-
deepStrictEqual(delayValues, [1000]);
1744+
deepStrictEqual(delayValues, [800]);
1745+
});
1746+
1747+
// Builds a cache whose delay, clock and randomness are all observable, so
1748+
// the backoff can be driven deterministically.
1749+
async function setupBackoffCache(opts: { delays: number[]; now?: () => number; rand?: () => number }) {
1750+
const fakeWatch = mock.mock(Watch);
1751+
const listObj = {
1752+
metadata: { resourceVersion: '12345' } as V1ListMeta,
1753+
items: [] as V1Namespace[],
1754+
} as V1NamespaceList;
1755+
const listFn: ListPromise<V1Namespace> = () => Promise.resolve(listObj);
1756+
1757+
const promise = new Promise((resolve) => {
1758+
mock.when(
1759+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
1760+
).thenCall(() => {
1761+
resolve(new AbortController());
1762+
return Promise.resolve(new AbortController());
1763+
});
1764+
});
1765+
1766+
// eslint-disable-next-line @typescript-eslint/no-unused-vars
1767+
const cache = new ListWatch(
1768+
'/some/path',
1769+
mock.instance(fakeWatch),
1770+
listFn,
1771+
true,
1772+
undefined,
1773+
undefined,
1774+
{
1775+
delayFn: (ms: number) => {
1776+
opts.delays.push(ms);
1777+
return Promise.resolve();
1778+
},
1779+
nowFn: opts.now,
1780+
randFn: opts.rand,
1781+
},
1782+
);
1783+
await promise;
1784+
const [, , , doneHandler] = mock.capture(fakeWatch.watch).last();
1785+
return { done: doneHandler };
1786+
}
1787+
1788+
it('should spread the backoff over a jittered range', async () => {
1789+
// Full jitter: each delay lands in [level, 2*level). Two clients that
1790+
// failed together therefore do not retry together.
1791+
const low: number[] = [];
1792+
const high: number[] = [];
1793+
const lowCache = await setupBackoffCache({ delays: low, rand: () => 0 });
1794+
const highCache = await setupBackoffCache({ delays: high, rand: () => 0.999 });
1795+
1796+
for (let i = 0; i < 3; i++) {
1797+
await lowCache.done(null);
1798+
await highCache.done(null);
1799+
}
1800+
1801+
deepStrictEqual(low, [800, 1600]);
1802+
ok(high[0] > 800 && high[0] < 1600, `expected [800, 1600), got ${high[0]}`);
1803+
ok(high[1] > 1600 && high[1] < 3200, `expected [1600, 3200), got ${high[1]}`);
1804+
});
1805+
1806+
it('should cap the backoff and reset it once idle', async () => {
1807+
const delays: number[] = [];
1808+
let now = 0;
1809+
const h = await setupBackoffCache({ delays, now: () => now, rand: () => 0 });
1810+
1811+
for (let i = 0; i < 12; i++) {
1812+
await h.done(null);
1813+
}
1814+
strictEqual(delays[delays.length - 1], 30000, 'expected the delay to cap at 30s');
1815+
1816+
// Two idle minutes means the previous failures no longer count.
1817+
now += 120000;
1818+
await h.done(null);
1819+
strictEqual(delays[delays.length - 1], 30000, 'the pending delay is still the capped one');
1820+
await h.done(null);
1821+
strictEqual(delays[delays.length - 1], 800, 'expected the backoff to start over');
17411822
});
17421823
});
17431824

0 commit comments

Comments
 (0)