Merge pull request #169 from brendandburns/cache

k8s-ci-robot · web-flow · commit 3783b772ba8d · 2019-01-22T16:13:35.000-08:00
Add an initial watch cache implementation.
diff --git a/examples/cache-example.js b/examples/cache-example.js
@@ -0,0 +1,33 @@
+const k8s = require('@kubernetes/client-node');
+
+const kc = new k8s.KubeConfig();
+kc.loadFromDefault();
+
+const k8sApi = kc.makeApiClient(k8s.Core_v1Api);
+
+const path = '/api/v1/namespaces/default/pods';
+const watch = new k8s.Watch(kc);
+const listFn = (fn) => {
+    k8sApi.listNamespacedPod('default')
+        .then((res) => {
+            fn(res.body.items);
+        })
+        .catch((err) => {
+            console.log(err);
+        });
+}
+const cache = new k8s.ListWatch(path, watch, listFn);
+
+const looper = () => {
+    const list = cache.list('default');
+    if (list) {
+        let names = [];
+        for (let i = 0; i < list.length; i++) {
+            names.push(list[i].metadata.name);
+        }
+        console.log(names.join(','));
+    }
+    setTimeout(looper, 2000);
+}
+
+looper();
diff --git a/src/cache.ts b/src/cache.ts
@@ -0,0 +1,105 @@
+import { KubernetesObject } from './types';
+import { Watch } from './watch';
+
+export interface ObjectCache<T> {
+    get(name: string, namespace?: string): T | undefined;
+    list(namespace?: string): ReadonlyArray<T>;
+}
+
+export type ListCallback<T extends KubernetesObject> = (list: T[]) => void;
+
+export class ListWatch<T extends KubernetesObject> implements ObjectCache<T> {
+    private objects: T[] = [];
+    private readonly indexCache: { [key: string]: T[] } = {};
+
+    public constructor(private readonly path: string,
+                       private readonly watch: Watch,
+                       private readonly listFn: (callback: ListCallback<T>) => void) {
+        this.watch = watch;
+        this.listFn = listFn;
+        this.doneHandler(null);
+    }
+
+    public get(name: string, namespace?: string): T | undefined {
+        return this.objects.find((obj: T): boolean => {
+            return (obj.metadata.name === name &&
+                    (!namespace || obj.metadata.namespace === namespace));
+        });
+    }
+
+    public list(namespace?: string | undefined): ReadonlyArray<T> {
+        if (!namespace) {
+            return this.objects;
+        }
+        return this.indexCache[namespace] as ReadonlyArray<T>;
+    }
+
+    private doneHandler(err: any) {
+        this.listFn((result: T[]) => {
+            this.objects = result;
+            for (const elt of this.objects) {
+                this.indexObj(elt);
+            }
+            this.watch.watch(this.path, {}, this.watchHandler.bind(this), this.doneHandler.bind(this));
+        });
+    }
+
+    private indexObj(obj: T) {
+        let namespaceList = this.indexCache[obj.metadata.namespace] as T[];
+        if (!namespaceList) {
+            namespaceList = [];
+            this.indexCache[obj.metadata.namespace] = namespaceList;
+        }
+        addOrUpdateObject(namespaceList, obj);
+    }
+
+    private watchHandler(phase: string, obj: T) {
+        switch (phase) {
+            case 'ADDED':
+            case 'MODIFIED':
+                addOrUpdateObject(this.objects, obj);
+                if (obj.metadata.namespace) {
+                    this.indexObj(obj);
+                }
+                break;
+            case 'DELETED':
+                deleteObject(this.objects, obj);
+                if (obj.metadata.namespace) {
+                    const namespaceList = this.indexCache[obj.metadata.namespace] as T[];
+                    if (namespaceList) {
+                        deleteObject(namespaceList, obj);
+                    }
+                }
+                break;
+        }
+    }
+}
+
+// Only public for testing.
+export function addOrUpdateObject<T extends KubernetesObject>(objects: T[], obj: T) {
+    const ix = findKubernetesObject(objects, obj);
+    if (ix === -1) {
+        objects.push(obj);
+    } else {
+        objects[ix] = obj;
+    }
+}
+
+function isSameObject<T extends KubernetesObject>(o1: T, o2: T): boolean {
+    return o1.metadata.name === o2.metadata.name &&
+           o1.metadata.namespace === o2.metadata.namespace;
+}
+
+function findKubernetesObject<T extends KubernetesObject>(objects: T[], obj: T): number {
+    return objects.findIndex((elt: T) => {
+        return isSameObject(elt, obj);
+    });
+}
+
+// Public for testing.
+export function deleteObject<T extends KubernetesObject>(objects: T[], obj: T) {
+    const ix = findKubernetesObject(objects, obj);
+    if (ix !== -1) {
+        objects.splice(ix, 1);
+    }
+}
diff --git a/src/cache_test.ts b/src/cache_test.ts
@@ -0,0 +1,180 @@
+import { expect } from 'chai';
+import * as mock from 'ts-mockito';
+
+import { V1Namespace, V1ObjectMeta, V1Pod } from './api';
+import { deleteObject, ListCallback, ListWatch } from './cache';
+import { Watch } from './watch';
+
+describe('ListWatchCache', () => {
+    it('should perform basic caching', () => {
+        const fakeWatch = mock.mock(Watch);
+        const list: V1Namespace[] = [
+            {
+                metadata: {
+                    name: 'name1',
+                } as V1ObjectMeta,
+            } as V1Namespace,
+            {
+                metadata: {
+                    name: 'name2',
+                } as V1ObjectMeta,
+            } as V1Namespace,
+        ];
+        const listFn = (callback: ListCallback<V1Namespace>) => {
+            callback(list);
+        };
+        const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
+        const [pathOut, , watchHandler] = mock.capture(fakeWatch.watch).last();
+        expect(pathOut).to.equal('/some/path');
+        expect(cache.list()).to.deep.equal(list);
+
+        expect(cache.get('name1')).to.equal(list[0]);
+        expect(cache.get('name2')).to.equal(list[1]);
+
+        watchHandler('ADDED',  {
+            metadata: {
+                name: 'name3',
+            } as V1ObjectMeta,
+        } as V1Namespace);
+
+        expect(cache.list().length).to.equal(3);
+        expect(cache.get('name3')).to.not.equal(null);
+
+        watchHandler('MODIFIED',  {
+            metadata: {
+                name: 'name3',
+                resourceVersion: 'baz',
+            } as V1ObjectMeta,
+        } as V1Namespace);
+        expect(cache.list().length).to.equal(3);
+        const obj3 = cache.get('name3');
+        expect(obj3).to.not.equal(null);
+        if (obj3) {
+            expect(obj3.metadata.name).to.equal('name3');
+            expect(obj3.metadata.resourceVersion).to.equal('baz');
+        }
+
+        watchHandler('DELETED', {
+            metadata: {
+                name: 'name2',
+            } as V1ObjectMeta,
+        } as V1Namespace);
+        expect(cache.list().length).to.equal(2);
+        expect(cache.get('name2')).to.equal(undefined);
+    });
+
+    it('should perform namespace caching', () => {
+        const fakeWatch = mock.mock(Watch);
+        const list: V1Pod[] = [
+            {
+                metadata: {
+                    name: 'name1',
+                    namespace: 'ns1',
+                } as V1ObjectMeta,
+            } as V1Pod,
+            {
+                metadata: {
+                    name: 'name2',
+                    namespace: 'ns2',
+                } as V1ObjectMeta,
+            } as V1Pod,
+        ];
+        const listFn = (callback: ListCallback<V1Pod>) => {
+            callback(list);
+        };
+        const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
+        const [pathOut, , watchHandler] = mock.capture(fakeWatch.watch).last();
+        expect(pathOut).to.equal('/some/path');
+        expect(cache.list()).to.deep.equal(list);
+
+        expect(cache.get('name1')).to.equal(list[0]);
+        expect(cache.get('name2')).to.equal(list[1]);
+
+        expect(cache.list('ns1').length).to.equal(1);
+        expect(cache.list('ns1')[0].metadata.name).to.equal('name1');
+
+        expect(cache.list('ns2').length).to.equal(1);
+        expect(cache.list('ns2')[0].metadata.name).to.equal('name2');
+
+        watchHandler('ADDED',  {
+            metadata: {
+                name: 'name3',
+                namespace: 'ns3',
+            } as V1ObjectMeta,
+        } as V1Pod);
+
+        expect(cache.list().length).to.equal(3);
+        expect(cache.get('name3', 'ns3')).to.not.equal(null);
+
+        watchHandler('MODIFIED',  {
+            metadata: {
+                name: 'name3',
+                namespace: 'ns3',
+                resourceVersion: 'baz',
+            } as V1ObjectMeta,
+        } as V1Pod);
+        expect(cache.list().length).to.equal(3);
+        const obj3 = cache.get('name3', 'ns3');
+        expect(obj3).to.not.equal(null);
+        if (obj3) {
+            expect(obj3.metadata.name).to.equal('name3');
+            expect(obj3.metadata.resourceVersion).to.equal('baz');
+        }
+
+        watchHandler('DELETED', {
+            metadata: {
+                name: 'name2',
+                namespace: 'other-ns',
+            } as V1ObjectMeta,
+        } as V1Pod);
+        expect(cache.list().length).to.equal(3);
+        expect(cache.get('name2')).to.not.equal(null);
+
+        watchHandler('DELETED', {
+            metadata: {
+                name: 'name2',
+                namespace: 'ns2',
+            } as V1ObjectMeta,
+        } as V1Pod);
+        expect(cache.list().length).to.equal(2);
+        expect(cache.list('ns2').length).to.equal(0);
+        expect(cache.get('name2', 'ns2')).to.equal(undefined);
+    });
+    it('should delete an object correctly', () => {
+        const list: V1Pod[] = [
+            {
+                metadata: {
+                    name: 'name1',
+                    namespace: 'ns1',
+                } as V1ObjectMeta,
+            } as V1Pod,
+            {
+                metadata: {
+                    name: 'name2',
+                    namespace: 'ns2',
+                } as V1ObjectMeta,
+            } as V1Pod,
+        ];
+        deleteObject(list, {
+            metadata: {
+                name: 'other',
+                namespace: 'ns1',
+            },
+        } as V1Pod);
+        expect(list.length).to.equal(2);
+        deleteObject(list, {
+            metadata: {
+                name: 'name1',
+                namespace: 'ns2',
+            },
+        } as V1Pod);
+        expect(list.length).to.equal(2);
+        deleteObject(list, {
+            metadata: {
+                name: 'name1',
+                namespace: 'ns1',
+            },
+        } as V1Pod);
+        expect(list.length).to.equal(1);
+    });
+});
diff --git a/src/index.ts b/src/index.ts
@@ -1,7 +1,9 @@
 export * from './config';
+export * from './cache';
 export * from './api';
 export * from './attach';
 export * from './watch';
 export * from './exec';
 export * from './portforward';
+export * from './types';
 export * from './yaml';
diff --git a/src/types.ts b/src/types.ts
@@ -0,0 +1,14 @@
+import { V1ListMeta, V1ObjectMeta } from './api';
+
+export interface KubernetesObject {
+    apiVersion: string;
+    kind: string;
+    metadata: V1ObjectMeta;
+}
+
+export interface KubernetesListObject {
+    apiVersion: string;
+    kind: string;
+    metadata: V1ListMeta;
+    items: KubernetesObject;
+}
