Skip to content

fix(cache): match client-go's reconnect backoff algorithm #3026

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
davidgamero wants to merge 1 commit into
base: main
Choose a base branch
from
+131 −16
Draft

fix(cache): match client-go's reconnect backoff algorithm #3026

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 43 additions & 9 deletions src/cache.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,22 @@ export type CacheMap<T extends KubernetesObject> = Map<string, Map<string, T>>;

export interface ListWatchOptions {
delayFn?: (ms: number) => Promise<void>;
// Clock source, injectable for testing.
nowFn?: () => number;
// Randomness source in [0, 1), injectable for testing.
randFn?: () => number;
}

export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, Informer<T> {
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<T> = new Map();
private resourceVersion: string;
Expand All @@ -39,8 +50,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, 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<void>;
private readonly nowFn: () => number;
private readonly randFn: () => number;
private readonly path: string;
private readonly watch: Watch;
private readonly listFn: ListPromise<T>;
Expand All @@ -63,6 +77,8 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, 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] = [];
Expand All @@ -78,6 +94,7 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
public async start(): Promise<void> {
this.stopped = false;
this.reconnectDelayMs = 0;
this.lastBackoffAt = undefined;
this.hasConnected = false;
await this.doneHandler(null);
}
Expand Down Expand Up @@ -160,6 +177,26 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, 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<void> {
this._stop();
if (
Expand Down Expand Up @@ -208,14 +245,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, 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(
Expand Down
95 changes: 88 additions & 7 deletions src/cache_test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -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;
Expand All @@ -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 () => {
Expand Down Expand Up @@ -1617,6 +1619,7 @@ describe('ListWatchCache', () => {
delayValues.push(ms);
return Promise.resolve();
},
randFn: () => 0,
},
);
await promise;
Expand All @@ -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,
Expand All @@ -1636,7 +1639,7 @@ describe('ListWatchCache', () => {
deepStrictEqual(delayValues, []);

await doneHandler(null);
deepStrictEqual(delayValues, [1000]);
deepStrictEqual(delayValues, [800]);
});

it('should reconnect on TimeoutError', async () => {
Expand Down Expand Up @@ -1718,6 +1721,7 @@ describe('ListWatchCache', () => {
delayValues.push(ms);
return Promise.resolve();
},
randFn: () => 0,
},
);
await promise;
Expand All @@ -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<V1Namespace> = () => 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');
});
});

Expand Down
Loading