fix: terminate nydusd left behind on startup failure

Author: qinfustu

pkg/filesystem/fs.go

@@ -109,6 +109,9 @@ func NewFileSystem(ctx context.Context, opt ...NewFSOpt) (*Filesystem, error) {
 	// TODO: We still need to consider shared daemon the time sequence of initializing daemon,
 	// start daemon commit its state to DB and retrieving its state.
 	if fscacheManager, ok := fs.enabledManagers[config.FsDriverFscache]; ok {
+		fscacheManager.IsSharedDaemonRetained = func(d *daemon.Daemon) bool {
+			return fs.fscacheSharedDaemon != nil && fs.fscacheSharedDaemon.ID() == d.ID()
+		}
 		if !hasFscacheSharedDaemon && fs.fscacheSharedDaemon == nil {
 			log.L.Infof("initializing shared nydus daemon for fscache")
 			if err := fs.initSharedDaemon(fscacheManager); err != nil {
@@ -119,6 +122,9 @@ func NewFileSystem(ctx context.Context, opt ...NewFSOpt) (*Filesystem, error) {
 		return nil, errors.Errorf("shared fscache daemon is present, but manager is missing")
 	}
 	if fusedevManager, ok := fs.enabledManagers[config.FsDriverFusedev]; ok {
+		fusedevManager.IsSharedDaemonRetained = func(d *daemon.Daemon) bool {
+			return fs.fusedevSharedDaemon != nil && fs.fusedevSharedDaemon.ID() == d.ID()
+		}
 		if config.IsFusedevSharedModeEnabled() && !hasFusedevSharedDaemon && fs.fusedevSharedDaemon == nil {
 			log.L.Infof("initializing shared nydus daemon for fusedev")
 			if err := fs.initSharedDaemon(fusedevManager); err != nil {
@@ -929,6 +935,10 @@ func (fs *Filesystem) initSharedDaemon(fsManager *manager.Manager) (err error) {
 		return errors.Wrap(err, "start shared daemon")
 	}
 
+	if err := d.WaitUntilState(types.DaemonStateRunning); err != nil {
+		return errors.Wrap(err, "wait shared daemon to become running")
+	}
+
 	fs.TryRetainSharedDaemon(d)
 
 	return

pkg/manager/daemon_adaptor.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"os/exec"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/containerd/log"
@@ -28,6 +29,11 @@ import (
 
 const endpointGetBackend string = "/api/v1/daemons/%s/backend"
 
+// defaultDaemonTerminationTimeout is how long we wait for a failed nydusd to
+// exit after SIGTERM before escalating to SIGKILL. It is a var so tests can
+// shorten it.
+var defaultDaemonTerminationTimeout = 5 * time.Second
+
 // Spawn a nydusd daemon to serve the daemon instance.
 //
 // When returning from `StartDaemon()` with out error:
@@ -81,22 +87,42 @@ func (m *Manager) StartDaemon(d *daemon.Daemon) error {
 		log.L.Errorf("Fail to update daemon info (%+v) to DB: %v", d, err)
 	}
 
-	// If nydusd fails startup, manager can't subscribe its death event.
-	// So we can ignore the subscribing error.
+	// Verify the daemon actually comes up in the background. If it does not,
+	// terminate the spawned process so it cannot linger as an orphan that keeps
+	// pulling from the registry (see #771). `proc` is captured here so we act on
+	// exactly the process we started, regardless of any concurrent pid changes.
+	proc := cmd.Process
 	go func() {
 		if err := daemon.WaitUntilSocketExisted(d.GetAPISock(), d.States.ProcessID); err != nil {
-			// FIXME: Should clean the daemon record in DB if the nydusd fails starting
-			log.L.Errorf("Nydusd %s probably not started", d.ID())
+			log.L.Errorf("Nydusd %s probably not started, terminating it: %v", d.ID(), err)
+			m.terminateFailedDaemon(d, proc)
 			return
 		}
 
 		if err = m.SubscribeDaemonEvent(d); err != nil {
-			log.L.Errorf("Nydusd %s probably not started", d.ID())
+			log.L.Errorf("Failed to subscribe nydusd %s events, terminating it: %v", d.ID(), err)
+			m.terminateFailedDaemon(d, proc)
 			return
 		}
 
 		if err := d.WaitUntilState(types.DaemonStateRunning); err != nil {
-			log.L.WithError(err).Errorf("daemon %s is not managed to reach RUNNING state", d.ID())
+			// A failover-managed daemon (recover_policy=failover) legitimately
+			// stays in INIT/READY while the failover/upgrade flow drives
+			// INIT -> TakeOver -> Start after StartDaemon returns. Do not treat
+			// that as a failed startup: killing or unsubscribing it here would
+			// wreck the takeover. Its cleanup belongs to the failover flow.
+			if d.Supervisor != nil {
+				log.L.WithError(err).Warnf("daemon %s did not reach RUNNING yet, leaving it to the failover flow", d.ID())
+				return
+			}
+			log.L.WithError(err).Errorf("daemon %s is not managed to reach RUNNING state, terminating it", d.ID())
+			// Unsubscribe before killing, otherwise the liveness monitor would
+			// observe the kill as a death event and (with recover_policy=restart)
+			// immediately respawn what we just decided to reap.
+			if uerr := m.UnsubscribeDaemonEvent(d); uerr != nil {
+				log.L.WithError(uerr).Warnf("unsubscribe daemon %s after startup failure", d.ID())
+			}
+			m.terminateFailedDaemon(d, proc)
 			return
 		}
 
@@ -119,6 +145,71 @@ func (m *Manager) StartDaemon(d *daemon.Daemon) error {
 	return nil
 }
 
+// terminateFailedDaemon best-effort stops a nydusd that failed startup
+// verification, so it stops pulling from the registry instead of lingering as
+// an orphan. SIGTERM is tried first and escalated to SIGKILL if the process,
+// for example one wedged on a slow registry, does not exit in time. The process
+// is finally reaped to avoid leaving a zombie. It operates on the captured
+// process handle, so it is safe against pid reuse and concurrent teardown.
+func (m *Manager) terminateFailedDaemon(d *daemon.Daemon, proc *os.Process) {
+	if proc == nil {
+		return
+	}
+
+	// A failover-managed daemon (recover_policy=failover) legitimately stays in
+	// INIT/READY while the failover flow drives INIT -> TakeOver -> Start after
+	// StartDaemon returns, so a "not RUNNING yet" verdict here may be a takeover
+	// in progress rather than a failed startup. Killing it would destroy the
+	// takeover, so leave its cleanup to the failover flow.
+	if d.Supervisor != nil {
+		log.L.Warnf("nydusd %s (pid %d) failed startup verification, leaving cleanup to the failover flow", d.ID(), proc.Pid)
+		return
+	}
+
+	// For a shared daemon, if it fails to reach RUNNING, it won't be retained
+	// by TryRetainSharedDaemon. We must kill it here so it doesn't linger.
+	// But if it is already retained (e.g., this is a spurious timeout or
+	// TryRetainSharedDaemon was called concurrently), killing it would drop
+	// the active shared daemon. We check IsSharedDaemon() and whether it is
+	// the currently active one.
+	if d.IsSharedDaemon() && m.isDaemonRetainedAsShared(d) {
+		log.L.Warnf("nydusd %s (pid %d) failed startup verification but is already retained as shared daemon, not killing it", d.ID(), proc.Pid)
+		return
+	}
+
+	log.L.Warnf("terminating nydusd %s (pid %d) that failed to start", d.ID(), proc.Pid)
+
+	done := make(chan struct{})
+	go func() {
+		_, _ = proc.Wait()
+		close(done)
+	}()
+
+	if err := proc.Signal(syscall.SIGTERM); err != nil && !errors.Is(err, os.ErrProcessDone) {
+		log.L.WithError(err).Warnf("send SIGTERM to nydusd %s (pid %d)", d.ID(), proc.Pid)
+	}
+
+	select {
+	case <-done:
+	case <-time.After(defaultDaemonTerminationTimeout):
+		log.L.Warnf("nydusd %s (pid %d) did not exit after SIGTERM, sending SIGKILL", d.ID(), proc.Pid)
+		if err := proc.Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) {
+			log.L.WithError(err).Warnf("send SIGKILL to nydusd %s (pid %d)", d.ID(), proc.Pid)
+		}
+		<-done
+	}
+}
+
+// isDaemonRetainedAsShared checks if the daemon is currently retained as the
+// active shared daemon in the filesystem manager. This is a callback provided
+// by the manager to avoid circular dependencies.
+func (m *Manager) isDaemonRetainedAsShared(d *daemon.Daemon) bool {
+	if m.IsSharedDaemonRetained != nil {
+		return m.IsSharedDaemonRetained(d)
+	}
+	return false
+}
+
 // Build commandline according to nydusd daemon configuration.
 func (m *Manager) BuildDaemonCommand(d *daemon.Daemon, bin string, upgrade bool) (*exec.Cmd, error) {
 	var cmdOpts []command.Opt

pkg/manager/daemon_adaptor_test.go

@@ -0,0 +1,64 @@
+/*
+ * Copyright (c) 2022. Nydus Developers. All rights reserved.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package manager
+
+import (
+	"os"
+	"os/exec"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/containerd/nydus-snapshotter/pkg/daemon"
+)
+
+func TestTerminateFailedDaemon(t *testing.T) {
+	orig := defaultDaemonTerminationTimeout
+	t.Cleanup(func() { defaultDaemonTerminationTimeout = orig })
+
+	m := &Manager{}
+
+	t.Run("nil process is a no-op", func(t *testing.T) {
+		d, err := daemon.NewDaemon()
+		require.NoError(t, err)
+		m.terminateFailedDaemon(d, nil)
+	})
+
+	t.Run("SIGTERM stops a well-behaved process", func(t *testing.T) {
+		defaultDaemonTerminationTimeout = 5 * time.Second
+
+		cmd := exec.Command("sleep", "300")
+		require.NoError(t, cmd.Start())
+		d, err := daemon.NewDaemon()
+		require.NoError(t, err)
+
+		start := time.Now()
+		m.terminateFailedDaemon(d, cmd.Process)
+
+		// The process is reaped inside the helper, so it is marked done.
+		assert.ErrorIs(t, cmd.Process.Signal(syscall.Signal(0)), os.ErrProcessDone)
+		// A SIGTERM-able process must exit well before the SIGKILL escalation.
+		assert.Less(t, time.Since(start), defaultDaemonTerminationTimeout)
+	})
+
+	t.Run("escalates to SIGKILL when SIGTERM is ignored", func(t *testing.T) {
+		defaultDaemonTerminationTimeout = 200 * time.Millisecond
+
+		// Ignore SIGTERM so only SIGKILL can stop it.
+		cmd := exec.Command("sh", "-c", "trap '' TERM; sleep 300")
+		require.NoError(t, cmd.Start())
+		d, err := daemon.NewDaemon()
+		require.NoError(t, err)
+
+		m.terminateFailedDaemon(d, cmd.Process)
+
+		assert.ErrorIs(t, cmd.Process.Signal(syscall.Signal(0)), os.ErrProcessDone)
+	})
+}

pkg/manager/manager.go

@@ -49,6 +49,11 @@ type Manager struct {
 	NydusdBinaryPath string
 	RecoverPolicy    config.DaemonRecoverPolicy
 	SupervisorSet    *supervisor.SupervisorsSet
+
+	// IsSharedDaemonRetained is a callback to check if a daemon is currently
+	// retained as the active shared daemon. It is used to avoid killing a
+	// shared daemon that has already been retained.
+	IsSharedDaemonRetained func(d *daemon.Daemon) bool
 }
 
 type Opt struct {