Add optional websocket keepalive pings for idle Exec sessions

src/exec.ts

@@ -7,6 +7,10 @@ import { KubeConfig } from './config.js';
 import { isResizable, ResizableStream, TerminalSizeQueue } from './terminal-size-queue.js';
 import { WebSocketHandler, WebSocketInterface } from './web-socket-handler.js';
 
+export interface ExecOptions {
+    pingIntervalMs?: number;
+}
+
 export class Exec {
     public 'handler': WebSocketInterface;
 
@@ -39,6 +43,7 @@ export class Exec {
         stdin: stream.Readable | null,
         tty: boolean,
         statusCallback?: (status: V1Status) => void,
+        options?: ExecOptions,
     ): Promise<WebSocket.WebSocket> {
         const query = {
             stdout: stdout != null,
@@ -60,6 +65,13 @@ export class Exec {
             }
             return true;
         });
+        const pingIntervalMs = options?.pingIntervalMs;
+        if (pingIntervalMs !== undefined) {
+            if (!Number.isInteger(pingIntervalMs) || pingIntervalMs <= 0) {
+                throw new Error('pingIntervalMs must be a positive integer');
+            }
+            this.setupPing(conn, pingIntervalMs);
+        }
         if (stdin != null) {
             WebSocketHandler.handleStandardInput(conn, stdin, WebSocketHandler.StdinStream);
         }
@@ -70,4 +82,37 @@ export class Exec {
         }
         return conn;
     }
+
+    private setupPing(conn: WebSocket.WebSocket, pingIntervalMs: number): void {
+        const socket = conn as WebSocket.WebSocket & {
+            ping?: () => void;
+            on?: (event: string, listener: (...args: unknown[]) => void) => void;
+            removeListener?: (event: string, listener: (...args: unknown[]) => void) => void;
+        };
+        if (typeof socket.ping !== 'function') {
+            return;
+        }
+
+        let awaitingPong = false;
+        const timer = setInterval(() => {
+            if (conn.readyState === WebSocket.OPEN) {
+                if (!awaitingPong) {
+                    awaitingPong = true;
+                    socket.ping!();
+                }
+            }
+        }, pingIntervalMs);
+        const onPong = () => {
+            awaitingPong = false;
+        };
+        const clearKeepAlive = () => {
+            clearInterval(timer);
+            socket.removeListener?.('pong', onPong);
+        };
+        if (typeof socket.on === 'function') {
+            socket.on('pong', onPong);
+            socket.on('close', clearKeepAlive);
+            socket.on('error', clearKeepAlive);
+        }
+    }
 }

src/exec_test.ts

@@ -1,5 +1,6 @@
 import { describe, it } from 'node:test';
-import { deepStrictEqual, strictEqual } from 'node:assert';
+import { deepStrictEqual, ok, strictEqual } from 'node:assert';
+import { setTimeout as setTimeoutPromise } from 'node:timers/promises';
 import WebSocket from 'isomorphic-ws';
 import { ReadableStreamBuffer, WritableStreamBuffer } from 'stream-buffers';
 import { anyFunction, anything, capture, instance, mock, verify, when } from 'ts-mockito';
@@ -156,5 +157,44 @@ describe('Exec', () => {
             await closePromise;
             verify(fakeWebSocket.close()).called();
         });
+
+        it('should optionally send websocket pings', async () => {
+            const pingIntervalMs = 5;
+            const waitForPingsMs = 20;
+            const kc = new KubeConfig();
+            const pingHandlers: Record<string, () => void> = {};
+            let pingCount = 0;
+            const fakeConn = {
+                readyState: WebSocket.OPEN,
+                ping: () => {
+                    pingCount++;
+                },
+                on: (event: string, listener: () => void) => {
+                    pingHandlers[event] = listener;
+                },
+            } as unknown as WebSocket.WebSocket;
+            const ws: WebSocketInterface = {
+                connect: async () => fakeConn,
+            };
+            const exec = new Exec(kc, ws);
+
+            await exec.exec('ns', 'pod', 'container', 'command', null, null, null, false, undefined, {
+                pingIntervalMs,
+            });
+            await setTimeoutPromise(waitForPingsMs);
+
+            strictEqual(pingCount, 1);
+            strictEqual(typeof pingHandlers.pong, 'function');
+            pingHandlers.pong();
+            await setTimeoutPromise(waitForPingsMs);
+            ok(pingCount > 1);
+
+            strictEqual(typeof pingHandlers.close, 'function');
+            pingHandlers.close();
+            const pingCountAtClose = pingCount;
+            await setTimeoutPromise(waitForPingsMs);
+
+            strictEqual(pingCount, pingCountAtClose);
+        });
     });
 });