Merge pull request #12 from alexellis/watchdog_env_timeout

Make watchdog configurable

watchdog/.gitignore

 fwatchdog
+watchdog

watchdog/config_test.go

+package main
+
+import (
+	"testing"
+	"time"
+)
+
+type EnvBucket struct {
+	Items map[string]string
+}
+
+func NewEnvBucket() EnvBucket {
+	return EnvBucket{
+		Items: make(map[string]string),
+	}
+}
+
+func (e EnvBucket) Getenv(key string) string {
+	return e.Items[key]
+}
+
+func (e EnvBucket) Setenv(key string, value string) {
+	e.Items[key] = value
+}
+func TestRead_WriteDebug_DefaultIsTrueConfig(t *testing.T) {
+	defaults := NewEnvBucket()
+	readConfig := ReadConfig{}
+
+	config := readConfig.Read(defaults)
+
+	if config.writeDebug != true {
+		t.Logf("writeDebug should have been true")
+		t.Fail()
+	}
+}
+func TestRead_WriteDebug_FalseConfig(t *testing.T) {
+	defaults := NewEnvBucket()
+	readConfig := ReadConfig{}
+	defaults.Setenv("writeDebug", "true")
+
+	config := readConfig.Read(defaults)
+
+	if config.writeDebug != true {
+		t.Logf("writeDebug should have been true")
+		t.Fail()
+	}
+}
+
+func TestRead_FprocessConfig(t *testing.T) {
+	defaults := NewEnvBucket()
+	readConfig := ReadConfig{}
+	defaults.Setenv("fprocess", "cat")
+
+	config := readConfig.Read(defaults)
+
+	if config.faasProcess != "cat" {
+		t.Logf("fprocess envVariable incorrect, got: %s.\n", config.faasProcess)
+		t.Fail()
+	}
+}
+
+func TestRead_EmptyTimeoutConfig(t *testing.T) {
+	defaults := NewEnvBucket()
+	readConfig := ReadConfig{}
+
+	config := readConfig.Read(defaults)
+
+	if (config.readTimeout) != time.Duration(5)*time.Second {
+		t.Log("readTimeout incorrect")
+		t.Fail()
+	}
+	if (config.writeTimeout) != time.Duration(5)*time.Second {
+		t.Log("writeTimeout incorrect")
+		t.Fail()
+	}
+}
+
+func TestRead_ReadAndWriteTimeoutConfig(t *testing.T) {
+	defaults := NewEnvBucket()
+	defaults.Setenv("read_timeout", "10")
+	defaults.Setenv("write_timeout", "60")
+
+	readConfig := ReadConfig{}
+	config := readConfig.Read(defaults)
+
+	if (config.readTimeout) != time.Duration(10)*time.Second {
+		t.Logf("readTimeout incorrect, got: %d\n", config.readTimeout)
+		t.Fail()
+	}
+	if (config.writeTimeout) != time.Duration(60)*time.Second {
+		t.Logf("writeTimeout incorrect, got: %d\n", config.writeTimeout)
+		t.Fail()
+	}
+}

watchdog/main.go

@@ -11,20 +11,19 @@ import (
 	"time"
 )
 
-func main() {
-	s := &http.Server{
-		Addr:           ":8080",
-		ReadTimeout:    5 * time.Second,
-		WriteTimeout:   5 * time.Second,
-		MaxHeaderBytes: 1 << 20,
-	}
-
-	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
-		if r.Method == "POST" {
+// OsEnv implements interface to wrap os.Getenv
+type OsEnv struct {
+}
 
-			process := os.Getenv("fprocess")
+// Getenv wraps os.Getenv
+func (OsEnv) Getenv(key string) string {
+	return os.Getenv(key)
+}
 
-			parts := strings.Split(process, " ")
+func makeRequestHandler(config *WatchdogConfig) func(http.ResponseWriter, *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == "POST" {
+			parts := strings.Split(config.faasProcess, " ")
 
 			targetCmd := exec.Command(parts[0], parts[1:]...)
 			writer, _ := targetCmd.StdinPipe()
@@ -34,10 +33,13 @@ func main() {
 			writer.Write(res)
 			writer.Close()
 
-			out, err := targetCmd.Output()
-			targetCmd.CombinedOutput()
+			out, err := targetCmd.CombinedOutput()
+
 			if err != nil {
-				log.Println(targetCmd, err)
+				if config.writeDebug == true {
+					log.Println(targetCmd, err)
+				}
+
 				w.WriteHeader(500)
 				response := bytes.NewBufferString(err.Error())
 				w.Write(response.Bytes())
@@ -45,11 +47,35 @@ func main() {
 			}
 			w.WriteHeader(200)
 
-			// TODO: consider stdout to container as configurable via env-variable.
-			os.Stdout.Write(out)
+			if config.writeDebug == true {
+				os.Stdout.Write(out)
+			}
 			w.Write(out)
 		}
-	})
+	}
+}
+
+func main() {
+	osEnv := OsEnv{}
+	readConfig := ReadConfig{}
+	config := readConfig.Read(osEnv)
+
+	if len(config.faasProcess) == 0 {
+		log.Panicln("Provide a valid process via fprocess environmental variable.")
+		return
+	}
+
+	readTimeout := time.Duration(config.readTimeout) * time.Second
+	writeTimeout := time.Duration(config.writeTimeout) * time.Second
+
+	s := &http.Server{
+		Addr:           ":8080",
+		ReadTimeout:    readTimeout,
+		WriteTimeout:   writeTimeout,
+		MaxHeaderBytes: 1 << 20,
+	}
+
+	http.HandleFunc("/", makeRequestHandler(&config))
 
 	log.Fatal(s.ListenAndServe())
 }

watchdog/readconfig.go

+package main
+
+import (
+	"strconv"
+	"time"
+)
+
+// HasEnv provides interface for os.Getenv
+type HasEnv interface {
+	Getenv(key string) string
+}
+
+// ReadConfig constitutes config from env variables
+type ReadConfig struct {
+}
+
+func parseBoolValue(val string) bool {
+	if val == "true" {
+		return true
+	}
+	return true
+}
+
+func parseIntValue(val string) int {
+	if len(val) > 0 {
+		parsedVal, parseErr := strconv.Atoi(val)
+
+		if parseErr == nil && parsedVal >= 0 {
+
+			return parsedVal
+		}
+	}
+	return 0
+}
+
+// Read fetches config from environmental variables.
+func (ReadConfig) Read(hasEnv HasEnv) WatchdogConfig {
+	cfg := WatchdogConfig{
+		writeDebug: true,
+	}
+
+	cfg.faasProcess = hasEnv.Getenv("fprocess")
+
+	readTimeout := parseIntValue(hasEnv.Getenv("read_timeout"))
+	writeTimeout := parseIntValue(hasEnv.Getenv("write_timeout"))
+
+	if readTimeout == 0 {
+		readTimeout = 5
+	}
+
+	if writeTimeout == 0 {
+		writeTimeout = 5
+	}
+
+	cfg.readTimeout = time.Duration(readTimeout) * time.Second
+	cfg.writeTimeout = time.Duration(writeTimeout) * time.Second
+
+	cfg.writeDebug = parseBoolValue(hasEnv.Getenv("write_debug"))
+
+	return cfg
+}
+
+// WatchdogConfig for the process.
+type WatchdogConfig struct {
+	readTimeout  time.Duration
+	writeTimeout time.Duration
+	// faasProcess is the process to exec
+	faasProcess string
+
+	// writeDebug write console stdout statements to the container
+	writeDebug bool
+}