Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Prometheus metrics improvements (#2398)

Browse Source 
- 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 <sanketsudake@gmail.com>
Co-authored-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Ankit Chawla
2022-04-13 21:49:46 +05:30
committed by GitHub
co-authored by Sanket Sudake
parent 231de707dc
commit b638a6d047
26 changed files with 359 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files
charts/fission-all/templates/buildermgr/deployment.yaml
+7
View File
@@ -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 }}
charts/fission-all/templates/controller/deployment.yaml
+6
View File
@@ -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 }}
charts/fission-all/templates/storagesvc/deployment.yaml
+6
View File
@@ -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 }}
pkg/buildermgr/pkgwatcher.go
+2
View File
@@ -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())
pkg/controller/api.go
+5 -1
View File
@@ -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))
}
pkg/controller/controller.go
+1 -1
View File
@@ -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)
}
pkg/executor/api.go
+2
View File
@@ -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")
pkg/executor/executor.go
+2 -12
View File
@@ -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
}
pkg/executor/executortype/container/containermgr.go
+3 -1
View File
@@ -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
}
pkg/executor/executortype/newdeploy/newdeploymgr.go
+3 -5
View File
@@ -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())
}()
}
}
pkg/executor/executortype/poolmgr/gp.go
+2 -1
View File
@@ -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),
pkg/executor/executortype/poolmgr/gpm.go
-4
View File
@@ -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())
}
}
}()
pkg/executor/fscache/functionServiceCache.go
+2 -6
View File
@@ -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].
pkg/executor/fscache/metrics.go
-100
View File
@@ -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)
}
pkg/executor/metrics/metrics.go
+51
View File
@@ -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,
)
)
pkg/router/functionHandler.go
+16 -16
View File
@@ -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 {
pkg/router/httpTriggers.go
+2
View File
@@ -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)
pkg/router/metrics.go
+5 -109
View File
@@ -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))
}
}
pkg/router/mutablemux_test.go
+4
View File
@@ -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)
pkg/router/router.go
+11 -16
View File
@@ -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))
pkg/storagesvc/client/storagesvc_test.go
+25 -27
View File
@@ -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)
pkg/storagesvc/metrics.go
+24
View File
@@ -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,
)
)
pkg/storagesvc/storagesvc.go
+20 -4
View File
@@ -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
pkg/storagesvc/stowClient.go
+12
View File
@@ -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
pkg/utils/metrics/http_metrics.go
+92
View File
@@ -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)
})
}
}
pkg/utils/metrics/server.go
+56
View File
@@ -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))
}
}