Add custom metrics

Custom metrics are added for HTTP traffic and exposed by default
on port 8081. This has been tested more extensively in the
classic watchdog and copied / duplicated into the of-watchdog.

Local e2e testing was done on MacOS outside of a container and
the new http_* fields and metrics were reported as expected.

Signed-off-by: Alex Ellis <[email protected]>

Author: alexellis

Dockerfile

@@ -1,11 +1,13 @@
-FROM golang:1.9.4
+FROM golang:1.10
 
 RUN mkdir -p /go/src/github.com/openfaas-incubator/of-watchdog
 WORKDIR /go/src/github.com/openfaas-incubator/of-watchdog
 
-COPY main.go    .
+COPY vendor     vendor
 COPY config     config
 COPY executor   executor
+COPY metrics    metrics
+COPY main.go    .
 
 # Run a gofmt and exclude all vendored code.
 RUN test -z "$(gofmt -l $(find . -type f -name '*.go' -not -path "./vendor/*"))"

Gopkg.lock

@@ -0,0 +1,34 @@
+# Gopkg.toml example
+#
+# Refer to https://golang.github.io/dep/docs/Gopkg.toml.html
+# for detailed Gopkg.toml documentation.
+#
+# required = ["github.com/user/thing/cmd/thing"]
+# ignored = ["github.com/user/project/pkgX", "bitbucket.org/user/project/pkgA/pkgY"]
+#
+# [[constraint]]
+#   name = "github.com/user/project"
+#   version = "1.0.0"
+#
+# [[constraint]]
+#   name = "github.com/user/project2"
+#   branch = "dev"
+#   source = "github.com/myfork/project2"
+#
+# [[override]]
+#   name = "github.com/x/y"
+#   version = "2.4.0"
+#
+# [prune]
+#   non-go = false
+#   go-tests = true
+#   unused-packages = true
+
+
+[[constraint]]
+  name = "github.com/prometheus/client_golang"
+  version = "0.9.2"
+
+[prune]
+  go-tests = true
+  unused-packages = true

Gopkg.toml

@@ -25,6 +25,9 @@ type WatchdogConfig struct {
 	// to prevent transfer type of chunked encoding
 	// which some servers do not support.
 	BufferHTTPBody bool
+
+	// MetricsPort TCP port on which to serve HTTP Prometheus metrics
+	MetricsPort int
 }
 
 // Process returns a string for the process and a slice for the arguments from the FunctionProcess.
@@ -77,6 +80,7 @@ func New(env []string) (WatchdogConfig, error) {
 		SuppressLock:     getBool(envMap, "suppress_lock"),
 		UpstreamURL:      upstreamURL,
 		BufferHTTPBody:   getBool(envMap, "buffer_http"),
+		MetricsPort:      8081,
 	}
 
 	if val := envMap["mode"]; len(val) > 0 {

config/config.go

@@ -15,6 +15,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/openfaas-incubator/of-watchdog/metrics"
+
 	"github.com/openfaas-incubator/of-watchdog/config"
 	"github.com/openfaas-incubator/of-watchdog/executor"
 )
@@ -41,11 +43,19 @@ func main() {
 
 	log.Printf("OperationalMode: %s\n", config.WatchdogMode(watchdogConfig.OperationalMode))
 
-	http.HandleFunc("/", requestHandler)
+	httpMetrics := metrics.NewHttp()
+	http.HandleFunc("/", metrics.InstrumentHandler(requestHandler, httpMetrics))
+
 	http.HandleFunc("/_/health", makeHealthHandler())
 
-	shutdownTimeout := watchdogConfig.HTTPWriteTimeout
+	metricsServer := metrics.MetricsServer{}
+	metricsServer.Register(watchdogConfig.MetricsPort)
+
+	cancel := make(chan bool)
 
+	go metricsServer.Serve(cancel)
+
+	shutdownTimeout := watchdogConfig.HTTPWriteTimeout
 	s := &http.Server{
 		Addr:           fmt.Sprintf(":%d", watchdogConfig.TCPPort),
 		ReadTimeout:    watchdogConfig.HTTPReadTimeout,
@@ -54,7 +64,6 @@ func main() {
 	}
 
 	listenUntilShutdown(shutdownTimeout, s, watchdogConfig.SuppressLock)
-
 }
 
 func markUnhealthy() error {

main.go

@@ -0,0 +1,27 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+type Http struct {
+	RequestsTotal            *prometheus.CounterVec
+	RequestDurationHistogram *prometheus.HistogramVec
+}
+
+func NewHttp() Http {
+	return Http{
+		RequestsTotal: promauto.NewCounterVec(prometheus.CounterOpts{
+			Subsystem: "http",
+			Name:      "requests_total",
+			Help:      "total HTTP requests processed",
+		}, []string{"code", "method"}),
+		RequestDurationHistogram: promauto.NewHistogramVec(prometheus.HistogramOpts{
+			Subsystem: "http",
+			Name:      "request_duration_seconds",
+			Help:      "Seconds spent serving HTTP requests.",
+			Buckets:   prometheus.DefBuckets,
+		}, []string{"code", "method"}),
+	}
+}

metrics/http.go

@@ -0,0 +1,65 @@
+package metrics
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+// MetricsServer provides instrumentation for HTTP calls
+type MetricsServer struct {
+	s    *http.Server
+	port int
+}
+
+// Register binds a HTTP server to expose Prometheus metrics
+func (m *MetricsServer) Register(metricsPort int) {
+
+	m.port = metricsPort
+
+	readTimeout := time.Millisecond * 500
+	writeTimeout := time.Millisecond * 500
+
+	metricsMux := http.NewServeMux()
+	metricsMux.Handle("/metrics", promhttp.Handler())
+
+	m.s = &http.Server{
+		Addr:           fmt.Sprintf(":%d", metricsPort),
+		ReadTimeout:    readTimeout,
+		WriteTimeout:   writeTimeout,
+		MaxHeaderBytes: 1 << 20, // Max header of 1MB
+		Handler:        metricsMux,
+	}
+
+}
+
+// Serve http traffic in go routine, non-blocking
+func (m *MetricsServer) Serve(cancel chan bool) {
+	log.Printf("Metrics server. Port: %d\n", m.port)
+
+	go func() {
+		if err := m.s.ListenAndServe(); err != http.ErrServerClosed {
+			panic(fmt.Sprintf("metrics error ListenAndServe: %v\n", err))
+		}
+	}()
+
+	go func() {
+		select {
+		case <-cancel:
+			log.Printf("metrics server shutdown\n")
+
+			m.s.Shutdown(context.Background())
+		}
+	}()
+}
+
+// InstrumentHandler returns a handler which records HTTP requests
+// as they are made
+func InstrumentHandler(next http.HandlerFunc, _http Http) http.HandlerFunc {
+	return promhttp.InstrumentHandlerCounter(_http.RequestsTotal,
+		promhttp.InstrumentHandlerDuration(_http.RequestDurationHistogram, next))
+}

metrics/metrics.go

@@ -0,0 +1,57 @@
+package metrics
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+	"time"
+)
+
+func Test_Register_ProvidesBytes(t *testing.T) {
+
+	metricsPort := 31111
+
+	metricsServer := MetricsServer{}
+	metricsServer.Register(metricsPort)
+
+	cancel := make(chan bool)
+	go metricsServer.Serve(cancel)
+
+	defer func() {
+		cancel <- true
+	}()
+
+	retries := 10
+
+	for i := 0; i < retries; i++ {
+		req, _ := http.NewRequest(http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/metrics", metricsPort), nil)
+
+		res, err := http.DefaultClient.Do(req)
+
+		if err != nil {
+			t.Logf("cannot get metrics, or not ready: %s", err.Error())
+
+			time.Sleep(time.Millisecond * 100)
+			continue
+		}
+
+		wantStatus := http.StatusOK
+		if res.StatusCode != wantStatus {
+			t.Errorf("metrics gave wrong status, want: %d, got: %d", wantStatus, res.StatusCode)
+			t.Fail()
+			return
+		}
+
+		if res.Body == nil {
+			t.Errorf("metrics response should have a body")
+			t.Fail()
+			return
+		}
+		defer res.Body.Close()
+
+		return
+	}
+
+	t.Errorf("unable to get expected response from metrics server")
+	t.Fail()
+}