Merge pull request #2727 from AkihiroSuda/fix-2726

logging: respect ctx

Author: AkihiroSuda

.github/workflows/test.yml

@@ -70,7 +70,7 @@ jobs:
 
   test-integration:
     runs-on: "ubuntu-${{ matrix.ubuntu }}"
-    timeout-minutes: 60
+    timeout-minutes: 40
     strategy:
       fail-fast: false
       matrix:

Dockerfile

@@ -313,7 +313,7 @@ RUN curl -L -o nydus-static.tgz "https://github.com/dragonflyoss/image-service/r
   mv nydus-static/nydus-image nydus-static/nydusd nydus-static/nydusify /usr/bin/ && \
   rm nydus-static.tgz
 CMD ["gotestsum", "--format=testname", "--rerun-fails=2", "--packages=github.com/containerd/nerdctl/cmd/nerdctl/...", \
-  "--", "-timeout=60m", "-args", "-test.kill-daemon"]
+  "--", "-timeout=30m", "-args", "-test.kill-daemon"]
 
 FROM test-integration AS test-integration-rootless
 # Install SSH for creating systemd user session.

cmd/nerdctl/container_run_test.go

@@ -510,3 +510,17 @@ func TestRunAddHostRemainsWhenAnotherContainerCreated(t *testing.T) {
 
 	base.Cmd("exec", containerName, "cat", "/etc/hosts").AssertOutWithFunc(checkEtcHosts)
 }
+
+// https://github.com/containerd/nerdctl/issues/2726
+func TestRunRmTime(t *testing.T) {
+	base := testutil.NewBase(t)
+	base.Cmd("pull", testutil.CommonImage)
+	t0 := time.Now()
+	base.Cmd("run", "--rm", testutil.CommonImage, "true").AssertOK()
+	t1 := time.Now()
+	took := t1.Sub(t0)
+	const deadline = 3 * time.Second
+	if took > deadline {
+		t.Fatalf("expected to have completed in %v, took %v", deadline, took)
+	}
+}

go.mod

@@ -40,6 +40,7 @@ require (
 	github.com/moby/sys/mount v0.3.3
 	github.com/moby/sys/signal v0.7.0
 	github.com/moby/term v0.5.0
+	github.com/muesli/cancelreader v0.2.2
 	github.com/opencontainers/go-digest v1.0.0
 	github.com/opencontainers/image-spec v1.1.0-rc5
 	github.com/opencontainers/runtime-spec v1.1.0

go.sum

@@ -214,6 +214,8 @@ github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
 github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y=
 github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
 github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
+github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELUXHmA=
+github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo=
 github.com/multiformats/go-base32 v0.1.0 h1:pVx9xoSPqEIQG8o+UbAe7DNi51oej1NtK+aGkbLYxPE=
 github.com/multiformats/go-base32 v0.1.0/go.mod h1:Kj3tFY6zNr+ABYMqeUNeGvkIC/UYgtWibDcT0rExnbI=
 github.com/multiformats/go-base36 v0.1.0 h1:JR6TyF7JjGd3m6FbLU2cOxhC0Li8z8dLNGQ89tUg4F4=

pkg/logging/logging.go

@@ -31,6 +31,7 @@ import (
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/runtime/v2/logging"
 	"github.com/containerd/log"
+	"github.com/muesli/cancelreader"
 )
 
 const (
@@ -140,10 +141,25 @@ func LoadLogConfig(dataStore, ns, id string) (LogConfig, error) {
 	return logConfig, nil
 }
 
-func loggingProcessAdapter(driver Driver, dataStore string, config *logging.Config) error {
+func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore string, config *logging.Config) error {
 	if err := driver.PreProcess(dataStore, config); err != nil {
 		return err
 	}
+
+	stdoutR, err := cancelreader.NewReader(config.Stdout)
+	if err != nil {
+		return err
+	}
+	stderrR, err := cancelreader.NewReader(config.Stderr)
+	if err != nil {
+		return err
+	}
+	go func() {
+		<-ctx.Done() // delivered on SIGTERM
+		stdoutR.Cancel()
+		stderrR.Cancel()
+	}()
+
 	var wg sync.WaitGroup
 	wg.Add(3)
 	stdout := make(chan string, 10000)
@@ -161,8 +177,8 @@ func loggingProcessAdapter(driver Driver, dataStore string, config *logging.Conf
 		}
 	}
 
-	go processLogFunc(config.Stdout, stdout)
-	go processLogFunc(config.Stderr, stderr)
+	go processLogFunc(stdoutR, stdout)
+	go processLogFunc(stderrR, stderr)
 	go func() {
 		defer wg.Done()
 		driver.Process(stdout, stderr)
@@ -175,7 +191,7 @@ func loggerFunc(dataStore string) (logging.LoggerFunc, error) {
 	if dataStore == "" {
 		return nil, errors.New("got empty data store")
 	}
-	return func(_ context.Context, config *logging.Config, ready func() error) error {
+	return func(ctx context.Context, config *logging.Config, ready func() error) error {
 		if config.Namespace == "" || config.ID == "" {
 			return errors.New("got invalid config")
 		}
@@ -193,7 +209,7 @@ func loggerFunc(dataStore string) (logging.LoggerFunc, error) {
 				return err
 			}
 
-			return loggingProcessAdapter(driver, dataStore, config)
+			return loggingProcessAdapter(ctx, driver, dataStore, config)
 		} else if !errors.Is(err, os.ErrNotExist) {
 			// the file does not exist if the container was created with nerdctl < 0.20
 			return err