diff --git a/runner/go.mod b/runner/go.mod index f338e78ed..95f3a2f42 100644 --- a/runner/go.mod +++ b/runner/go.mod @@ -7,7 +7,6 @@ require ( github.com/alexellis/go-execute/v2 v2.2.1 github.com/bluekeyes/go-gitdiff v0.7.2 github.com/codeclysm/extract/v4 v4.0.0 - github.com/creack/pty v1.1.24 github.com/docker/docker v26.0.0+incompatible github.com/docker/go-connections v0.5.0 github.com/docker/go-units v0.5.0 diff --git a/runner/go.sum b/runner/go.sum index 655ea59dc..9eefc2a07 100644 --- a/runner/go.sum +++ b/runner/go.sum @@ -34,8 +34,6 @@ github.com/codeclysm/extract/v4 v4.0.0 h1:H87LFsUNaJTu2e/8p/oiuiUsOK/TaPQ5wxsjPn github.com/codeclysm/extract/v4 v4.0.0/go.mod h1:SFju1lj6as7FvUgalpSct7torJE0zttbJUWtryPRG6s= github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I= github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo= -github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= -github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= github.com/cyphar/filepath-securejoin v0.2.4 h1:Ugdm7cg7i6ZK6x3xDF1oEu1nfkyfH53EtKeQYTC3kyg= github.com/cyphar/filepath-securejoin v0.2.4/go.mod h1:aPGpWjXOXUn2NCNjFvBE6aRxGGx79pTxQpKOJNYHHl4= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/runner/internal/runner/executor/executor.go b/runner/internal/runner/executor/executor.go index 45cb42963..d78bf629f 100644 --- a/runner/internal/runner/executor/executor.go +++ b/runner/internal/runner/executor/executor.go @@ -17,7 +17,6 @@ import ( "syscall" "time" - "github.com/creack/pty" "github.com/dstackai/ansistrip" "github.com/prometheus/procfs" "github.com/sirupsen/logrus" @@ -85,7 +84,10 @@ type RunExecutor struct { runnerLogs *appendWriter timestamp *MonotonicTimestamp - killDelay time.Duration + killDelay time.Duration + // How long output may go on being copied after the command has exited, before the pty + // master is closed. Only reached when the job leaves processes holding the terminal open. + logsDrainDelay time.Duration connectionTracker ConnectionTracker } @@ -121,6 +123,7 @@ func NewRunExecutor(tempDir string, dstackDir string, currentUser linuxuser.User timestamp: timestamp, killDelay: 10 * time.Second, + logsDrainDelay: 2 * time.Second, connectionTracker: connectionTracker, }, nil } @@ -595,38 +598,21 @@ func (ex *RunExecutor) execJob(ctx context.Context, jobLogFile io.Writer) error return fmt.Errorf("start command: %w", err) } defer func() { _ = ptm.Close() }() - defer func() { _ = cmd.Wait() }() // release resources if copy fails stripper := ansistrip.NewWriter(ex.jobLogs, AnsiStripFlushInterval, AnsiStripMaxDelay, MaxBufferSize) logger := io.MultiWriter(jobLogFile, ex.jobWsLogs, stripper) - if err := ex.copyOutputWithQuota(cmd, ptm, stripper, logger); err != nil { - return err - } - if err = cmd.Wait(); err != nil { - return fmt.Errorf("wait for command: %w", err) - } - return nil -} - -// copyOutputWithQuota streams process output through the log pipeline and -// monitors for log quota exceeded. The quota signal is out-of-band (via channel) -// because the ansistrip writer is async and swallows downstream write errors. -func (ex *RunExecutor) copyOutputWithQuota(cmd *exec.Cmd, ptm io.Reader, stripper io.Closer, logger io.Writer) error { copyDone := make(chan error, 1) go func() { - _, err := io.Copy(logger, ptm) - copyDone <- err + _, copyErr := io.Copy(logger, ptm) + copyDone <- copyErr }() - // Wait for either io.Copy to finish or quota to be exceeded. - var copyErr error - select { - case copyErr = <-copyDone: - case <-ex.jobLogs.QuotaExceeded(): - _ = cmd.Process.Kill() - <-copyDone - } + stopQuotaWatch := watchLogQuota(cmd, ex.jobLogs.QuotaExceeded()) + defer stopQuotaWatch() + + waitErr := cmd.Wait() + copyErr := ex.finishOutputCopy(ctx, ptm, copyDone) // Flush the ansistrip buffer — may also trigger quota exceeded. _ = stripper.Close() @@ -636,13 +622,52 @@ func (ex *RunExecutor) copyOutputWithQuota(cmd *exec.Cmd, ptm io.Reader, strippe return ErrLogQuotaExceeded default: } - if copyErr != nil && !isPtyError(copyErr) { return fmt.Errorf("copy command output: %w", copyErr) } + if waitErr != nil { + return fmt.Errorf("wait for command: %w", waitErr) + } return nil } +// finishOutputCopy waits for the output copy to finish, bounding how long it may run after +// the command has exited. +// +// A read on the pty master returns EIO only once every process holding the slave has closed +// it. A job that leaves a process behind -- a `cmd &` job, a daemon -- would otherwise keep +// the copy running forever, and with it the executor: the job state would never be reported +// and the run would hang until the container is destroyed. Give the output the command has +// already written a moment to drain, then close the master, which unblocks the read. +func (ex *RunExecutor) finishOutputCopy(ctx context.Context, ptm *os.File, copyDone <-chan error) error { + select { + case copyErr := <-copyDone: + return copyErr + case <-time.After(ex.logsDrainDelay): + } + log.Warning(ctx, "The job left processes holding the terminal open, stopped reading output") + _ = ptm.Close() + <-copyDone // fails with os.ErrClosed, which is what closing the master is for + return nil +} + +// watchLogQuota kills the command if the job exceeds its log quota. Output keeps being copied +// until the command exits, so a full pty buffer cannot keep it from exiting. +// +// The quota signal is out-of-band (via channel) because the ansistrip writer is async and +// swallows downstream write errors. +func watchLogQuota(cmd *exec.Cmd, quotaExceeded <-chan struct{}) (stop func()) { + done := make(chan struct{}) + go func() { + select { + case <-quotaExceeded: + _ = cmd.Process.Kill() + case <-done: + } + }() + return func() { close(done) } +} + // setupGitCredentials must be called from Run after setJobUser func (ex *RunExecutor) setupGitCredentials(ctx context.Context) (func(), error) { if ex.repoCredentials == nil { @@ -704,6 +729,37 @@ func (ex *RunExecutor) setupGitCredentials(ctx context.Context) (func(), error) return nil, fmt.Errorf("unknown protocol %s", ex.repoCredentials.GetProtocol()) } +// openPty opens a new pty pair. +// +// The master is opened non-blocking so that Go registers it with the runtime poller. A +// blocking os.File never reaches the poller, and closing one does not interrupt a Read already +// in flight -- the close is deferred until that read returns, which may be never. execJob +// relies on closing the master to stop reading output. +func openPty() (*os.File, *os.File, error) { + ptmFd, err := unix.Open("/dev/ptmx", unix.O_RDWR|unix.O_NOCTTY|unix.O_NONBLOCK, 0) + if err != nil { + return nil, nil, fmt.Errorf("open pty master: %w", err) + } + ptm := os.NewFile(uintptr(ptmFd), "/dev/ptmx") + + if err := unix.IoctlSetPointerInt(ptmFd, unix.TIOCSPTLCK, 0); err != nil { + _ = ptm.Close() + return nil, nil, fmt.Errorf("unlock pty slave: %w", err) + } + ptsNum, err := unix.IoctlGetInt(ptmFd, unix.TIOCGPTN) + if err != nil { + _ = ptm.Close() + return nil, nil, fmt.Errorf("get pty slave number: %w", err) + } + ptsName := fmt.Sprintf("/dev/pts/%d", ptsNum) + pts, err := os.OpenFile(ptsName, os.O_RDWR|unix.O_NOCTTY, 0) + if err != nil { + _ = ptm.Close() + return nil, nil, fmt.Errorf("open pty slave: %w", err) + } + return ptm, pts, nil +} + func isPtyError(err error) bool { /* read /dev/ptmx: input/output error */ var e *os.PathError @@ -715,9 +771,9 @@ func isPtyError(err error) bool { // * controlling terminal is properly set (cmd.Extrafiles, Cmd.SysProcAttr.Ctty) // * owner of slave pty is changed to the child process uid func startCommand(cmd *exec.Cmd) (*os.File, error) { - ptm, pts, err := pty.Open() + ptm, pts, err := openPty() if err != nil { - return nil, fmt.Errorf("open pty: %w", err) + return nil, err } defer func() { _ = pts.Close() }() diff --git a/runner/internal/runner/executor/executor_test.go b/runner/internal/runner/executor/executor_test.go index 90a942328..acbf49381 100644 --- a/runner/internal/runner/executor/executor_test.go +++ b/runner/internal/runner/executor/executor_test.go @@ -10,7 +10,9 @@ import ( "os/exec" "path" "path/filepath" + "strconv" "strings" + "syscall" "testing" "time" @@ -162,6 +164,49 @@ func TestExecutor_LogQuota(t *testing.T) { assert.Equal(t, schemas.JobStateFailed, lastState.State) } +// A job that leaves a process behind keeps the pty slave open, so reading the master never +// returns EIO. The executor must stop reading anyway instead of hanging forever. +func TestExecutor_SurvivingProcessDoesNotHangRun(t *testing.T) { + if testing.Short() { + t.Skip() + } + + ex := makeTestExecutor(t) + ex.logsDrainDelay = 500 * time.Millisecond + // `-i` as the server sends it: job control puts the backgrounded process in its own + // process group, so it does not get the SIGHUP the kernel sends to the foreground group + // when the shell exits, and goes on holding the pty slave open. It must outlive the + // assertion below, or the executor would be let off the hook by the process exiting. + pidPath := filepath.Join(t.TempDir(), "survivor.pid") + ex.jobSpec.Commands = []string{ + "/bin/bash", "-i", "-c", + fmt.Sprintf("sleep 300 & echo $! > %s; echo done", pidPath), + } + t.Cleanup(func() { killRecordedPid(t, pidPath) }) + makeCodeTar(t, ex) + + runDone := make(chan error, 1) + go func() { runDone <- ex.Run(t.Context()) }() + + select { + case err := <-runDone: + assert.NoError(t, err) + case <-time.After(20 * time.Second): + t.Fatal("Run did not return while a process left by the job held the terminal open") + } + + history := ex.GetHistory(0) + lastState := history.JobStates[len(history.JobStates)-1] + assert.Equal(t, schemas.JobStateDone, lastState.State) + + // Output written before the command exited must still be drained. + var logs strings.Builder + for _, event := range history.JobLogs { + logs.Write(event.Message) + } + assert.Contains(t, logs.String(), "done") +} + func TestExecutor_RemoteRepo(t *testing.T) { if testing.Short() { t.Skip() @@ -482,3 +527,16 @@ func combineLogMessages(logHistory []schemas.LogEvent) string { } return logOutput.String() } + +func killRecordedPid(t *testing.T, pidPath string) { + t.Helper() + data, err := os.ReadFile(pidPath) + if err != nil { + return + } + pid, err := strconv.Atoi(strings.TrimSpace(string(data))) + if err != nil { + return + } + _ = syscall.Kill(pid, syscall.SIGKILL) +}