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 {