From 9b235722e42230de91a9af25cd709278d90453dc Mon Sep 17 00:00:00 2001 From: Weilu Jia Date: Thu, 11 Jun 2026 23:35:30 -0700 Subject: [PATCH] fix(orchestrator): harden Firecracker process shutdown Use stdlib pidfd-backed signaling (os.Process is pidfd-backed since Go 1.23) so FC processes cannot be signaled after PID reuse, kill the sandbox cgroup on stop and wait for it to empty via poll on cgroup.events instead of polling cgroup.procs, and keep memory shutdown running even when the exit wait context is canceled so UFFD can exit. --- .../pkg/sandbox/cgroup/manager.go | 142 ++++++++++++++++- .../pkg/sandbox/cgroup/manager_test.go | 147 +++++++++++++++--- .../orchestrator/pkg/sandbox/fc/process.go | 89 +++++------ .../pkg/sandbox/fc/process_test.go | 115 ++++++++++++++ packages/orchestrator/pkg/sandbox/sandbox.go | 17 +- 5 files changed, 433 insertions(+), 77 deletions(-) create mode 100644 packages/orchestrator/pkg/sandbox/fc/process_test.go diff --git a/packages/orchestrator/pkg/sandbox/cgroup/manager.go b/packages/orchestrator/pkg/sandbox/cgroup/manager.go index d5a979dbbf..8325b681a5 100644 --- a/packages/orchestrator/pkg/sandbox/cgroup/manager.go +++ b/packages/orchestrator/pkg/sandbox/cgroup/manager.go @@ -12,6 +12,7 @@ import ( "time" "go.uber.org/zap" + "golang.org/x/sys/unix" "github.com/e2b-dev/infra/packages/shared/pkg/logger" ) @@ -26,6 +27,9 @@ const ( // NoCgroupFD is a sentinel value indicating that no cgroup file descriptor // is available (e.g. cgroup accounting is disabled or the FD has been released). NoCgroupFD = -1 + + cgroupKillTimeout = 2 * time.Second + cgroupKillPollInterval = 100 * time.Millisecond ) // Stats contains resource usage statistics from a cgroup @@ -103,6 +107,88 @@ func (h *CgroupHandle) GetStats(ctx context.Context) (*Stats, error) { return h.manager.getStatsForPath(ctx, h.path, h.memoryPeakFile) } +// Kill terminates all processes currently in this cgroup. +// Safe to call multiple times. Returns nil if the cgroup is already empty or gone. +func (h *CgroupHandle) Kill(ctx context.Context) error { + if h == nil || h.noop || h.removed { + return nil + } + + return h.kill(ctx) +} + +func (h *CgroupHandle) kill(ctx context.Context) error { + if h == nil || h.noop { + return nil + } + + populated, err := h.populated() + if err != nil { + return err + } + if !populated { + return nil + } + + if err := os.WriteFile(filepath.Join(h.path, "cgroup.kill"), []byte("1"), 0); err != nil { + if os.IsNotExist(err) { + return nil + } + + return fmt.Errorf("failed to write cgroup.kill: %w", err) + } + + events, err := os.Open(filepath.Join(h.path, "cgroup.events")) + if os.IsNotExist(err) { + return nil + } + if err != nil { + return fmt.Errorf("failed to open cgroup.events: %w", err) + } + defer events.Close() + + deadline := time.Now().Add(cgroupKillTimeout) + + for { + populated, err := cgroupEventsPopulated(events) + if err != nil { + return err + } + if !populated { + return nil + } + + if err := ctx.Err(); err != nil { + return err + } + + remaining := time.Until(deadline) + if remaining <= 0 { + return fmt.Errorf("cgroup %s still has processes after cgroup.kill", h.cgroupName) + } + + pollTimeout := min(remaining, cgroupKillPollInterval) + pollTimeoutMillis := int(pollTimeout / time.Millisecond) + if pollTimeoutMillis == 0 { + pollTimeoutMillis = 1 + } + + fds := []unix.PollFd{{ + Fd: int32(events.Fd()), + Events: unix.POLLPRI | unix.POLLERR, + }} + + _, err = unix.Poll(fds, pollTimeoutMillis) + if err != nil { + if errors.Is(err, unix.EINTR) { + continue + } + + return fmt.Errorf("failed to poll cgroup.events: %w", err) + } + } +} + // Remove closes all open FDs and deletes the cgroup directory. // The handle should not be used after calling Remove. // Safe to call multiple times. Returns error if removal fails @@ -144,8 +230,8 @@ func (h *CgroupHandle) Remove(ctx context.Context) error { zap.String("path", h.path), zap.Error(rmErr)) - if err := os.WriteFile(filepath.Join(h.path, "cgroup.kill"), []byte("1"), 0); err != nil && !os.IsNotExist(err) { - logger.L().Warn(ctx, "failed to write cgroup.kill", + if err := h.kill(ctx); err != nil { + logger.L().Warn(ctx, "failed to kill cgroup processes", zap.String("cgroup_name", h.cgroupName), zap.String("path", h.path), zap.Error(err)) @@ -192,6 +278,58 @@ func (h *CgroupHandle) CgroupName() string { return h.cgroupName } +func (h *CgroupHandle) populated() (bool, error) { + data, err := os.ReadFile(filepath.Join(h.path, "cgroup.events")) + if os.IsNotExist(err) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("failed to read cgroup.events: %w", err) + } + + return parseCgroupEventsPopulated(data) +} + +func cgroupEventsPopulated(file *os.File) (bool, error) { + if _, err := file.Seek(0, io.SeekStart); err != nil { + if os.IsNotExist(err) { + return false, nil + } + + return false, fmt.Errorf("failed to seek cgroup.events: %w", err) + } + + data, err := io.ReadAll(file) + if os.IsNotExist(err) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("failed to read cgroup.events: %w", err) + } + + return parseCgroupEventsPopulated(data) +} + +func parseCgroupEventsPopulated(data []byte) (bool, error) { + for line := range strings.SplitSeq(string(data), "\n") { + fields := strings.Fields(line) + if len(fields) != 2 || fields[0] != "populated" { + continue + } + + switch fields[1] { + case "0": + return false, nil + case "1": + return true, nil + default: + return false, fmt.Errorf("invalid populated value in cgroup.events: %q", fields[1]) + } + } + + return false, errors.New("missing populated value in cgroup.events") +} + // Manager handles initialization and creation of cgroups // Individual cgroup operations are performed through CgroupHandle type Manager interface { diff --git a/packages/orchestrator/pkg/sandbox/cgroup/manager_test.go b/packages/orchestrator/pkg/sandbox/cgroup/manager_test.go index 46942b8e58..cb1ea5e4ae 100644 --- a/packages/orchestrator/pkg/sandbox/cgroup/manager_test.go +++ b/packages/orchestrator/pkg/sandbox/cgroup/manager_test.go @@ -16,6 +16,32 @@ import ( "github.com/stretchr/testify/require" ) +func requireWritableCgroup(t *testing.T) { + t.Helper() + + if os.Geteuid() != 0 { + t.Skip("test requires root privileges") + } + + probePath := filepath.Join(cgroupV2MountPoint, fmt.Sprintf("e2b-probe-%d-%d", os.Getpid(), time.Now().UnixNano())) + if err := os.Mkdir(probePath, 0o755); err != nil { + t.Skipf("test requires writable cgroup v2 filesystem: %v", err) + } + + subtreeControlPath := filepath.Join(probePath, "cgroup.subtree_control") + if _, err := os.Stat(subtreeControlPath); err != nil { + _ = os.Remove(probePath) + t.Skipf("test requires usable cgroup v2 control files: %v", err) + } + if err := os.WriteFile(subtreeControlPath, []byte("+cpu +memory"), 0); err != nil { + _ = os.Remove(probePath) + t.Skipf("test requires writable cgroup v2 subtree control: %v", err) + } + if err := os.Remove(probePath); err != nil { + t.Skipf("test requires removable cgroup v2 filesystem: %v", err) + } +} + func TestNewManager(t *testing.T) { t.Parallel() @@ -31,9 +57,7 @@ func TestNewManager(t *testing.T) { func TestManagerInitialize(t *testing.T) { t.Parallel() - if os.Geteuid() != 0 { - t.Skip("test requires root privileges") - } + requireWritableCgroup(t) ctx := t.Context() @@ -58,9 +82,7 @@ func TestManagerInitialize(t *testing.T) { func TestCgroupHandleLifecycle(t *testing.T) { t.Parallel() - if os.Geteuid() != 0 { - t.Skip("test requires root privileges") - } + requireWritableCgroup(t) ctx := t.Context() mgr, err := NewManager() @@ -96,9 +118,7 @@ func TestCgroupHandleLifecycle(t *testing.T) { func TestCgroupHandleWithProcessCreation(t *testing.T) { t.Parallel() - if os.Geteuid() != 0 { - t.Skip("test requires root privileges") - } + requireWritableCgroup(t) ctx := t.Context() mgr, err := NewManager() @@ -151,9 +171,7 @@ func TestCgroupHandleWithProcessCreation(t *testing.T) { func TestCgroupHandleNoRaceOnQuickExit(t *testing.T) { t.Parallel() - if os.Geteuid() != 0 { - t.Skip("test requires root privileges") - } + requireWritableCgroup(t) ctx := t.Context() mgr, err := NewManager() @@ -188,9 +206,7 @@ func TestCgroupHandleNoRaceOnQuickExit(t *testing.T) { func TestCgroupHandleGetStats(t *testing.T) { t.Parallel() - if os.Geteuid() != 0 { - t.Skip("test requires root privileges") - } + requireWritableCgroup(t) ctx := t.Context() mgr, err := NewManager() @@ -237,9 +253,7 @@ func TestCgroupHandleGetStats(t *testing.T) { func TestCgroupHandleGetStatsNonExistent(t *testing.T) { t.Parallel() - if os.Geteuid() != 0 { - t.Skip("test requires root privileges") - } + requireWritableCgroup(t) ctx := t.Context() mgr, err := NewManager() @@ -262,12 +276,99 @@ func TestCgroupHandleGetStatsNonExistent(t *testing.T) { assert.Contains(t, err.Error(), "failed to read cpu.stat") } -func TestCgroupHandleRemoveNonExistent(t *testing.T) { +func TestCgroupHandleKillNoProcesses(t *testing.T) { t.Parallel() - if os.Geteuid() != 0 { - t.Skip("test requires root privileges") + cgroupPath := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(cgroupPath, "cgroup.events"), []byte("populated 0\n"), 0o644)) + + handle := &CgroupHandle{ + cgroupName: "test-empty-kill", + path: cgroupPath, + } + + require.NoError(t, handle.Kill(t.Context())) +} + +func TestCgroupHandlePopulated(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + content string + write bool + want bool + }{ + { + name: "populated", + content: "populated 1\n", + write: true, + want: true, + }, + { + name: "empty", + content: "populated 0\n", + write: true, + }, + { + name: "missing file", + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + cgroupPath := t.TempDir() + if tc.write { + require.NoError(t, os.WriteFile(filepath.Join(cgroupPath, "cgroup.events"), []byte(tc.content), 0o644)) + } + + handle := &CgroupHandle{path: cgroupPath} + + got, err := handle.populated() + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) } +} + +func TestCgroupHandleKillTerminatesProcesses(t *testing.T) { + t.Parallel() + + requireWritableCgroup(t) + + ctx := t.Context() + mgr, err := NewManager() + require.NoError(t, err) + + err = mgr.Initialize(ctx) + require.NoError(t, err) + + handle, err := mgr.Create(ctx, "test-cgroup-kill") + require.NoError(t, err) + defer handle.Remove(ctx) + defer handle.ReleaseCgroupFD() + + cmd := exec.CommandContext(ctx, "sleep", "60") + cmd.SysProcAttr = &syscall.SysProcAttr{ + UseCgroupFD: true, + CgroupFD: handle.GetFD(), + } + + require.NoError(t, cmd.Start()) + t.Cleanup(func() { + _ = cmd.Process.Kill() + _ = cmd.Wait() + }) + + require.NoError(t, handle.ReleaseCgroupFD()) + require.NoError(t, handle.Kill(ctx)) + require.Error(t, cmd.Wait()) +} + +func TestCgroupHandleRemoveNonExistent(t *testing.T) { + t.Parallel() + + requireWritableCgroup(t) ctx := t.Context() mgr, err := NewManager() @@ -329,9 +430,7 @@ burst_usec 0` func TestCgroupHandlePeakReset(t *testing.T) { t.Parallel() - if os.Geteuid() != 0 { - t.Skip("test requires root privileges") - } + requireWritableCgroup(t) ctx := t.Context() mgr, err := NewManager() diff --git a/packages/orchestrator/pkg/sandbox/fc/process.go b/packages/orchestrator/pkg/sandbox/fc/process.go index 0074176a48..761a0f138d 100644 --- a/packages/orchestrator/pkg/sandbox/fc/process.go +++ b/packages/orchestrator/pkg/sandbox/fc/process.go @@ -14,7 +14,6 @@ import ( "syscall" "time" - "github.com/shirou/gopsutil/v4/process" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" @@ -187,11 +186,7 @@ func NewProcess( startScript.Value, ) - cmd.SysProcAttr = &syscall.SysProcAttr{ - Setsid: true, // Create a new session - } - - return &Process{ + p := &Process{ Versions: versions, Exit: utils.NewErrorOnce(), cmd: cmd, @@ -205,7 +200,13 @@ func NewProcess( kernelPath: startScript.KernelPath, rootfsPath: startScript.RootfsPath, - }, nil + } + + cmd.SysProcAttr = &syscall.SysProcAttr{ + Setsid: true, // Create a new session + } + + return p, nil } func (p *Process) configure( @@ -638,18 +639,6 @@ func (p *Process) Pid() (int, error) { return p.cmd.Process.Pid, nil } -// getProcessStatus returns the process status using gopsutil. -// Return values: R (running), S (sleep), T (stop), I (idle), -// Z (zombie), W (wait), L (lock), D (disk sleep / uninterruptible). -func getProcessStatus(pid int) ([]string, error) { - proc, err := process.NewProcess(int32(pid)) - if err != nil { - return nil, fmt.Errorf("process %d not found: %w", pid, err) - } - - return proc.Status() -} - func (p *Process) Stop(ctx context.Context) error { if p.cmd.Process == nil { return errors.New("fc process not started") @@ -661,7 +650,10 @@ func (p *Process) Stop(ctx context.Context) error { logger.L().Warn(ctx, "failed to remove fc metrics FIFO", zap.Error(removeErr), logger.WithSandboxID(p.files.SandboxID)) } - // Check if process has already exited. + pid := p.cmd.Process.Pid + + // Check if the Firecracker leader has already exited. Descendant cleanup is + // handled by the sandbox cgroup so Stop never signals a numeric process group. select { case <-p.Exit.Done(): logger.L().Info(ctx, "fc process already exited", logger.WithSandboxID(p.files.SandboxID)) @@ -673,6 +665,7 @@ func (p *Process) Stop(ctx context.Context) error { // this function should never fail b/c a previous context was canceled. ctx = context.WithoutCancel(ctx) + // On Linux >= 5.4, Go backs os.Process with pidfd, so Signal is safe against PID reuse. err := p.cmd.Process.Signal(syscall.SIGTERM) if err != nil { if errors.Is(err, os.ErrProcessDone) { @@ -684,38 +677,38 @@ func (p *Process) Stop(ctx context.Context) error { logger.L().Warn(ctx, "failed to send SIGTERM to fc process", zap.Error(err), logger.WithSandboxID(p.files.SandboxID)) } - go func() { - select { - // Wait 10 sec for the FC process to exit, if it doesn't, send SIGKILL. - case <-time.After(10 * time.Second): - // Check process status right before Kill — the pre-SIGTERM status - // captured above is 10s stale and no longer useful here. - status, stateErr := getProcessStatus(p.cmd.Process.Pid) - if errors.Is(stateErr, process.ErrorProcessNotRunning) { - // Process already exited, no need to send SIGKILL. - return - } else if stateErr != nil { - logger.L().Warn(ctx, "failed to get fc process status before SIGKILL", zap.Error(stateErr), logger.WithSandboxID(p.files.SandboxID)) - } + termDeadline := time.NewTimer(10 * time.Second) + defer termDeadline.Stop() - err := p.cmd.Process.Kill() - if err == nil { - logger.L().Info(ctx, "sent SIGKILL to fc process because it was not responding to SIGTERM for 10 seconds", - zap.Strings("status", status), - logger.WithSandboxID(p.files.SandboxID), - ) - } - if err != nil && !errors.Is(err, os.ErrProcessDone) { - logger.L().Warn(ctx, "failed to send SIGKILL to fc process", zap.Error(err), logger.WithSandboxID(p.files.SandboxID)) - } + select { + case <-p.Exit.Done(): + return nil + case <-termDeadline.C: + killErr := p.cmd.Process.Kill() + if killErr == nil { + logger.L().Info(ctx, "sent SIGKILL to fc process because it was not responding to SIGTERM for 10 seconds", + logger.WithSandboxID(p.files.SandboxID), + ) + } + if errors.Is(killErr, os.ErrProcessDone) { + logger.L().Info(ctx, "fc process already exited", logger.WithSandboxID(p.files.SandboxID)) - // If the FC process exited, we can return. - case <-p.Exit.Done(): - return + return nil + } + if killErr != nil && !errors.Is(killErr, os.ErrProcessDone) { + logger.L().Warn(ctx, "failed to send SIGKILL to fc process", zap.Error(killErr), logger.WithSandboxID(p.files.SandboxID)) } - }() - return nil + killDeadline := time.NewTimer(time.Second) + defer killDeadline.Stop() + + select { + case <-p.Exit.Done(): + return nil + case <-killDeadline.C: + return fmt.Errorf("fc process %d still exists after SIGKILL", pid) + } + } } func (p *Process) Pause(ctx context.Context) error { diff --git a/packages/orchestrator/pkg/sandbox/fc/process_test.go b/packages/orchestrator/pkg/sandbox/fc/process_test.go new file mode 100644 index 0000000000..84051966ef --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/fc/process_test.go @@ -0,0 +1,115 @@ +//go:build linux + +package fc + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/e2b-dev/infra/packages/shared/pkg/storage" + "github.com/e2b-dev/infra/packages/shared/pkg/utils" +) + +func TestProcessStopIsIdempotent(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + + cmd := startSetsidCommand(t, ctx, "sleep", "60") + pid := cmd.Process.Pid + + exit := utils.NewErrorOnce() + waitDone := make(chan struct{}) + go func() { + defer close(waitDone) + _ = cmd.Wait() + exit.SetError(nil) + }() + t.Cleanup(func() { + _ = syscall.Kill(-pid, syscall.SIGKILL) + <-waitDone + }) + + p := &Process{ + cmd: cmd, + Exit: exit, + files: &storage.SandboxFiles{SandboxID: "test"}, + metricsPath: t.TempDir() + "/metrics.fifo", + } + require.NoError(t, p.Stop(ctx)) + require.NoError(t, p.Stop(ctx)) +} + +func TestProcessStopDoesNotSignalProcessGroupAfterLeaderExit(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(t.Context(), time.Minute) + defer cancel() + + childPidPath := filepath.Join(t.TempDir(), "child.pid") + cmd := startSetsidCommand(t, ctx, "bash", "-c", "sleep 60 & echo $! > \"$1\"", "sh", childPidPath) + pid := cmd.Process.Pid + + exit := utils.NewErrorOnce() + waitDone := make(chan struct{}) + go func() { + defer close(waitDone) + _ = cmd.Wait() + exit.SetError(nil) + }() + + var childPid int + require.Eventually(t, func() bool { + data, err := os.ReadFile(childPidPath) + if err != nil { + return false + } + + childPid, err = strconv.Atoi(strings.TrimSpace(string(data))) + + return err == nil && childPid > 0 + }, time.Second, 10*time.Millisecond) + t.Cleanup(func() { + _ = syscall.Kill(childPid, syscall.SIGKILL) + _ = syscall.Kill(-pid, syscall.SIGKILL) + <-waitDone + }) + + require.Eventually(t, func() bool { + select { + case <-waitDone: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + p := &Process{ + cmd: cmd, + Exit: exit, + files: &storage.SandboxFiles{SandboxID: "test"}, + metricsPath: filepath.Join(t.TempDir(), "metrics.fifo"), + } + require.NoError(t, p.Stop(ctx)) + require.NoError(t, syscall.Kill(childPid, 0)) +} + +func startSetsidCommand(t *testing.T, ctx context.Context, name string, args ...string) *exec.Cmd { + t.Helper() + + cmd := exec.CommandContext(ctx, name, args...) + cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} + require.NoError(t, cmd.Start()) + + return cmd +} diff --git a/packages/orchestrator/pkg/sandbox/sandbox.go b/packages/orchestrator/pkg/sandbox/sandbox.go index a97344f481..7e6500a6e2 100644 --- a/packages/orchestrator/pkg/sandbox/sandbox.go +++ b/packages/orchestrator/pkg/sandbox/sandbox.go @@ -1018,9 +1018,20 @@ func (s *Sandbox) doStop(ctx context.Context) error { errs = append(errs, fmt.Errorf("failed to stop FC: %w", fcStopErr)) } - // The process exited, we can continue with the rest of the cleanup. - // We could use select with ctx.Done() to wait for cancellation, but if the process is not exited the whole cleanup will be in a bad state and will result in unexpected behavior. - <-s.process.Exit.Done() + cgroupKillErr := s.cgroupHandle.Kill(ctx) + if cgroupKillErr != nil { + errs = append(errs, fmt.Errorf("failed to kill sandbox cgroup: %w", cgroupKillErr)) + } + + // The process should exit before the rest of cleanup, but memory shutdown + // must still run if the wait context is canceled so UFFD can exit. + // FC's own exit error is reported via the exit waiters, not as a stop + // failure, so only a canceled wait counts as an error here. + select { + case <-s.process.Exit.Done(): + case <-ctx.Done(): + errs = append(errs, fmt.Errorf("failed waiting for FC exit: %w", ctx.Err())) + } uffdStopErr := s.Resources.memory.Stop() if uffdStopErr != nil {