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)) + } +}