Launch process in a separate session and kill its spawned subprocesses

client/commands.go

@@ -156,14 +156,15 @@ func GetDeployment(name string) (server.DeploymentFullInfo, error) {
 	return fetch.Get[server.DeploymentFullInfo]("/deployments/" + name)
 }
 
-func DeleteDeployment(name string) {
+func DeleteDeployment(name string) error {
 	versionsWarning()
 	_, err := fetch.Delete[fetch.Empty]("/deployments/" + name)
 	if err != nil {
 		fmt.Println(err)
 	} else {
 		fmt.Printf("Successfully deleted '%s' deployment\n", name)
 	}
+	return err
 }
 
 func Apply(path string) []error {

common/unix/commands.go

@@ -3,6 +3,7 @@ package unix
 import (
 	"context"
 	"fmt"
+	xunix "golang.org/x/sys/unix"
 	"io"
 	"log"
 	"os"
@@ -20,7 +21,6 @@ func TerminateProcessTimeout(pid int, timeout time.Duration) error {
 	return TerminateProcess(ctx, pid)
 }
 
-// Blocking. Once context expires, it sends SIGKILL.
 func TerminateProcess(ctx context.Context, pid int) error {
 	process, err := os.FindProcess(pid)
 	if err != nil {
@@ -52,6 +52,37 @@ func TerminateProcess(ctx context.Context, pid int) error {
 	}
 }
 
+// Blocking. Once context expires, it sends SIGKILL.
+func TerminateSession(ctx context.Context, parentPid int) error {
+	// syscall doesn't have Getsid for Linix, and it has been frozen.
+	sid, err := xunix.Getsid(parentPid)
+	if err != nil {
+		return err
+	}
+	err = syscall.Kill(-sid, syscall.SIGTERM)
+	if err != nil {
+		return err
+	}
+
+	// Wait until the process terminates, but think of the children!
+	for {
+		select {
+		case <-ctx.Done():
+			err = syscall.Kill(-sid, syscall.SIGKILL)
+			if err != nil {
+				log.Printf("context deadline exceeded: failed to kill %d session: %s\n", sid, err)
+				return err
+			}
+			return context.DeadlineExceeded
+		default:
+			if !IsProcessAlive(parentPid) {
+				return nil
+			}
+			time.Sleep(5 * time.Millisecond)
+		}
+	}
+}
+
 func IsProcessAlive(pid int) bool {
 	// 'ps -o pid= -p $PID' command works on MacOS and Linux
 	res, err := exec.Command("ps", "-o", "pid=", "-o", "command=", "-p", strconv.Itoa(pid)).Output()

common/unix/commands_test.go

@@ -38,12 +38,17 @@ func TestTerminateProcess(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error launching process: %s", err)
 	}
+	pid := cmd.Process.Pid
 
 	ctx, _ := context.WithTimeout(context.Background(), 2*time.Second)
 	err = TerminateProcess(ctx, cmd.Process.Pid)
 	if err != nil {
 		t.Fatalf("failed to terminated the process: %s", err)
 	}
+
+	if IsProcessAlive(pid) {
+		t.Errorf("process is still alive")
+	}
 }
 
 func TestKill(t *testing.T) {

common/version.go

@@ -1,3 +1,3 @@
 package common
 
-const YetisVersion = "v0.4.1"
+const YetisVersion = "v0.4.2"

go.mod

@@ -4,4 +4,7 @@ go 1.23
 
 require sigs.k8s.io/yaml v1.4.0
 
-require github.com/glossd/fetch v1.0.1 // indirect
+require (
+	github.com/glossd/fetch v1.0.1 // indirect
+	golang.org/x/sys v0.29.0 // indirect
+)

go.sum

@@ -36,6 +36,8 @@ github.com/glossd/fetch v1.0.1 h1:k43IPNBDDjzDOpGPmj9jukU3zKT8yWtyGK4IMMkHl3I=
 github.com/glossd/fetch v1.0.1/go.mod h1:zIV0m9x5g9z4b0urwELyLJRI+FisrnnAH0I/pE+vJ1k=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
+golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E=

itests/server_test.go

@@ -144,3 +144,30 @@ func TestShutdown_DeleteDeployments(t *testing.T) {
 		t.Fatal("server should've stopped")
 	}
 }
+
+func TestDeleteAllDeploymentChildProcesses(t *testing.T) {
+	//unix.KillByPort(27000, true)
+	//unix.KillByPort(27001, true)
+	go server.Run("")
+	t.Cleanup(server.Stop)
+	time.Sleep(time.Millisecond)
+	errs := client.Apply(pwd(t) + "/specs/subproc.yaml")
+	if len(errs) != 0 {
+		t.Fatalf("apply errors: %v", errs)
+	}
+
+	checkDeploymentRunning(t, "subproc")
+	// check subprocess is running
+	if !common.IsPortOpenRetry(27001, 20*time.Millisecond, 5) {
+		t.Fatal("subprocess isn't running")
+	}
+	err := client.DeleteDeployment("subproc")
+	assert(t, err, nil)
+
+	if !common.IsPortCloseRetry(27000, 20*time.Millisecond, 5) {
+		t.Errorf("main process should be dead")
+	}
+	if !common.IsPortCloseRetry(27001, 20*time.Millisecond, 5) {
+		t.Errorf("subprocess should be dead")
+	}
+}

itests/specs/cmd/subproc/main.go

@@ -0,0 +1,17 @@
+package main
+
+import (
+	"net/http"
+	"os/exec"
+)
+
+func main() {
+	err := exec.Command("nc", "-lk", "27001").Start()
+	if err != nil {
+		panic(err)
+	}
+	err = http.ListenAndServe(":27000", nil)
+	if err != nil {
+		panic(err)
+	}
+}

itests/specs/subproc.yaml

@@ -0,0 +1,10 @@
+spec:
+  name: subproc
+  preCmd: go build -o main cmd/subproc/main.go
+  cmd: ./main
+  logdir: stdout
+  livenessProbe:
+    tcpSocket:
+      port: 27000
+    initialDelaySeconds: 1
+    periodSeconds: 0.1

server/handlers_deployment.go

@@ -50,6 +50,12 @@ func CreateOrRestartDeployment(req fetch.Request[common.DeploymentSpec]) (*CRDep
 		return &CRDeploymentResponse{Existed: true}, nil
 	}
 
+	if spec.LivenessProbe.Port() > 0 {
+		if common.IsPortOpen(spec.LivenessProbe.Port()) {
+			return nil, fmt.Errorf("port of livenessProbe %d is already busy", spec.LivenessProbe.Port())
+		}
+	}
+
 	// Begin creating the deployment
 	spec, err := setYetisPortEnv(spec.WithDefaults().(common.DeploymentSpec))
 	if err != nil {
@@ -238,7 +244,7 @@ func DeleteDeployment(r fetch.Request[fetch.Empty]) error {
 
 	updateDeploymentStatus(name, Terminating)
 
-	err := terminateProcess(r.Context, d)
+	err := terminateProcess(r.Context, d.pid)
 	if err != nil {
 		return err
 	}
@@ -331,7 +337,7 @@ func restartDeployment(ctx context.Context, name string, reapplySpec *common.Dep
 		}
 
 	} else {
-		err := terminateProcess(ctx, oldDeployment)
+		err := terminateProcess(ctx, oldDeployment.pid)
 		if err != nil {
 			return fmt.Errorf("failed to terminate deployment's process: %s", err)
 		}

server/liveness.go

@@ -141,7 +141,7 @@ func heartbeat(deploymentName string, restartsLimit int) heartbeatResult {
 		updateDeploymentStatus(oldSpec.Name, Terminating)
 		ctx, cancelCtx := context.WithTimeout(context.Background(), oldSpec.LivenessProbe.PeriodDuration())
 		defer cancelCtx()
-		err := terminateProcess(ctx, p)
+		err := terminateProcess(ctx, p.pid)
 		if err != nil {
 			log.Printf("failed to terminate process, deployment=%s, pid=%d\n", oldSpec.Name, p.pid)
 		} else {

server/process.go

@@ -13,6 +13,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"syscall"
 )
 
 var logNamePattern = regexp.MustCompile("^[a-zA-Z]+-(\\d+).log$")
@@ -69,6 +70,7 @@ func launchProcessWithOut(c common.DeploymentSpec, w io.Writer, wait bool) (int,
 		cmd.Stderr = w
 	}
 	cmd.Dir = c.Workdir
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
 	if wait {
 		err = cmd.Run()
 	} else {
@@ -126,14 +128,12 @@ func getLogCounter(name, logDir string) int {
 	return highest
 }
 
-func terminateProcess(ctx context.Context, r resource) error {
-	if r.getPid() != 0 {
-		err := unix.TerminateProcess(ctx, r.getPid())
+func terminateProcess(ctx context.Context, pid int) error {
+	if pid != 0 {
+		err := unix.TerminateSession(ctx, pid)
 		if err != nil && err != context.DeadlineExceeded {
 			return err
 		}
 	}
-	// todo instead of killing by port, terminate function should terminate all children as well.
-	unix.KillByPort(r.getPort(), false)
 	return nil
 }