fix(kubectl/shell): zombie pods and websocket connection error bugfixes EE-1520 (#5562)

* - fixed zombie kubectl shell pod issue
- fixed bug with websocket connection error

* exec error if not websocket disconnect

* StartExecProcess updated to write error to channel

api/http/handler/websocket/pod.go

@@ -135,21 +135,25 @@ func (handler *Handler) hijackPodExecStartOperation(
 	stdoutReader, stdoutWriter := io.Pipe()
 	defer stdoutWriter.Close()
 
+	// errorChan is used to propagate errors from the go routines to the caller.
 	errorChan := make(chan error, 1)
 	go streamFromWebsocketToWriter(websocketConn, stdinWriter, errorChan)
 	go streamFromReaderToWebsocket(websocketConn, stdoutReader, errorChan)
 
-	err = cli.StartExecProcess(serviceAccountToken, isAdminToken, namespace, podName, containerName, commandArray, stdinReader, stdoutWriter)
-	if err != nil {
-		return &httperror.HandlerError{http.StatusInternalServerError, "Unable to start exec process inside container", err}
-	}
+	// StartExecProcess is a blocking operation which streams IO to/from pod;
+	// this must execute in asynchronously, since the websocketConn could return errors (e.g. client disconnects) before
+	// the blocking operation is completed.
+	go cli.StartExecProcess(serviceAccountToken, isAdminToken, namespace, podName, containerName, commandArray, stdinReader, stdoutWriter, errorChan)
 
 	err = <-errorChan
+
+	// websocket client successfully disconnected
 	if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNoStatusReceived) {
 		log.Printf("websocket error: %s \n", err.Error())
+		return nil
 	}
 
-	return nil
+	return &httperror.HandlerError{http.StatusInternalServerError, "Unable to start exec process inside container", err}
 }
 
 func (handler *Handler) getToken(request *http.Request, endpoint *portainer.Endpoint, setLocalAdminToken bool) (string, bool, error) {

api/kubernetes/cli/exec.go

@@ -4,7 +4,7 @@ import (
 	"errors"
 	"io"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/remotecommand"
@@ -15,10 +15,12 @@ import (
 // using the specified command. The stdin parameter will be bound to the stdin process and the stdout process will write
 // to the stdout parameter.
 // This function only works against a local endpoint using an in-cluster config with the user's SA token.
-func (kcl *KubeClient) StartExecProcess(token string, useAdminToken bool, namespace, podName, containerName string, command []string, stdin io.Reader, stdout io.Writer) error {
+// This is a blocking operation.
+func (kcl *KubeClient) StartExecProcess(token string, useAdminToken bool, namespace, podName, containerName string, command []string, stdin io.Reader, stdout io.Writer, errChan chan error) {
 	config, err := rest.InClusterConfig()
 	if err != nil {
-		return err
+		errChan <- err
+		return
 	}
 
 	if !useAdminToken {
@@ -44,7 +46,8 @@ func (kcl *KubeClient) StartExecProcess(token string, useAdminToken bool, namesp
 
 	exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
 	if err != nil {
-		return err
+		errChan <- err
+		return
 	}
 
 	err = exec.Stream(remotecommand.StreamOptions{
@@ -54,9 +57,7 @@ func (kcl *KubeClient) StartExecProcess(token string, useAdminToken bool, namesp
 	})
 	if err != nil {
 		if _, ok := err.(utilexec.ExitError); !ok {
-			return errors.New("unable to start exec process")
+			errChan <- errors.New("unable to start exec process")
 		}
 	}
-
-	return nil
 }

api/portainer.go

@@ -1228,7 +1228,7 @@ type (
 		GetServiceAccount(tokendata *TokenData) (*v1.ServiceAccount, error)
 		GetServiceAccountBearerToken(userID int) (string, error)
 		CreateUserShellPod(ctx context.Context, serviceAccountName string) (*KubernetesShellPod, error)
-		StartExecProcess(token string, useAdminToken bool, namespace, podName, containerName string, command []string, stdin io.Reader, stdout io.Writer) error
+		StartExecProcess(token string, useAdminToken bool, namespace, podName, containerName string, command []string, stdin io.Reader, stdout io.Writer, errChan chan error)
 		NamespaceAccessPoliciesDeleteNamespace(namespace string) error
 		GetNodesLimits() (K8sNodesLimits, error)
 		GetNamespaceAccessPolicies() (map[string]K8sNamespaceAccessPolicy, error)