fix(k8s): reconnect port-forwards automatically

kubectl port-forward errors are now caught and the process killed, and
the Garden proxy now handles such errors. This handles cases where Pods
are terminated for whatever reason, including after re-deployment.

It would be possible to further improve this by moving to use
the API directly instead of kubectl (which I spiked and looks like quite
a bit of work), and potentially to somehow signal to the proxy that the
underlying port forward has been terminated. It works fairly well as-is
though, so I'll leave those improvements for later.

Fixes #1345

garden-service/src/plugins/kubernetes/container/deployment.ts

@@ -31,6 +31,7 @@ import { millicpuToString, kilobytesToString, prepareEnvVars, workloadTypes } fr
 import { gardenAnnotationKey } from "../../../util/string"
 import { RuntimeContext } from "../../../runtime-context"
 import { resolve } from "path"
+import { killPortForwards } from "../port-forward"
 
 export const DEFAULT_CPU_REQUEST = "10m"
 export const DEFAULT_MEMORY_REQUEST = "64Mi"
@@ -74,7 +75,12 @@ export async function deployContainerServiceRolling(
     log,
   })
 
-  return getContainerServiceStatus(params)
+  const status = await getContainerServiceStatus(params)
+
+  // Make sure port forwards work after redeployment
+  killPortForwards(service, status.forwardablePorts || [], log)
+
+  return status
 }
 
 export async function deployContainerServiceBlueGreen(
@@ -178,7 +184,13 @@ export async function deployContainerServiceBlueGreen(
       selector: `${gardenAnnotationKey("service")}=${service.name},` + `${versionKey}!=${newVersion}`,
     })
   }
-  return getContainerServiceStatus(params)
+
+  const status = await getContainerServiceStatus(params)
+
+  // Make sure port forwards work after redeployment
+  killPortForwards(service, status.forwardablePorts || [], log)
+
+  return status
 }
 
 export async function createContainerManifests(

garden-service/src/plugins/kubernetes/helm/deployment.ts

@@ -26,7 +26,7 @@ import { ContainerHotReloadSpec } from "../../container/config"
 import { getHotReloadSpec } from "./hot-reload"
 import { DeployServiceParams } from "../../../types/plugin/service/deployService"
 import { DeleteServiceParams } from "../../../types/plugin/service/deleteService"
-import { getForwardablePorts } from "../port-forward"
+import { getForwardablePorts, killPortForwards } from "../port-forward"
 
 export async function deployHelmService({
   ctx,
@@ -110,6 +110,9 @@ export async function deployHelmService({
 
   const forwardablePorts = getForwardablePorts(chartResources)
 
+  // Make sure port forwards work after redeployment
+  killPortForwards(service, forwardablePorts || [], log)
+
   return {
     forwardablePorts,
     state: "ready",

garden-service/src/plugins/kubernetes/kubernetes-module/handlers.ts

@@ -28,7 +28,7 @@ import { DeployServiceParams } from "../../../types/plugin/service/deployService
 import { DeleteServiceParams } from "../../../types/plugin/service/deleteService"
 import { GetServiceLogsParams } from "../../../types/plugin/service/getServiceLogs"
 import { gardenAnnotationKey } from "../../../util/string"
-import { getForwardablePorts, getPortForwardHandler } from "../port-forward"
+import { getForwardablePorts, getPortForwardHandler, killPortForwards } from "../port-forward"
 import { LogEntry } from "../../../logger/log-entry"
 
 export const kubernetesHandlers: Partial<ModuleAndRuntimeActionHandlers<KubernetesModule>> = {
@@ -106,7 +106,12 @@ async function deployService(params: DeployServiceParams<KubernetesModule>): Pro
     log,
   })
 
-  return getServiceStatus(params)
+  const status = await getServiceStatus(params)
+
+  // Make sure port forwards work after redeployment
+  killPortForwards(service, status.forwardablePorts || [], log)
+
+  return status
 }
 
 async function deleteService(params: DeleteServiceParams): Promise<KubernetesServiceStatus> {

garden-service/src/plugins/kubernetes/port-forward.ts

@@ -19,7 +19,7 @@ import { registerCleanupFunction } from "../../util/util"
 import { PluginContext } from "../../plugin-context"
 import { kubectl } from "./kubectl"
 import { KubernetesResource } from "./types"
-import { ForwardablePort } from "../../types/service"
+import { ForwardablePort, Service } from "../../types/service"
 import { isBuiltIn } from "./util"
 import { LogEntry } from "../../logger/log-entry"
 import { RuntimeError } from "../../exceptions"
@@ -42,15 +42,24 @@ registerCleanupFunction("kill-port-forward-procs", () => {
   }
 })
 
-export function killPortForward(targetResource: string, port: number) {
-  const key = getPortForwardKey(targetResource, port)
+export function killPortForward(targetResource: string, targetPort: number, log?: LogEntry) {
+  const key = getPortForwardKey(targetResource, targetPort)
   const fwd = registeredPortForwards[key]
   if (fwd) {
+    log?.debug(`Terminating port forward ${key}`)
     const { proc } = fwd
     !proc.killed && proc.kill()
   }
 }
 
+export function killPortForwards(service: Service, forwardablePorts: ForwardablePort[], log: LogEntry) {
+  const targetResource = getTargetResource(service)
+
+  for (const port of forwardablePorts) {
+    killPortForward(targetResource, port.targetPort, log)
+  }
+}
+
 function getPortForwardKey(targetResource: string, port: number) {
   return `${targetResource}:${port}`
 }
@@ -141,6 +150,17 @@ export async function getPortForward({
       proc.stderr!.on("data", (line) => {
         log.silly(`[${targetResource} port forwarder] ${line}`)
         output += line
+        // tslint:disable-next-line: max-line-length
+        // Following this: https://github.com/nkubala/skaffold/blob/0d52436f792b862e06311c42065afd8e2363771c/pkg/skaffold/kubernetes/portforward/kubectl_forwarder.go#L177
+        // Note: It'd be much more robust to avoid kubectl here, but it's more work to implement.
+        if (
+          line.includes("error forwarding port") ||
+          line.includes("unable to forward") ||
+          line.includes("error upgrading connection")
+        ) {
+          // Terminate the forward, which will trigger the Garden proxy to create a new one.
+          !proc.killed && proc.kill()
+        }
       })
     })
   })
@@ -154,7 +174,7 @@ export async function getPortForwardHandler({
 }: GetPortForwardParams): Promise<GetPortForwardResult> {
   const provider = ctx.provider as KubernetesProvider
   const namespace = await getAppNamespace(ctx, log, provider)
-  const targetResource = `Service/${service.name}`
+  const targetResource = getTargetResource(service)
 
   const fwd = await getPortForward({ ctx, log, namespace, targetResource, port: targetPort })
 
@@ -164,6 +184,10 @@ export async function getPortForwardHandler({
   }
 }
 
+function getTargetResource(service: Service) {
+  return `Service/${service.name}`
+}
+
 /**
  * Returns a list of forwardable ports based on the specified resources.
  */

garden-service/src/proxy.ts

@@ -59,11 +59,10 @@ async function startPortProxy(garden: Garden, log: LogEntry, service: Service, s
   return proxy
 }
 
-// TODO: handle dead port forwards
 async function createProxy(garden: Garden, log: LogEntry, service: Service, spec: ForwardablePort): Promise<PortProxy> {
   const actions = await garden.getActionRouter()
   const key = getPortKey(service, spec)
-  let fwd: GetPortForwardResult
+  let fwd: GetPortForwardResult | null = null
 
   // get preferred IP from db
   const localAddress = await LocalAddress.resolve({
@@ -117,12 +116,9 @@ async function createProxy(garden: Garden, log: LogEntry, service: Service, spec
 
     const getRemote = async () => {
       if (!_remote) {
-        const { hostname, port } = await getPortForward()
-
-        log.debug(`Connecting to ${key} port forward at ${hostname}:${port}`)
+        const forwardResult = await getPortForward()
 
         _remote = new Socket()
-        _remote.connect(port, hostname)
 
         _remote.on("data", (data) => {
           if (!local.writable) {
@@ -145,8 +141,18 @@ async function createProxy(garden: Garden, log: LogEntry, service: Service, spec
 
         _remote.on("error", (err) => {
           log.debug(`Remote socket error: ${err.message}`)
+          // Existing port forward doesn't seem to be healthy, retry forward on next connection
+          // TODO: it would be nice (but much more intricate) to hold the local connection open while reconnecting,
+          // plus we don't really know if the connection is dead until a connection attempt is made.
+          fwd = null
         })
 
+        if (forwardResult) {
+          const { port, hostname } = forwardResult
+          log.debug(`Connecting to ${key} port forward at ${hostname}:${port}`)
+          _remote.connect(port, hostname)
+        }
+
         // Local connection was closed while remote connection was being created
         if (localDidClose) {
           _remote.end()