Merge pull request #646 from StefanCenusa/feature/listwatch-labelselector

Watch with labelSelector

Author: k8s-ci-robot

examples/typescript/informer/informer-with-label-selector.ts

@@ -0,0 +1,83 @@
+// tslint:disable:no-console
+import * as k8s from '@kubernetes/client-node';
+
+const kc = new k8s.KubeConfig();
+kc.loadFromDefault();
+
+const k8sApi = kc.makeApiClient(k8s.CoreV1Api);
+
+const APP_LABEL_SELECTOR = 'app=foo';
+
+const listFn = () => k8sApi.listNamespacedPod(
+    'default',
+    undefined,
+    undefined,
+    undefined,
+    undefined,
+    APP_LABEL_SELECTOR,
+);
+
+const createPod = async (name, app) => {
+    const appPodContainer = {
+        name: 'nginx',
+        image: 'nginx:latest',
+    } as k8s.V1Container;
+
+    const appPod = {
+        metadata: {
+            name,
+            labels: {
+                app,
+            },
+        },
+        spec: {
+            containers: [appPodContainer],
+        },
+    } as k8s.V1Pod;
+    await k8sApi.createNamespacedPod('default', appPod).catch((e) => console.error(e));
+    console.log('create', name);
+};
+
+const deletePod = async (name, namespace) => {
+    await k8sApi.deleteNamespacedPod(name, namespace);
+    console.log('delete', name);
+};
+
+const delay = (ms) => {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+};
+
+const informer = k8s.makeInformer(
+    kc,
+    '/api/v1/namespaces/default/pods',
+    listFn,
+    APP_LABEL_SELECTOR,
+);
+
+informer.on('add', (obj: k8s.V1Pod) => {
+    console.log(`Added: ${obj.metadata!.name}`);
+});
+informer.on('update', (obj: k8s.V1Pod) => {
+    console.log(`Updated: ${obj.metadata!.name}`);
+});
+informer.on('delete', (obj: k8s.V1Pod) => {
+    console.log(`Deleted: ${obj.metadata!.name}`);
+});
+informer.on('error', (err: k8s.V1Pod) => {
+    console.error(err);
+    // Restart informer after 5sec
+    setTimeout(() => {
+        informer.start();
+    }, 5000);
+});
+
+informer.start().then(() => {
+    setTimeout(async () => {
+        await createPod('server-foo', 'foo');
+        await delay(5000);
+        await createPod('server-bar', 'bar');
+        await delay(5000);
+        await deletePod('server-foo', 'default');
+        await deletePod('server-bar', 'default');
+    }, 5000);
+});

src/cache.ts

@@ -1,3 +1,4 @@
+import { ObjectSerializer } from './api';
 import {
     ADD,
     CHANGE,
@@ -15,6 +16,7 @@ import { RequestResult, Watch } from './watch';
 
 export interface ObjectCache<T> {
     get(name: string, namespace?: string): T | undefined;
+
     list(namespace?: string): ReadonlyArray<T>;
 }
 
@@ -31,6 +33,7 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
         private readonly watch: Watch,
         private readonly listFn: ListPromise<T>,
         autoStart: boolean = true,
+        private readonly labelSelector?: string,
     ) {
         this.callbackCache[ADD] = [];
         this.callbackCache[UPDATE] = [];
@@ -142,9 +145,18 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
             }
         });
         this.addOrUpdateItems(list.items);
+        const queryParams = {
+            resourceVersion: list.metadata!.resourceVersion,
+        } as {
+            resourceVersion: string | undefined;
+            labelSelector: string | undefined;
+        };
+        if (this.labelSelector !== undefined) {
+            queryParams.labelSelector = ObjectSerializer.serialize(this.labelSelector, 'string');
+        }
         this.request = await this.watch.watch(
             this.path,
-            { resourceVersion: list.metadata!.resourceVersion },
+            queryParams,
             this.watchHandler.bind(this),
             this.doneHandler.bind(this),
         );

src/cache_test.ts

@@ -1,4 +1,5 @@
 import { expect, use } from 'chai';
+import * as request from 'request';
 import chaiAsPromised = require('chai-as-promised');
 
 import * as mock from 'ts-mockito';
@@ -9,18 +10,46 @@ import { EventEmitter } from 'ws';
 
 import { V1Namespace, V1NamespaceList, V1ObjectMeta, V1Pod, V1ListMeta } from './api';
 import { deleteObject, ListWatch, deleteItems } from './cache';
-import { ListPromise } from './informer';
+import { KubeConfig } from './config';
+import { Cluster, Context, User } from './config_types';
+import { ADD, UPDATE, DELETE, ERROR, ListPromise, CHANGE } from './informer';
 
 use(chaiAsPromised);
 
-import { RequestResult, Watch } from './watch';
+import { DefaultRequest, RequestResult, Watch } from './watch';
 
 // Object replacing real Request object in the test
 class FakeRequest extends EventEmitter implements RequestResult {
     pipe(stream: Duplex): void {}
     abort() {}
 }
 
+const server = 'foo.company.com';
+
+const fakeConfig: {
+    clusters: Cluster[];
+    contexts: Context[];
+    users: User[];
+} = {
+    clusters: [
+        {
+            name: 'cluster',
+            server,
+        } as Cluster,
+    ],
+    contexts: [
+        {
+            cluster: 'cluster',
+            user: 'user',
+        } as Context,
+    ],
+    users: [
+        {
+            name: 'user',
+        } as User,
+    ],
+};
+
 describe('ListWatchCache', () => {
     it('should throw on unknown update', () => {
         const fake = mock.mock(Watch);
@@ -1024,52 +1053,67 @@ describe('ListWatchCache', () => {
         ).once();
         expect(errorEmitted).to.equal(true);
     });
-});
 
-describe('delete items', () => {
-    it('should remove correctly', () => {
-        const listA: V1Pod[] = [
+    it('should send label selector', async () => {
+        const APP_LABEL_SELECTOR = 'app=foo';
+
+        const list: V1Namespace[] = [
             {
                 metadata: {
                     name: 'name1',
-                    namespace: 'ns1',
+                    labels: {
+                        app: 'foo',
+                    },
                 } as V1ObjectMeta,
-            } as V1Pod,
+            } as V1Namespace,
             {
                 metadata: {
                     name: 'name2',
-                    namespace: 'ns2',
-                } as V1ObjectMeta,
-            } as V1Pod,
-        ];
-        const listB: V1Pod[] = [
-            {
-                metadata: {
-                    name: 'name1',
-                    namespace: 'ns1',
+                    labels: {
+                        app: 'foo',
+                    },
                 } as V1ObjectMeta,
-            } as V1Pod,
-            {
-                metadata: {
-                    name: 'name3',
-                    namespace: 'ns3',
-                } as V1ObjectMeta,
-            } as V1Pod,
-        ];
-        const expected: V1Pod[] = [
-            {
-                metadata: {
-                    name: 'name1',
-                    namespace: 'ns1',
-                } as V1ObjectMeta,
-            } as V1Pod,
+            } as V1Namespace,
         ];
+        const listObj = {
+            metadata: {
+                resourceVersion: '12345',
+            } as V1ListMeta,
+            items: list,
+        } as V1NamespaceList;
 
-        const output = deleteItems(listA, listB);
-        expect(output).to.deep.equal(expected);
+        const listFn: ListPromise<V1Namespace> = function(): Promise<{
+            response: http.IncomingMessage;
+            body: V1NamespaceList;
+        }> {
+            return new Promise<{ response: http.IncomingMessage; body: V1NamespaceList }>(
+                (resolve, reject) => {
+                    resolve({ response: {} as http.IncomingMessage, body: listObj });
+                },
+            );
+        };
+
+        const kc = new KubeConfig();
+        Object.assign(kc, fakeConfig);
+        const fakeRequestor = mock.mock(DefaultRequest);
+        const watch = new Watch(kc, mock.instance(fakeRequestor));
+
+        const fakeRequest = new FakeRequest();
+        mock.when(fakeRequestor.webRequest(mock.anything())).thenReturn(fakeRequest);
+
+        const informer = new ListWatch('/some/path', watch, listFn, false, APP_LABEL_SELECTOR);
+
+        await informer.start();
+
+        mock.verify(fakeRequestor.webRequest(mock.anything()));
+        const [opts] = mock.capture(fakeRequestor.webRequest).last();
+        const reqOpts: request.OptionsWithUri = opts as request.OptionsWithUri;
+        expect(reqOpts.qs.labelSelector).to.equal(APP_LABEL_SELECTOR);
     });
+});
 
-    it('should callback correctly', () => {
+describe('delete items', () => {
+    it('should remove correctly', () => {
         const listA: V1Pod[] = [
             {
                 metadata: {

src/informer.ts

@@ -35,7 +35,8 @@ export function makeInformer<T>(
     kubeconfig: KubeConfig,
     path: string,
     listPromiseFn: ListPromise<T>,
+    labelSelector?: string,
 ): Informer<T> {
     const watch = new Watch(kubeconfig);
-    return new ListWatch<T>(path, watch, listPromiseFn, false);
+    return new ListWatch<T>(path, watch, listPromiseFn, false, labelSelector);
 }