Add Port forwarding.

Author: brendandburns

examples/typescript/port-forward/port-forward.ts

@@ -0,0 +1,19 @@
+import k8s = require('@kubernetes/client-node');
+import * as net from 'net';
+
+const command = process.argv[2];
+
+const kc = new k8s.KubeConfig();
+kc.loadFromDefault();
+
+const forward = new k8s.PortForward(kc);
+
+// This simple server just forwards traffic from itself to a service running in kubernetes
+// -> localhost:8080 -> port-forward-tunnel -> kubernetes-pod
+// This is basically equivalent to 'kubectl port-forward ...' but in TypeScript.
+const server = net.createServer(function(socket) {
+    forward.portForward('default', 'simple-node-simple-node-d49cc9d69-8ls5q', [8000], socket, null, socket);
+});
+
+server.listen(8080, '127.0.0.1');
+

node-client/package.json

@@ -26,7 +26,7 @@
     ],
     "exclude": [
       "src/api.ts",
-      "src/config_test.ts"
+      "src/*_test.ts"
     ],
     "extension": [
       ".ts"
@@ -71,6 +71,8 @@
     "mock-fs": "^4.6.0",
     "nyc": "^12.0.2",
     "source-map-support": "^0.5.9",
+    "stream-buffers": "^3.0.2",
+    "ts-mockito": "^2.3.1",
     "ts-node": "^3.1.0",
     "tslint": "^5.10.0",
     "typescript": "^2.6.2"

node-client/src/attach.ts

@@ -23,8 +23,9 @@ export class Attach {
         };
         const queryStr = querystring.stringify(query);
         const path = `/api/v1/namespaces/${namespace}/pods/${podName}/attach?${queryStr}`;
-        const promise = this.handler.connect(path, () => { return; }, (streamNum: number, buff: Buffer) => {
+        const promise = this.handler.connect(path, null, (streamNum: number, buff: Buffer): boolean => {
             WebSocketHandler.handleStandardStreams(streamNum, buff, stdout, stderr);
+            return true;
         });
         const result = new Promise<void>((resolvePromise, reject) => {
             promise.then(() => resolvePromise(), (err) => reject(err));

node-client/src/exec.ts

@@ -26,8 +26,9 @@ export class Exec {
         };
         const queryStr = querystring.stringify(query);
         const path = `/api/v1/namespaces/${namespace}/pods/${podName}/exec?${queryStr}`;
-        const conn = await this.handler.connect(path, (text) => { return; }, (streamNum: number, buff: Buffer) => {
+        const conn = await this.handler.connect(path, null, (streamNum: number, buff: Buffer): boolean => {
             WebSocketHandler.handleStandardStreams(streamNum, buff, stdout, stderr);
+            return true;
         });
         if (stdin != null) {
             WebSocketHandler.handleStandardInput(conn, stdin);

node-client/src/index.ts

@@ -3,3 +3,4 @@ export * from './api';
 export * from './attach';
 export * from './watch';
 export * from './exec';
+export * from './portforward';

node-client/src/portforward.ts

@@ -0,0 +1,65 @@
+import querystring = require('querystring');
+import stream = require('stream');
+import ws = require('websocket');
+
+import { KubeConfig } from './config';
+import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
+
+export class PortForward {
+    private readonly handler: WebSocketInterface;
+    private readonly disconnectOnErr: boolean;
+
+    // handler is a parameter really only for injecting for testing.
+    constructor(config: KubeConfig, disconnectOnErr?: boolean, handler?: WebSocketInterface) {
+        if (!handler) {
+            this.handler = new WebSocketHandler(config);
+        } else {
+            this.handler = handler;
+        }
+        this.disconnectOnErr = true;
+    }
+
+    // TODO: support multiple ports for real...
+    public async portForward(namespace: string, podName: string, targetPorts: number[],
+                             output: stream.Writable, err: stream.Writable,
+                             input: stream.Readable): Promise<ws.connection> {
+        if (targetPorts.length === 0) {
+            throw new Error('You must provide at least one port to forward to.');
+        }
+        if (targetPorts.length > 1) {
+            throw(new Error('ERROR: Only one port is currently supported for port-forward'));
+        }
+        const query = {
+            ports: targetPorts[0],
+        };
+        const queryStr = querystring.stringify(query);
+        const needsToReadPortNumber: boolean[] = [];
+        targetPorts.forEach((value: number, index: number) => {
+            needsToReadPortNumber[index * 2] = true;
+            needsToReadPortNumber[index * 2 + 1] = true;
+        });
+        const path = `/api/v1/namespaces/${namespace}/pods/${podName}/portforward?${queryStr}`;
+        const conn = await this.handler.connect(path, null, (streamNum: number, buff: Buffer | string): boolean => {
+            if (streamNum >= targetPorts.length * 2) {
+                if (this.disconnectOnErr) {
+                    return false;
+                }
+            }
+            // First two bytes of each stream are the port number
+            if (needsToReadPortNumber[streamNum]) {
+                buff = buff.slice(2);
+                needsToReadPortNumber[streamNum] = false;
+            }
+            if (streamNum % 2 === 1) {
+                if (err) {
+                    err.write(buff);
+                }
+            } else {
+                output.write(buff);
+            }
+            return true;
+        });
+        WebSocketHandler.handleStandardInput(conn, input, 0);
+        return conn;
+    }
+}

node-client/src/portforward_test.ts

@@ -0,0 +1,70 @@
+import { expect } from 'chai';
+import { ReadableStreamBuffer, WritableStreamBuffer } from 'stream-buffers';
+import { anyFunction, capture, instance, mock, reset, verify } from 'ts-mockito';
+
+import { KubeConfig } from './config';
+import { PortForward } from './portforward';
+import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
+
+describe('PortForward', () => {
+    describe('basic', () => {
+        it('should correctly port-forward to a url', async () => {
+            const kc = new KubeConfig();
+            const fakeWebSocket: WebSocketInterface = mock(WebSocketHandler);
+            const portForward = new PortForward(kc, true, instance(fakeWebSocket));
+            const osStream = new WritableStreamBuffer();
+            const errStream = new WritableStreamBuffer();
+            const isStream = new ReadableStreamBuffer();
+
+            const namespace = 'somenamespace';
+            const pod = 'somepod';
+            const port = 8080;
+
+            await portForward.portForward(
+                namespace, pod, [port], osStream, errStream, isStream);
+
+            const path = `/api/v1/namespaces/${namespace}/pods/${pod}/portforward?ports=${port}`;
+            verify(fakeWebSocket.connect(path, null,  anyFunction())).called();
+        });
+
+        it('should correctly port-forward streams', async () => {
+            const kc = new KubeConfig();
+            const fakeWebSocket: WebSocketInterface = mock(WebSocketHandler);
+            const portForward = new PortForward(kc, true, instance(fakeWebSocket));
+            const osStream = new WritableStreamBuffer();
+            const errStream = new WritableStreamBuffer();
+            const isStream = new ReadableStreamBuffer();
+
+            await portForward.portForward(
+                'ns', 'p', [8000], osStream, errStream, isStream);
+
+            const [ , , outputFn] = capture(fakeWebSocket.connect).last();
+
+            /* tslint:disable:no-unused-expression */
+            expect(outputFn).to.not.be.null;
+            // this is redundant but needed for the compiler, sigh...
+            if (outputFn) {
+                const buffer = Buffer.alloc(1024, 10);
+
+                outputFn(0, buffer);
+                // first time, drop two bytes for the port number.
+                expect(osStream.size()).to.equal(1022);
+
+                outputFn(0, buffer);
+                expect(osStream.size()).to.equal(2046);
+
+                // error stream, drop two bytes for the port number.
+                outputFn(1, buffer);
+                expect(errStream.size()).to.equal(1022);
+
+                outputFn(1, buffer);
+                expect(errStream.size()).to.equal(2046);
+
+                // unknown stream, shouldn't change anything.
+                outputFn(2, buffer);
+                expect(osStream.size()).to.equal(2046);
+                expect(errStream.size()).to.equal(2046);
+            }
+        });
+    });
+});

node-client/src/proto-client.ts

@@ -0,0 +1,36 @@
+import http = require('http');
+import url = require('url');
+
+import { KubeConfig } from './config';
+
+export class ProtoClient {
+    public readonly 'config': KubeConfig;
+
+    public async get(msgType: any, requestPath: string): Promise<any> {
+        const server = this.config.getCurrentCluster().server;
+        const u = new url.URL(server);
+        const options = {
+            path: requestPath,
+            hostname: u.hostname,
+            protocol: u.protocol,
+        };
+        this.config.applytoHTTPSOptions(options);
+        const req = http.request(options);
+
+        const result = new Promise<any>((resolve, reject) => {
+            let data = '';
+            req.on('data', (chunk) => {
+                data = data + chunk;
+            });
+            req.on('end', () => {
+                const obj = msgType.deserializeBinary(data);
+                resolve(obj);
+            });
+            req.on('error', (err) => {
+                reject(err);
+            });
+        });
+        req.end();
+        return result;
+    }
+}

node-client/src/web-socket-handler.ts

@@ -1,7 +1,7 @@
 import https = require('https');
 import stream = require('stream');
-
 import ws = require('websocket');
+
 import { V1Status } from './api';
 import { KubeConfig } from './config';
 
@@ -12,13 +12,20 @@ const protocols = [
     'channel.k8s.io',
 ];
 
-export class WebSocketHandler {
+export interface WebSocketInterface {
+    connect(path: string,
+            textHandler: ((text: string) => boolean) | null,
+            binaryHandler: ((stream: number, buff: Buffer) => boolean) | null): Promise<ws.connection>;
+}
+
+export class WebSocketHandler implements WebSocketInterface {
     public static readonly StdinStream = 0;
     public static readonly StdoutStream = 1;
     public static readonly StderrStream = 2;
     public static readonly StatusStream = 3;
 
-    public static handleStandardStreams(streamNum: number, buff: Buffer, stdout: any, stderr: any): V1Status | null {
+    public static handleStandardStreams(streamNum: number, buff: Buffer,
+                                        stdout: stream.Writable, stderr: stream.Writable): V1Status | null {
         if (buff.length < 1) {
             return null;
         }
@@ -41,10 +48,11 @@ export class WebSocketHandler {
         return null;
     }
 
-    public static handleStandardInput(conn: ws.connection, stdin: stream.Readable | any) {
+    public static handleStandardInput(conn: ws.connection,
+                                      stdin: stream.Readable | any, streamNum: number = 0): boolean {
         stdin.on('data', (data) => {
             const buff = Buffer.alloc(data.length + 1);
-            buff.writeInt8(0, 0);
+            buff.writeInt8(streamNum, 0);
             if (data instanceof Buffer) {
                 data.copy(buff, 1);
             } else {
@@ -56,6 +64,8 @@ export class WebSocketHandler {
         stdin.on('end', () => {
             conn.close();
         });
+        // Keep the stream open
+        return true;
     }
 
     public 'config': KubeConfig;
@@ -64,9 +74,17 @@ export class WebSocketHandler {
         this.config = config;
     }
 
+    /**
+     * Connect to a web socket endpoint.
+     * @param path The HTTP Path to connect to on the server.
+     * @param textHandler Callback for text over the web socket.
+     *      Returns true if the connection should be kept alive, false to disconnect.
+     * @param binaryHandler Callback for binary data over the web socket.
+     *      Returns true if the connection should be kept alive, false to disconnect.
+     */
     public connect(path: string,
-                   textHandler: (text: string) => void,
-                   binaryHandler: (stream: number, buff: Buffer) => void): Promise<ws.connection> {
+                   textHandler: ((text: string) => boolean) | null,
+                   binaryHandler: ((stream: number, buff: Buffer) => boolean) | null): Promise<ws.connection> {
 
         const server = this.config.getCurrentCluster().server;
         const ssl = server.startsWith('https://');
@@ -78,19 +96,23 @@ export class WebSocketHandler {
         // TODO: This doesn't set insecureSSL if skipTLSVerify is set...
         this.config.applytoHTTPSOptions(opts);
 
-        const client = new ws.client({ tlsOptions: opts } );
+        const client = new ws.client({ tlsOptions: opts });
 
         return new Promise((resolve, reject) => {
             client.on('connect', (connection) => {
                 connection.on('message', (message: ws.IMessage) => {
                     if (message.type === 'utf8' && message.utf8Data) {
                         if (textHandler) {
-                            textHandler(message.utf8Data);
+                            if (!textHandler(message.utf8Data)) {
+                                connection.close();
+                            }
                         }
                     } else if (message.type === 'binary' && message.binaryData) {
                         if (binaryHandler) {
                             const streamNum = message.binaryData.readInt8(0);
-                            binaryHandler(streamNum, message.binaryData.slice(1));
+                            if (!binaryHandler(streamNum, message.binaryData.slice(1))) {
+                                connection.close();
+                            }
                         }
                     }
                 });

node-client/src/web-socket-handler_test.ts

@@ -0,0 +1,38 @@
+import { expect } from 'chai';
+import { ReadableStreamBuffer, WritableStreamBuffer } from 'stream-buffers';
+import { anyFunction, capture, instance, mock, reset, verify } from 'ts-mockito';
+
+import { KubeConfig } from './config';
+import { PortForward } from './portforward';
+import { WebSocketHandler, WebSocketInterface } from './web-socket-handler';
+
+describe('WebSocket', () => {
+    it('should handle output streams', () => {
+        const osStream = new WritableStreamBuffer();
+        const errStream = new WritableStreamBuffer();
+
+        const fill1 = 1;
+        const fill2 = 2;
+
+        const buff1 = Buffer.alloc(1024, fill1);
+        const buff2 = Buffer.alloc(512, fill2);
+
+        WebSocketHandler.handleStandardStreams(WebSocketHandler.StdoutStream, buff1, osStream, errStream);
+
+        expect(osStream.size()).to.equal(1024);
+        expect(errStream.size()).to.equal(0);
+
+        WebSocketHandler.handleStandardStreams(WebSocketHandler.StderrStream, buff2, osStream, errStream);
+
+        expect(osStream.size()).to.equal(1024);
+        expect(errStream.size()).to.equal(512);
+
+        for (let i = 0; i < 1024; i++) {
+            expect(osStream.getContents()[0]).to.equal(fill1);
+        }
+
+        for (let i = 0; i < 512; i++) {
+            expect(errStream.getContents()[0]).to.equal(fill2);
+        }
+    });
+});