add metric to common controller - basic metrics utilities

use a different port, use a longer delta

Author: yuxiangqian

go.mod

@@ -10,11 +10,15 @@ require (
 	github.com/imdario/mergo v0.3.7 // indirect
 	github.com/kubernetes-csi/csi-lib-utils v0.7.0
 	github.com/kubernetes-csi/csi-test v2.0.0+incompatible
+	github.com/prometheus/client_golang v1.0.0
+	github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4
+	github.com/prometheus/common v0.4.1
 	google.golang.org/grpc v1.26.0
 	k8s.io/api v0.17.0
 	k8s.io/apimachinery v0.17.1-beta.0
 	k8s.io/client-go v0.17.0
 	k8s.io/code-generator v0.0.0-20191121015212-c4c8f8345c7e
+	k8s.io/component-base v0.17.0
 	k8s.io/klog v1.0.0
 	k8s.io/kubernetes v1.14.0
 )

pkg/metrics/metrics.go

@@ -0,0 +1,190 @@
+/*
+Copyright 2020 The Kubernetes 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"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"k8s.io/apimachinery/pkg/types"
+	k8smetrics "k8s.io/component-base/metrics"
+	"k8s.io/klog"
+)
+
+const (
+	labelDriverName               = "driver_name"
+	labelOperationName            = "operation_name"
+	labelOperationStatus          = "operation_status"
+	snapshotController            = "snapshot_controller"
+	operationLatencyMetricName    = "operation_total_seconds"
+	operationLatencyMetricHelpMsg = "Total number of seconds spent by the snapshot controller for an operation to complete end to end"
+)
+
+type OperationState string
+
+const (
+	// InterimFailure is an operation state which means the controller will keep
+	// on retrying the operation upon encountering this type of error.
+	InterimFailure OperationState = "InterimFailure"
+
+	// TerminatingFailure is an operation state which means the controller has encountered
+	// an error which it's not able to resume the operation and has marked the
+	// permanently failure of the operation.
+	TerminatingFailure OperationState = "TerminatingFailure"
+
+	// Success states that the controller has successfully executed the operation.
+	Success OperationState = "Success"
+)
+
+var operationLatencyMetricBuckets = []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 30, 60, 120, 300, 600}
+
+type MetricsManager interface {
+	// StartServing starts the metrics endpoint at the specified addr/pattern for
+	// metrics managed by this MetricsManager. It spawn a goroutine to listen to
+	// and serve HTTP requests received on addr/pattern
+	// if the "pattern" is empty (i.e., ""), no endpoint will be started. An error
+	// will be returned if there is any.
+	StartServing(pattern, addr string, logger promhttp.Logger) error
+
+	// OperationStart takes in an operation and cache it's start time.
+	// if the operation already exists, it's an no-op.
+	OperationStart(op Operation)
+
+	// RecordMetrics records a metric point. Note that it will be an no-op if an
+	// operation has not been marked "Started" previously via invoking "OperationStart".
+	// op - the Operation which the metric is associated with.
+	// state - metrics manager operations differently to the cache based on the state
+	//         of the Operation:
+	//   InterimFailure: record a failure metric however keep operation's start time in cache
+	//   TerminatingFailure: record a failure metric and remove operation's start time from cache
+	//   Success: record a success metric and remove the operation's start time from cache
+	RecordMetrics(op Operation, state OperationState)
+}
+
+// Operation is a structure which holds information to identifier a snapshot
+// related operation
+type Operation struct {
+	// the name of the operation, for example: "CreateSnapshot", "DeleteSnapshot"
+	Name string
+	// the name of the driver which executes the operation
+	Driver string
+	// the resource UID to which the operation has been executed against
+	ResourceID types.UID
+}
+
+func (op *Operation) key() string {
+	return fmt.Sprintf("%s-%s-%s", op.Name, op.Driver, op.ResourceID)
+}
+
+type operationTs struct {
+	operation Operation
+	startTime time.Time
+}
+
+type operationMetricsManager struct {
+	// cache is a concurrent-safe map which stores start timestamps for all
+	// ongoing operations.
+	// key is
+	// value is
+	cache sync.Map
+
+	registry k8smetrics.KubeRegistry
+
+	opLatencyMetric *k8smetrics.HistogramVec
+}
+
+func NewMetricsManager() MetricsManager {
+	mgr := &operationMetricsManager{
+		cache: sync.Map{},
+	}
+	mgr.init()
+	return mgr
+}
+
+func (opMgr *operationMetricsManager) OperationStart(op Operation) {
+	opTs := operationTs{
+		operation: op,
+		startTime: time.Now(),
+	}
+	opMgr.cache.LoadOrStore(op.key(), &opTs)
+}
+
+func (opMgr *operationMetricsManager) RecordMetrics(op Operation, state OperationState) {
+	key := op.key()
+	obj, exists := opMgr.cache.Load(key)
+	if !exists {
+		// the operation has not been cached, return directly
+		return
+	}
+	ts, ok := obj.(*operationTs)
+	if !ok {
+		// the cached item is not an operationTs, should NEVER happen, clean and return
+		klog.Errorf("Invalid cache entry for key %s", key)
+		opMgr.cache.Delete(key)
+		return
+	}
+	duration := time.Since(ts.startTime).Seconds()
+	opMgr.opLatencyMetric.WithLabelValues(op.Driver, op.Name, string(state)).Observe(duration)
+	switch state {
+	case Success, TerminatingFailure:
+		opMgr.cache.Delete(key)
+	case InterimFailure:
+		// do nothing
+	default:
+		// log error
+		klog.Errorf("Not supported operation state %s", state)
+	}
+}
+
+func (opMgr *operationMetricsManager) init() {
+	opMgr.registry = k8smetrics.NewKubeRegistry()
+	opMgr.opLatencyMetric = k8smetrics.NewHistogramVec(
+		&k8smetrics.HistogramOpts{
+			Subsystem: snapshotController,
+			Name:      operationLatencyMetricName,
+			Help:      operationLatencyMetricHelpMsg,
+			Buckets:   operationLatencyMetricBuckets,
+		},
+		[]string{labelDriverName, labelOperationName, labelOperationStatus},
+	)
+	opMgr.registry.MustRegister(opMgr.opLatencyMetric)
+}
+
+func (opMgr *operationMetricsManager) StartServing(pattern, addr string, logger promhttp.Logger) error {
+	if addr == "" {
+		return fmt.Errorf("metrics endpoint will not be started as endpoint address is not specified")
+	}
+
+	http.Handle(pattern, k8smetrics.HandlerFor(
+		opMgr.registry,
+		k8smetrics.HandlerOpts{
+			ErrorLog:      logger,
+			ErrorHandling: k8smetrics.ContinueOnError,
+		}))
+
+	// start serving the endpoint
+	go func() {
+		err := http.ListenAndServe(addr, nil)
+		if err != nil {
+			klog.Fatalf("failed to start endpoint at:%s/%s, error: %v", addr, pattern, err)
+		}
+	}()
+	return nil
+}