Fission metrics integration (#677)

Add prometheus metrics collection endpoints to router and executor. Add prometheus annotations to router and executor pods.

charts/fission-all/templates/deployment.yaml

@@ -168,6 +168,10 @@ spec:
       labels:
         application: fission-router
         svc: router
+      annotations:
+        prometheus.io/scrape: "true"
+        prometheus.io/path: "/metrics"
+        prometheus.io/port: "8080"
     spec:
       containers:
       - name: router
@@ -188,6 +192,11 @@ spec:
             port: 8888
           initialDelaySeconds: 35
           periodSeconds: 5
+        ports:
+        - containerPort: 8080
+          name: metrics
+        - containerPort: 8888
+          name: http
       serviceAccount: fission-svc
 
 ---
@@ -219,6 +228,10 @@ spec:
     metadata:
       labels:
         svc: executor
+      annotations:
+        prometheus.io/scrape: "true"
+        prometheus.io/path: "/metrics"
+        prometheus.io/port: "8080"
     spec:
       containers:
       - name: executor
@@ -248,6 +261,11 @@ spec:
             port: 8888
           initialDelaySeconds: 35
           periodSeconds: 5
+        ports:
+        - containerPort: 8080
+          name: metrics
+        - containerPort: 8888
+          name: http
       serviceAccount: fission-svc
 
 ---

executor/api.go

@@ -89,6 +89,7 @@ func (executor *Executor) getServiceForFunction(m *metav1.ObjectMeta) (string, e
 	if resp.err != nil {
 		return "", resp.err
 	}
+	executor.fsCache.IncreaseColdStarts(m.Name, string(m.UID))
 	return resp.funcSvc.Address, resp.err
 }
 

executor/executor.go

@@ -18,12 +18,14 @@ package executor
 
 import (
 	"log"
+	"net/http"
 	"runtime/debug"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/dchest/uniuri"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/fission/fission"
@@ -202,8 +204,15 @@ func dumpStackTrace() {
 	debug.PrintStack()
 }
 
-// StartExecutor Starts executor and the executor components such as Poolmgr,
+func serveMetric() {
-// deploymgr and potential future executor types
+	// Expose the registered metrics via HTTP.
+	metricAddr := ":8080"
+	http.Handle("/metrics", promhttp.Handler())
+	log.Fatal(http.ListenAndServe(metricAddr, nil))
+}
+
+// StartExecutor Starts executor and the backend components that executor uses such as Poolmgr,
+// deploymgr and potential future backends
 func StartExecutor(fissionNamespace string, functionNamespace string, port int) error {
 	// setup a signal handler for SIGTERM
 	fission.SetupStackTraceHandler()
@@ -238,6 +247,7 @@ func StartExecutor(fissionNamespace string, functionNamespace string, port int)
 	api := MakeExecutor(gpm, ndm, fissionClient, fsCache)
 
 	go api.Serve(port)
+	go serveMetric()
 
 	return nil
 }

executor/fscache/functionServiceCache.go

@@ -216,6 +216,7 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
 		return nil, err
 	}
 
+	fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), true)
 	return nil, nil
 }
 
@@ -249,6 +250,10 @@ func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) {
 	fsc.byFunction.Delete(crd.CacheKey(fsvc.Function))
 	fsc.byAddress.Delete(fsvc.Address)
 	fsc.byFunctionUID.Delete(fsvc.Function.UID)
+
+	fsc.observeFuncRunningTime(fsvc.Function.Name, string(fsvc.Function.UID), fsvc.Atime.Sub(fsvc.Ctime).Seconds())
+	fsc.observeFuncAliveTime(fsvc.Function.Name, string(fsvc.Function.UID), time.Now().Sub(fsvc.Ctime).Seconds())
+	fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), false)
 }
 
 func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) (bool, error) {

executor/fscache/metrics.go

@@ -0,0 +1,70 @@
+package fscache
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	metricAddr = ":8080"
+
+	// funcname: the function's name
+	// funcuid: the function's version id
+	coldStarts = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "fission_cold_starts_total",
+			Help: "How many cold starts are made by funcname, funcuid.",
+		},
+		[]string{"funcname", "funcuid"},
+	)
+	funcRunningSummary = prometheus.NewSummaryVec(
+		prometheus.SummaryOpts{
+			Name:       "fission_func_running_seconds_summary",
+			Help:       "The running time (last access - create) in seconds of the function.",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		[]string{"funcname", "funcuid"},
+	)
+	funcAliveSummary = prometheus.NewSummaryVec(
+		prometheus.SummaryOpts{
+			Name:       "fission_func_alive_seconds_summary",
+			Help:       "The alive time in seconds of the function.",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		[]string{"funcname", "funcuid"},
+	)
+	funcIsAlive = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "fission_func_is_alive",
+			Help: "A binary value indicating is the funcname, funcuid alive",
+		},
+		[]string{"funcname", "funcuid"},
+	)
+)
+
+func init() {
+	// Register the function calls counter with Prometheus's default registry.
+	prometheus.MustRegister(coldStarts)
+	prometheus.MustRegister(funcRunningSummary)
+	prometheus.MustRegister(funcAliveSummary)
+	prometheus.MustRegister(funcIsAlive)
+}
+
+func (fsc *FunctionServiceCache) IncreaseColdStarts(funcname, funcuid string) {
+	coldStarts.WithLabelValues(funcname, funcuid).Inc()
+}
+
+func (fsc *FunctionServiceCache) observeFuncRunningTime(funcname, funcuid string, running float64) {
+	funcRunningSummary.WithLabelValues(funcname, funcuid).Observe(running)
+}
+
+func (fsc *FunctionServiceCache) observeFuncAliveTime(funcname, funcuid string, alive float64) {
+	funcAliveSummary.WithLabelValues(funcname, funcuid).Observe(alive)
+}
+
+func (fsc *FunctionServiceCache) setFuncAlive(funcname, funcuid string, isAlive bool) {
+	count := 0
+	if isAlive {
+		count = 1
+	}
+	funcIsAlive.WithLabelValues(funcname, funcuid).Set(float64(count))
+}

glide.lock

@@ -3,6 +3,12 @@ updated: 2018-03-23T17:49:04.748453691-07:00
 imports:
 - name: cloud.google.com/go
   version: 3b1ae45394a234c385be014e9a488f2bb6eef821
+- name: github.com/beorn7/perks
+  version: 4c0e84591b9aa9e6dcfdf3e020114cd81f89d5f9
+  subpackages:
+  - quantile
+- name: github.com/blang/semver
+  version: 60ec3488bfea7cca02b021d106d9911120d25fe9
   subpackages:
   - compute/metadata
   - internal
@@ -153,6 +159,31 @@ imports:
   - xxHash32
 - name: github.com/pkg/errors
   version: f15c970de5b76fac0b59abb32d62c17cc7bed265
+- name: github.com/matttproud/golang_protobuf_extensions
+  version: c12348ce28de40eed0136aa2b644d0ee0650e56c
+  subpackages:
+  - pbutil
+- name: github.com/pborman/uuid
+  version: 3d4f2ba23642d3cfd06bd4b54cf03d99d95c0f1b
+- name: github.com/prometheus/client_golang
+  version: c5b7fccd204277076155f10851dad72b76a49317
+  subpackages:
+  - prometheus
+  - prometheus/promhttp
+- name: github.com/prometheus/client_model
+  version: 6f3806018612930941127f2a7c6c453ba2c527d2
+  subpackages:
+  - go
+- name: github.com/prometheus/common
+  version: 13ba4ddd0caa9c28ca7b7bffe1dfa9ed8d5ef207
+  subpackages:
+  - expfmt
+  - internal/bitbucket.org/ww/goautoneg
+  - model
+- name: github.com/prometheus/procfs
+  version: 65c1f6f8f0fc1e2185eb9863a3bc751496404259
+  subpackages:
+  - xfs
 - name: github.com/PuerkitoBio/purell
   version: 8a290539e2e8629dbc4e6bad948158f790ec31f4
 - name: github.com/PuerkitoBio/urlesc

glide.yaml

@@ -62,3 +62,6 @@ import:
   version: ~0.3.2
 - package: github.com/hashicorp/go-multierror
 - package: github.com/hashicorp/errwrap
+- package: github.com/prometheus/client_golang
+  version: v0.8.0
+

router/functionHandler.go

@@ -29,13 +29,15 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/fission/fission"
+	"github.com/fission/fission/crd"
 	executorClient "github.com/fission/fission/executor/client"
 )
 
 type functionHandler struct {
-	fmap     *functionServiceMap
+	fmap        *functionServiceMap
-	executor *executorClient.Client
+	executor    *executorClient.Client
-	function *metav1.ObjectMeta
+	function    *metav1.ObjectMeta
+	httpTrigger *crd.HTTPTrigger
 }
 
 // A layer on top of http.DefaultTransport, with retries.
@@ -75,6 +77,20 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
 	var needExecutor, serviceUrlFromExecutor bool
 	var serviceUrl *url.URL
 
+	// Metrics stuff
+	startTime := time.Now()
+	funcMetricLabels := &functionLabels{
+		namespace: roundTripper.funcHandler.function.Namespace,
+		name:      roundTripper.funcHandler.function.Name,
+	}
+	httpMetricLabels := &httpLabels{
+		method: req.Method,
+	}
+	if roundTripper.funcHandler.httpTrigger != nil {
+		httpMetricLabels.host = roundTripper.funcHandler.httpTrigger.Spec.Host
+		httpMetricLabels.path = roundTripper.funcHandler.httpTrigger.Spec.RelativeURL
+	}
+
 	// set the timeout for transport context
 	timeout := roundTripper.initialTimeout
 	transport := http.DefaultTransport.(*http.Transport)
@@ -135,13 +151,23 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
 			KeepAlive: 30 * time.Second,
 		}).DialContext
 
+		overhead := time.Since(startTime)
+
 		// forward the request to the function service
 		resp, err = transport.RoundTrip(req)
 		if err == nil {
+			// Track metrics
+			httpMetricLabels.code = resp.StatusCode
+			funcMetricLabels.cached = !serviceUrlFromExecutor
+
+			functionCallCompleted(funcMetricLabels, httpMetricLabels,
+				overhead, time.Since(startTime), resp.ContentLength)
+
 			// if transport.RoundTrip succeeds and it was a cached entry, then tapService
 			if !serviceUrlFromExecutor {
 				go roundTripper.funcHandler.tapService(serviceUrl)
 			}
+
 			// return response back to user
 			return resp, nil
 		}
@@ -190,11 +216,9 @@ func (fh *functionHandler) handler(responseWriter http.ResponseWriter, request *
 		request.Header.Add(fmt.Sprintf("X-Fission-Params-%v", k), v)
 	}
 
-	// System Params
+	// system params
 	MetadataToHeaders(HEADERS_FISSION_FUNCTION_PREFIX, fh.function, request)
 
-	// TODO: As an optimization we may want to cache proxies too -- this might get us
-	// connection reuse and possibly better performance
 	director := func(req *http.Request) {
 		if _, ok := req.Header["User-Agent"]; !ok {
 			// explicitly disable User-Agent so it's not set to default value

router/httpTriggers.go

@@ -117,9 +117,10 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
 		}
 
 		fh := &functionHandler{
-			fmap:     ts.functionServiceMap,
+			fmap:        ts.functionServiceMap,
-			function: rr.functionMetadata,
+			function:    rr.functionMetadata,
-			executor: ts.executor,
+			executor:    ts.executor,
+			httpTrigger: &trigger,
 		}
 
 		ht := muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler)

router/metrics.go

@@ -0,0 +1,139 @@
+package router
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type (
+	// functionLabels is the set of metrics labels that relate to
+	// functions.
+	//
+	// cached indicates whether or not the function call hit the
+	// cache in this service.
+	//
+	// namespace and name are the metadata of the function.
+	functionLabels struct {
+		cached    bool
+		namespace string
+		name      string
+	}
+
+	// httpLabels is the set of metrics labels that relate to HTTP
+	// requests.
+	//
+	// host is the host that the HTTP request was made to
+	// path is the relative URL of the request
+	// method is the HTTP method ("GET", "POST", ...)
+	// code is the HTTP status code
+	httpLabels struct {
+		host   string
+		path   string
+		method string
+		code   int
+	}
+)
+
+var (
+	metricAddr = ":8080"
+
+	// function + http labels as strings
+	labelsStrings = []string{"cached", "namespace", "name", "host", "path", "method", "code"}
+
+	// Function http calls count
+	// cached: true | false, is this function service address cached locally
+	// namespace: function namespace
+	// name: function name
+	// code: http status code
+	// path: the client call the function on which http path
+	// method: the function's http method
+	functionCalls = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "fission_function_calls_total",
+			Help: "Count of Fission function calls",
+		},
+		labelsStrings,
+	)
+	functionCallErrors = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "fission_function_errors_total",
+			Help: "Count of Fission function errors",
+		},
+		labelsStrings,
+	)
+	functionCallDuration = prometheus.NewSummaryVec(
+		prometheus.SummaryOpts{
+			Name:       "fission_function_duration_seconds",
+			Help:       "Runtime duration of the Fission function.",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		labelsStrings,
+	)
+	functionCallOverhead = prometheus.NewSummaryVec(
+		prometheus.SummaryOpts{
+			Name:       "fission_function_overhead_seconds",
+			Help:       "The function call delay caused by fission.",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		labelsStrings,
+	)
+	functionCallResponseSize = prometheus.NewSummaryVec(
+		prometheus.SummaryOpts{
+			Name:       "fission_function_response_size_bytes",
+			Help:       "The response size of the http call to target function.",
+			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+		},
+		labelsStrings,
+	)
+)
+
+func init() {
+	prometheus.MustRegister(functionCalls)
+	prometheus.MustRegister(functionCallErrors)
+	prometheus.MustRegister(functionCallDuration)
+	prometheus.MustRegister(functionCallOverhead)
+	prometheus.MustRegister(functionCallResponseSize)
+}
+
+func labelsToStrings(f *functionLabels, h *httpLabels) []string {
+	var cached string
+	if f.cached {
+		cached = "true"
+	} else {
+		cached = "false"
+	}
+	return []string{
+		cached,
+		f.namespace,
+		f.name,
+		h.host,
+		h.path,
+		h.method,
+		fmt.Sprint(h.code),
+	}
+}
+
+func functionCallCompleted(f *functionLabels, h *httpLabels, overhead, duration time.Duration, respSize int64) {
+	l := labelsToStrings(f, h)
+
+	// overhead: time from request ingress into router upto proxing into function pod
+	functionCallOverhead.WithLabelValues(l...).Observe(float64(overhead.Nanoseconds()) / 1e9)
+
+	// total function call counter
+	functionCalls.WithLabelValues(l...).Inc()
+
+	// error counter
+	if h.code >= 400 {
+		functionCallErrors.WithLabelValues(l...).Inc()
+	}
+
+	// duration summary
+	functionCallDuration.WithLabelValues(l...).Observe(float64(duration.Nanoseconds()) / 1e9)
+
+	// Response size.  -1 means the size unknown, in which case we don't report it.
+	if respSize != -1 {
+		functionCallResponseSize.WithLabelValues(l...).Observe(float64(respSize))
+	}
+}

router/router.go

@@ -47,6 +47,7 @@ import (
 	"time"
 
 	"github.com/gorilla/mux"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 
 	"github.com/fission/fission"
 	"github.com/fission/fission/crd"
@@ -71,6 +72,12 @@ func serve(ctx context.Context, port int, httpTriggerSet *HTTPTriggerSet, resolv
 	http.ListenAndServe(url, mr)
 }
 
+func serveMetric() {
+	// Expose the registered metrics via HTTP.
+	http.Handle("/metrics", promhttp.Handler())
+	log.Fatal(http.ListenAndServe(metricAddr, nil))
+}
+
 func Start(port int, executorUrl string) {
 	// setup a signal handler for SIGTERM
 	fission.SetupStackTraceHandler()
@@ -93,6 +100,8 @@ func Start(port int, executorUrl string) {
 	triggers, _, fnStore := makeHTTPTriggerSet(fmap, fissionClient, executor, restClient)
 	resolver := makeFunctionReferenceResolver(fnStore)
 
+	go serveMetric()
+
 	log.Printf("Starting router at port %v\n", port)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()