add lastFailTime into cache

Author: yuxiangqian

pkg/metrics/metrics.go

@@ -18,6 +18,7 @@ package metrics
 
 import (
 	"fmt"
+	"net"
 	"net/http"
 	"sync"
 	"time"
@@ -34,7 +35,7 @@ const (
 	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"
+	operationLatencyMetricHelpMsg = "Total number of seconds spent by the controller on an operation from end to end"
 )
 
 type OperationState string
@@ -45,8 +46,8 @@ const (
 	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 a
-	// permanent failure of the operation.
+	// an error which is not recoverable and the controller has marked a permanent
+	// failure of the operation.
 	TerminatingFailure OperationState = "TerminatingFailure"
 
 	// Success states that the controller has successfully executed the operation.
@@ -61,12 +62,16 @@ type MetricsManager interface {
 	// 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
+	StartServing(pattern, addr string, logger promhttp.Logger, wg *sync.WaitGroup) (*http.Server, 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)
 
+	// DropOperation removes an operation from cache.
+	// if the operation does not exist, it's an no-op.
+	DropOperation(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.
@@ -89,11 +94,23 @@ type Operation struct {
 	ResourceID types.UID
 }
 
+type operationTs struct {
+	// startTime is the timestamp when an operation has been picked up by the
+	// controller for processing
+	startTime time.Time
+
+	// lastFailTime is the timestamp of the last interim failure on an operation
+	// if this field is not specified, i.e., lastFailTime.IsZero == true, it means
+	// the operation has never failed previously
+	lastFailTime time.Time
+}
+
 type operationMetricsManager struct {
 	// cache is a concurrent-safe map which stores start timestamps for all
 	// ongoing operations.
 	// key is an Operation
-	// value is the Operation's start time
+	// value is the operationTs which records the start time and last failing time
+	// of the operation
 	cache sync.Map
 
 	// registry is a wrapper around Prometheus Registry
@@ -112,7 +129,14 @@ func NewMetricsManager() MetricsManager {
 }
 
 func (opMgr *operationMetricsManager) OperationStart(op Operation) {
-	opMgr.cache.LoadOrStore(op, time.Now())
+	opTs := operationTs{
+		startTime: time.Now(),
+	}
+	opMgr.cache.LoadOrStore(op, opTs)
+}
+
+func (opMgr *operationMetricsManager) DropOperation(op Operation) {
+	opMgr.cache.Delete(op)
 }
 
 func (opMgr *operationMetricsManager) RecordMetrics(op Operation, state OperationState) {
@@ -122,24 +146,34 @@ func (opMgr *operationMetricsManager) RecordMetrics(op Operation, state Operatio
 		// the operation has not been cached, return directly
 		return
 	}
-	ts, ok := obj.(time.Time)
+	ts, ok := obj.(operationTs)
 	if !ok {
 		// the cached item is not a time.Time, should NEVER happen, clean and return
 		klog.Errorf("Invalid cache entry for key %v", op)
 		opMgr.cache.Delete(op)
 		return
 	}
-	duration := time.Since(ts).Seconds()
-	opMgr.opLatencyMetrics.WithLabelValues(op.Driver, op.Name, string(state)).Observe(duration)
+	duration := time.Since(ts.startTime).Seconds()
 	switch state {
 	case Success, TerminatingFailure:
 		opMgr.cache.Delete(op)
 	case InterimFailure:
-		// do nothing
+		if !ts.lastFailTime.IsZero() {
+			// override duration
+			duration = time.Since(ts.lastFailTime).Seconds()
+		}
+		// override lastFailTime
+		newTs := operationTs{
+			startTime:    ts.startTime,
+			lastFailTime: time.Now(),
+		}
+		opMgr.cache.Store(op, newTs)
 	default:
 		// log error
 		klog.Errorf("Not supported operation state %s", state)
+		return
 	}
+	opMgr.opLatencyMetrics.WithLabelValues(op.Driver, op.Name, string(state)).Observe(duration)
 }
 
 func (opMgr *operationMetricsManager) init() {
@@ -156,11 +190,16 @@ func (opMgr *operationMetricsManager) init() {
 	opMgr.registry.MustRegister(opMgr.opLatencyMetrics)
 }
 
-func (opMgr *operationMetricsManager) StartServing(pattern, addr string, logger promhttp.Logger) error {
+func (opMgr *operationMetricsManager) StartServing(pattern, addr string, logger promhttp.Logger, wg *sync.WaitGroup) (*http.Server, error) {
 	if addr == "" {
-		return fmt.Errorf("metrics endpoint will not be started as endpoint address is not specified")
+		return nil, fmt.Errorf("metrics endpoint will not be started as endpoint address is not specified")
 	}
-
+	// start listening
+	l, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to listen on address[%s], error[%v]", addr, err)
+	}
+	srv := &http.Server{Addr: addr}
 	http.Handle(pattern, k8smetrics.HandlerFor(
 		opMgr.registry,
 		k8smetrics.HandlerOpts{
@@ -170,10 +209,10 @@ func (opMgr *operationMetricsManager) StartServing(pattern, addr string, logger
 
 	// start serving the endpoint
 	go func() {
-		err := http.ListenAndServe(addr, nil)
-		if err != nil {
+		defer wg.Done()
+		if err := srv.Serve(l); err != http.ErrServerClosed {
 			klog.Fatalf("failed to start endpoint at:%s/%s, error: %v", addr, pattern, err)
 		}
 	}()
-	return nil
+	return srv, nil
 }