add lastFailTime into cache

Author: yuxiangqian

pkg/metrics/metrics.go

@@ -34,7 +34,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 +45,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 +61,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 +93,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 +128,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 +145,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 +189,11 @@ 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")
 	}
-
+	srv := &http.Server{Addr: addr}
 	http.Handle(pattern, k8smetrics.HandlerFor(
 		opMgr.registry,
 		k8smetrics.HandlerOpts{
@@ -170,10 +203,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.ListenAndServe(); err != http.ErrServerClosed {
 			klog.Fatalf("failed to start endpoint at:%s/%s, error: %v", addr, pattern, err)
 		}
 	}()
-	return nil
+	return srv, nil
 }

pkg/metrics/metrics_test.go

@@ -17,9 +17,12 @@ limitations under the License.
 package metrics
 
 import (
+	"context"
 	"io/ioutil"
+	"log"
 	"math"
 	"net/http"
+	"os"
 	"reflect"
 	"strings"
 	"sync"
@@ -32,38 +35,54 @@ import (
 )
 
 var (
-	mgr  MetricsManager
-	once sync.Once
+	mgr MetricsManager
+	srv *http.Server
+	wg  *sync.WaitGroup
 )
 
 const (
 	httpPattern = "/metrics"
-	addr        = ":12120"
-	httpAddr    = "http://localhost:12120/metrics"
+	addr        = "localhost:12134"
+	httpAddr    = "http://" + addr + httpPattern
 )
 
 func initMgr() {
+	wg = &sync.WaitGroup{}
+	wg.Add(1)
 	mgr = NewMetricsManager()
-	if err := mgr.StartServing(httpPattern, addr, nil); err != nil {
+	var err error
+	if srv, err = mgr.StartServing(httpPattern, addr, nil, wg); err != nil {
+		log.Fatalf("failed to start serving%v", err)
 	}
 }
 
+func shutdown() {
+	if err := srv.Shutdown(context.Background()); err != nil {
+		panic(err)
+	}
+	wg.Wait()
+}
+
+func TestMain(m *testing.M) {
+	initMgr()
+	c := m.Run()
+	shutdown()
+	os.Exit(c)
+}
+
 func TestNew(t *testing.T) {
-	once.Do(initMgr)
 	if mgr == nil {
 		t.Errorf("failed testing new")
 	}
 }
 
 func TestStartServing(t *testing.T) {
-	once.Do(initMgr)
 	if _, err := http.Get(httpAddr); err != nil {
 		t.Errorf("failed to get response from server %v", err)
 	}
 }
 
 func TestOperationStart(t *testing.T) {
-	once.Do(initMgr)
 	// add an operation
 	op := Operation{
 		Name:       "op1",
@@ -103,7 +122,7 @@ func TestOperationStart(t *testing.T) {
 	}
 
 	expected :=
-		`# HELP snapshot_controller_operation_total_seconds [ALPHA] Total number of seconds spent by the snapshot controller for an operation to complete end to end
+		`# HELP snapshot_controller_operation_total_seconds [ALPHA] Total number of seconds spent by the controller on an operation from end to end
 # TYPE snapshot_controller_operation_total_seconds histogram
 snapshot_controller_operation_total_seconds_bucket{driver_name="driver1",operation_name="op1",operation_status="InterimFailure",le="0.1"} 0
 snapshot_controller_operation_total_seconds_bucket{driver_name="driver1",operation_name="op1",operation_status="InterimFailure",le="0.25"} 0