From b638a6d04779843cea33f952a77fe41cdd3ad5d6 Mon Sep 17 00:00:00 2001 From: Ankit Chawla Date: Wed, 13 Apr 2022 21:49:46 +0530 Subject: [PATCH] Prometheus metrics improvements (#2398) - Enabled metrics in storagesvc, buildermgr and controller. - Added a middleware in storagesvc, router, executor and controller to monitor total number of http requests, each request's duration and number of requests that are currently being served. These requests can be filtered on their path, method or statuscode. - Removed functionCallDuration and functionCallResponseSize metrics from router. - Removed funcAliveSummary, funcIsAlive, funcReapTime and idleTime metrics. - Replaced function calls for collecting metrics to direct metric calls. Signed-off-by: Sanket Sudake Co-authored-by: Sanket Sudake --- .../templates/buildermgr/deployment.yaml | 7 ++ .../templates/controller/deployment.yaml | 6 + .../templates/storagesvc/deployment.yaml | 6 + pkg/buildermgr/pkgwatcher.go | 2 + pkg/controller/api.go | 6 +- pkg/controller/controller.go | 2 +- pkg/executor/api.go | 2 + pkg/executor/executor.go | 14 +-- .../executortype/container/containermgr.go | 4 +- .../executortype/newdeploy/newdeploymgr.go | 8 +- pkg/executor/executortype/poolmgr/gp.go | 3 +- pkg/executor/executortype/poolmgr/gpm.go | 4 - pkg/executor/fscache/functionServiceCache.go | 8 +- pkg/executor/fscache/metrics.go | 100 --------------- pkg/executor/metrics/metrics.go | 51 ++++++++ pkg/router/functionHandler.go | 32 ++--- pkg/router/httpTriggers.go | 2 + pkg/router/metrics.go | 114 +----------------- pkg/router/mutablemux_test.go | 4 + pkg/router/router.go | 27 ++--- pkg/storagesvc/client/storagesvc_test.go | 52 ++++---- pkg/storagesvc/metrics.go | 24 ++++ pkg/storagesvc/storagesvc.go | 24 +++- pkg/storagesvc/stowClient.go | 12 ++ pkg/utils/metrics/http_metrics.go | 92 ++++++++++++++ pkg/utils/metrics/server.go | 56 +++++++++ 26 files changed, 359 insertions(+), 303 deletions(-) delete mode 100644 pkg/executor/fscache/metrics.go create mode 100644 pkg/executor/metrics/metrics.go create mode 100644 pkg/storagesvc/metrics.go create mode 100644 pkg/utils/metrics/http_metrics.go create mode 100644 pkg/utils/metrics/server.go diff --git a/charts/fission-all/templates/buildermgr/deployment.yaml b/charts/fission-all/templates/buildermgr/deployment.yaml index 5e606265..aa28bcd4 100644 --- a/charts/fission-all/templates/buildermgr/deployment.yaml +++ b/charts/fission-all/templates/buildermgr/deployment.yaml @@ -14,6 +14,10 @@ spec: metadata: labels: svc: buildermgr + annotations: + prometheus.io/scrape: "true" + prometheus.io/path: "/metrics" + prometheus.io/port: "8080" spec: containers: - name: buildermgr @@ -50,6 +54,9 @@ spec: value: {{ .Release.Name | quote }} {{- include "opentracing.envs" . | indent 8 }} {{- include "opentelemtry.envs" . | indent 8 }} + ports: + - containerPort: 8080 + name: metrics resources: {{- toYaml .Values.buildermgr.resources | nindent 10 }} {{- if .Values.terminationMessagePath }} diff --git a/charts/fission-all/templates/controller/deployment.yaml b/charts/fission-all/templates/controller/deployment.yaml index 3703953b..cbb98b60 100644 --- a/charts/fission-all/templates/controller/deployment.yaml +++ b/charts/fission-all/templates/controller/deployment.yaml @@ -17,6 +17,10 @@ spec: labels: svc: controller application: fission-api + annotations: + prometheus.io/scrape: "true" + prometheus.io/path: "/metrics" + prometheus.io/port: "8080" spec: containers: - name: controller @@ -63,6 +67,8 @@ spec: mountPath: /etc/config/config.yaml subPath: config.yaml ports: + - containerPort: 8080 + name: metrics - containerPort: 8888 name: http {{- if .Values.pprof.enabled }} diff --git a/charts/fission-all/templates/storagesvc/deployment.yaml b/charts/fission-all/templates/storagesvc/deployment.yaml index 1a522975..e66e1033 100644 --- a/charts/fission-all/templates/storagesvc/deployment.yaml +++ b/charts/fission-all/templates/storagesvc/deployment.yaml @@ -17,6 +17,10 @@ spec: labels: svc: storagesvc application: fission-storage + annotations: + prometheus.io/scrape: "true" + prometheus.io/path: "/metrics" + prometheus.io/port: "8080" spec: containers: - name: storagesvc @@ -72,6 +76,8 @@ spec: initialDelaySeconds: 35 periodSeconds: 5 ports: + - containerPort: 8080 + name: metrics - containerPort: 8000 name: http {{- if .Values.pprof.enabled }} diff --git a/pkg/buildermgr/pkgwatcher.go b/pkg/buildermgr/pkgwatcher.go index cd15038c..12f3d6b6 100644 --- a/pkg/buildermgr/pkgwatcher.go +++ b/pkg/buildermgr/pkgwatcher.go @@ -32,6 +32,7 @@ import ( "github.com/fission/fission/pkg/cache" "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/utils" + "github.com/fission/fission/pkg/utils/metrics" ) type ( @@ -323,6 +324,7 @@ func (pkgw *packageWatcher) packageInformerHandler() k8sCache.ResourceEventHandl } func (pkgw *packageWatcher) Run(ctx context.Context) { + go metrics.ServeMetrics(ctx, pkgw.logger) go (*pkgw.podInformer).Run(ctx.Done()) (*pkgw.pkgInformer).AddEventHandler(pkgw.packageInformerHandler()) (*pkgw.pkgInformer).Run(ctx.Done()) diff --git a/pkg/controller/api.go b/pkg/controller/api.go index eaf07e30..c4fea0e6 100644 --- a/pkg/controller/api.go +++ b/pkg/controller/api.go @@ -35,6 +35,7 @@ import ( ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/fission-cli/logdb" "github.com/fission/fission/pkg/info" + "github.com/fission/fission/pkg/utils/metrics" "github.com/fission/fission/pkg/utils/otel" ) @@ -198,6 +199,7 @@ func (api *API) GetSvcName(w http.ResponseWriter, r *http.Request) { func (api *API) GetHandler() http.Handler { r := mux.NewRouter() + r.Use(metrics.HTTPMetricMiddleware()) r.HandleFunc("/healthz", api.HealthHandler).Methods("GET") // Give a useful error message if an older CLI attempts to make a request r.HandleFunc(`/v1/{rest:[a-zA-Z0-9=\-\/]+}`, api.ApiVersionMismatchHandler) @@ -267,7 +269,7 @@ func (api *API) GetHandler() http.Handler { return r } -func (api *API) Serve(port int, openTracingEnabled bool) { +func (api *API) Serve(ctx context.Context, port int, openTracingEnabled bool) { address := fmt.Sprintf(":%v", port) api.logger.Info("server started", zap.Int("port", port)) @@ -277,6 +279,8 @@ func (api *API) Serve(port int, openTracingEnabled bool) { } else { handler = otel.GetHandlerWithOTEL(api.GetHandler(), "fission-controller", otel.UrlsToIgnore("/healthz")) } + + go metrics.ServeMetrics(ctx, api.logger) err := http.ListenAndServe(address, handler) api.logger.Fatal("done listening", zap.Error(err)) } diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index 83bebf8b..a7f44413 100644 --- a/pkg/controller/controller.go +++ b/pkg/controller/controller.go @@ -51,5 +51,5 @@ func Start(ctx context.Context, logger *zap.Logger, port int, unitTestFlag bool, if err != nil { cLogger.Fatal("failed to start controller", zap.Error(err)) } - api.Serve(port, openTracingEnabled) + api.Serve(ctx, port, openTracingEnabled) } diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 4e4aab5a..b5a07222 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -34,6 +34,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/executor/client" + "github.com/fission/fission/pkg/utils/metrics" otelUtils "github.com/fission/fission/pkg/utils/otel" ) @@ -251,6 +252,7 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { // GetHandler returns an http.Handler. func (executor *Executor) GetHandler() http.Handler { r := mux.NewRouter() + r.Use(metrics.HTTPMetricMiddleware()) r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionAPI).Methods("POST") r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") // for backward compatibility r.HandleFunc("/v2/tapServices", executor.tapServices).Methods("POST") diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index b14a341b..8b7a60ae 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -19,7 +19,6 @@ package executor import ( "context" "fmt" - "net/http" "os" "strconv" "strings" @@ -28,7 +27,6 @@ import ( "github.com/dchest/uniuri" "github.com/pkg/errors" - "github.com/prometheus/client_golang/prometheus/promhttp" "go.uber.org/zap" k8sInformers "k8s.io/client-go/informers" k8sCache "k8s.io/client-go/tools/cache" @@ -46,6 +44,7 @@ import ( fetcherConfig "github.com/fission/fission/pkg/fetcher/config" genInformer "github.com/fission/fission/pkg/generated/informers/externalversions" "github.com/fission/fission/pkg/utils" + "github.com/fission/fission/pkg/utils/metrics" otelUtils "github.com/fission/fission/pkg/utils/otel" ) @@ -250,15 +249,6 @@ func (executor *Executor) getFunctionServiceFromCache(ctx context.Context, fn *f return e.GetFuncSvcFromCache(ctx, fn) } -func serveMetric(logger *zap.Logger) { - // Expose the registered metrics via HTTP. - metricAddr := ":8080" - http.Handle("/metrics", promhttp.Handler()) - err := http.ListenAndServe(metricAddr, nil) - - logger.Fatal("done listening on metrics endpoint", zap.Error(err)) -} - // StartExecutor Starts executor and the executor components such as Poolmgr, // deploymgr and potential future executor types func StartExecutor(ctx context.Context, logger *zap.Logger, functionNamespace string, envBuilderNamespace string, port int, openTracingEnabled bool) error { @@ -379,8 +369,8 @@ func StartExecutor(ctx context.Context, logger *zap.Logger, functionNamespace st return err } go reaper.CleanupRoleBindings(ctx, logger, kubernetesClient, fissionClient, functionNamespace, envBuilderNamespace, time.Minute*30) + go metrics.ServeMetrics(ctx, logger) go api.Serve(port, openTracingEnabled) - go serveMetric(logger) return nil } diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index 168b63ba..b5109c71 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -45,6 +45,7 @@ import ( "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/executortype" "github.com/fission/fission/pkg/executor/fscache" + "github.com/fission/fission/pkg/executor/metrics" "github.com/fission/fission/pkg/executor/reaper" finformerv1 "github.com/fission/fission/pkg/generated/informers/externalversions/core/v1" "github.com/fission/fission/pkg/throttler" @@ -440,10 +441,11 @@ func (caaf *Container) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache _, err = caaf.fsCache.Add(*fsvc) if err != nil { caaf.logger.Error("error adding function to cache", zap.Error(err), zap.Any("function", fsvc.Function)) + metrics.FuncError.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc() return fsvc, err } - caaf.fsCache.IncreaseColdStarts(fn.ObjectMeta.Name, string(fn.ObjectMeta.UID)) + metrics.ColdStarts.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc() return fsvc, nil } diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 6104f6b3..af767ebd 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -45,6 +45,7 @@ import ( "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/executortype" "github.com/fission/fission/pkg/executor/fscache" + "github.com/fission/fission/pkg/executor/metrics" "github.com/fission/fission/pkg/executor/reaper" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" finformerv1 "github.com/fission/fission/pkg/generated/informers/externalversions/core/v1" @@ -480,10 +481,11 @@ func (deploy *NewDeploy) fnCreate(ctx context.Context, fn *fv1.Function) (*fscac _, err = deploy.fsCache.Add(*fsvc) if err != nil { deploy.logger.Error("error adding function to cache", zap.Error(err), zap.Any("function", fsvc.Function)) + metrics.FuncError.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc() return fsvc, err } - deploy.fsCache.IncreaseColdStarts(fn.ObjectMeta.Name, string(fn.ObjectMeta.UID)) + metrics.ColdStarts.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc() return fsvc, nil } @@ -804,10 +806,7 @@ func (deploy *NewDeploy) idleObjectReaper(ctx context.Context) { continue } - deploy.fsCache.IdleTime(fsvc.Name, fsvc.Address, float64(time.Since(fsvc.Atime)-idlePodReapTime)) - go func() { - startTime := time.Now() deployObj := getDeploymentObj(fsvc.KubernetesObjects) if deployObj == nil { deploy.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) @@ -832,7 +831,6 @@ func (deploy *NewDeploy) idleObjectReaper(ctx context.Context) { if err != nil { deploy.logger.Error("error scaling down function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) } - deploy.fsCache.ReapTime(fsvc.Function.Name, fsvc.Address, time.Since(startTime).Seconds()) }() } } diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 7ed07899..33bc5ab5 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -46,6 +46,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/fscache" + "github.com/fission/fission/pkg/executor/metrics" fetcherClient "github.com/fission/fission/pkg/fetcher/client" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" "github.com/fission/fission/pkg/utils" @@ -583,7 +584,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address}) gp.fsCache.AddFunc(ctx, *fsvc) - gp.fsCache.IncreaseColdStarts(fn.ObjectMeta.Name, string(fn.ObjectMeta.UID)) + metrics.ColdStarts.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc() logger.Info("added function service", zap.String("pod", pod.ObjectMeta.Name), diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index a718a361..30b53705 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -631,11 +631,8 @@ func (gpm *GenericPoolManager) idleObjectReaper() { if time.Since(fsvc.Atime) < idlePodReapTime { continue } - idleTime := (time.Since(fsvc.Atime) - idlePodReapTime).Seconds() - gpm.fsCache.IdleTime(fsvc.Name, fsvc.Address, idleTime) go func() { - startTime := time.Now() deleted, err := gpm.fsCache.DeleteOldPoolCache(ctx, fsvc, idlePodReapTime) if err != nil { gpm.logger.Error("error deleting Kubernetes objects for function service", @@ -652,7 +649,6 @@ func (gpm *GenericPoolManager) idleObjectReaper() { ) reaper.CleanupKubeObject(ctx, gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i]) time.Sleep(50 * time.Millisecond) - gpm.fsCache.ReapTime(fsvc.Function.Name, fsvc.Address, time.Since(startTime).Seconds()) } } }() diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 6a62f03a..ae7e8d01 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -34,6 +34,7 @@ import ( "github.com/fission/fission/pkg/cache" "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" + "github.com/fission/fission/pkg/executor/metrics" "github.com/fission/fission/pkg/poolcache" ) @@ -227,8 +228,6 @@ func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc) { now := time.Now() fsvc.Ctime = now fsvc.Atime = now - - fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), true) } // SetCPUUtilizaton updates/sets CPUutilization in the pool cache @@ -284,7 +283,6 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { return nil, err } - fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), true) return nil, nil } @@ -345,9 +343,7 @@ func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) { ) } - 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.Since(fsvc.Ctime).Seconds()) - fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), false) + metrics.FuncRunningSummary.WithLabelValues(fsvc.Function.Name, fsvc.Function.Namespace).Observe(fsvc.Atime.Sub(fsvc.Ctime).Seconds()) } // DeleteFunctionSvc deletes a function service at key composed of [function][address]. diff --git a/pkg/executor/fscache/metrics.go b/pkg/executor/fscache/metrics.go deleted file mode 100644 index 93b0abe5..00000000 --- a/pkg/executor/fscache/metrics.go +++ /dev/null @@ -1,100 +0,0 @@ -package fscache - -import ( - "github.com/prometheus/client_golang/prometheus" -) - -var ( - // function_name: the function's name - // function_uid: the function's version id - // function_address: the address of the pod from which the function was called - functionLabels = []string{"function_name", "function_uid"} - functionPodLabels = []string{"function_name", "function_address"} - coldStarts = prometheus.NewCounterVec( - prometheus.CounterOpts{ - Name: "fission_function_cold_starts_total", - Help: "How many cold starts are made by function_name, function_uid.", - }, - functionLabels, - ) - funcRunningSummary = prometheus.NewSummaryVec( - prometheus.SummaryOpts{ - Name: "fission_function_running_seconds", - 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}, - }, - functionLabels, - ) - funcAliveSummary = prometheus.NewSummaryVec( - prometheus.SummaryOpts{ - Name: "fission_function_alive_seconds", - 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}, - }, - functionLabels, - ) - funcIsAlive = prometheus.NewGaugeVec( - prometheus.GaugeOpts{ - Name: "fission_function_is_alive", - Help: "A binary value indicating is the function_name, function_uid alive", - }, - functionLabels, - ) - funcReapTime = prometheus.NewSummaryVec( - prometheus.SummaryOpts{ - Name: "fission_function_pod_reaptime_seconds", - Help: "Amount of seconds to reap a pod", - Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, - }, - functionPodLabels, - ) - idleTime = prometheus.NewSummaryVec( - prometheus.SummaryOpts{ - Name: "fission_function_idle_pod_time", - Help: "Number of seconds it took for Reaper to detect the pod was idle", - Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, - }, - functionPodLabels, - ) -) - -func init() { - // Register the function calls counter with Prometheus's default registry. - prometheus.MustRegister(coldStarts) - prometheus.MustRegister(funcRunningSummary) - prometheus.MustRegister(funcAliveSummary) - prometheus.MustRegister(funcIsAlive) - prometheus.MustRegister(funcReapTime) - prometheus.MustRegister(idleTime) -} - -// IncreaseColdStarts increments the counter by 1. -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)) -} - -// ReapTime is the amount of time taken to reap a pod -func (fsc *FunctionServiceCache) ReapTime(funcName, funcAddress string, time float64) { - funcReapTime.WithLabelValues(funcName, funcAddress).Observe(time) -} - -// IdleTime is the amount of time it took Reaper to find out the pod was idle -func (fsc *FunctionServiceCache) IdleTime(funcName, funcAddress string, time float64) { - idleTime.WithLabelValues(funcName, funcAddress).Observe(time) -} diff --git a/pkg/executor/metrics/metrics.go b/pkg/executor/metrics/metrics.go new file mode 100644 index 00000000..ad806ce4 --- /dev/null +++ b/pkg/executor/metrics/metrics.go @@ -0,0 +1,51 @@ +/* +Copyright 2022 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +var ( + // function_name: the function's name + // function_uid: the function's version id + // function_address: the address of the pod from which the function was called + functionLabels = []string{"function_name", "function_namespace"} + ColdStarts = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "fission_function_cold_starts_total", + Help: "How many cold starts are made by function_name, function_uid.", + }, + functionLabels, + ) + FuncRunningSummary = promauto.NewSummaryVec( + prometheus.SummaryOpts{ + Name: "fission_function_running_seconds", + 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}, + }, + functionLabels, + ) + FuncError = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "fission_function_cold_start_errors_total", + Help: "Count of fission cold start errors", + }, + functionLabels, + ) +) diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index 97923d55..28c4aab9 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -786,30 +786,30 @@ func (fh functionHandler) getProxyErrorHandler(start time.Time, rrt *RetryingRou func (fh functionHandler) collectFunctionMetric(start time.Time, rrt *RetryingRoundTripper, req *http.Request, resp *http.Response) { duration := time.Since(start) + var path string - // Metrics stuff - funcMetricLabels := &functionLabels{ - namespace: fh.function.ObjectMeta.Namespace, - name: fh.function.ObjectMeta.Name, - } - httpMetricLabels := &httpLabels{ - method: req.Method, - } if fh.httpTrigger != nil { - httpMetricLabels.host = fh.httpTrigger.Spec.Host if fh.httpTrigger.Spec.Prefix != nil && *fh.httpTrigger.Spec.Prefix != "" { - httpMetricLabels.path = *fh.httpTrigger.Spec.Prefix + path = *fh.httpTrigger.Spec.Prefix } else { - httpMetricLabels.path = fh.httpTrigger.Spec.RelativeURL + path = fh.httpTrigger.Spec.RelativeURL } } - // Track metrics - httpMetricLabels.code = resp.StatusCode - funcMetricLabels.cached = rrt.urlFromCache + functionCalls.WithLabelValues(fh.function.ObjectMeta.Namespace, + fh.function.ObjectMeta.Name, path, req.Method, + fmt.Sprint(resp.StatusCode)).Inc() - functionCallCompleted(funcMetricLabels, httpMetricLabels, - duration, duration, resp.ContentLength) + if resp.StatusCode >= 400 { + functionCallErrors.WithLabelValues(fh.function.ObjectMeta.Namespace, + fh.function.ObjectMeta.Name, path, req.Method, + fmt.Sprint(resp.StatusCode)).Inc() + } + + functionCallOverhead.WithLabelValues(fh.function.ObjectMeta.Namespace, + fh.function.ObjectMeta.Name, path, req.Method, + fmt.Sprint(resp.StatusCode)). + Observe(float64(duration.Nanoseconds()) / 1e9) // tapService before invoking roundTrip for the serviceUrl if rrt.urlFromCache { diff --git a/pkg/router/httpTriggers.go b/pkg/router/httpTriggers.go index 9bf62d68..0a0ab146 100644 --- a/pkg/router/httpTriggers.go +++ b/pkg/router/httpTriggers.go @@ -41,6 +41,7 @@ import ( genInformer "github.com/fission/fission/pkg/generated/informers/externalversions" "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils" + "github.com/fission/fission/pkg/utils/metrics" "github.com/fission/fission/pkg/utils/otel" "github.com/fission/fission/pkg/utils/tracing" ) @@ -217,6 +218,7 @@ func authLoginHandler(w http.ResponseWriter, r *http.Request) { func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router { muxRouter := mux.NewRouter() + muxRouter.Use(metrics.HTTPMetricMiddleware()) openTracingEnabled := tracing.TracingEnabled(ts.logger) diff --git a/pkg/router/metrics.go b/pkg/router/metrics.go index 99092b6a..fb08e424 100644 --- a/pkg/router/metrics.go +++ b/pkg/router/metrics.go @@ -1,80 +1,35 @@ package router import ( - "fmt" - "sync/atomic" - "time" - "github.com/prometheus/client_golang/prometheus" -) - -var globalFunctionCallCount uint64 - -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 - } + "github.com/prometheus/client_golang/prometheus/promauto" ) var ( - metricAddr = ":8080" - // function + http labels as strings - labelsStrings = []string{"cached", "function_namespace", "function_name", "host", "path", "method", "code"} + labelsStrings = []string{"function_namespace", "function_name", "path", "method", "code"} // Function http calls count - // cached: true | false, is this function service address cached locally // function_namespace: function namespace // function_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( + functionCalls = promauto.NewCounterVec( prometheus.CounterOpts{ Name: "fission_function_calls_total", Help: "Count of Fission function calls", }, labelsStrings, ) - functionCallErrors = prometheus.NewCounterVec( + functionCallErrors = promauto.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( + functionCallOverhead = promauto.NewSummaryVec( prometheus.SummaryOpts{ Name: "fission_function_overhead_seconds", Help: "The function call delay caused by fission.", @@ -82,63 +37,4 @@ var ( }, 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) { - atomic.AddUint64(&globalFunctionCallCount, 1) - - l := labelsToStrings(f, h) - - // overhead: time from request ingress into router up to 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/pkg/router/mutablemux_test.go b/pkg/router/mutablemux_test.go index ca951f16..344a46c7 100644 --- a/pkg/router/mutablemux_test.go +++ b/pkg/router/mutablemux_test.go @@ -25,6 +25,8 @@ import ( "github.com/gorilla/mux" "go.uber.org/zap" "go.uber.org/zap/zapcore" + + "github.com/fission/fission/pkg/utils/metrics" ) func OldHandler(responseWriter http.ResponseWriter, request *http.Request) { @@ -73,6 +75,7 @@ func TestMutableMux(t *testing.T) { // make a simple mutable router log.Print("Create mutable router") muxRouter := mux.NewRouter() + muxRouter.Use(metrics.HTTPMetricMiddleware()) muxRouter.HandleFunc("/", OldHandler) config := zap.NewDevelopmentConfig() config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder @@ -99,6 +102,7 @@ func TestMutableMux(t *testing.T) { // change the muxer log.Print("Change mux router") newMuxRouter := mux.NewRouter() + muxRouter.Use(metrics.HTTPMetricMiddleware()) newMuxRouter.HandleFunc("/", NewHandler) mr.updateRouter(newMuxRouter) diff --git a/pkg/router/router.go b/pkg/router/router.go index 79b57de8..3f7d7b7c 100644 --- a/pkg/router/router.go +++ b/pkg/router/router.go @@ -50,7 +50,6 @@ import ( "time" "github.com/gorilla/mux" - "github.com/prometheus/client_golang/prometheus/promhttp" "go.opencensus.io/plugin/ochttp" "go.opencensus.io/trace" "go.opentelemetry.io/otel" @@ -59,6 +58,7 @@ import ( "github.com/fission/fission/pkg/crd" executorClient "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/throttler" + "github.com/fission/fission/pkg/utils/metrics" otelUtils "github.com/fission/fission/pkg/utils/otel" ) @@ -68,13 +68,15 @@ import ( func router(ctx context.Context, logger *zap.Logger, httpTriggerSet *HTTPTriggerSet) *mutableRouter { var mr *mutableRouter + mux := mux.NewRouter() + mux.Use(metrics.HTTPMetricMiddleware()) // see issue https://github.com/fission/fission/issues/1317 useEncodedPath, _ := strconv.ParseBool(os.Getenv("USE_ENCODED_PATH")) if useEncodedPath { - mr = newMutableRouter(logger, mux.NewRouter().UseEncodedPath()) + mr = newMutableRouter(logger, mux.UseEncodedPath()) } else { - mr = newMutableRouter(logger, mux.NewRouter()) + mr = newMutableRouter(logger, mux) } httpTriggerSet.subscribeRouter(ctx, mr) @@ -86,9 +88,9 @@ func serve(ctx context.Context, logger *zap.Logger, port int, tracingSamplingRat mr := router(ctx, logger, httpTriggerSet) url := fmt.Sprintf(":%v", port) - var err error + var handler http.Handler if openTracingEnabled { - err = http.ListenAndServe(url, &ochttp.Handler{ + handler = &ochttp.Handler{ Handler: mr, GetStartOptions: func(r *http.Request) trace.StartOptions { // do not trace router healthz endpoint @@ -108,23 +110,16 @@ func serve(ctx context.Context, logger *zap.Logger, port int, tracingSamplingRat Sampler: trace.ProbabilitySampler(tracingSamplingRate), } }, - }) + } } else { - err = http.ListenAndServe(url, otelUtils.GetHandlerWithOTEL(mr, "fission-router", otelUtils.UrlsToIgnore("/router-healthz"))) + handler = otelUtils.GetHandlerWithOTEL(mr, "fission-router", otelUtils.UrlsToIgnore("/router-healthz")) } + err := http.ListenAndServe(url, handler) if err != nil { logger.Error("HTTP server error", zap.Error(err)) } } -func serveMetric(logger *zap.Logger) { - // Expose the registered metrics via HTTP. - http.Handle("/metrics", promhttp.Handler()) - err := http.ListenAndServe(metricAddr, nil) - - logger.Fatal("done listening on metrics endpoint", zap.Error(err)) -} - // Start starts a router func Start(ctx context.Context, logger *zap.Logger, port int, executorURL string, openTracingEnabled bool) { fmap := makeFunctionServiceMap(logger, time.Minute) @@ -253,7 +248,7 @@ func Start(ctx context.Context, logger *zap.Logger, port int, executorURL string svcAddrRetryCount: svcAddrRetryCount, }, isDebugEnv, unTapServiceTimeout, throttler.MakeThrottler(svcAddrUpdateTimeout)) - go serveMetric(logger) + go metrics.ServeMetrics(ctx, logger) logger.Info("starting router", zap.Int("port", port)) diff --git a/pkg/storagesvc/client/storagesvc_test.go b/pkg/storagesvc/client/storagesvc_test.go index d44d66f7..562a754a 100644 --- a/pkg/storagesvc/client/storagesvc_test.go +++ b/pkg/storagesvc/client/storagesvc_test.go @@ -76,27 +76,6 @@ func runMinioDockerContainer(pool *dockertest.Pool) *dockertest.Resource { return resource } -func startS3StorageService(ctx context.Context, endpoint, bucketName, subDir string) { - // testID := uniuri.NewLen(8) - port := 8081 - - config := zap.NewDevelopmentConfig() - config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder - logger, err := config.Build() - panicIf(err) - - log.Println("starting storage svc") - os.Setenv("STORAGE_S3_ENDPOINT", endpoint) - os.Setenv("STORAGE_S3_BUCKET_NAME", bucketName) - os.Setenv("STORAGE_S3_SUB_DIR", subDir) - os.Setenv("STORAGE_S3_ACCESS_KEY_ID", minioAccessKeyID) - os.Setenv("STORAGE_S3_SECRET_ACCESS_KEY", minioSecretAccessKey) - os.Setenv("STORAGE_S3_REGION", minioRegion) - - storage := storagesvc.NewS3Storage() - _ = storagesvc.Start(ctx, logger, storage, port, true) -} - func TestS3StorageService(t *testing.T) { fmt.Println("Test S3 Storage service") var minioClient *minio.Client @@ -135,7 +114,26 @@ func TestS3StorageService(t *testing.T) { // Start storagesvc bucketName := "test-s3-service" subDir := "x/y/z" - startS3StorageService(context.Background(), endpoint, bucketName, subDir) + // testID := uniuri.NewLen(8) + port := 8081 + + config := zap.NewDevelopmentConfig() + config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder + logger, err := config.Build() + panicIf(err) + + log.Println("starting storage svc") + os.Setenv("STORAGE_S3_ENDPOINT", endpoint) + os.Setenv("STORAGE_S3_BUCKET_NAME", bucketName) + os.Setenv("STORAGE_S3_SUB_DIR", subDir) + os.Setenv("STORAGE_S3_ACCESS_KEY_ID", minioAccessKeyID) + os.Setenv("STORAGE_S3_SECRET_ACCESS_KEY", minioSecretAccessKey) + os.Setenv("STORAGE_S3_REGION", minioRegion) + + storage := storagesvc.NewS3Storage() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + _ = storagesvc.Start(ctx, logger, storage, port, true) time.Sleep(time.Second) client := MakeClient(fmt.Sprintf("http://localhost:%v/", 8081)) @@ -146,7 +144,6 @@ func TestS3StorageService(t *testing.T) { // store it metadata := make(map[string]string) - ctx := context.Background() fileID, err := client.Upload(ctx, tmpfile.Name(), &metadata) panicIf(err) @@ -195,12 +192,11 @@ func TestS3StorageService(t *testing.T) { if err == nil { log.Panic("Download succeeded but file isn't supposed to exist") } - } func TestLocalStorageService(t *testing.T) { testID := uniuri.NewLen(8) - port := 8080 + port := 8082 config := zap.NewDevelopmentConfig() config.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder @@ -211,7 +207,10 @@ func TestLocalStorageService(t *testing.T) { localPath := fmt.Sprintf("/tmp/%v", testID) _ = os.Mkdir(localPath, os.ModePerm) storage := storagesvc.NewLocalStorage(localPath) - _ = storagesvc.Start(context.Background(), logger, storage, port, true) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + os.Setenv("METRICS_ADDR", ":8083") + _ = storagesvc.Start(ctx, logger, storage, port, true) time.Sleep(time.Second) client := MakeClient(fmt.Sprintf("http://localhost:%v/", port)) @@ -222,7 +221,6 @@ func TestLocalStorageService(t *testing.T) { // store it metadata := make(map[string]string) - ctx := context.Background() fileID, err := client.Upload(ctx, tmpfile.Name(), &metadata) panicIf(err) diff --git a/pkg/storagesvc/metrics.go b/pkg/storagesvc/metrics.go new file mode 100644 index 00000000..35a94ba0 --- /dev/null +++ b/pkg/storagesvc/metrics.go @@ -0,0 +1,24 @@ +package storagesvc + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +var ( + functionLabels = []string{} + totalArchives = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "fission_archives_total", + Help: "Number of archives stored", + }, + functionLabels, + ) + totalMemoryUsage = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "fission_archive_memory_bytes", + Help: "Amount of memory consumed by archives", + }, + functionLabels, + ) +) diff --git a/pkg/storagesvc/storagesvc.go b/pkg/storagesvc/storagesvc.go index 8fadd452..7500bb0c 100644 --- a/pkg/storagesvc/storagesvc.go +++ b/pkg/storagesvc/storagesvc.go @@ -31,6 +31,7 @@ import ( "go.opencensus.io/plugin/ochttp" "go.uber.org/zap" + "github.com/fission/fission/pkg/utils/metrics" "github.com/fission/fission/pkg/utils/otel" ) @@ -115,6 +116,8 @@ func (ss *StorageService) uploadHandler(w http.ResponseWriter, r *http.Request) return } + totalMemoryUsage.WithLabelValues().Add(float64(fileSize)) + // respond with an ID that can be used to retrieve the file ur := &UploadResponse{ ID: id, @@ -135,6 +138,8 @@ func (ss *StorageService) uploadHandler(w http.ResponseWriter, r *http.Request) zap.String("filename", handler.Filename), ) } + + totalArchives.WithLabelValues().Inc() } func (ss *StorageService) getIdFromRequest(r *http.Request) (string, error) { @@ -154,12 +159,20 @@ func (ss *StorageService) deleteHandler(w http.ResponseWriter, r *http.Request) return } + filesize, err := ss.storageClient.getFileSize(fileId) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + } + err = ss.storageClient.removeFileByID(fileId) if err != nil { msg := fmt.Sprintf("Error deleting item: %v", err) http.Error(w, msg, http.StatusInternalServerError) return } + + totalArchives.WithLabelValues().Dec() + totalMemoryUsage.WithLabelValues().Sub(float64(filesize)) w.WriteHeader(http.StatusOK) } @@ -203,6 +216,7 @@ func MakeStorageService(logger *zap.Logger, storageClient *StowClient, port int) func (ss *StorageService) Start(port int, openTracingEnabled bool) { r := mux.NewRouter() + r.Use(metrics.HTTPMetricMiddleware()) r.HandleFunc("/v1/archive", ss.uploadHandler).Methods("POST") r.HandleFunc("/v1/archive", ss.downloadHandler).Methods("GET") r.HandleFunc("/v1/archive", ss.deleteHandler).Methods("DELETE") @@ -210,14 +224,15 @@ func (ss *StorageService) Start(port int, openTracingEnabled bool) { address := fmt.Sprintf(":%v", port) - var err error + var handler http.Handler if openTracingEnabled { - err = http.ListenAndServe(address, &ochttp.Handler{ + handler = &ochttp.Handler{ Handler: r, - }) + } } else { - err = http.ListenAndServe(address, otel.GetHandlerWithOTEL(r, "fission-storagesvc", otel.UrlsToIgnore("/healthz"))) + handler = otel.GetHandlerWithOTEL(r, "fission-storagesvc", otel.UrlsToIgnore("/healthz")) } + err := http.ListenAndServe(address, handler) ss.logger.Fatal("done listening", zap.Error(err)) } @@ -232,6 +247,7 @@ func Start(ctx context.Context, logger *zap.Logger, storage Storage, port int, o // create http handlers storageService := MakeStorageService(logger, storageClient, port) + go metrics.ServeMetrics(ctx, logger) go storageService.Start(port, openTracingEnabled) // enablePruner prevents storagesvc unit test from needing to talk to kubernetes diff --git a/pkg/storagesvc/stowClient.go b/pkg/storagesvc/stowClient.go index f1e43308..d1ea1d3d 100644 --- a/pkg/storagesvc/stowClient.go +++ b/pkg/storagesvc/stowClient.go @@ -182,6 +182,18 @@ func (client *StowClient) removeFileByID(itemID string) error { return client.container.RemoveItem(itemID) } +func (client *StowClient) getFileSize(itemID string) (int64, error) { + item, err := client.container.Item(itemID) + if err != nil { + if err == stow.ErrNotFound { + return 0, ErrNotFound + } else { + return 0, ErrRetrievingItem + } + } + return item.Size() +} + // filter defines an interface to filter out items from a set of items type filter func(stow.Item, interface{}) bool diff --git a/pkg/utils/metrics/http_metrics.go b/pkg/utils/metrics/http_metrics.go new file mode 100644 index 00000000..4775ef5e --- /dev/null +++ b/pkg/utils/metrics/http_metrics.go @@ -0,0 +1,92 @@ +/* +Copyright 2022 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package metrics + +import ( + "fmt" + "net/http" + + "github.com/gorilla/mux" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + + "github.com/fission/fission/pkg/router/util" +) + +type ResponseWriterWrapper struct { + http.ResponseWriter + statusCode int +} + +var ( + httpRequestsTotal = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "http_requests_total", + Help: "Number of requests by path, method and status code.", + }, + []string{"path", "method", "code"}, + ) + httpRequestDuration = promauto.NewSummaryVec( + prometheus.SummaryOpts{ + Name: "http_requests_duration_seconds", + Help: "Time taken to serve the request by path and method.", + Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, + }, + []string{"path", "method"}, + ) + httpRequestInFlight = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "http_requests_in_flight", + Help: "Number of requests currently being served by path and method.", + }, + []string{"path", "method"}, + ) +) + +func (rw *ResponseWriterWrapper) WriteHeader(statuscode int) { + rw.statusCode = statuscode + rw.ResponseWriter.WriteHeader(statuscode) +} + +func HTTPMetricMiddleware() mux.MiddlewareFunc { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if util.IsWebsocketRequest(r) { + next.ServeHTTP(w, r) + return + } + labels := make(prometheus.Labels, 0) + labels["path"] = r.URL.Path + if route := mux.CurrentRoute(r); route != nil { + if routePath, err := route.GetPathTemplate(); err == nil { + labels["path"] = routePath + } + } + labels["method"] = r.Method + rw := ResponseWriterWrapper{w, http.StatusOK} + httpRequestInFlight.With(labels).Inc() + httpRequestDuration := prometheus.NewTimer(httpRequestDuration.With(labels)) + defer func() { + httpRequestDuration.ObserveDuration() + httpRequestInFlight.With(labels).Dec() + labels["code"] = fmt.Sprintf("%d", rw.statusCode) + httpRequestsTotal.With(labels).Inc() + }() + next.ServeHTTP(&rw, r) + }) + } +} diff --git a/pkg/utils/metrics/server.go b/pkg/utils/metrics/server.go new file mode 100644 index 00000000..e5205160 --- /dev/null +++ b/pkg/utils/metrics/server.go @@ -0,0 +1,56 @@ +/* +Copyright 2022 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package metrics + +import ( + "context" + "net/http" + "os" + + "github.com/prometheus/client_golang/prometheus/promhttp" + "go.uber.org/zap" +) + +func ServeMetrics(ctx context.Context, logger *zap.Logger) { + metricsAddr := os.Getenv("METRICS_ADDR") + if metricsAddr == "" { + metricsAddr = ":8080" + } + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.Handler()) + s := &http.Server{ + Addr: metricsAddr, + Handler: mux, + } + logger.Info("Starting metrics server", zap.String("address", metricsAddr)) + go func() { + if err := s.ListenAndServe(); err != nil { + if err != http.ErrServerClosed { + logger.Error("Metrics server error", zap.Error(err)) + } + } + }() + <-ctx.Done() + logger.Info("Shutting down metrics server") + err := s.Shutdown(ctx) + if err == context.DeadlineExceeded || err == context.Canceled { + return + } + if err != nil { + logger.Error("Failed to shutdown metrics server", zap.Error(err)) + } +}