Merge pull request #1605 from Ace-Tang/exec

bugfix: fix exec stuck when exec get error

apis/server/exec_bridge.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/go-openapi/strfmt"
 	"github.com/gorilla/mux"
+	"github.com/sirupsen/logrus"
 )
 
 func (s *Server) createContainerExec(ctx context.Context, rw http.ResponseWriter, req *http.Request) error {
@@ -55,8 +56,13 @@ func (s *Server) startContainerExec(ctx context.Context, rw http.ResponseWriter,
 	name := mux.Vars(req)["name"]
 	_, upgrade := req.Header["Upgrade"]
 
+	if err := s.ContainerMgr.CheckExecExist(ctx, name); err != nil {
+		return err
+	}
+
 	var attach *mgr.AttachConfig
 
+	// TODO(huamin.thm): support detach exec process through http post method
 	if !config.Detach {
 		hijacker, ok := rw.(http.Hijacker)
 		if !ok {
@@ -72,7 +78,11 @@ func (s *Server) startContainerExec(ctx context.Context, rw http.ResponseWriter,
 		}
 	}
 
-	return s.ContainerMgr.StartExec(ctx, name, config, attach)
+	if err := s.ContainerMgr.StartExec(ctx, name, attach); err != nil {
+		logrus.Errorf("failed to run exec process: %s", err)
+	}
+
+	return nil
 }
 
 func (s *Server) getExecInfo(ctx context.Context, rw http.ResponseWriter, req *http.Request) error {

apis/server/router.go

@@ -186,6 +186,7 @@ func filter(handler handler, s *Server) http.HandlerFunc {
 			return
 		}
 		// Handle error if request handling fails.
+		logrus.Errorf("Handler for %s %s, client %s returns error: %s", req.Method, req.URL.RequestURI(), clientInfo, err)
 		HandleErrorResponse(w, err)
 	}
 }

cli/exec.go

@@ -1,15 +1,17 @@
 package main
 
 import (
+	"bufio"
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"os"
-	"sync"
 
 	"github.com/alibaba/pouch/apis/types"
-	"github.com/docker/docker/pkg/stdcopy"
 
+	"github.com/docker/docker/pkg/stdcopy"
+	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 )
 
@@ -61,6 +63,7 @@ func (e *ExecCommand) runExec(args []string) error {
 	id := args[0]
 	command := args[1:]
 
+	// TODO(huamin.thm): exec detach not implement now, detach mode not hijack connect
 	createExecConfig := &types.ExecCreateConfig{
 		Cmd:          command,
 		Tty:          e.Terminal,
@@ -85,53 +88,81 @@ func (e *ExecCommand) runExec(args []string) error {
 
 	conn, reader, err := apiClient.ContainerStartExec(ctx, createResp.ID, startExecConfig)
 	if err != nil {
-		return fmt.Errorf("failed to create exec: %v", err)
+		return fmt.Errorf("failed to start exec: %v", err)
 	}
-	defer conn.Close()
 
 	// handle stdio.
-	var wg sync.WaitGroup
-
-	if createExecConfig.AttachStderr || createExecConfig.AttachStdout {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			if !e.Terminal {
-				stdcopy.StdCopy(os.Stdout, os.Stderr, reader)
-			} else {
-				io.Copy(os.Stdout, reader)
+	if err := holdHijackConnection(ctx, conn, reader, createExecConfig.AttachStdin, createExecConfig.AttachStdout, createExecConfig.AttachStderr, e.Terminal); err != nil {
+		return err
+	}
+
+	execInfo, err := apiClient.ContainerExecInspect(ctx, createResp.ID)
+	if err != nil {
+		return err
+	}
+
+	code := execInfo.ExitCode
+	if code != 0 {
+		return ExitError{Code: int(code)}
+	}
+
+	return nil
+}
+
+func holdHijackConnection(ctx context.Context, conn net.Conn, reader *bufio.Reader, stdin, stdout, stderr, tty bool) error {
+	if stdin && tty {
+		in, out, err := setRawMode(true, false)
+		if err != nil {
+			return fmt.Errorf("failed to set raw mode")
+		}
+		defer func() {
+			if err := restoreMode(in, out); err != nil {
+				fmt.Fprintf(os.Stderr, "failed to restore term mode")
 			}
 		}()
 	}
 
-	if createExecConfig.AttachStdin {
-		if e.Terminal {
-			in, out, err := setRawMode(true, false)
-			if err != nil {
-				return fmt.Errorf("failed to set raw mode")
+	stdoutDone := make(chan error, 1)
+	go func() {
+		var err error
+		if stderr || stdout {
+			if !tty {
+				_, err = stdcopy.StdCopy(os.Stdout, os.Stderr, reader)
+			} else {
+				_, err = io.Copy(os.Stdout, reader)
 			}
-			defer func() {
-				if err := restoreMode(in, out); err != nil {
-					fmt.Fprintf(os.Stderr, "failed to restore term mode")
-				}
-			}()
 		}
+		stdoutDone <- err
+	}()
 
-		go func() {
+	stdinDone := make(chan struct{})
+	go func() {
+		if stdin {
 			io.Copy(conn, os.Stdin)
-		}()
-	}
+		}
 
-	wg.Wait()
+		// TODO: close write side of conn
+		close(stdinDone)
+	}()
 
-	execInfo, err := apiClient.ContainerExecInspect(ctx, createResp.ID)
-	if err != nil {
-		return err
-	}
+	select {
+	case err := <-stdoutDone:
+		if err != nil {
+			logrus.Debugf("receive stdout error: %s", err)
+			return err
+		}
 
-	code := execInfo.ExitCode
-	if code != 0 {
-		return ExitError{Code: int(code)}
+	case <-stdinDone:
+		if stdout || stderr {
+			select {
+			case err := <-stdoutDone:
+				logrus.Debugf("receive stdout error: %s", err)
+				return err
+			case <-ctx.Done():
+			}
+		}
+
+	case <-ctx.Done():
 	}
 
 	return nil

cri/v1alpha1/cri.go

@@ -781,9 +781,7 @@ func (c *CriManager) ExecSync(ctx context.Context, r *runtime.ExecSyncRequest) (
 		MuxDisabled: true,
 	}
 
-	startConfig := &apitypes.ExecStartConfig{}
-
-	err = c.ContainerMgr.StartExec(ctx, execid, startConfig, attachConfig)
+	err = c.ContainerMgr.StartExec(ctx, execid, attachConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to start exec for container %q: %v", id, err)
 	}

cri/v1alpha1/cri_stream.go

@@ -57,13 +57,12 @@ func (s *streamRuntime) Exec(containerID string, cmd []string, streamOpts *remot
 		return 0, fmt.Errorf("failed to create exec for container %q: %v", containerID, err)
 	}
 
-	startConfig := &apitypes.ExecStartConfig{}
 	attachConfig := &mgr.AttachConfig{
 		Streams:     streams,
 		MuxDisabled: true,
 	}
 
-	err = s.containerMgr.StartExec(ctx, execid, startConfig, attachConfig)
+	err = s.containerMgr.StartExec(ctx, execid, attachConfig)
 	if err != nil {
 		return 0, fmt.Errorf("failed to start exec for container %q: %v", containerID, err)
 	}

cri/v1alpha2/cri.go

@@ -787,9 +787,7 @@ func (c *CriManager) ExecSync(ctx context.Context, r *runtime.ExecSyncRequest) (
 		MuxDisabled: true,
 	}
 
-	startConfig := &apitypes.ExecStartConfig{}
-
-	err = c.ContainerMgr.StartExec(ctx, execid, startConfig, attachConfig)
+	err = c.ContainerMgr.StartExec(ctx, execid, attachConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to start exec for container %q: %v", id, err)
 	}

cri/v1alpha2/cri_stream.go

@@ -57,13 +57,12 @@ func (s *streamRuntime) Exec(containerID string, cmd []string, streamOpts *remot
 		return 0, fmt.Errorf("failed to create exec for container %q: %v", containerID, err)
 	}
 
-	startConfig := &apitypes.ExecStartConfig{}
 	attachConfig := &mgr.AttachConfig{
 		Streams:     streams,
 		MuxDisabled: true,
 	}
 
-	err = s.containerMgr.StartExec(ctx, execid, startConfig, attachConfig)
+	err = s.containerMgr.StartExec(ctx, execid, attachConfig)
 	if err != nil {
 		return 0, fmt.Errorf("failed to start exec for container %q: %v", containerID, err)
 	}

daemon/mgr/container.go

@@ -108,14 +108,17 @@ type ContainerMgr interface {
 	CreateExec(ctx context.Context, name string, config *types.ExecCreateConfig) (string, error)
 
 	// StartExec executes a new process in container.
-	StartExec(ctx context.Context, execid string, config *types.ExecStartConfig, attach *AttachConfig) error
+	StartExec(ctx context.Context, execid string, attach *AttachConfig) error
 
 	// InspectExec returns low-level information about exec command.
 	InspectExec(ctx context.Context, execid string) (*types.ContainerExecInspect, error)
 
 	// GetExecConfig returns execonfig of a exec process inside container.
 	GetExecConfig(ctx context.Context, execid string) (*ContainerExecConfig, error)
 
+	// CheckExecExist check if exec process `name` exist
+	CheckExecExist(ctx context.Context, name string) error
+
 	// 3. The following two function is related to network management.
 	// TODO: inconsistency, Connect/Disconnect operation is in newtork_bridge.go in upper API layer.
 	// Here we encapsualted them in container manager, inconsistency exists.

daemon/mgr/container_exec.go

@@ -3,12 +3,14 @@ package mgr
 import (
 	"context"
 	"fmt"
+	"io"
 
 	"github.com/alibaba/pouch/apis/types"
 	"github.com/alibaba/pouch/ctrd"
 	"github.com/alibaba/pouch/pkg/errtypes"
 	"github.com/alibaba/pouch/pkg/randomid"
 
+	"github.com/docker/docker/pkg/stdcopy"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
 )
@@ -37,25 +39,40 @@ func (mgr *ContainerManager) CreateExec(ctx context.Context, name string, config
 }
 
 // StartExec executes a new process in container.
-func (mgr *ContainerManager) StartExec(ctx context.Context, execid string, config *types.ExecStartConfig, attach *AttachConfig) error {
-	v, ok := mgr.ExecProcesses.Get(execid).Result()
-	if !ok {
-		return errors.Wrap(errtypes.ErrNotfound, "to be exec process: "+execid)
+func (mgr *ContainerManager) StartExec(ctx context.Context, execid string, attach *AttachConfig) (err error) {
+	// GetExecConfig should not error, since we have done this before call StartExec
+	execConfig, err := mgr.GetExecConfig(ctx, execid)
+	if err != nil {
+		return err
 	}
-	execConfig, ok := v.(*ContainerExecConfig)
-	if !ok {
-		return fmt.Errorf("invalid exec config type")
+
+	eio, err := mgr.openExecIO(execid, attach)
+	if err != nil {
+		return err
 	}
 
+	defer func() {
+		if err != nil {
+			var stdout io.Writer = eio.Stdout
+			if !execConfig.Tty && !eio.MuxDisabled {
+				stdout = stdcopy.NewStdWriter(stdout, stdcopy.Stdout)
+			}
+			stdout.Write([]byte(err.Error() + "\r\n"))
+			// close io to make hijack connection exit
+			eio.Close()
+			mgr.IOs.Remove(execid)
+			// set exec exit status
+			execConfig.Running = false
+			exitCode := 126
+			execConfig.ExitCode = int64(exitCode)
+		}
+		mgr.ExecProcesses.Put(execid, execConfig)
+	}()
+
 	if attach != nil {
 		attach.Stdin = execConfig.AttachStdin
 	}
 
-	io, err := mgr.openExecIO(execid, attach)
-	if err != nil {
-		return err
-	}
-
 	c, err := mgr.container(execConfig.ContainerID)
 	if err != nil {
 		return err
@@ -86,19 +103,11 @@ func (mgr *ContainerManager) StartExec(ctx context.Context, execid string, confi
 	c.Unlock()
 
 	execConfig.Running = true
-	defer func() {
-		if err != nil {
-			execConfig.Running = false
-			exitCode := 126
-			execConfig.ExitCode = int64(exitCode)
-		}
-		mgr.ExecProcesses.Put(execid, execConfig)
-	}()
 
 	err = mgr.Client.ExecContainer(ctx, &ctrd.Process{
 		ContainerID: execConfig.ContainerID,
 		ExecID:      execid,
-		IO:          io,
+		IO:          eio,
 		P:           process,
 	})
 
@@ -144,6 +153,12 @@ func (mgr *ContainerManager) GetExecConfig(ctx context.Context, execid string) (
 	return execConfig, nil
 }
 
+// CheckExecExist check if exec process `name` exist
+func (mgr *ContainerManager) CheckExecExist(ctx context.Context, name string) error {
+	_, err := mgr.GetExecConfig(ctx, name)
+	return err
+}
+
 func (mgr *ContainerManager) getEntrypointAndArgs(cmd []string) (string, []string) {
 	if len(cmd) == 0 {
 		return "", []string{}

test/cli_exec_test.go

@@ -143,3 +143,11 @@ func (suite *PouchExecSuite) TestExecExitCode(c *check.C) {
 	command.PouchRun("exec", name, "sh", "-c", "exit 101").Assert(c, icmd.Expected{ExitCode: 101})
 	command.PouchRun("exec", name, "sh", "-c", "exit 0").Assert(c, icmd.Success)
 }
+
+// TestExecFail test exec fail should not hang.
+func (suite *PouchExecSuite) TestExecFail(c *check.C) {
+	name := "TestExecFail"
+	res := command.PouchRun("run", "-d", "--name", name, "-u", name, busyboxImage, "top")
+	defer DelContainerForceMultyTime(c, name)
+	c.Assert(res.Stderr(), check.NotNil)
+}