Perform graceful shutdown according to number of connections

Prior to this change the watchdog would attempt to wait for
the write_timeout value twice, even if there were no connections
in flight.

This adds a new metric for in-flight connections and removes
the explicit wait.

Testing with killall of-watchdog -s SIGTERM performs as
expected, and clients that already connected complete their
work before the termination.

This change is a prerequisite for a change in faas-netes
openfaas/faas-netes#853

Signed-off-by: Alex Ellis (OpenFaaS Ltd) <[email protected]>

Author: alexellis

.gitignore

@@ -25,3 +25,5 @@ template
 **/*.sha256
 
 bin
+/handler
+/Dockerfile2

README.md

@@ -136,6 +136,7 @@ Environmental variables:
 | `static_path`               | Yes          | Absolute or relative path to the directory that will be served if `mode="static"` |
 | `read_timeout`              | Yes          | HTTP timeout for reading the payload from the client caller (in seconds) |
 | `write_timeout`             | Yes          | HTTP timeout for writing a response body from your function (in seconds)  |
+| `healthcheck_interval`      | Yes          | Interval (in seconds) for HTTP healthcheck by container orchestrator i.e. kubelet. Used for graceful shutdowns.  |
 | `exec_timeout`              | Yes          | Exec timeout for process exec'd for each incoming request (in seconds). Disabled if set to 0. |
 | `port`                      | Yes          | Specify an alternative TCP port for testing. Default: `8080` |
 | `write_debug`               | No           | Write all output, error messages, and additional information to the logs. Default is `false`. |

config/config.go

@@ -4,6 +4,7 @@
 package config
 
 import (
+	"fmt"
 	"log"
 	"strconv"
 	"strings"
@@ -12,10 +13,11 @@ import (
 
 // WatchdogConfig configuration for a watchdog.
 type WatchdogConfig struct {
-	TCPPort          int
-	HTTPReadTimeout  time.Duration
-	HTTPWriteTimeout time.Duration
-	ExecTimeout      time.Duration
+	TCPPort             int
+	HTTPReadTimeout     time.Duration
+	HTTPWriteTimeout    time.Duration
+	ExecTimeout         time.Duration
+	HealthcheckInterval time.Duration
 
 	FunctionProcess  string
 	ContentType      string
@@ -56,7 +58,7 @@ func (w WatchdogConfig) Process() (string, []string) {
 }
 
 // New create config based upon environmental variables.
-func New(env []string) WatchdogConfig {
+func New(env []string) (WatchdogConfig, error) {
 
 	envMap := mapEnv(env)
 
@@ -100,29 +102,43 @@ func New(env []string) WatchdogConfig {
 		staticPath = val
 	}
 
-	config := WatchdogConfig{
-		TCPPort:          getInt(envMap, "port", 8080),
-		HTTPReadTimeout:  getDuration(envMap, "read_timeout", time.Second*10),
-		HTTPWriteTimeout: getDuration(envMap, "write_timeout", time.Second*10),
-		FunctionProcess:  functionProcess,
-		StaticPath:       staticPath,
-		InjectCGIHeaders: true,
-		ExecTimeout:      getDuration(envMap, "exec_timeout", time.Second*10),
-		OperationalMode:  ModeStreaming,
-		ContentType:      contentType,
-		SuppressLock:     getBool(envMap, "suppress_lock"),
-		UpstreamURL:      upstreamURL,
-		BufferHTTPBody:   getBools(envMap, "buffer_http", "http_buffer_req_body"),
-		MetricsPort:      8081,
-		MaxInflight:      getInt(envMap, "max_inflight", 0),
-		PrefixLogs:       prefixLogs,
+	writeTimeout := getDuration(envMap, "write_timeout", time.Second*10)
+	healthcheckInterval := writeTimeout
+	if val, exists := envMap["healthcheck_interval"]; exists {
+		healthcheckInterval = parseIntOrDurationValue(val, writeTimeout)
+	}
+
+	c := WatchdogConfig{
+		TCPPort:             getInt(envMap, "port", 8080),
+		HTTPReadTimeout:     getDuration(envMap, "read_timeout", time.Second*10),
+		HTTPWriteTimeout:    writeTimeout,
+		HealthcheckInterval: healthcheckInterval,
+		FunctionProcess:     functionProcess,
+		StaticPath:          staticPath,
+		InjectCGIHeaders:    true,
+		ExecTimeout:         getDuration(envMap, "exec_timeout", time.Second*10),
+		OperationalMode:     ModeStreaming,
+		ContentType:         contentType,
+		SuppressLock:        getBool(envMap, "suppress_lock"),
+		UpstreamURL:         upstreamURL,
+		BufferHTTPBody:      getBools(envMap, "buffer_http", "http_buffer_req_body"),
+		MetricsPort:         8081,
+		MaxInflight:         getInt(envMap, "max_inflight", 0),
+		PrefixLogs:          prefixLogs,
 	}
-
 	if val := envMap["mode"]; len(val) > 0 {
-		config.OperationalMode = WatchdogModeConst(val)
+		c.OperationalMode = WatchdogModeConst(val)
+	}
+
+	if writeTimeout == 0 {
+		return c, fmt.Errorf("HTTP write timeout must be over 0s")
+	}
+
+	if len(c.FunctionProcess) == 0 && c.OperationalMode != ModeStatic {
+		return c, fmt.Errorf(`provide a "function_process" or "fprocess" environmental variable for your function`)
 	}
 
-	return config
+	return c, nil
 }
 
 func mapEnv(env []string) map[string]string {

config/config_test.go

@@ -10,14 +10,14 @@ import (
 )
 
 func TestNew(t *testing.T) {
-	defaults := New([]string{})
+	defaults, _ := New([]string{})
 	if defaults.TCPPort != 8080 {
 		t.Errorf("Want TCPPort: 8080, got: %d", defaults.TCPPort)
 	}
 }
 
 func Test_OperationalMode_Default(t *testing.T) {
-	defaults := New([]string{})
+	defaults, _ := New([]string{})
 	if defaults.OperationalMode != ModeStreaming {
 		t.Errorf("Want %s. got: %s", WatchdogMode(ModeStreaming), WatchdogMode(defaults.OperationalMode))
 	}
@@ -26,7 +26,7 @@ func Test_OperationalMode_Default(t *testing.T) {
 func Test_BufferHttpModeDefaultsToFalse(t *testing.T) {
 	env := []string{}
 
-	actual := New(env)
+	actual, _ := New(env)
 	want := false
 	if actual.BufferHTTPBody != want {
 		t.Errorf("Want %v. got: %v", want, actual.BufferHTTPBody)
@@ -39,7 +39,7 @@ func Test_UpstreamURL(t *testing.T) {
 		fmt.Sprintf("upstream_url=%s", urlVal),
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 	want := urlVal
 	if actual.UpstreamURL != want {
 		t.Errorf("Want %v. got: %v", want, actual.UpstreamURL)
@@ -52,7 +52,7 @@ func Test_UpstreamURLVerbose(t *testing.T) {
 		fmt.Sprintf("http_upstream_url=%s", urlVal),
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 	want := urlVal
 	if actual.UpstreamURL != want {
 		t.Errorf("Want %v. got: %v", want, actual.UpstreamURL)
@@ -64,7 +64,7 @@ func Test_BufferHttpMode_CanBeSetToTrue(t *testing.T) {
 		"http_buffer_req_body=true",
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 	want := true
 	if actual.BufferHTTPBody != want {
 		t.Errorf("Want %v. got: %v", want, actual.BufferHTTPBody)
@@ -76,7 +76,7 @@ func Test_OperationalMode_AfterBurn(t *testing.T) {
 		"mode=afterburn",
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 
 	if actual.OperationalMode != ModeAfterBurn {
 		t.Errorf("Want %s. got: %s", WatchdogMode(ModeAfterBurn), WatchdogMode(actual.OperationalMode))
@@ -88,7 +88,7 @@ func Test_OperationalMode_Static(t *testing.T) {
 		"mode=static",
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 
 	if actual.OperationalMode != ModeStatic {
 		t.Errorf("Want %s. got: %s", WatchdogMode(ModeStatic), WatchdogMode(actual.OperationalMode))
@@ -98,7 +98,7 @@ func Test_OperationalMode_Static(t *testing.T) {
 func Test_ContentType_Default(t *testing.T) {
 	env := []string{}
 
-	actual := New(env)
+	actual, _ := New(env)
 
 	if actual.ContentType != "application/octet-stream" {
 		t.Errorf("Default (ContentType) Want %s. got: %s", actual.ContentType, "octet-stream")
@@ -110,7 +110,7 @@ func Test_ContentType_Override(t *testing.T) {
 		"content_type=application/json",
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 
 	if actual.ContentType != "application/json" {
 		t.Errorf("(ContentType) Want %s. got: %s", actual.ContentType, "application/json")
@@ -122,7 +122,7 @@ func Test_FunctionProcessLegacyName(t *testing.T) {
 		"fprocess=env",
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 
 	if actual.FunctionProcess != "env" {
 		t.Errorf("Want %s. got: %s", "env", actual.FunctionProcess)
@@ -134,7 +134,7 @@ func Test_FunctionProcessAlternativeName(t *testing.T) {
 		"function_process=env",
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 
 	if actual.FunctionProcess != "env" {
 		t.Errorf("Want %s. got: %s", "env", actual.FunctionProcess)
@@ -177,7 +177,7 @@ func Test_FunctionProcess_Arguments(t *testing.T) {
 
 	for _, testCase := range cases {
 		t.Run(testCase.scenario, func(t *testing.T) {
-			actual := New([]string{testCase.env})
+			actual, _ := New([]string{testCase.env})
 
 			process, args := actual.Process()
 			if process != testCase.wantProcess {
@@ -205,7 +205,7 @@ func Test_PortOverride(t *testing.T) {
 		"port=8081",
 	}
 
-	actual := New(env)
+	actual, _ := New(env)
 
 	if actual.TCPPort != 8081 {
 		t.Errorf("Want %d. got: %d", 8081, actual.TCPPort)
@@ -251,7 +251,7 @@ func Test_Timeouts(t *testing.T) {
 	}
 
 	for _, testCase := range cases {
-		actual := New(testCase.env)
+		actual, _ := New(testCase.env)
 		if testCase.readTimeout != actual.HTTPReadTimeout {
 			t.Errorf("(%s) HTTPReadTimeout want: %s, got: %s", testCase.name, actual.HTTPReadTimeout, testCase.readTimeout)
 		}

executor/http_runner.go

@@ -68,7 +68,6 @@ func (f *HTTPFunctionRunner) Start() error {
 
 		<-sig
 		cmd.Process.Signal(syscall.SIGTERM)
-
 	}()
 
 	err := cmd.Start()
@@ -100,7 +99,11 @@ func (f *HTTPFunctionRunner) Run(req FunctionRequest, contentLength int64, r *ht
 		body = r.Body
 	}
 
-	request, _ := http.NewRequest(r.Method, upstreamURL, body)
+	request, err := http.NewRequest(r.Method, upstreamURL, body)
+	if err != nil {
+		return err
+	}
+
 	for h := range r.Header {
 		request.Header.Set(h, r.Header.Get(h))
 	}
@@ -116,14 +119,11 @@ func (f *HTTPFunctionRunner) Run(req FunctionRequest, contentLength int64, r *ht
 	} else {
 		reqCtx = r.Context()
 		cancel = func() {
-
 		}
 	}
-
 	defer cancel()
 
 	res, err := f.Client.Do(request.WithContext(reqCtx))
-
 	if err != nil {
 		log.Printf("Upstream HTTP request error: %s\n", err.Error())
 
@@ -136,19 +136,15 @@ func (f *HTTPFunctionRunner) Run(req FunctionRequest, contentLength int64, r *ht
 			return nil
 		}
 
-		select {
-		case <-reqCtx.Done():
-			{
-				if reqCtx.Err() != nil {
-					// Error due to timeout / deadline
-					log.Printf("Upstream HTTP killed due to exec_timeout: %s\n", f.ExecTimeout)
-					w.Header().Set("X-Duration-Seconds", fmt.Sprintf("%f", time.Since(startedTime).Seconds()))
+		<-reqCtx.Done()
 
-					w.WriteHeader(http.StatusGatewayTimeout)
-					return nil
-				}
+		if reqCtx.Err() != nil {
+			// Error due to timeout / deadline
+			log.Printf("Upstream HTTP killed due to exec_timeout: %s\n", f.ExecTimeout)
+			w.Header().Set("X-Duration-Seconds", fmt.Sprintf("%f", time.Since(startedTime).Seconds()))
 
-			}
+			w.WriteHeader(http.StatusGatewayTimeout)
+			return nil
 		}
 
 		w.Header().Set("X-Duration-Seconds", fmt.Sprintf("%f", time.Since(startedTime).Seconds()))

go.mod

@@ -5,4 +5,5 @@ go 1.13
 require (
 	github.com/openfaas/faas-middleware v1.0.0
 	github.com/prometheus/client_golang v1.9.0
+	github.com/prometheus/client_model v0.2.0 // indirect
 )