Merge pull request #33 from pfnet-research/kill-task-subprocesses-cle…

…anly

Worker should kill task handler process and its descendant processes cleanly when timeout

.github/workflows/ci.yaml

@@ -18,7 +18,7 @@ jobs:
     - name: Set up Go 1.x
       uses: actions/setup-go@v2
       with:
-        go-version: ^1.13
+        go-version: ^1.14
       id: go
     - name: Check out code into the Go module directory
       uses: actions/checkout@v2
@@ -50,7 +50,7 @@ jobs:
       - name: Set up Go 1.x
         uses: actions/setup-go@v2
         with:
-          go-version: ^1.13
+          go-version: ^1.14
         id: go
       - name: Check out
         uses: actions/checkout@v2

.github/workflows/release.yaml

@@ -17,7 +17,7 @@ jobs:
       - name: Set up Go 1.x
         uses: actions/setup-go@v2
         with:
-          go-version: ^1.13
+          go-version: ^1.14
         id: go
       - name: Check out
         uses: actions/checkout@v2
@@ -42,7 +42,7 @@ jobs:
       - name: Set up Go 1.x
         uses: actions/setup-go@v2
         with:
-          go-version: ^1.13
+          go-version: ^1.14
         id: go
       - name: Check out
         uses: actions/checkout@v2

.goreleaser.yaml

@@ -9,7 +9,6 @@ builds:
   binary: pftaskqueue
   goos:
   - linux
-  - windows
   - darwin
   goarch:
   - amd64

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.13.5 as builder
+FROM golang:1.14 as builder
 ARG RELEASE
 ARG VERSION
 WORKDIR /workspace

go.mod

@@ -1,6 +1,6 @@
 module github.com/pfnet-research/pftaskqueue
 
-go 1.13
+go 1.14
 
 require (
 	github.com/MakeNowJust/heredoc/v2 v2.0.1

pkg/worker/worker.go

@@ -31,6 +31,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"syscall"
 	"time"
 
 	"github.com/pfnet-research/pftaskqueue/pkg/apis/taskqueue"
@@ -301,7 +302,11 @@ func (w *Worker) runCommand(logger zerolog.Logger, t *task.Task) (task.TaskResul
 
 	cmdCtx, cmdCtxCancel := context.WithTimeout(w.ctx, t.Spec.ActualTimeout(w.config.TaskHandler.DefaultCommandTimeout))
 	defer cmdCtxCancel()
-	cmd := exec.CommandContext(cmdCtx, w.config.TaskHandler.Commands[0], w.config.TaskHandler.Commands[1:]...)
+
+	cmd := exec.Command(w.config.TaskHandler.Commands[0], w.config.TaskHandler.Commands[1:]...)
+	cmd.SysProcAttr = &syscall.SysProcAttr{
+		Setpgid: true,
+	}
 
 	// Inject workspace path to stdin
 	rStdout, wStdout := io.Pipe()
@@ -330,6 +335,9 @@ func (w *Worker) runCommand(logger zerolog.Logger, t *task.Task) (task.TaskResul
 	}()
 	select {
 	case <-cmdCtx.Done():
+		if err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL); err != nil {
+			streamLogger.Error().Int("pid", cmd.Process.Pid).Err(err).Msg("Failed to kill the process and its descendants")
+		}
 		cmdErr = cmdCtx.Err()
 	case err := <-cmdDone:
 		cmdErr = err