Skip to content

Commit 8d3e933

Browse files
author
Jan Kryl
committed
cache: Fix connection leak.
Make sure that done callback of a watcher isn't called more than once and call abort() on request object to close the connection when done.
1 parent dc1260c commit 8d3e933

File tree

4 files changed

+227
-220
lines changed

4 files changed

+227
-220
lines changed

‎src/cache.ts

Lines changed: 11 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
import { ADD, DELETE, ERROR, Informer, ListPromise, ObjectCallback, UPDATE } from './informer';
22
import { KubernetesObject } from './types';
3-
import { Watch } from './watch';
3+
import { RequestResult, Watch } from './watch';
44

55
export interface ObjectCache<T> {
66
get(name: string, namespace?: string): T | undefined;
@@ -12,7 +12,7 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
1212
private resourceVersion: string;
1313
private readonly indexCache: { [key: string]: T[] } = {};
1414
private readonly callbackCache: { [key: string]: Array<ObjectCallback<T>> } = {};
15-
private stopped: boolean;
15+
private request: RequestResult | undefined;
1616

1717
public constructor(
1818
private readonly path: string,
@@ -25,19 +25,20 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
2525
this.callbackCache[DELETE] = [];
2626
this.callbackCache[ERROR] = [];
2727
this.resourceVersion = '';
28-
this.stopped = true;
2928
if (autoStart) {
30-
this.start();
29+
this.doneHandler(null);
3130
}
3231
}
3332

3433
public async start(): Promise<void> {
35-
this.stopped = false;
36-
await this.doneHandler();
34+
await this.doneHandler(null);
3735
}
3836

3937
public stop(): void {
40-
this.stopped = true;
38+
if (this.request) {
39+
this.request.abort();
40+
this.request = undefined;
41+
}
4142
}
4243

4344
public on(verb: string, cb: ObjectCallback<T>): void {
@@ -79,15 +80,10 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
7980
return this.resourceVersion;
8081
}
8182

82-
private async errorHandler(err: any): Promise<void> {
83+
private async doneHandler(err: any): Promise<any> {
84+
this.stop();
8385
if (err) {
8486
this.callbackCache[ERROR].forEach((elt: ObjectCallback<T>) => elt(err));
85-
}
86-
this.stopped = true;
87-
}
88-
89-
private async doneHandler(): Promise<any> {
90-
if (this.stopped) {
9187
return;
9288
}
9389
// TODO: Don't always list here for efficiency
@@ -106,12 +102,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
106102
}
107103
});
108104
this.addOrUpdateItems(list.items);
109-
await this.watch.watch(
105+
this.request = await this.watch.watch(
110106
this.path,
111107
{ resourceVersion: list.metadata!.resourceVersion },
112108
this.watchHandler.bind(this),
113109
this.doneHandler.bind(this),
114-
this.errorHandler.bind(this),
115110
);
116111
}
117112

‎src/cache_test.ts

Lines changed: 38 additions & 91 deletions
Original file line numberDiff line numberDiff line change
@@ -4,14 +4,22 @@ import chaiAsPromised = require('chai-as-promised');
44
import * as mock from 'ts-mockito';
55

66
import http = require('http');
7+
import { Duplex } from 'stream';
8+
import { EventEmitter } from 'ws';
79

810
import { V1Namespace, V1NamespaceList, V1ObjectMeta, V1Pod, V1ListMeta } from './api';
911
import { deleteObject, ListWatch, deleteItems } from './cache';
10-
import { ListCallback, ADD, UPDATE, DELETE, ListPromise } from './informer';
12+
import { ADD, UPDATE, DELETE, ListPromise } from './informer';
1113

1214
use(chaiAsPromised);
1315

14-
import { Watch } from './watch';
16+
import { RequestResult, Watch } from './watch';
17+
18+
// Object replacing real Request object in the test
19+
class FakeRequest extends EventEmitter implements RequestResult {
20+
pipe(stream: Duplex): void {}
21+
abort() {}
22+
}
1523

1624
describe('ListWatchCache', () => {
1725
it('should throw on unknown update', () => {
@@ -24,7 +32,12 @@ describe('ListWatchCache', () => {
2432
(resolve, reject) => {
2533
resolve({
2634
response: {} as http.IncomingMessage,
27-
body: {} as V1NamespaceList,
35+
body: {
36+
metadata: {
37+
resourceVersion: '12345',
38+
} as V1ListMeta,
39+
items: [],
40+
} as V1NamespaceList,
2841
});
2942
},
3043
);
@@ -88,15 +101,9 @@ describe('ListWatchCache', () => {
88101
};
89102
const promise = new Promise((resolve) => {
90103
mock.when(
91-
fakeWatch.watch(
92-
mock.anything(),
93-
mock.anything(),
94-
mock.anything(),
95-
mock.anything(),
96-
mock.anything(),
97-
),
104+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
98105
).thenCall(() => {
99-
resolve(null);
106+
resolve(new FakeRequest());
100107
});
101108
});
102109
const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
@@ -159,7 +166,7 @@ describe('ListWatchCache', () => {
159166
} as V1ObjectMeta,
160167
} as V1Namespace);
161168

162-
await doneHandler();
169+
await doneHandler(null);
163170
expect(cache.list().length, 'all namespace list').to.equal(1);
164171
expect(cache.list('default').length, 'default namespace list').to.equal(1);
165172
expect(cache.list('other'), 'other namespace list').to.be.undefined;
@@ -198,15 +205,9 @@ describe('ListWatchCache', () => {
198205
};
199206
const promise = new Promise((resolve) => {
200207
mock.when(
201-
fakeWatch.watch(
202-
mock.anything(),
203-
mock.anything(),
204-
mock.anything(),
205-
mock.anything(),
206-
mock.anything(),
207-
),
208+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
208209
).thenCall(() => {
209-
resolve(null);
210+
resolve(new FakeRequest());
210211
});
211212
});
212213
const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
@@ -285,15 +286,9 @@ describe('ListWatchCache', () => {
285286
};
286287
const promise = new Promise((resolve) => {
287288
mock.when(
288-
fakeWatch.watch(
289-
mock.anything(),
290-
mock.anything(),
291-
mock.anything(),
292-
mock.anything(),
293-
mock.anything(),
294-
),
289+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
295290
).thenCall(() => {
296-
resolve(null);
291+
resolve(new FakeRequest());
297292
});
298293
});
299294
const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
@@ -362,15 +357,9 @@ describe('ListWatchCache', () => {
362357
};
363358
let promise = new Promise((resolve) => {
364359
mock.when(
365-
fakeWatch.watch(
366-
mock.anything(),
367-
mock.anything(),
368-
mock.anything(),
369-
mock.anything(),
370-
mock.anything(),
371-
),
360+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
372361
).thenCall(() => {
373-
resolve(null);
362+
resolve(new FakeRequest());
374363
});
375364
});
376365
const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
@@ -390,18 +379,12 @@ describe('ListWatchCache', () => {
390379

391380
promise = new Promise((resolve) => {
392381
mock.when(
393-
fakeWatch.watch(
394-
mock.anything(),
395-
mock.anything(),
396-
mock.anything(),
397-
mock.anything(),
398-
mock.anything(),
399-
),
382+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
400383
).thenCall(() => {
401-
resolve(null);
384+
resolve(new FakeRequest());
402385
});
403386
});
404-
doneHandler();
387+
doneHandler(null);
405388
await promise;
406389
expect(addObjects).to.deep.equal(list);
407390
expect(updateObjects).to.deep.equal(list);
@@ -447,15 +430,9 @@ describe('ListWatchCache', () => {
447430
};
448431
let promise = new Promise((resolve) => {
449432
mock.when(
450-
fakeWatch.watch(
451-
mock.anything(),
452-
mock.anything(),
453-
mock.anything(),
454-
mock.anything(),
455-
mock.anything(),
456-
),
433+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
457434
).thenCall(() => {
458-
resolve(null);
435+
resolve(new FakeRequest());
459436
});
460437
});
461438
const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
@@ -476,19 +453,13 @@ describe('ListWatchCache', () => {
476453

477454
promise = new Promise((resolve) => {
478455
mock.when(
479-
fakeWatch.watch(
480-
mock.anything(),
481-
mock.anything(),
482-
mock.anything(),
483-
mock.anything(),
484-
mock.anything(),
485-
),
456+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
486457
).thenCall(() => {
487-
resolve(null);
458+
resolve(new FakeRequest());
488459
});
489460
});
490461
listObj.items = list2;
491-
doneHandler();
462+
doneHandler(null);
492463
await promise;
493464
expect(addObjects).to.deep.equal(list);
494465
expect(updateObjects).to.deep.equal(list2);
@@ -536,15 +507,9 @@ describe('ListWatchCache', () => {
536507
};
537508
const promise = new Promise((resolve) => {
538509
mock.when(
539-
fakeWatch.watch(
540-
mock.anything(),
541-
mock.anything(),
542-
mock.anything(),
543-
mock.anything(),
544-
mock.anything(),
545-
),
510+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
546511
).thenCall(() => {
547-
resolve(null);
512+
resolve(new FakeRequest());
548513
});
549514
});
550515
const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
@@ -662,13 +627,7 @@ describe('ListWatchCache', () => {
662627
};
663628
const watchCalled = new Promise((resolve) => {
664629
mock.when(
665-
fakeWatch.watch(
666-
mock.anything(),
667-
mock.anything(),
668-
mock.anything(),
669-
mock.anything(),
670-
mock.anything(),
671-
),
630+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
672631
).thenCall(resolve);
673632
});
674633
const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
@@ -727,13 +686,7 @@ describe('ListWatchCache', () => {
727686
};
728687
const watchCalled = new Promise((resolve) => {
729688
mock.when(
730-
fakeWatch.watch(
731-
mock.anything(),
732-
mock.anything(),
733-
mock.anything(),
734-
mock.anything(),
735-
mock.anything(),
736-
),
689+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
737690
).thenCall(resolve);
738691
});
739692
const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
@@ -820,13 +773,7 @@ describe('ListWatchCache', () => {
820773
};
821774
const watchCalled = new Promise((resolve) => {
822775
mock.when(
823-
fakeWatch.watch(
824-
mock.anything(),
825-
mock.anything(),
826-
mock.anything(),
827-
mock.anything(),
828-
mock.anything(),
829-
),
776+
fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
830777
).thenCall(resolve);
831778
});
832779
const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);

0 commit comments

Comments
 (0)