//go:build !windows package devstack import ( "errors" "fmt" "io" "os" "os/exec" "path/filepath" "strconv" "strings" "syscall" "time" log "github.com/sirupsen/logrus" "golang.org/x/sys/unix" ) // startService launches sp in its own process group with its output in // logPath, and returns the record that identifies the process. func startService(sp Spec, logPath string) (ServiceRecord, error) { record, _, err := launchProcess(sp, logPath, ServiceIO{}, false) return record, err } func launchService(sp Spec, logPath string, streams ServiceIO) (ServiceRecord, <-chan error, error) { return launchProcess(sp, logPath, streams, true) } type serviceOutputCopy struct { reader *os.File writer *os.File output io.Writer done chan error } func launchProcess(sp Spec, logPath string, streams ServiceIO, foreground bool) (ServiceRecord, <-chan error, error) { if err := os.MkdirAll(filepath.Dir(logPath), 0o755); err != nil { return ServiceRecord{}, nil, err } logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o644) if err != nil { return ServiceRecord{}, nil, err } if _, err := fmt.Fprintf(logFile, "cwd: %s\ncommand: %s\n\n", sp.Dir, strings.Join(sp.Command, " ")); err != nil { _ = logFile.Close() return ServiceRecord{}, nil, err } startedAt := time.Now().UTC().Truncate(time.Second) cmd := exec.Command(sp.Command[0], sp.Command[1:]...) cmd.Dir = sp.Dir cmd.Env = sp.Env cmd.Stdout = logFile cmd.Stderr = logFile var copies []serviceOutputCopy defer func() { for _, copy := range copies { _ = copy.writer.Close() } }() if streams.Stdout != nil { reader, writer, err := os.Pipe() if err != nil { _ = logFile.Close() return ServiceRecord{}, nil, err } copies = append(copies, serviceOutputCopy{reader, writer, io.MultiWriter(logFile, streams.Stdout), make(chan error, 1)}) cmd.Stdout = writer } if streams.Stderr != nil { reader, writer, err := os.Pipe() if err != nil { for _, copy := range copies { _ = copy.reader.Close() } _ = logFile.Close() return ServiceRecord{}, nil, err } copies = append(copies, serviceOutputCopy{reader, writer, io.MultiWriter(logFile, streams.Stderr), make(chan error, 1)}) cmd.Stderr = writer } cmd.Stdin = streams.Stdin cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} terminalInput := -1 var inputReader, inputWriter *os.File if input, ok := streams.Stdin.(*os.File); ok && input == os.Stdin { fd := int(input.Fd()) foreground, err := unix.IoctlGetInt(fd, unix.TIOCGPGRP) if err == nil && foreground == syscall.Getpgrp() { // Keep terminal signals on ODS so it can enforce the stop deadline. terminalInput, err = unix.Open("/dev/tty", unix.O_RDONLY|unix.O_NONBLOCK|unix.O_CLOEXEC, 0) if err == nil { inputReader, inputWriter, err = os.Pipe() } if err != nil { if terminalInput >= 0 { _ = unix.Close(terminalInput) } for _, copy := range copies { _ = copy.reader.Close() } _ = logFile.Close() return ServiceRecord{}, nil, err } cmd.Stdin = inputReader } } if err := cmd.Start(); err != nil { if inputReader != nil { _ = inputReader.Close() _ = inputWriter.Close() _ = unix.Close(terminalInput) } for _, copy := range copies { _ = copy.reader.Close() } _ = logFile.Close() return ServiceRecord{}, nil, fmt.Errorf("start %s: %w", sp.Name, err) } stopInput := func() {} if terminalInput >= 0 { _ = inputReader.Close() stop := make(chan struct{}) inputDone := make(chan struct{}) go func() { defer close(inputDone) defer func() { _ = inputWriter.Close() }() defer func() { _ = unix.Close(terminalInput) }() // macOS cannot poll /dev/tty; poll stdin and read without blocking. poll := []unix.PollFd{{Fd: int32(os.Stdin.Fd()), Events: unix.POLLIN}} buffer := make([]byte, 4096) for { select { case <-stop: return default: } _, err := unix.Poll(poll, 100) if errors.Is(err, unix.EINTR) { continue } if err != nil || poll[0].Revents&(unix.POLLERR|unix.POLLNVAL) != 0 { return } if poll[0].Revents&(unix.POLLIN|unix.POLLHUP) == 0 { continue } n, err := unix.Read(terminalInput, buffer) if errors.Is(err, unix.EINTR) || errors.Is(err, unix.EAGAIN) { continue } if err != nil || n == 0 { return } if _, err := inputWriter.Write(buffer[:n]); err != nil { return } } }() stopInput = func() { close(stop) _ = inputWriter.Close() <-inputDone } } pid := cmd.Process.Pid record := ServiceRecord{ Name: sp.Name, Slug: sp.Slug, Command: sp.Command, Dir: sp.Dir, SpecHash: sp.Hash(), Port: sp.Port, Health: sp.Health, Optional: sp.Optional, PID: pid, PGID: pid, StartMarker: startMarker(pid), StartedAt: startedAt, Log: logPath, } for _, copy := range copies { go func() { _, err := io.Copy(copy.output, copy.reader) _ = copy.reader.Close() copy.done <- err }() } if foreground { rememberGroupMembers(&record) } supervised := record supervised.Members = make(map[int]string, len(record.Members)) for pid, marker := range record.Members { supervised.Members[pid] = marker } done := make(chan error, 1) go func() { var err error if foreground { err = waitForegroundProcess(cmd, supervised, copies) } else { err = cmd.Wait() } stopInput() _ = logFile.Close() done <- err close(done) }() return record, done, nil } func waitForegroundProcess(cmd *exec.Cmd, record ServiceRecord, copies []serviceOutputCopy) error { for pidAlive(record.PID) { time.Sleep(100 * time.Millisecond) } // Our unreaped child reserves its PID and PGID until this final capture. rememberGroupMembers(&record) err := cmd.Wait() if left := stopRecordsWithKillWait([]ServiceRecord{record}, 250*time.Millisecond, time.Second); len(left) > 0 { err = errors.Join(err, &serviceCleanupError{record: record}) } deadline := time.NewTimer(time.Second) defer deadline.Stop() for _, copy := range copies { select { case copyErr := <-copy.done: err = errors.Join(err, copyErr) case <-deadline.C: for _, pending := range copies { _ = pending.reader.Close() } return errors.Join(err, fmt.Errorf("service output did not drain before timeout")) } } return err } func signalService(rec ServiceRecord, signal os.Signal) error { if !groupIsOurs(rec) { return nil } err := syscall.Kill(-rec.PGID, signal.(syscall.Signal)) if errors.Is(err, syscall.ESRCH) { return nil } return err } // isCurrent reports whether the recorded leader process is still the one // that was started: alive, in its own group, with the same start time. func isCurrent(rec ServiceRecord) bool { if rec.PID <= 0 || !pidAlive(rec.PID) { return false } pgid, err := syscall.Getpgid(rec.PID) if err != nil || pgid != rec.PGID { return false } return rec.StartMarker != "" && startMarker(rec.PID) == rec.StartMarker } // pidAlive reports whether pid exists and is not a zombie. func pidAlive(pid int) bool { err := syscall.Kill(pid, 0) if err != nil && !errors.Is(err, syscall.EPERM) { return false } out, err := exec.Command("ps", "-p", strconv.Itoa(pid), "-o", "stat=").Output() if err != nil { return false } return !strings.HasPrefix(strings.TrimSpace(string(out)), "Z") } // groupAlive reports whether any live process is left in the group. func groupAlive(pgid int) bool { if pgid <= 1 { return false } err := syscall.Kill(-pgid, 0) if err != nil && !errors.Is(err, syscall.EPERM) { return false } return len(groupMembers(pgid)) > 0 } // startMarker returns the process start time as ps prints it, which with the // pid identifies one process incarnation. func startMarker(pid int) string { out, err := exec.Command("ps", "-p", strconv.Itoa(pid), "-o", "lstart=").Output() if err != nil { return "" } return strings.TrimSpace(string(out)) } // groupIsOurs decides whether signalling the group is safe. The leader // proves it while alive. After the leader exits, the group is still ours // while a member we recorded (pid and start time) is still running: a // process group id cannot be reused until the group is empty. func groupIsOurs(rec ServiceRecord) bool { if isCurrent(rec) { return true } if pidAlive(rec.PID) { return false } members := groupMembers(rec.PGID) for _, pid := range members { if marker := rec.Members[pid]; marker != "" && marker == startMarker(pid) { return true } } return false } func rememberGroup(rec *ServiceRecord) { if !isCurrent(*rec) { return } rememberGroupMembers(rec) } func rememberGroupMembers(rec *ServiceRecord) { if rec.Members == nil { rec.Members = map[int]string{} } for _, pid := range groupMembers(rec.PGID) { if marker := startMarker(pid); marker != "" { rec.Members[pid] = marker } } } // groupMembers lists the live (non-zombie) pids in a process group. func groupMembers(pgid int) []int { out, err := exec.Command("ps", "-A", "-o", "pid=", "-o", "pgid=", "-o", "stat=").Output() if err != nil { return nil } var pids []int for _, line := range strings.Split(string(out), "\n") { fields := strings.Fields(line) if len(fields) < 3 || strings.HasPrefix(fields[2], "Z") { continue } pid, err1 := strconv.Atoi(fields[0]) gid, err2 := strconv.Atoi(fields[1]) if err1 == nil && err2 == nil && gid == pgid { pids = append(pids, pid) } } return pids } // stopRecords terminates the groups of recs that are still ours and waits // for every member to exit, escalating to SIGKILL after timeout. It returns // the records that still have members alive. func stopRecords(recs []ServiceRecord, timeout time.Duration) []ServiceRecord { postKill := 5 * time.Second if timeout == 0 { postKill = 0 } return stopRecordsWithKillWait(recs, timeout, postKill) } func stopRecordsWithKillWait(recs []ServiceRecord, timeout, postKill time.Duration) []ServiceRecord { var active []ServiceRecord var unresolved []ServiceRecord for _, rec := range recs { rememberGroup(&rec) if rec.PGID <= 1 || !groupIsOurs(rec) { if rec.PGID > 1 && !pidAlive(rec.PID) && groupAlive(rec.PGID) { log.Warnf("Cannot prove ownership of surviving process group %d for %s", rec.PGID, rec.Name) unresolved = append(unresolved, rec) } continue } log.Debugf("Stopping %s (pgid %d)", rec.Name, rec.PGID) _ = syscall.Kill(-rec.PGID, syscall.SIGTERM) active = append(active, rec) } var killed []ServiceRecord for _, rec := range waitGroups(active, timeout) { if !groupIsOurs(rec) { // Still alive, but no recorded member proves the group is ours // any more: report it rather than kill or forget it. log.Warnf("Cannot prove ownership of surviving process group %d for %s", rec.PGID, rec.Name) unresolved = append(unresolved, rec) continue } log.Warnf("%s did not stop in %s, killing its process group", rec.Name, timeout) _ = syscall.Kill(-rec.PGID, syscall.SIGKILL) killed = append(killed, rec) } return append(waitGroups(killed, postKill), unresolved...) } func waitGroups(recs []ServiceRecord, timeout time.Duration) []ServiceRecord { deadline := time.Now().Add(timeout) for len(recs) > 0 { // A group stays tracked while any member lives, whether or not // ownership can still be proven. var left []ServiceRecord for _, rec := range recs { if groupAlive(rec.PGID) { left = append(left, rec) } } recs = left if len(recs) == 0 || time.Now().After(deadline) { break } time.Sleep(100 * time.Millisecond) } return recs } // portHeldByGroup reports whether a process in the group listens on port. // Without lsof it falls back to the port being bound at all. func portHeldByGroup(port, pgid int) bool { out, err := exec.Command("lsof", "-t", "-nP", "-iTCP:"+strconv.Itoa(port), "-sTCP:LISTEN").Output() if err != nil { if _, lookErr := exec.LookPath("lsof"); lookErr != nil { return !bindable(port) } return false } members := map[int]bool{} for _, pid := range groupMembers(pgid) { members[pid] = true } for _, field := range strings.Fields(string(out)) { if pid, err := strconv.Atoi(field); err == nil && members[pid] { return true } } return false }