feature: add exec timeout support

Signed-off-by: zhangyue <zy675793960@yeah.net>
（cherry pick from #2946）

Signed-off-by: huaihuan.zy <huaihuan.zy@alibaba-inc.com>

apis/server/exec_bridge.go

@@ -101,7 +101,7 @@ func (s *Server) startContainerExec(ctx context.Context, rw http.ResponseWriter,
 		}
 	}
 
-	if err := s.ContainerMgr.StartExec(ctx, name, attach); err != nil {
+	if err := s.ContainerMgr.StartExec(ctx, name, attach, 0); err != nil {
 		if config.Detach {
 			return err
 		}

cri/stream/runtime.go

@@ -69,7 +69,7 @@ func (s *streamRuntime) Exec(ctx context.Context, containerID string, cmd []stri
 		Terminal:  createConfig.Tty,
 	}
 
-	if err := s.containerMgr.StartExec(ctx, execid, attachCfg); err != nil {
+	if err := s.containerMgr.StartExec(ctx, execid, attachCfg, 0); err != nil {
 		return 0, fmt.Errorf("failed to exec for container %q: %v", containerID, err)
 	}
 

cri/v1alpha2/cri.go

@@ -1193,15 +1193,6 @@ func (c *CriManager) ReopenContainerLog(ctx context.Context, r *runtime.ReopenCo
 func (c *CriManager) ExecSync(ctx context.Context, r *runtime.ExecSyncRequest) (*runtime.ExecSyncResponse, error) {
 	id := r.GetContainerId()
 
-	timeout := time.Duration(r.GetTimeout()) * time.Second
-	var cancel context.CancelFunc
-	if timeout == 0 {
-		ctx, cancel = context.WithCancel(ctx)
-	} else {
-		ctx, cancel = context.WithTimeout(ctx, timeout)
-	}
-	defer cancel()
-
 	createConfig := &apitypes.ExecCreateConfig{
 		Cmd: r.GetCmd(),
 	}
@@ -1217,7 +1208,8 @@ func (c *CriManager) ExecSync(ctx context.Context, r *runtime.ExecSyncRequest) (
 		UseStderr: true,
 		Stderr:    stderrBuf,
 	}
-	if err := c.ContainerMgr.StartExec(ctx, execid, attachCfg); err != nil {
+
+	if err := c.ContainerMgr.StartExec(ctx, execid, attachCfg, int(r.GetTimeout())); err != nil {
 		return nil, fmt.Errorf("failed to start exec for container %q: %v", id, err)
 	}
 

ctrd/container.go

@@ -80,27 +80,22 @@ func (c *Client) containerStats(ctx context.Context, id string) (*containerdtype
 }
 
 // ExecContainer executes a process in container.
-func (c *Client) ExecContainer(ctx context.Context, process *Process) error {
-	if err := c.execContainer(ctx, process); err != nil {
+func (c *Client) ExecContainer(ctx context.Context, process *Process, timeout int) error {
+	if err := c.execContainer(ctx, process, timeout); err != nil {
 		return convertCtrdErr(err)
 	}
 	return nil
 }
 
 // execContainer executes a process in container.
-func (c *Client) execContainer(ctx context.Context, process *Process) error {
+func (c *Client) execContainer(ctx context.Context, process *Process, timeout int) error {
 	pack, err := c.watch.get(process.ContainerID)
 	if err != nil {
 		return err
 	}
 
 	closeStdinCh := make(chan struct{})
 
-	// make sure the closeStdinCh has been closed.
-	defer func() {
-		close(closeStdinCh)
-	}()
-
 	var (
 		cntrID, execID          = pack.container.ID(), process.ExecID
 		withStdin, withTerminal = process.IO.Stream().Stdin() != nil, process.P.Terminal
@@ -131,29 +126,10 @@ func (c *Client) execContainer(ctx context.Context, process *Process) error {
 		return errors.Wrap(err, "failed to exec process")
 	}
 
-	errCh := make(chan error, 1)
-	defer close(errCh)
-
-	go func() {
-		var msg *Message
-
-		if startErr := <-errCh; startErr != nil {
-			msg = &Message{
-				err:      startErr,
-				exitCode: 126,
-				exitTime: time.Now().UTC(),
-			}
-		}
-
-		// success to start which means the cmd is valid and wait
-		// for process.
+	var msg *Message
+	defer func() {
 		if msg == nil {
-			status := <-exitStatus
-			msg = &Message{
-				err:      status.Error(),
-				exitCode: status.ExitCode(),
-				exitTime: status.ExitTime(),
-			}
+			return
 		}
 
 		// XXX: if exec process get run, io should be closed in this function,
@@ -168,13 +144,53 @@ func (c *Client) execContainer(ctx context.Context, process *Process) error {
 		if _, err := execProcess.Delete(context.TODO()); err != nil {
 			logrus.Warnf("failed to delete exec process %s: %s", process.ExecID, err)
 		}
+
 	}()
 
 	// start the exec process
 	if err := execProcess.Start(ctx); err != nil {
-		errCh <- err
-		return err
+		msg = &Message{
+			err:      err,
+			exitCode: 126,
+			exitTime: time.Now().UTC(),
+		}
+		return errors.Wrapf(err, "failed to start exec, exec id %s", execID)
+	}
+	// make sure the closeStdinCh has been closed.
+	close(closeStdinCh)
+
+	t := time.Duration(timeout) * time.Second
+	var timeCh <-chan time.Time
+	if t == 0 {
+		timeCh = make(chan time.Time)
+	} else {
+		timeCh = time.After(t)
+	}
+
+	select {
+	case status := <-exitStatus:
+		msg = &Message{
+			err:      status.Error(),
+			exitCode: status.ExitCode(),
+			exitTime: status.ExitTime(),
+		}
+	case <-timeCh:
+		// ignore the not found error because the process may exit itself before kill
+		if err := execProcess.Kill(ctx, syscall.SIGKILL); err != nil && !errdefs.IsNotFound(err) {
+			// try to force kill the exec process
+			if err := execProcess.Kill(ctx, syscall.SIGTERM); err != nil && !errdefs.IsNotFound(err) {
+				return errors.Wrapf(err, "failed to kill the exec process")
+			}
+		}
+		// wait for process to be killed
+		status := <-exitStatus
+		msg = &Message{
+			err:      errors.Wrapf(status.Error(), "failed to exec process %s, timeout", execID),
+			exitCode: status.ExitCode(),
+			exitTime: status.ExitTime(),
+		}
 	}
+
 	return nil
 }
 

ctrd/interface.go

@@ -46,7 +46,7 @@ type ContainerAPIClient interface {
 	// ContainerStats returns stats of the container.
 	ContainerStats(ctx context.Context, id string) (*containerdtypes.Metric, error)
 	// ExecContainer executes a process in container.
-	ExecContainer(ctx context.Context, process *Process) error
+	ExecContainer(ctx context.Context, process *Process, timeout int) error
 	// ResizeContainer changes the size of the TTY of the exec process running
 	// in the container to the given height and width.
 	ResizeExec(ctx context.Context, id string, execid string, opts types.ResizeOptions) error

daemon/mgr/container.go

@@ -122,7 +122,7 @@ 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, cfg *streams.AttachConfig) error
+	StartExec(ctx context.Context, execid string, cfg *streams.AttachConfig, timeout int) error
 
 	// InspectExec returns low-level information about exec command.
 	InspectExec(ctx context.Context, execid string) (*types.ContainerExecInspect, error)

daemon/mgr/container_exec.go

@@ -58,7 +58,8 @@ func (mgr *ContainerManager) ResizeExec(ctx context.Context, execid string, opts
 }
 
 // StartExec executes a new process in container.
-func (mgr *ContainerManager) StartExec(ctx context.Context, execid string, cfg *streams.AttachConfig) (err0 error) {
+// timeout = 0 means no timeout
+func (mgr *ContainerManager) StartExec(ctx context.Context, execid string, cfg *streams.AttachConfig, timeout int) (err0 error) {
 	// GetExecConfig should not error, since we have done this before call StartExec
 	execConfig, err := mgr.GetExecConfig(ctx, execid)
 	if err != nil {
@@ -161,7 +162,7 @@ func (mgr *ContainerManager) StartExec(ctx context.Context, execid string, cfg *
 		ExecID:      execid,
 		IO:          eio,
 		P:           process,
-	}); err != nil {
+	}, timeout); err != nil {
 		return err
 	}
 	return <-attachErrCh

hack/testing/run_daemon_cri_integration.sh

@@ -24,7 +24,6 @@ DEFAULT_CRI_SKIP+="|image"
 # TODO: support failed test
 DEFAULT_CRI_SKIP+="|runtime should support Privileged is false"
 DEFAULT_CRI_SKIP+="|runtime should support reopening container log"
-DEFAULT_CRI_SKIP+="|runtime should support execSync with timeout"
 CRI_SKIP="${CRI_SKIP:-"${DEFAULT_CRI_SKIP}"}"
 
 # CRI_FOCUS focuses the test to run.