feat: Implement tunnel-based APIs (Exec, Attach, Portforward) (#6)

* fix(generation): Implement quirked argv parameters

It seems like Kubernetes should OpenAPI should be changed to present
this argument properly. In any non-trivial case, it will be a list.

* Collapse param building loop for lists

* Properly implement PodExec with examples

* Improve PodExec interface and examples

* Add missing method on PortforwardTunnel

* Drop Deno v1.22 from CI - lacks Deno.consoleSize()

* Update /x/kubernetes_client to v0.7.0

* Put 'tunnel' into tunnel API names

* Fix array arg of portforward API too

* Get PortForward going with WebSockets

* Rename ChannelTunnel to StdioTunnel

* Add a basic test for each tunnel utility class

* Make test green on somewhat older Denos (v1.28)

* Update README

Author: danopia

.github/workflows/deno-ci.yaml

@@ -13,7 +13,6 @@ jobs:
     strategy:
       matrix:
         deno-version:
-        - v1.22
         - v1.28
         - v1.32
         - v1.36
@@ -49,3 +48,6 @@ jobs:
 
     - name: Check lib/examples/*.ts
       run: time deno check lib/examples/*.ts
+
+    - name: Test
+      run: time deno test

generation/codegen-mod.ts

@@ -21,6 +21,7 @@ export function generateModuleTypescript(surface: SurfaceMap, api: SurfaceApi):
   chunks.push(`import * as c from "../../common.ts";`);
   chunks.push(`import * as operations from "../../operations.ts";`);
   chunks.push(`import * as ${api.friendlyName} from "./structs.ts";`);
+  const tunnelsImport = `import * as tunnels from "../../tunnels.ts";`;
   chunks.push('');
 
   const foreignApis = new Set<SurfaceApi>();
@@ -114,6 +115,9 @@ export function generateModuleTypescript(surface: SurfaceMap, api: SurfaceApi):
       } else throw new Error(`Unknown param.in ${param.in}`);
     }
 
+    let funcName = op.operationName;
+    let expectsTunnel: 'PortforwardTunnel' | 'StdioTunnel' | null = null;
+
     // Entirely specialcase and collapse each method's proxy functions into one
     if (op['x-kubernetes-action'] === 'connect' && op.operationName.endsWith('Proxy')) {
       if (op.method !== 'get') return; // only emit the GET function, and make it generic
@@ -135,6 +139,29 @@ export function generateModuleTypescript(surface: SurfaceMap, api: SurfaceApi):
       chunks.push(`    return await this.#client.performRequest({ ...opts, path });`);
       chunks.push(`  }\n`);
       return;
+
+    // Specialcase bidirectionally-tunneled APIs (these use either SPDY/3.1 or WebSockets)
+    } else if (op['x-kubernetes-action'] === 'connect') {
+      if (op.method !== 'get') return; // only emit the GET function, method doesn't matter at this level
+
+      const middleName = op.operationName.slice('connectGet'.length);
+      funcName = `tunnel${middleName}`;
+
+      if (middleName == 'PodAttach' || middleName == 'PodExec') {
+        expectsTunnel = 'StdioTunnel';
+        // Make several extra params required
+        const commandArg = opts.find(x => x[0].name == 'command');
+        if (commandArg) commandArg[0].required = true;
+        const stdoutArg = opts.find(x => x[0].name == 'stdout');
+        if (stdoutArg) stdoutArg[0].required = true;
+      }
+      if (middleName == 'PodPortforward') {
+        expectsTunnel = 'PortforwardTunnel';
+      }
+
+      if (!expectsTunnel) {
+        throw new Error(`TODO: connect action was unexpected: ${funcName}`);
+      }
     }
 
     let accept = 'application/json';
@@ -158,7 +185,7 @@ export function generateModuleTypescript(surface: SurfaceMap, api: SurfaceApi):
     //   return AcmeCertManagerIoV1.toOrder(resp);
     // }
 
-    chunks.push(`  async ${op.operationName}(${writeSig(args, opts, '  ')}) {`);
+    chunks.push(`  async ${funcName}(${writeSig(args, opts, '  ')}) {`);
     const isWatch = op.operationName.startsWith('watch');
     const isStream = op.operationName.startsWith('stream');
 
@@ -186,6 +213,17 @@ export function generateModuleTypescript(surface: SurfaceMap, api: SurfaceApi):
           case 'number':
             chunks.push(`    ${maybeIf}query.append(${idStr}, String(opts[${idStr}]));`);
             break;
+          case 'list': {
+            const loop = `for (const item of opts[${idStr}]${opt[0].required ? '' : ' ?? []'}) `;
+            if (opt[1].inner.type == 'string') {
+              chunks.push(`    ${loop}query.append(${idStr}, item);`);
+              break;
+            }
+            if (opt[1].inner.type == 'number') {
+              chunks.push(`    ${loop}query.append(${idStr}, String(item));`);
+              break;
+            }
+          } /* falls through */
           default:
             chunks.push(`    // TODO: ${opt[0].in} ${opt[0].name} ${opt[0].required} ${opt[0].type} ${JSON.stringify(opt[1])}`);
         }
@@ -195,7 +233,10 @@ export function generateModuleTypescript(surface: SurfaceMap, api: SurfaceApi):
     chunks.push(`    const resp = await this.#client.performRequest({`);
     chunks.push(`      method: ${JSON.stringify(op.method.toUpperCase())},`);
     chunks.push(`      path: \`\${this.#root}${JSON.stringify(opPath).slice(1,-1).replace(/{/g, '${')}\`,`);
-    if (accept === 'application/json') {
+    if (expectsTunnel) {
+      if (!chunks.includes(tunnelsImport)) chunks.splice(6, 0, tunnelsImport);
+      chunks.push(`      expectTunnel: tunnels.${expectsTunnel}.supportedProtocols,`);
+    } else if (accept === 'application/json') {
       chunks.push(`      expectJson: true,`);
     }
     if (isWatch || isStream) {
@@ -219,6 +260,13 @@ export function generateModuleTypescript(surface: SurfaceMap, api: SurfaceApi):
     chunks.push(`      abortSignal: opts.abortSignal,`);
     chunks.push(`    });`);
 
+    if (expectsTunnel) {
+      chunks.push(`\n    const tunnel = new tunnels.${expectsTunnel}(resp, query);`);
+      chunks.push(`    await tunnel.ready;`);
+      chunks.push(`    return tunnel;`);
+      chunks.push(`  }\n`);
+      return;
+    }
     if (isStream) {
       if (accept === 'text/plain') {
         chunks.push(`    return resp.pipeThrough(new TextDecoderStream('utf-8'));`);
@@ -285,6 +333,8 @@ export function generateModuleTypescript(surface: SurfaceMap, api: SurfaceApi):
         return `${api.friendlyName}.${shape.reference}`;
       case 'foreign':
         return `${shape.api.friendlyName}.${shape.name}`;
+      case 'list':
+        return `Array<${writeType(shape.inner)}>`;
       case 'special':
         return `c.${shape.name}`;
     }

generation/describe-surface.ts

@@ -174,6 +174,28 @@ export function describeSurface(wholeSpec: OpenAPI2) {
         const allParams = new Array<OpenAPI2RequestParameter>()
           .concat(methodObj.parameters ?? [], pathObj.parameters ?? []);
 
+        // Special-case for PodExec/PodAttach which do not type 'command' as a list.
+        const commandArg = allParams.find(x => x.name == 'command' && x.description?.includes('argv array'));
+        if (commandArg) {
+          commandArg.schema = {
+            type: 'array',
+            items: {
+              type: 'string',
+            },
+          };
+          commandArg.type = undefined;
+        }
+        const portArg = allParams.find(x => x.name == 'ports' && x.description?.includes('List of ports'));
+        if (portArg) {
+          portArg.schema = {
+            type: 'array',
+            items: {
+              type: 'number',
+            },
+          };
+          portArg.type = undefined;
+        }
+
         if (opName == 'getPodLog') {
           // Add a streaming variant for pod logs
           api.operations.push({

lib/README.md

@@ -32,8 +32,13 @@ see `/x/kubernetes_client` for more information.
 
 ## Changelog
 
-* `v0.5.0` on `2023-08-??`:
-  * Updating `/x/kubernetes_client` API contract to `v0.6.0`.
+* `v0.5.0` on `2023-08-19`:
+  * Updating `/x/kubernetes_client` API contract to `v0.7.0`.
+  * Actually implement PodExec, PodAttach, PodPortForward APIs with a new tunnel implementation.
+  * Includes 'builtin' APIs generated from K8s `v1.28.0`.
+    * New APIs: `admissionregistration.k8s.io/v1beta1`, `certificates.k8s.io/v1alpha1`.
+    * Also, API additions for sidecar containers and `SelfSubjectReview`.
+  * Fix several structures incorrectly typed as `{}` instead of `JSONValue`.
 
 * `v0.4.0` on `2023-02-10`:
   * Updating `/x/kubernetes_client` API contract to `v0.5.0`.

lib/builtin/core@v1/mod.ts

@@ -8,6 +8,7 @@ import * as AutoscalingV1 from "../autoscaling@v1/structs.ts";
 import * as PolicyV1 from "../policy@v1/structs.ts";
 import * as MetaV1 from "../meta@v1/structs.ts";
 import * as CoreV1 from "./structs.ts";
+import * as tunnels from "../../tunnels.ts";
 
 export class CoreV1Api {
   #client: c.RestClient;
@@ -1332,50 +1333,31 @@ export class CoreV1NamespacedApi {
     return CoreV1.toPod(resp);
   }
 
-  async connectGetPodAttach(name: string, opts: {
+  async tunnelPodAttach(name: string, opts: {
     container?: string;
     stderr?: boolean;
     stdin?: boolean;
-    stdout?: boolean;
+    stdout: boolean;
     tty?: boolean;
     abortSignal?: AbortSignal;
-  } = {}) {
+  }) {
     const query = new URLSearchParams;
     if (opts["container"] != null) query.append("container", opts["container"]);
     if (opts["stderr"] != null) query.append("stderr", opts["stderr"] ? '1' : '0');
     if (opts["stdin"] != null) query.append("stdin", opts["stdin"] ? '1' : '0');
-    if (opts["stdout"] != null) query.append("stdout", opts["stdout"] ? '1' : '0');
+    query.append("stdout", opts["stdout"] ? '1' : '0');
     if (opts["tty"] != null) query.append("tty", opts["tty"] ? '1' : '0');
     const resp = await this.#client.performRequest({
       method: "GET",
       path: `${this.#root}pods/${name}/attach`,
-      expectJson: true,
+      expectTunnel: tunnels.StdioTunnel.supportedProtocols,
       querystring: query,
       abortSignal: opts.abortSignal,
     });
-  }
 
-  async connectPostPodAttach(name: string, opts: {
-    container?: string;
-    stderr?: boolean;
-    stdin?: boolean;
-    stdout?: boolean;
-    tty?: boolean;
-    abortSignal?: AbortSignal;
-  } = {}) {
-    const query = new URLSearchParams;
-    if (opts["container"] != null) query.append("container", opts["container"]);
-    if (opts["stderr"] != null) query.append("stderr", opts["stderr"] ? '1' : '0');
-    if (opts["stdin"] != null) query.append("stdin", opts["stdin"] ? '1' : '0');
-    if (opts["stdout"] != null) query.append("stdout", opts["stdout"] ? '1' : '0');
-    if (opts["tty"] != null) query.append("tty", opts["tty"] ? '1' : '0');
-    const resp = await this.#client.performRequest({
-      method: "POST",
-      path: `${this.#root}pods/${name}/attach`,
-      expectJson: true,
-      querystring: query,
-      abortSignal: opts.abortSignal,
-    });
+    const tunnel = new tunnels.StdioTunnel(resp, query);
+    await tunnel.ready;
+    return tunnel;
   }
 
   async createPodBinding(name: string, body: CoreV1.Binding, opts: operations.PutOpts = {}) {
@@ -1437,54 +1419,33 @@ export class CoreV1NamespacedApi {
     return PolicyV1.toEviction(resp);
   }
 
-  async connectGetPodExec(name: string, opts: {
-    command?: string;
+  async tunnelPodExec(name: string, opts: {
+    command: Array<string>;
     container?: string;
     stderr?: boolean;
     stdin?: boolean;
-    stdout?: boolean;
+    stdout: boolean;
     tty?: boolean;
     abortSignal?: AbortSignal;
-  } = {}) {
+  }) {
     const query = new URLSearchParams;
-    if (opts["command"] != null) query.append("command", opts["command"]);
+    for (const item of opts["command"]) query.append("command", item);
     if (opts["container"] != null) query.append("container", opts["container"]);
     if (opts["stderr"] != null) query.append("stderr", opts["stderr"] ? '1' : '0');
     if (opts["stdin"] != null) query.append("stdin", opts["stdin"] ? '1' : '0');
-    if (opts["stdout"] != null) query.append("stdout", opts["stdout"] ? '1' : '0');
+    query.append("stdout", opts["stdout"] ? '1' : '0');
     if (opts["tty"] != null) query.append("tty", opts["tty"] ? '1' : '0');
     const resp = await this.#client.performRequest({
       method: "GET",
       path: `${this.#root}pods/${name}/exec`,
-      expectJson: true,
+      expectTunnel: tunnels.StdioTunnel.supportedProtocols,
       querystring: query,
       abortSignal: opts.abortSignal,
     });
-  }
 
-  async connectPostPodExec(name: string, opts: {
-    command?: string;
-    container?: string;
-    stderr?: boolean;
-    stdin?: boolean;
-    stdout?: boolean;
-    tty?: boolean;
-    abortSignal?: AbortSignal;
-  } = {}) {
-    const query = new URLSearchParams;
-    if (opts["command"] != null) query.append("command", opts["command"]);
-    if (opts["container"] != null) query.append("container", opts["container"]);
-    if (opts["stderr"] != null) query.append("stderr", opts["stderr"] ? '1' : '0');
-    if (opts["stdin"] != null) query.append("stdin", opts["stdin"] ? '1' : '0');
-    if (opts["stdout"] != null) query.append("stdout", opts["stdout"] ? '1' : '0');
-    if (opts["tty"] != null) query.append("tty", opts["tty"] ? '1' : '0');
-    const resp = await this.#client.performRequest({
-      method: "POST",
-      path: `${this.#root}pods/${name}/exec`,
-      expectJson: true,
-      querystring: query,
-      abortSignal: opts.abortSignal,
-    });
+    const tunnel = new tunnels.StdioTunnel(resp, query);
+    await tunnel.ready;
+    return tunnel;
   }
 
   async streamPodLog(name: string, opts: {
@@ -1544,34 +1505,23 @@ export class CoreV1NamespacedApi {
     return new TextDecoder('utf-8').decode(resp);
   }
 
-  async connectGetPodPortforward(name: string, opts: {
-    ports?: number;
+  async tunnelPodPortforward(name: string, opts: {
+    ports?: Array<number>;
     abortSignal?: AbortSignal;
   } = {}) {
     const query = new URLSearchParams;
-    if (opts["ports"] != null) query.append("ports", String(opts["ports"]));
+    for (const item of opts["ports"] ?? []) query.append("ports", String(item));
     const resp = await this.#client.performRequest({
       method: "GET",
       path: `${this.#root}pods/${name}/portforward`,
-      expectJson: true,
+      expectTunnel: tunnels.PortforwardTunnel.supportedProtocols,
       querystring: query,
       abortSignal: opts.abortSignal,
     });
-  }
 
-  async connectPostPodPortforward(name: string, opts: {
-    ports?: number;
-    abortSignal?: AbortSignal;
-  } = {}) {
-    const query = new URLSearchParams;
-    if (opts["ports"] != null) query.append("ports", String(opts["ports"]));
-    const resp = await this.#client.performRequest({
-      method: "POST",
-      path: `${this.#root}pods/${name}/portforward`,
-      expectJson: true,
-      querystring: query,
-      abortSignal: opts.abortSignal,
-    });
+    const tunnel = new tunnels.PortforwardTunnel(resp, query);
+    await tunnel.ready;
+    return tunnel;
   }
 
   proxyPodRequest(podName: string, opts: c.ProxyOptions & {expectStream: true; expectJson: true}): Promise<ReadableStream<c.JSONValue>>;

lib/client.ts

@@ -3,3 +3,4 @@
 // so this is provided an optional utility (as opposed to deps.ts)
 
 export * from "https://deno.land/x/kubernetes_client@v0.7.0/mod.ts";
+export * as tunnelBeta from "https://deno.land/x/kubernetes_client@v0.7.0/tunnel-beta/via-websocket.ts";

lib/examples/pod-exec-output.ts

@@ -0,0 +1,26 @@
+#!/usr/bin/env -S deno run --allow-net --allow-read --allow-env --unstable
+
+import { tunnelBeta, makeClientProviderChain } from '../client.ts';
+import { CoreV1Api } from '../builtin/core@v1/mod.ts';
+
+// Set up an experimental client which can use Websockets
+const client = await makeClientProviderChain(tunnelBeta.WebsocketRestClient).getClient();
+const coreApi = new CoreV1Api(client);
+
+// Launch a process into a particular container
+const tunnel = await coreApi
+  .namespace('media')
+  .tunnelPodExec('sabnzbd-srv-0', {
+    command: ['uname', '-a'],
+    stdout: true,
+    stderr: true,
+  });
+
+// Buffer & print the contents of stdout
+const output = await tunnel.output();
+console.log(new TextDecoder().decode(output.stdout).trimEnd());
+
+// Print any error that occurred
+if (output.status !== 'Success') {
+  console.error(output.message);
+}