Adding reversePort function

packages/kdot/lib/commands/fwd.js

@@ -5,6 +5,7 @@ import { oneLine } from 'common-tags'
 import { PortForward, kc } from '../k8s.js'
 import getRunningPods from '../getRunningPods.js'
 import configure from '../configure/index.js'
+import reversePort from '../reversePort.js'
 
 const logger = createLogger({ namespace: 'kdot.fwd', level: 'info' })
 const pollConfig = { interval: 1000, timeout: 300000, limit: 1 }
@@ -96,7 +97,15 @@ export default async function fwd (input) {
         const pod = await getRunningPods(namespace, app.name, pollConfig)
         for (const [name, portConfig] of Object.entries(app.ports)) {
           portConfig.name = name
-          await forwardPort(app, pod, portConfig)
+          if (portConfig.reversePort) {
+            await reversePort({
+              ktunnelPort: app.ports.grpc?.port,
+              port: portConfig.port,
+              reversePort: portConfig.reversePort
+            })
+          } else {
+            await forwardPort(app, pod, portConfig)
+          }
         }
       } catch (err) {
         logger.error(err)

packages/kdot/lib/reversePort.js

@@ -0,0 +1,76 @@
+import net from 'net'
+import grpc from '@grpc/grpc-js'
+import protoLoader from '@grpc/proto-loader'
+import { createLogger } from '@generates/logger'
+
+const logger = createLogger({ level: 'info', namespace: 'kdot.fwd' })
+
+// Read the protocol buffer.
+const packageDefinition = protoLoader.loadSync(
+  './tunnel.proto',
+  {
+    keepCase: true,
+    longs: String,
+    enums: String,
+    defaults: true,
+    oneofs: true
+  }
+)
+
+// Load the tunnel RPC service.
+const { grpc_tunnel: tunnel } = grpc.loadPackageDefinition(packageDefinition)
+
+// Create the default credentials.
+const credentials = grpc.credentials.createInsecure()
+
+export default function reversePort (config) {
+  const { ktunnelPort = '28688' } = config
+
+  // Create the gRPC client to communicate with the ktunnel server.
+  const client = new tunnel.Tunnel(`localhost:${ktunnelPort}`, credentials)
+
+  // Initialize the tunnel to get the duplex stream used to communicate.
+  const stream = client.initTunnel()
+
+  // Open the connection to ktunnel and tell it what port to listen on by
+  // sending an initial message on the stream.
+  stream.write({ port: config.port })
+
+  // Listen for requests from ktunnel.
+  stream.on('data', ({ requestId, data, shouldClose }) => {
+    logger.debug('Request', { requestId })
+
+    if (shouldClose) {
+      // If the client closed the connection, close the session.
+      logger.debug('Close request', { requestId })
+      stream.write({ requestId, shouldClose })
+    } else {
+      // Otherwise forward the request to the local server.
+      const client = net.createConnection({ port: config.reversePort }, () => {
+        logger.debug('Connected to server', { requestId })
+
+        // Forward the data received from ktunnel to the local server.
+        client.write(data, err => {
+          logger.debug('Client write', { requestId })
+          if (err) {
+            logger.error('Client write error', requestId, err)
+            stream.write({ requestId, hasErr: true })
+          }
+        })
+      })
+
+      // Pass data returned by the local server to ktunnel.
+      client.on('data', async data => {
+        logger.debug('Client data', { requestId })
+        stream.write({ requestId, data })
+      })
+
+      // When the local server closes the connection, tell ktunnel to close it's
+      // connection to the requesting client.
+      client.on('end', () => {
+        logger.debug('Client end', { requestId })
+        stream.write({ requestId, shouldClose: true })
+      })
+    }
+  })
+}

packages/kdot/package.json

@@ -50,6 +50,8 @@
     "@generates/merger": "^0.1.3",
     "@generates/package-updater": "^0.0.2",
     "@generates/prompt": "^0.0.2",
+    "@grpc/grpc-js": "^1.2.11",
+    "@grpc/proto-loader": "^0.5.6",
     "@ianwalter/bff": "^10.0.5",
     "@ianwalter/requester": "^2.5.0",
     "@kubernetes/client-node": "^0.14.0",
@@ -72,5 +74,9 @@
     "test": {
       "v": "0.0.1"
     }
+  },
+  "optionalDependencies": {
+    "bufferutil": "^4.0.3",
+    "utf-8-validate": "^5.0.4"
   }
 }