Skip to content

Retry transient informer failures #3034

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

Open
brendandburns wants to merge 1 commit into
base: main
Choose a base branch
from
+147 −17
Open

Retry transient informer failures #3034

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
61 changes: 49 additions & 12 deletions src/cache.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,8 +197,31 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, 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<boolean> {
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<void> {
this._stop();
let hasBackedOff = false;
if (
err &&
((err as { statusCode?: number }).statusCode === 410 || (err as { code?: number }).code === 410)
Expand All @@ -212,25 +235,36 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, 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
return;
}
this.callbackCache[CONNECT].forEach((elt: ErrorCallback) => elt(undefined));
if (!this.resourceVersion) {
let list: KubernetesListObject<T>;
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<T>;
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,
Expand All @@ -245,10 +279,13 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, 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;
Expand Down
103 changes: 98 additions & 5 deletions src/cache_test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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[] = [
{
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -1523,6 +1523,7 @@ describe('ListWatchCache', () => {
},
});

await informer.stop();
mockAgent.assertNoPendingInterceptors();
});

Expand Down Expand Up @@ -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<V1Namespace> = () => 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<V1Namespace> = () => {
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 = {
Expand Down Expand Up @@ -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[] = [
{
Expand Down Expand Up @@ -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(
Expand All @@ -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<V1Namespace> = function (): Promise<V1NamespaceList> {
return new Promise<V1NamespaceList>((resolve, reject) => {
reject(requestErr);
Expand Down