feature:support live-restore

Signed-off-by: ningmingxiao <ning.mingxiao@zte.com.cn>

Author: ningmingxiao

cmd/nerdctl/container/container_run_restart_linux_test.go

@@ -53,7 +53,8 @@ func TestRunRestart(t *testing.T) {
 		"--name", testContainerName,
 		"-p", fmt.Sprintf("127.0.0.1:%d:80", hostPort),
 		testutil.NginxAlpineImage).AssertOK()
-
+	inspectedContainer := base.InspectContainer(testContainerName)
+	pid := inspectedContainer.State.Pid
 	check := func(httpGetRetry int) error {
 		resp, err := nettestutil.HTTPGet(fmt.Sprintf("http://127.0.0.1:%d", hostPort), httpGetRetry, false)
 		if err != nil {
@@ -87,6 +88,9 @@ func TestRunRestart(t *testing.T) {
 		}
 		time.Sleep(sleep)
 	}
+	inspectedContainer = base.InspectContainer(testContainerName)
+	assert.Equal(t, inspectedContainer.State.Status, "running")
+	assert.Equal(t, inspectedContainer.State.Pid, pid)
 	base.DumpDaemonLogs(10)
 	t.Fatalf("the container does not seem to be restarted")
 }

pkg/logging/logging.go

@@ -207,7 +207,7 @@ func getContainerWait(ctx context.Context, address string, config *logging.Confi
 
 type ContainerWaitFunc func(ctx context.Context, address string, config *logging.Config) (<-chan containerd.ExitStatus, error)
 
-func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, address string, getContainerWait ContainerWaitFunc, config *logging.Config) error {
+func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore string, config *logging.Config) error {
 	if err := driver.PreProcess(ctx, dataStore, config); err != nil {
 		return err
 	}
@@ -220,6 +220,17 @@ func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, addres
 	if err != nil {
 		return err
 	}
+
+	stdoutChan, err := waitIOClose(config.Stdout)
+	if err != nil {
+		return err
+	}
+
+	stderrChan, err := waitIOClose(config.Stderr)
+	if err != nil {
+		return err
+	}
+
 	go func() {
 		<-ctx.Done() // delivered on SIGTERM
 		stdoutR.Cancel()
@@ -273,13 +284,8 @@ func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, addres
 		// close pipeStdoutW and pipeStderrW upon container exit
 		defer pipeStdoutW.Close()
 		defer pipeStderrW.Close()
-
-		exitCh, err := getContainerWait(ctx, address, config)
-		if err != nil {
-			log.G(ctx).Errorf("failed to get container task wait channel: %v", err)
-			return
-		}
-		<-exitCh
+		<-stdoutChan
+		<-stderrChan
 	}()
 	wg.Wait()
 	return driver.PostProcess()
@@ -314,7 +320,7 @@ func loggerFunc(dataStore string) (logging.LoggerFunc, error) {
 					return err
 				}
 				// getContainerWait is extracted as parameter to allow mocking in tests.
-				return loggingProcessAdapter(ctx, driver, dataStore, logConfig.Address, getContainerWait, 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

pkg/logging/logging_bsd.go

@@ -0,0 +1,68 @@
+//go:build darwin || freebsd || netbsd || openbsd || dragonfly
+// +build darwin freebsd netbsd openbsd dragonfly
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package logging
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/muesli/cancelreader"
+	"golang.org/x/sys/unix"
+)
+
+func waitIOClose(reader io.Reader) (chan struct{}, error) {
+	closeIO := make(chan struct{})
+
+	file, ok := reader.(cancelreader.File)
+	if !ok {
+		return nil, fmt.Errorf("reader is not an cancelreader.File")
+	}
+
+	kq, err := unix.Kqueue()
+	if err != nil {
+		return nil, fmt.Errorf("create kqueue: %w", err)
+	}
+	kev := unix.Kevent_t{
+		Ident:  uint64(file.Fd()),
+		Filter: unix.EVFILT_READ,
+		Flags:  unix.EV_ADD | unix.EV_ENABLE,
+	}
+
+	events := make([]unix.Kevent_t, 1)
+	_, err = unix.Kevent(kq, []unix.Kevent_t{kev}, events, nil)
+	if err != nil {
+		return nil, err
+	}
+	go func() {
+		for {
+			n, err := unix.Kevent(kq, nil, events, nil)
+			if err != nil {
+				continue
+			}
+			for i := 0; i < n; i++ {
+				if events[i].Flags&unix.EV_EOF != 0 {
+					close(closeIO)
+					break
+				}
+			}
+		}
+	}()
+	return closeIO, nil
+}

pkg/logging/logging_linux.go

@@ -0,0 +1,64 @@
+//go:build linux
+// +build linux
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package logging
+
+import (
+	"fmt"
+	"io"
+
+	"github.com/muesli/cancelreader"
+	"golang.org/x/sys/unix"
+)
+
+func waitIOClose(reader io.Reader) (chan struct{}, error) {
+	closeIO := make(chan struct{})
+	epfd, err := unix.EpollCreate1(0)
+	if err != nil {
+		return nil, err
+	}
+	file, ok := reader.(cancelreader.File)
+	if !ok {
+		return nil, fmt.Errorf("reader is not an cancelreader.File")
+	}
+	fd := file.Fd()
+	event := unix.EpollEvent{
+		Events: unix.EPOLLHUP,
+		Fd:     int32(fd),
+	}
+	if err := unix.EpollCtl(epfd, unix.EPOLL_CTL_ADD, int(fd), &event); err != nil {
+		return nil, err
+	}
+	events := make([]unix.EpollEvent, 1)
+	go func() {
+		for {
+			n, err := unix.EpollWait(epfd, events, -1)
+			if err != nil {
+				continue
+			}
+			for i := 0; i < n; i++ {
+				if events[i].Events&unix.EPOLLHUP != 0 {
+					close(closeIO)
+					break
+				}
+			}
+		}
+	}()
+	return closeIO, nil
+}

pkg/logging/logging_test.go

@@ -25,7 +25,6 @@ import (
 	"testing"
 	"time"
 
-	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/core/runtime/v2/logging"
 )
 
@@ -78,15 +77,7 @@ func TestLoggingProcessAdapter(t *testing.T) {
 	// Execute the logging process adapter
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-
-	var getContainerWaitMock ContainerWaitFunc = func(ctx context.Context, address string, config *logging.Config) (<-chan containerd.ExitStatus, error) {
-		exitChan := make(chan containerd.ExitStatus, 1)
-		time.Sleep(50 * time.Millisecond)
-		exitChan <- containerd.ExitStatus{}
-		return exitChan, nil
-	}
-
-	err := loggingProcessAdapter(ctx, driver, "testDataStore", "", getContainerWaitMock, config)
+	err := loggingProcessAdapter(ctx, driver, "testDataStore", config)
 	if err != nil {
 		t.Fatal(err)
 	}

pkg/logging/logging_windows.go

@@ -0,0 +1,24 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package logging
+
+import "io"
+
+// TODO: support windows
+func waitIOClose(reader io.Reader) (chan struct{}, error) {
+	return nil, nil
+}