Merge branch 'master' of https://github.com/alibaba/pouch into cri-compatibility

Author: starnop

CONTRIBUTORS

@@ -43,6 +43,7 @@ Runrioter Wung <[email protected]>
 Sam Xie <[email protected]>
 skoo <[email protected]>
 skoo87 <[email protected]>
+soarpenguin <[email protected]>
 Starnop <[email protected]>
 Tao Qingyun <[email protected]>
 Tiramisu 1993 <[email protected]>

apis/swagger.yml

@@ -2894,7 +2894,10 @@ definitions:
       AppArmorProfile:
         type: "string"
       ExecIDs:
-        type: "string"
+        description: "exec ids of container"
+        type: "array"
+        items:
+          type: "string"
       HostConfig:
         $ref: "#/definitions/HostConfig"
       SizeRw:

apis/types/container_json.go

@@ -138,7 +138,10 @@ func (e *ExecCommand) runExec(args []string) error {
 }
 
 // execExample shows examples in exec command, and is used in auto-generated cli docs.
-// TODO: add example
 func execExample() string {
-	return ""
+	return `$ pouch exec -it 25bf50 ps
+PID   USER     TIME  COMMAND
+    1 root      0:00 /bin/sh
+   38 root      0:00 ps
+`
 }

cli/exec.go

@@ -62,7 +62,7 @@ func (top *TopCommand) runTop(args []string) error {
 // topExamples shows examples in top command, and is used in auto-generated cli docs.
 func topExamples() string {
 	return `$ pouch top 44f675
-	UID     PID      PPID     C    STIME    TTY    TIME        CMD
-	root    28725    28714    0    3月14     ?      00:00:00    sh
-	`
+UID     PID      PPID     C    STIME    TTY    TIME        CMD
+root    28725    28714    0    3月14     ?      00:00:00    sh
+`
 }

cli/top.go

@@ -19,7 +19,7 @@ var (
 	TokenLen = 8
 )
 
-// requestCache caches streaming (exec/attach/port-forward) requests and generates a single-use
+// RequestCache caches streaming (exec/attach/port-forward) requests and generates a single-use
 // random token for their retrieval. The requestCache is used for building streaming URLs without
 // the need to encode every request parameter in the URL.
 type RequestCache struct {
@@ -31,15 +31,16 @@ type RequestCache struct {
 	lock sync.Mutex
 }
 
-// Type representing an *ExecRequest, *AttachRequest, or *PortForwardRequest.
-type request interface{}
+// Request representing an *ExecRequest, *AttachRequest, or *PortForwardRequest Type.
+type Request interface{}
 
 type cacheEntry struct {
 	token      string
-	req        request
+	req        Request
 	expireTime time.Time
 }
 
+// NewRequestCache return a RequestCache
 func NewRequestCache() *RequestCache {
 	return &RequestCache{
 		ll:     list.New(),
@@ -48,7 +49,7 @@ func NewRequestCache() *RequestCache {
 }
 
 // Insert the given request into the cache and returns the token used for fetching it out.
-func (c *RequestCache) Insert(req request) (token string, err error) {
+func (c *RequestCache) Insert(req Request) (token string, err error) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
@@ -69,7 +70,7 @@ func (c *RequestCache) Insert(req request) (token string, err error) {
 }
 
 // Consume the token (remove it from the cache) and return the cached request, if found.
-func (c *RequestCache) Consume(token string) (req request, found bool) {
+func (c *RequestCache) Consume(token string) (req Request, found bool) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	ele, ok := c.tokens[token]

cri/stream/request_cache.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"path"
 	"path/filepath"
@@ -695,55 +696,74 @@ func (c *CriManager) UpdateContainerResources(ctx context.Context, r *runtime.Up
 // ExecSync executes a command in the container, and returns the stdout output.
 // If command exits with a non-zero exit code, an error is returned.
 func (c *CriManager) ExecSync(ctx context.Context, r *runtime.ExecSyncRequest) (*runtime.ExecSyncResponse, error) {
-	// TODO: handle timeout.
 	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(),
 	}
-
 	execid, err := c.ContainerMgr.CreateExec(ctx, id, createConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create exec for container %q: %v", id, err)
 	}
 
-	var output bytes.Buffer
-	startConfig := &apitypes.ExecStartConfig{}
+	reader, writer := io.Pipe()
+	defer writer.Close()
+
 	attachConfig := &mgr.AttachConfig{
 		Stdout:      true,
 		Stderr:      true,
-		MemBuffer:   &output,
+		Pipe:        writer,
 		MuxDisabled: true,
 	}
 
+	startConfig := &apitypes.ExecStartConfig{}
+
 	err = c.ContainerMgr.StartExec(ctx, execid, startConfig, attachConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to start exec for container %q: %v", id, err)
 	}
 
-	var execConfig *mgr.ContainerExecConfig
-	for {
-		execConfig, err = c.ContainerMgr.GetExecConfig(ctx, execid)
+	readWaitCh := make(chan error, 1)
+	var recv bytes.Buffer
+	go func() {
+		defer reader.Close()
+		_, err = io.Copy(&recv, reader)
+		readWaitCh <- err
+	}()
+
+	select {
+	case <-ctx.Done():
+		//TODO maybe stop the execution?
+		return nil, fmt.Errorf("timeout %v exceeded", timeout)
+	case readWaitErr := <-readWaitCh:
+		if readWaitErr != nil {
+			return nil, fmt.Errorf("failed to read data from the pipe: %v", err)
+		}
+		execConfig, err := c.ContainerMgr.GetExecConfig(ctx, execid)
 		if err != nil {
 			return nil, fmt.Errorf("failed to inspect exec for container %q: %v", id, err)
 		}
-		// Loop until exec finished.
-		if !execConfig.Running {
-			break
+
+		var stderr []byte
+		if execConfig.Error != nil {
+			stderr = []byte(execConfig.Error.Error())
 		}
-		time.Sleep(100 * time.Millisecond)
-	}
 
-	var stderr []byte
-	if execConfig.Error != nil {
-		stderr = []byte(execConfig.Error.Error())
+		return &runtime.ExecSyncResponse{
+			Stdout:   recv.Bytes(),
+			Stderr:   stderr,
+			ExitCode: int32(execConfig.ExitCode),
+		}, nil
 	}
-
-	return &runtime.ExecSyncResponse{
-		Stdout:   output.Bytes(),
-		Stderr:   stderr,
-		ExitCode: int32(execConfig.ExitCode),
-	}, nil
 }
 
 // Exec prepares a streaming endpoint to execute a command in the container, and returns the address.

cri/v1alpha1/cri.go

@@ -282,7 +282,7 @@ func Test_makeSandboxPouchConfig(t *testing.T) {
 		want    *apitypes.ContainerCreateConfig
 		wantErr bool
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -342,7 +342,7 @@ func Test_toCriSandbox(t *testing.T) {
 		want    *runtime.PodSandbox
 		wantErr bool
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -558,7 +558,7 @@ func Test_makeContainerName(t *testing.T) {
 		args args
 		want string
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -579,7 +579,7 @@ func Test_modifyContainerNamespaceOptions(t *testing.T) {
 		name string
 		args args
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -600,7 +600,7 @@ func Test_applyContainerSecurityContext(t *testing.T) {
 		args    args
 		wantErr bool
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -628,7 +628,7 @@ func TestCriManager_updateCreateConfig(t *testing.T) {
 		args    args
 		wantErr bool
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -653,7 +653,7 @@ func Test_toCriContainer(t *testing.T) {
 		want    *runtime.Container
 		wantErr bool
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -680,7 +680,7 @@ func Test_imageToCriImage(t *testing.T) {
 		want    *runtime.Image
 		wantErr bool
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -711,7 +711,7 @@ func TestCriManager_ensureSandboxImageExists(t *testing.T) {
 		args    args
 		wantErr bool
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -736,7 +736,7 @@ func Test_getUserFromImageUser(t *testing.T) {
 		want  *int64
 		want1 string
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -760,7 +760,7 @@ func Test_parseUserFromImageUser(t *testing.T) {
 		args args
 		want string
 	}{
-		// TODO: Add test cases.
+	// TODO: Add test cases.
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

cri/v1alpha1/cri_utils_test.go

@@ -10,7 +10,6 @@ import (
 
 	"google.golang.org/grpc"
 	"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
-
 )
 
 // Service serves the kubelet runtime grpc api which will be consumed by kubelet.