diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index 5f782c09..02212ec9 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/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 --- diff --git a/executor/api.go b/executor/api.go index 23631e10..7e91d3a4 100644 --- a/executor/api.go +++ b/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 } diff --git a/executor/executor.go b/executor/executor.go index a476f45d..ac7fc7a6 100644 --- a/executor/executor.go +++ b/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, -// deploymgr and potential future executor types +func serveMetric() { + // 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 } diff --git a/executor/fscache/functionServiceCache.go b/executor/fscache/functionServiceCache.go index 8ea2180a..fb898231 100644 --- a/executor/fscache/functionServiceCache.go +++ b/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) { diff --git a/executor/fscache/metrics.go b/executor/fscache/metrics.go new file mode 100644 index 00000000..a66e1689 --- /dev/null +++ b/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)) +} diff --git a/glide.lock b/glide.lock index bd343307..ceab21f7 100644 --- a/glide.lock +++ b/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 diff --git a/glide.yaml b/glide.yaml index e56abff9..930f6262 100644 --- a/glide.yaml +++ b/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 + diff --git a/router/functionHandler.go b/router/functionHandler.go index 0fb221d1..8689b419 100644 --- a/router/functionHandler.go +++ b/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 - executor *executorClient.Client - function *metav1.ObjectMeta + fmap *functionServiceMap + executor *executorClient.Client + 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 diff --git a/router/httpTriggers.go b/router/httpTriggers.go index 5088f926..9014a719 100644 --- a/router/httpTriggers.go +++ b/router/httpTriggers.go @@ -117,9 +117,10 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router { } fh := &functionHandler{ - fmap: ts.functionServiceMap, - function: rr.functionMetadata, - executor: ts.executor, + fmap: ts.functionServiceMap, + function: rr.functionMetadata, + executor: ts.executor, + httpTrigger: &trigger, } ht := muxRouter.HandleFunc(trigger.Spec.RelativeURL, fh.handler) diff --git a/router/metrics.go b/router/metrics.go new file mode 100644 index 00000000..3733cd61 --- /dev/null +++ b/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)) + } +} diff --git a/router/router.go b/router/router.go index 50fce5a7..fe579ae3 100644 --- a/router/router.go +++ b/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()