Configurable job timeout (#906)

Configurable job timeout

* validate job timeout if it's specified

Co-authored-by: Andrejs Golevs <a.golevs@dynatech.lv>

builtin/bins/dkron-executor-shell/shell.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"encoding/base64"
+	"errors"
+	"fmt"
 	"log"
 	"os"
 	"os/exec"
@@ -74,12 +76,6 @@ func (s *Shell) ExecuteImpl(args *dktypes.ExecuteRequest, cb dkplugin.StatusHelp
 	cmd.Stderr = reportingWriter{buffer: output, cb: cb, isError: true}
 	cmd.Stdout = reportingWriter{buffer: output, cb: cb}
 
-	// Start a timer to warn about slow handlers
-	slowTimer := time.AfterFunc(2*time.Hour, func() {
-		log.Printf("shell: Script '%s' slow, execution exceeding %v", command, 2*time.Hour)
-	})
-	defer slowTimer.Stop()
-
 	stdin, err := cmd.StdinPipe()
 	if err != nil {
 		return nil, err
@@ -96,18 +92,53 @@ func (s *Shell) ExecuteImpl(args *dktypes.ExecuteRequest, cb dkplugin.StatusHelp
 	stdin.Close()
 
 	log.Printf("shell: going to run %s", command)
+
+	jobTimeout := args.Config["timeout"]
+	var jt time.Duration
+
+	if jobTimeout != "" {
+		jt, err = time.ParseDuration(jobTimeout)
+		if err != nil {
+			return nil, errors.New("shell: Error parsing job timeout")
+		}
+	}
+
 	err = cmd.Start()
 	if err != nil {
 		return nil, err
 	}
 
-	// Warn if buffer is overritten
+	var jobTimeoutMessage string
+	var jobTimedOut bool
+
+	slowTimer := time.AfterFunc(jt, func() {
+		err = cmd.Process.Kill()
+		if err != nil {
+			jobTimeoutMessage = fmt.Sprintf("shell: Job '%s' execution time exceeding defined timeout %v. SIGKILL returned error. Job may not have been killed", command, jt)
+		} else {
+			jobTimeoutMessage = fmt.Sprintf("shell: Job '%s' execution time exceeding defined timeout %v. Job was killed", command, jt)
+		}
+
+		jobTimedOut = true
+		return
+	})
+
+	defer slowTimer.Stop()
+
+	// Warn if buffer is overwritten
 	if output.TotalWritten() > output.Size() {
 		log.Printf("shell: Script '%s' generated %d bytes of output, truncated to %d", command, output.TotalWritten(), output.Size())
 	}
 
 	err = cmd.Wait()
 
+	if jobTimedOut {
+		_, err := output.Write([]byte(jobTimeoutMessage))
+		if err != nil {
+			log.Printf("Error writing output on timeout event: %v", err)
+		}
+	}
+
 	// Always log output
 	log.Printf("shell: Command output %s", output)
 

dkron/api_test.go

@@ -236,6 +236,18 @@ func TestAPIJobCreateUpdateValidationBadTimezone(t *testing.T) {
 	assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
 }
 
+func TestAPIJobCreateUpdateValidationBadShellExecutorTimeout(t *testing.T) {
+	resp := postJob(t, "8099", []byte(`{
+		"name": "testjob",
+		"schedule": "@every 1m",
+		"executor": "shell",
+		"executor_config": {"command": "date", "timeout": "foreverandever"},
+		"disabled": true
+	}`))
+
+	assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
+}
+
 func TestAPIGetNonExistentJobReturnsNotFound(t *testing.T) {
 	port := "8096"
 	baseURL := fmt.Sprintf("http://localhost:%s/v1", port)

dkron/job.go

@@ -357,6 +357,13 @@ func (j *Job) Validate() error {
 		return err
 	}
 
+	if j.Executor == "shell" && j.ExecutorConfig["timeout"] != "" {
+		_, err := time.ParseDuration(j.ExecutorConfig["timeout"])
+		if err != nil {
+			return fmt.Errorf("Error parsing job timeout value")
+		}
+	}
+
 	return nil
 }
 

website/content/usage/executors/shell.md

@@ -14,6 +14,7 @@ shell: Run this command using a shell environment
 command: The command to run
 env: Env vars separated by comma
 cwd: Chdir before command run
+timeout: Force kill job after specified time. Format: https://golang.org/pkg/time/#ParseDuration.
 ```
 
 Example
@@ -25,7 +26,8 @@ Example
       "shell": "true",
       "command": "my_command",
       "env": "ENV_VAR=va1,ANOTHER_ENV_VAR=var2",
-      "cwd": "/app"
+      "cwd": "/app",
+      "timeout": "24h"
   }
 }
 ```