Use executor.StreamWithContext instead of deprecated executor.Stream

Author: surik

cmd/crictl/attach.go

@@ -59,12 +59,15 @@ var runtimeAttachCommand = &cli.Command{
 			return err
 		}
 
+		ctx, cancel := context.WithCancel(c.Context)
+		defer cancel()
+
 		var opts = attachOptions{
 			id:    id,
 			tty:   c.Bool("tty"),
 			stdin: c.Bool("stdin"),
 		}
-		if err = Attach(runtimeClient, opts); err != nil {
+		if err = Attach(ctx, runtimeClient, opts); err != nil {
 			return fmt.Errorf("attaching running container failed: %w", err)
 
 		}
@@ -73,7 +76,7 @@ var runtimeAttachCommand = &cli.Command{
 }
 
 // Attach sends an AttachRequest to server, and parses the returned AttachResponse
-func Attach(client internalapi.RuntimeService, opts attachOptions) error {
+func Attach(ctx context.Context, client internalapi.RuntimeService, opts attachOptions) error {
 	if opts.id == "" {
 		return fmt.Errorf("ID cannot be empty")
 
@@ -86,7 +89,7 @@ func Attach(client internalapi.RuntimeService, opts attachOptions) error {
 		Stderr:      !opts.tty,
 	}
 	logrus.Debugf("AttachRequest: %v", request)
-	r, err := client.Attach(context.TODO(), request)
+	r, err := client.Attach(ctx, request)
 	logrus.Debugf("AttachResponse: %v", r)
 	if err != nil {
 		return err
@@ -106,5 +109,5 @@ func Attach(client internalapi.RuntimeService, opts attachOptions) error {
 	}
 
 	logrus.Debugf("Attach URL: %v", URL)
-	return stream(opts.stdin, opts.tty, URL)
+	return stream(ctx, opts.stdin, opts.tty, URL)
 }

cmd/crictl/exec.go

@@ -95,7 +95,11 @@ var runtimeExecCommand = &cli.Command{
 			}
 			return nil
 		}
-		err = Exec(runtimeClient, opts)
+
+		ctx, cancel := context.WithCancel(c.Context)
+		defer cancel()
+
+		err = Exec(ctx, runtimeClient, opts)
 		if err != nil {
 			return fmt.Errorf("execing command in container: %w", err)
 		}
@@ -124,7 +128,7 @@ func ExecSync(client internalapi.RuntimeService, opts execOptions) (int, error)
 }
 
 // Exec sends an ExecRequest to server, and parses the returned ExecResponse
-func Exec(client internalapi.RuntimeService, opts execOptions) error {
+func Exec(ctx context.Context, client internalapi.RuntimeService, opts execOptions) error {
 	request := &pb.ExecRequest{
 		ContainerId: opts.id,
 		Cmd:         opts.cmd,
@@ -135,7 +139,7 @@ func Exec(client internalapi.RuntimeService, opts execOptions) error {
 	}
 
 	logrus.Debugf("ExecRequest: %v", request)
-	r, err := client.Exec(context.TODO(), request)
+	r, err := client.Exec(ctx, request)
 	logrus.Debugf("ExecResponse: %v", r)
 	if err != nil {
 		return err
@@ -156,10 +160,10 @@ func Exec(client internalapi.RuntimeService, opts execOptions) error {
 	}
 
 	logrus.Debugf("Exec URL: %v", URL)
-	return stream(opts.stdin, opts.tty, URL)
+	return stream(ctx, opts.stdin, opts.tty, URL)
 }
 
-func stream(in, tty bool, url *url.URL) error {
+func stream(ctx context.Context, in, tty bool, url *url.URL) error {
 	executor, err := remoteclient.NewSPDYExecutor(&restclient.Config{TLSClientConfig: restclient.TLSClientConfig{Insecure: true}}, "POST", url)
 	if err != nil {
 		return err
@@ -176,7 +180,7 @@ func stream(in, tty bool, url *url.URL) error {
 	}
 	logrus.Debugf("StreamOptions: %v", streamOptions)
 	if !tty {
-		return executor.Stream(streamOptions)
+		return executor.StreamWithContext(ctx, streamOptions)
 	} else {
 		var detachKeys []byte
 		detachKeys, err = mobyterm.ToBytes(detachSequence)
@@ -198,5 +202,5 @@ func stream(in, tty bool, url *url.URL) error {
 		return fmt.Errorf("input is not a terminal")
 	}
 	streamOptions.TerminalSizeQueue = t.MonitorSize(t.GetSize())
-	return t.Safe(func() error { return executor.Stream(streamOptions) })
+	return t.Safe(func() error { return executor.StreamWithContext(ctx, streamOptions) })
 }

pkg/validate/streaming.go

@@ -198,7 +198,11 @@ func checkExec(c internalapi.RuntimeService, execServerURL, stdout string, stdou
 		streamOptions.Stdin = localInRead
 		streamOptions.Tty = true
 	}
-	err = e.Stream(streamOptions)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	err = e.StreamWithContext(ctx, streamOptions)
 	framework.ExpectNoError(err, "failed to open streamer for %q", execServerURL)
 
 	if stdoutExactMatch {
@@ -296,7 +300,10 @@ func checkAttach(c internalapi.RuntimeService, attachServerURL string) {
 	e, err := remoteclient.NewSPDYExecutor(&rest.Config{TLSClientConfig: rest.TLSClientConfig{Insecure: true}}, "POST", url)
 	framework.ExpectNoError(err, "failed to create executor for %q", attachServerURL)
 
-	err = e.Stream(remoteclient.StreamOptions{
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	err = e.StreamWithContext(ctx, remoteclient.StreamOptions{
 		Stdin:  reader,
 		Stdout: localOut,
 		Stderr: localErr,