@gempesaw feat(pod-exec): add initial support for command execution

examples/pod-exec.js

@@ -0,0 +1,39 @@
+//
+// Execute commands non-interactively in a pod
+//
+const Client = require('kubernetes-client').Client;
+const config = require('kubernetes-client').config;
+
+async function main() {
+  try {
+
+    const client = new Client({ config: config.fromKubeconfig(), version: '1.9' });
+
+    // Pod with single container
+    let res = await client.api.v1.namespaces('namespace_name').pods('pod_name').exec.post({
+      qs: {
+        command: ['ls', '-al'],
+        stdout: true,
+        stderr: true
+      }
+    });
+    console.log(res.body);
+    console.log(res.messages);
+
+    // Pod with multiple containers /must/ specify a container
+    res = await client.api.v1.namespaces('namespace_name').pods('pod_name').exec.post({
+      qs: {
+        command: ['ls', '-al'],
+        container: 'container_name',
+        stdout: true,
+        stderr: true
+      }
+    });
+    console.log(res.body);
+
+  } catch (err) {
+    console.error('Error: ', err);
+  }
+}
+
+main();

lib/request.js

@@ -1,6 +1,8 @@
 'use strict';
 
 const request = require('request');
+const qs = require('qs');
+const WebSocket = require('ws');
 
 /**
  * Refresh whatever authentication {type} is.
@@ -23,6 +25,63 @@ function refreshAuth(type, config) {
   });
 }
 
+const execChannels = [
+  'stdin',
+  'stdout',
+  'stderr',
+  'error',
+  'resize'
+];
+
+/**
+ * Determine whether a failed Kubernetes API response is asking for an upgrade
+ * @param {object} body - response body object from Kubernetes
+ * @property {string} status - request status
+ * @property {number} code - previous request's response code
+ * @property {message} message - previous request response message
+ * @returns {boolean} Upgrade the request
+ */
+
+function isUpgradeRequired(body) {
+  return body.status === 'Failure'
+    && body.code === 400
+    && body.message === 'Upgrade request required';
+}
+
+/**
+ * Upgrade a request into a Websocket transaction & process the result
+ * @param {ApiRequestOptions} options - Options object
+ * @param {callback} cb - The callback that handles the response
+ */
+
+function upgradeRequest(options, cb) {
+  const queryParams = qs.stringify(options.qs, { indices: false });
+  const wsUrl = `${options.baseUrl}/${options.uri}?${queryParams}`;
+  const protocol = 'base64.channel.k8s.io';
+  const ws = new WebSocket(wsUrl, protocol, options);
+
+  const messages = [];
+  ws.on('message', (msg) => {
+    const channel = execChannels[msg.slice(0, 1)];
+    const message = Buffer.from(msg.slice(1), 'base64').toString('ascii');
+    messages.push({ channel, message });
+  });
+
+  ws.on('error', (err) => {
+    err.messages = messages;
+    cb(err, messages);
+  });
+
+  ws.on('close', (code, reason) => cb(null, {
+    messages,
+    body: messages.map(({ message }) => message).join(''),
+    code,
+    reason
+  }));
+
+  return ws;
+}
+
 
 class Request {
   /**
@@ -92,6 +151,10 @@ class Request {
     return request(requestOptions, (err, res, body) => {
       if (err) return cb(err);
 
+      if (isUpgradeRequired(body)) {
+        return upgradeRequest(requestOptions, cb);
+      }
+
       // Refresh auth if 401
       if (res.statusCode === 401 && auth.type) {
         return refreshAuth(auth.type, auth.config)

package.json

@@ -36,7 +36,9 @@
     "fluent-openapi": "0.1.1",
     "js-yaml": "^3.10.0",
     "openid-client": "^2.0.0",
-    "request": "^2.83.0"
+    "qs": "^6.5.2",
+    "request": "^2.83.0",
+    "ws": "^6.0.0"
   },
   "devDependencies": {
     "@types/node": "^10.3.5",