Merge pull request #78061 from yuxiangqian/metrics

pkg/controller/volume/persistentvolume/framework_test.go

@@ -419,6 +419,26 @@ func claimWithAnnotation(name, value string, claims []*v1.PersistentVolumeClaim)
 	return claims
 }
 
+// volumeWithAnnotation saves given annotation into given volume.
+// Meant to be used to compose volume specified inline in a test.
+func volumeWithAnnotation(name, value string, volume *v1.PersistentVolume) *v1.PersistentVolume {
+	if volume.Annotations == nil {
+		volume.Annotations = map[string]string{name: value}
+	} else {
+		volume.Annotations[name] = value
+	}
+	return volume
+}
+
+// volumesWithAnnotation saves given annotation into given volumes.
+// Meant to be used to compose volumes specified inline in a test.
+func volumesWithAnnotation(name, value string, volumes []*v1.PersistentVolume) []*v1.PersistentVolume {
+	for _, volume := range volumes {
+		volumeWithAnnotation(name, value, volume)
+	}
+	return volumes
+}
+
 // claimWithAccessMode saves given access into given claims.
 // Meant to be used to compose claims specified inline in a test.
 func claimWithAccessMode(modes []v1.PersistentVolumeAccessMode, claims []*v1.PersistentVolumeClaim) []*v1.PersistentVolumeClaim {

pkg/controller/volume/persistentvolume/metrics/BUILD

@@ -6,6 +6,7 @@ go_library(
     importpath = "k8s.io/kubernetes/pkg/controller/volume/persistentvolume/metrics",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/volume/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",

pkg/controller/volume/persistentvolume/metrics/metrics.go

@@ -18,11 +18,12 @@ package metrics
 
 import (
 	"sync"
-
-	"k8s.io/api/core/v1"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/klog"
+	metricutil "k8s.io/kubernetes/pkg/volume/util"
 )
 
 const (
@@ -56,7 +57,6 @@ type PVCLister interface {
 func Register(pvLister PVLister, pvcLister PVCLister) {
 	registerMetrics.Do(func() {
 		prometheus.MustRegister(newPVAndPVCCountCollector(pvLister, pvcLister))
-		prometheus.MustRegister(volumeOperationMetric)
 		prometheus.MustRegister(volumeOperationErrorsMetric)
 	})
 }
@@ -92,12 +92,6 @@ var (
 		"Gauge measuring number of persistent volume claim currently unbound",
 		[]string{namespaceLabel}, nil)
 
-	volumeOperationMetric = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Name: "volume_operation_total_seconds",
-			Help: "Total volume operation time",
-		},
-		[]string{"plugin_name", "operation_name"})
 	volumeOperationErrorsMetric = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Name: "volume_operation_total_errors",
@@ -198,14 +192,79 @@ func (collector *pvAndPVCCountCollector) pvcCollect(ch chan<- prometheus.Metric)
 	}
 }
 
-// RecordVolumeOperationMetric records the latency and errors of volume operations.
-func RecordVolumeOperationMetric(pluginName, opName string, timeTaken float64, err error) {
+// RecordVolumeOperationErrorMetric records error count into metric
+// volume_operation_total_errors for provisioning/deletion operations
+func RecordVolumeOperationErrorMetric(pluginName, opName string) {
 	if pluginName == "" {
 		pluginName = "N/A"
 	}
-	if err != nil {
-		volumeOperationErrorsMetric.WithLabelValues(pluginName, opName).Inc()
+	volumeOperationErrorsMetric.WithLabelValues(pluginName, opName).Inc()
+}
+
+// operationTimestamp stores the start time of an operation by a plugin
+type operationTimestamp struct {
+	pluginName string
+	operation  string
+	startTs    time.Time
+}
+
+func newOperationTimestamp(pluginName, operationName string) *operationTimestamp {
+	return &operationTimestamp{
+		pluginName: pluginName,
+		operation:  operationName,
+		startTs:    time.Now(),
+	}
+}
+
+// OperationStartTimeCache concurrent safe cache for operation start timestamps
+type OperationStartTimeCache struct {
+	cache sync.Map // [string]operationTimestamp
+}
+
+// NewOperationStartTimeCache creates a operation timestamp cache
+func NewOperationStartTimeCache() OperationStartTimeCache {
+	return OperationStartTimeCache{
+		cache: sync.Map{}, //[string]operationTimestamp {}
+	}
+}
+
+// AddIfNotExist returns directly if there exists an entry with the key. Otherwise, it
+// creates a new operation timestamp using operationName, pluginName, and current timestamp
+// and stores the operation timestamp with the key
+func (c *OperationStartTimeCache) AddIfNotExist(key, pluginName, operationName string) {
+	ts := newOperationTimestamp(pluginName, operationName)
+	c.cache.LoadOrStore(key, ts)
+}
+
+// Delete deletes a value for a key.
+func (c *OperationStartTimeCache) Delete(key string) {
+	c.cache.Delete(key)
+}
+
+// Has returns a bool value indicates the existence of a key in the cache
+func (c *OperationStartTimeCache) Has(key string) bool {
+	_, exists := c.cache.Load(key)
+	return exists
+}
+
+// RecordMetric records either an error count metric or a latency metric if there
+// exists a start timestamp entry in the cache. For a successful operation, i.e.,
+// err == nil, the corresponding timestamp entry will be removed from cache
+func RecordMetric(key string, c *OperationStartTimeCache, err error) {
+	obj, exists := c.cache.Load(key)
+	if !exists {
+		return
+	}
+	ts, ok := obj.(*operationTimestamp)
+	if !ok {
 		return
 	}
-	volumeOperationMetric.WithLabelValues(pluginName, opName).Observe(timeTaken)
+	if err != nil {
+		RecordVolumeOperationErrorMetric(ts.pluginName, ts.operation)
+	} else {
+		timeTaken := time.Since(ts.startTs).Seconds()
+		metricutil.RecordOperationLatencyMetric(ts.pluginName, ts.operation, timeTaken)
+		// end of this operation, remove the timestamp entry from cache
+		c.Delete(key)
+	}
 }

pkg/controller/volume/persistentvolume/provision_test.go

@@ -20,7 +20,7 @@ import (
 	"errors"
 	"testing"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	storage "k8s.io/api/storage/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -432,6 +432,17 @@ func TestProvisionSync(t *testing.T) {
 			[]string{"Normal ExternalProvisioning"},
 			noerrors, wrapTestWithCSIMigrationProvisionCalls(testSyncClaim),
 		},
+		{
+			// volume provisioned and available
+			// in this case, NO normal event with external provisioner should be issued
+			"11-22 - external provisioner with volume available",
+			newVolumeArray("volume11-22", "1Gi", "", "", v1.VolumeAvailable, v1.PersistentVolumeReclaimRetain, classExternal),
+			newVolumeArray("volume11-22", "1Gi", "uid11-22", "claim11-22", v1.VolumeBound, v1.PersistentVolumeReclaimRetain, classExternal, pvutil.AnnBoundByController),
+			newClaimArray("claim11-22", "uid11-22", "1Gi", "", v1.ClaimPending, &classExternal),
+			newClaimArray("claim11-22", "uid11-22", "1Gi", "volume11-22", v1.ClaimBound, &classExternal, pvutil.AnnBoundByController, pvutil.AnnBindCompleted),
+			noevents,
+			noerrors, wrapTestWithProvisionCalls([]provisionCall{}, testSyncClaim),
+		},
 	}
 	runSyncTests(t, tests, storageClasses, []*v1.Pod{})
 }
@@ -461,6 +472,68 @@ func TestProvisionMultiSync(t *testing.T) {
 			newClaimArray("claim12-1", "uid12-1", "1Gi", "pvc-uid12-1", v1.ClaimBound, &classGold, pvutil.AnnBoundByController, pvutil.AnnBindCompleted, pvutil.AnnStorageProvisioner),
 			noevents, noerrors, wrapTestWithProvisionCalls([]provisionCall{provision1Success}, testSyncClaim),
 		},
+		{
+			// provision a volume (external provisioner) and binding + normal event with external provisioner
+			"12-2 - external provisioner with volume provisioned success",
+			novolumes,
+			newVolumeArray("pvc-uid12-2", "1Gi", "uid12-2", "claim12-2", v1.VolumeBound, v1.PersistentVolumeReclaimRetain, classExternal, pvutil.AnnBoundByController),
+			newClaimArray("claim12-2", "uid12-2", "1Gi", "", v1.ClaimPending, &classExternal),
+			claimWithAnnotation(pvutil.AnnStorageProvisioner, "vendor.com/my-volume",
+				newClaimArray("claim12-2", "uid12-2", "1Gi", "pvc-uid12-2", v1.ClaimBound, &classExternal, pvutil.AnnBoundByController, pvutil.AnnBindCompleted)),
+			[]string{"Normal ExternalProvisioning"},
+			noerrors,
+			wrapTestWithInjectedOperation(wrapTestWithProvisionCalls([]provisionCall{}, testSyncClaim), func(ctrl *PersistentVolumeController, reactor *pvtesting.VolumeReactor) {
+				// Create a volume before syncClaim tries to bind a PV to PVC
+				// This simulates external provisioner creating a volume while the controller
+				// is waiting for a volume to bind to the existed claim
+				// the external provisioner workflow implemented in "provisionClaimOperationCSI"
+				// should issue an ExternalProvisioning event to signal that some external provisioner
+				// is working on provisioning the PV, also add the operation start timestamp into local cache
+				// operationTimestamps. Rely on the existences of the start time stamp to create a PV for binding
+				if ctrl.operationTimestamps.Has("default/claim12-2") {
+					volume := newVolume("pvc-uid12-2", "1Gi", "", "", v1.VolumeAvailable, v1.PersistentVolumeReclaimRetain, classExternal)
+					ctrl.volumes.store.Add(volume) // add the volume to controller
+					reactor.AddVolume(volume)
+				}
+			}),
+		},
+		{
+			// provision a volume (external provisioner) but binding will not happen + normal event with external provisioner
+			"12-3 - external provisioner with volume to be provisioned",
+			novolumes,
+			novolumes,
+			newClaimArray("claim12-3", "uid12-3", "1Gi", "", v1.ClaimPending, &classExternal),
+			claimWithAnnotation(pvutil.AnnStorageProvisioner, "vendor.com/my-volume",
+				newClaimArray("claim12-3", "uid12-3", "1Gi", "", v1.ClaimPending, &classExternal)),
+			[]string{"Normal ExternalProvisioning"},
+			noerrors,
+			wrapTestWithProvisionCalls([]provisionCall{provision1Success}, testSyncClaim),
+		},
+		{
+			// provision a volume (external provisioner) and binding + normal event with external provisioner
+			"12-4 - external provisioner with volume provisioned/bound success",
+			novolumes,
+			newVolumeArray("pvc-uid12-4", "1Gi", "uid12-4", "claim12-4", v1.VolumeBound, v1.PersistentVolumeReclaimRetain, classExternal, pvutil.AnnBoundByController),
+			newClaimArray("claim12-4", "uid12-4", "1Gi", "", v1.ClaimPending, &classExternal),
+			claimWithAnnotation(pvutil.AnnStorageProvisioner, "vendor.com/my-volume",
+				newClaimArray("claim12-4", "uid12-4", "1Gi", "pvc-uid12-4", v1.ClaimBound, &classExternal, pvutil.AnnBoundByController, pvutil.AnnBindCompleted)),
+			[]string{"Normal ExternalProvisioning"},
+			noerrors,
+			wrapTestWithInjectedOperation(wrapTestWithProvisionCalls([]provisionCall{}, testSyncClaim), func(ctrl *PersistentVolumeController, reactor *pvtesting.VolumeReactor) {
+				// Create a volume before syncClaim tries to bind a PV to PVC
+				// This simulates external provisioner creating a volume while the controller
+				// is waiting for a volume to bind to the existed claim
+				// the external provisioner workflow implemented in "provisionClaimOperationCSI"
+				// should issue an ExternalProvisioning event to signal that some external provisioner
+				// is working on provisioning the PV, also add the operation start timestamp into local cache
+				// operationTimestamps. Rely on the existences of the start time stamp to create a PV for binding
+				if ctrl.operationTimestamps.Has("default/claim12-4") {
+					volume := newVolume("pvc-uid12-4", "1Gi", "uid12-4", "claim12-4", v1.VolumeBound, v1.PersistentVolumeReclaimRetain, classExternal, pvutil.AnnBoundByController)
+					ctrl.volumes.store.Add(volume) // add the volume to controller
+					reactor.AddVolume(volume)
+				}
+			}),
+		},
 	}
 
 	runMultisyncTests(t, tests, storageClasses, storageClasses[0].Name)

pkg/controller/volume/persistentvolume/pv_controller_base.go

@@ -21,7 +21,7 @@ import (
 	"strconv"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -92,6 +92,7 @@ func NewController(p ControllerParameters) (*PersistentVolumeController, error)
 		claimQueue:                    workqueue.NewNamed("claims"),
 		volumeQueue:                   workqueue.NewNamed("volumes"),
 		resyncPeriod:                  p.SyncPeriod,
+		operationTimestamps:           metrics.NewOperationStartTimeCache(),
 	}
 
 	// Prober is nil because PV is not aware of Flexvolume.
@@ -209,6 +210,10 @@ func (ctrl *PersistentVolumeController) updateVolume(volume *v1.PersistentVolume
 func (ctrl *PersistentVolumeController) deleteVolume(volume *v1.PersistentVolume) {
 	_ = ctrl.volumes.store.Delete(volume)
 	klog.V(4).Infof("volume %q deleted", volume.Name)
+	// record deletion metric if a deletion start timestamp is in the cache
+	// the following calls will be a no-op if there is nothing for this volume in the cache
+	// end of timestamp cache entry lifecycle, "RecordMetric" will do the clean
+	metrics.RecordMetric(volume.Name, &ctrl.operationTimestamps, nil)
 
 	if volume.Spec.ClaimRef == nil {
 		return
@@ -245,20 +250,26 @@ func (ctrl *PersistentVolumeController) updateClaim(claim *v1.PersistentVolumeCl
 	}
 }
 
+// Unit test [5-5] [5-6] [5-7]
 // deleteClaim runs in worker thread and handles "claim deleted" event.
 func (ctrl *PersistentVolumeController) deleteClaim(claim *v1.PersistentVolumeClaim) {
 	_ = ctrl.claims.Delete(claim)
-	klog.V(4).Infof("claim %q deleted", claimToClaimKey(claim))
+	claimKey := claimToClaimKey(claim)
+	klog.V(4).Infof("claim %q deleted", claimKey)
+	// clean any possible unfinished provision start timestamp from cache
+	// Unit test [5-8] [5-9]
+	ctrl.operationTimestamps.Delete(claimKey)
 
 	volumeName := claim.Spec.VolumeName
 	if volumeName == "" {
-		klog.V(5).Infof("deleteClaim[%q]: volume not bound", claimToClaimKey(claim))
+		klog.V(5).Infof("deleteClaim[%q]: volume not bound", claimKey)
 		return
 	}
+
 	// sync the volume when its claim is deleted.  Explicitly sync'ing the
 	// volume here in response to claim deletion prevents the volume from
 	// waiting until the next sync period for its Release.
-	klog.V(5).Infof("deleteClaim[%q]: scheduling sync of volume %s", claimToClaimKey(claim), volumeName)
+	klog.V(5).Infof("deleteClaim[%q]: scheduling sync of volume %s", claimKey, volumeName)
 	ctrl.volumeQueue.Add(volumeName)
 }
 

pkg/controller/volume/persistentvolume/pv_controller_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package persistentvolume
 
 import (
+	"errors"
 	"testing"
 	"time"
 
@@ -26,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
+	storagelisters "k8s.io/client-go/listers/storage/v1"
 	core "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog"
@@ -100,6 +102,130 @@ func TestControllerSync(t *testing.T) {
 				return nil
 			},
 		},
+		{
+			// deleteClaim with a bound claim makes bound volume released with external deleter.
+			// delete the corresponding volume from apiserver, and report latency metric
+			"5-5 - delete claim and delete volume report metric",
+			volumesWithAnnotation(pvutil.AnnDynamicallyProvisioned, "gcr.io/vendor-csi",
+				newVolumeArray("volume5-6", "10Gi", "uid5-6", "claim5-6", v1.VolumeBound, v1.PersistentVolumeReclaimDelete, classExternal, pvutil.AnnBoundByController)),
+			novolumes,
+			claimWithAnnotation(pvutil.AnnStorageProvisioner, "gcr.io/vendor-csi",
+				newClaimArray("claim5-5", "uid5-5", "1Gi", "volume5-5", v1.ClaimBound, &classExternal, pvutil.AnnBoundByController, pvutil.AnnBindCompleted)),
+			noclaims,
+			noevents, noerrors,
+			// Custom test function that generates a delete claim event which should have been caught by
+			// "deleteClaim" to remove the claim from controller's cache, after that, a volume deleted
+			// event will be generated to trigger "deleteVolume" call for metric reporting
+			func(ctrl *PersistentVolumeController, reactor *pvtesting.VolumeReactor, test controllerTest) error {
+				test.initialVolumes[0].Annotations[pvutil.AnnDynamicallyProvisioned] = "gcr.io/vendor-csi"
+				obj := ctrl.claims.List()[0]
+				claim := obj.(*v1.PersistentVolumeClaim)
+				reactor.DeleteClaimEvent(claim)
+				for len(ctrl.claims.ListKeys()) > 0 {
+					time.Sleep(10 * time.Millisecond)
+				}
+				// claim has been removed from controller's cache, generate a volume deleted event
+				volume := ctrl.volumes.store.List()[0].(*v1.PersistentVolume)
+				reactor.DeleteVolumeEvent(volume)
+				return nil
+			},
+		},
+		{
+			// deleteClaim with a bound claim makes bound volume released with external deleter pending
+			// there should be an entry in operation timestamps cache in controller
+			"5-6 - delete claim and waiting for external volume deletion",
+			volumesWithAnnotation(pvutil.AnnDynamicallyProvisioned, "gcr.io/vendor-csi",
+				newVolumeArray("volume5-6", "10Gi", "uid5-6", "claim5-6", v1.VolumeBound, v1.PersistentVolumeReclaimDelete, classExternal, pvutil.AnnBoundByController)),
+			volumesWithAnnotation(pvutil.AnnDynamicallyProvisioned, "gcr.io/vendor-csi",
+				newVolumeArray("volume5-6", "10Gi", "uid5-6", "claim5-6", v1.VolumeReleased, v1.PersistentVolumeReclaimDelete, classExternal, pvutil.AnnBoundByController)),
+			claimWithAnnotation(pvutil.AnnStorageProvisioner, "gcr.io/vendor-csi",
+				newClaimArray("claim5-6", "uid5-6", "1Gi", "volume5-6", v1.ClaimBound, &classExternal, pvutil.AnnBoundByController, pvutil.AnnBindCompleted)),
+			noclaims,
+			noevents, noerrors,
+			// Custom test function that generates a delete claim event which should have been caught by
+			// "deleteClaim" to remove the claim from controller's cache and mark bound volume to be released
+			func(ctrl *PersistentVolumeController, reactor *pvtesting.VolumeReactor, test controllerTest) error {
+				// should have been provisioned by external provisioner
+				obj := ctrl.claims.List()[0]
+				claim := obj.(*v1.PersistentVolumeClaim)
+				reactor.DeleteClaimEvent(claim)
+				// wait until claim is cleared from cache, i.e., deleteClaim is called
+				for len(ctrl.claims.ListKeys()) > 0 {
+					time.Sleep(10 * time.Millisecond)
+				}
+				// make sure the operation timestamp cache is NOT empty
+				if !ctrl.operationTimestamps.Has("volume5-6") {
+					return errors.New("failed checking timestamp cache: should not be empty")
+				}
+				return nil
+			},
+		},
+		{
+			// deleteVolume event issued before deleteClaim, no metric should have been reported
+			// and no delete operation start timestamp should be inserted into controller.operationTimestamps cache
+			"5-7 - delete volume event makes claim lost, delete claim event will not report metric",
+			newVolumeArray("volume5-7", "10Gi", "uid5-7", "claim5-7", v1.VolumeBound, v1.PersistentVolumeReclaimDelete, classExternal, pvutil.AnnBoundByController, pvutil.AnnDynamicallyProvisioned),
+			novolumes,
+			claimWithAnnotation(pvutil.AnnStorageProvisioner, "gcr.io/vendor-csi",
+				newClaimArray("claim5-7", "uid5-7", "1Gi", "volume5-7", v1.ClaimBound, &classExternal, pvutil.AnnBoundByController, pvutil.AnnBindCompleted)),
+			noclaims,
+			[]string{"Warning ClaimLost"},
+			noerrors,
+			// Custom test function that generates a delete claim event which should have been caught by
+			// "deleteClaim" to remove the claim from controller's cache and mark bound volume to be released
+			func(ctrl *PersistentVolumeController, reactor *pvtesting.VolumeReactor, test controllerTest) error {
+				volume := ctrl.volumes.store.List()[0].(*v1.PersistentVolume)
+				reactor.DeleteVolumeEvent(volume)
+				for len(ctrl.volumes.store.ListKeys()) > 0 {
+					time.Sleep(10 * time.Millisecond)
+				}
+				// trying to remove the claim as well
+				obj := ctrl.claims.List()[0]
+				claim := obj.(*v1.PersistentVolumeClaim)
+				reactor.DeleteClaimEvent(claim)
+				// wait until claim is cleared from cache, i.e., deleteClaim is called
+				for len(ctrl.claims.ListKeys()) > 0 {
+					time.Sleep(10 * time.Millisecond)
+				}
+				// make sure operation timestamp cache is empty
+				if ctrl.operationTimestamps.Has("volume5-7") {
+					return errors.New("failed checking timestamp cache")
+				}
+				return nil
+			},
+		},
+		{
+			// delete a claim waiting for being bound cleans up provision(volume ref == "") entry from timestamp cache
+			"5-8 - delete claim cleans up operation timestamp cache for provision",
+			novolumes,
+			novolumes,
+			claimWithAnnotation(pvutil.AnnStorageProvisioner, "gcr.io/vendor-csi",
+				newClaimArray("claim5-8", "uid5-8", "1Gi", "", v1.ClaimPending, &classExternal)),
+			noclaims,
+			[]string{"Normal ExternalProvisioning"},
+			noerrors,
+			// Custom test function that generates a delete claim event which should have been caught by
+			// "deleteClaim" to remove the claim from controller's cache and mark bound volume to be released
+			func(ctrl *PersistentVolumeController, reactor *pvtesting.VolumeReactor, test controllerTest) error {
+				// wait until the provision timestamp has been inserted
+				for !ctrl.operationTimestamps.Has("default/claim5-8") {
+					time.Sleep(10 * time.Millisecond)
+				}
+				// delete the claim
+				obj := ctrl.claims.List()[0]
+				claim := obj.(*v1.PersistentVolumeClaim)
+				reactor.DeleteClaimEvent(claim)
+				// wait until claim is cleared from cache, i.e., deleteClaim is called
+				for len(ctrl.claims.ListKeys()) > 0 {
+					time.Sleep(10 * time.Millisecond)
+				}
+				// make sure operation timestamp cache is empty
+				if ctrl.operationTimestamps.Has("default/claim5-8") {
+					return errors.New("failed checking timestamp cache")
+				}
+				return nil
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -120,6 +246,18 @@ func TestControllerSync(t *testing.T) {
 			t.Fatalf("Test %q construct persistent volume failed: %v", test.name, err)
 		}
 
+		// Inject storage classes into controller via a custom lister for test [5-5]
+		storageClasses := []*storagev1.StorageClass{
+			makeStorageClass(classExternal, &modeImmediate),
+		}
+
+		storageClasses[0].Provisioner = "gcr.io/vendor-csi"
+		indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
+		for _, class := range storageClasses {
+			indexer.Add(class)
+		}
+		ctrl.classLister = storagelisters.NewStorageClassLister(indexer)
+
 		reactor := newVolumeReactor(client, ctrl, fakeVolumeWatch, fakeClaimWatch, test.errors)
 		for _, claim := range test.initialClaims {
 			reactor.AddClaim(claim)

pkg/volume/util/metrics.go

@@ -54,6 +54,15 @@ var storageOperationStatusMetric = prometheus.NewCounterVec(
 	[]string{"volume_plugin", "operation_name", "status"},
 )
 
+var storageOperationEndToEndLatencyMetric = prometheus.NewHistogramVec(
+	prometheus.HistogramOpts{
+		Name:    "volume_operation_total_seconds",
+		Help:    "Storage operation end to end duration in seconds",
+		Buckets: []float64{.1, .25, .5, 1, 2.5, 5, 10, 15, 25, 50, 120, 300, 600},
+	},
+	[]string{"volume_plugin", "operation_name"},
+)
+
 func init() {
 	registerMetrics()
 }
@@ -62,6 +71,7 @@ func registerMetrics() {
 	prometheus.MustRegister(storageOperationMetric)
 	prometheus.MustRegister(storageOperationErrorMetric)
 	prometheus.MustRegister(storageOperationStatusMetric)
+	prometheus.MustRegister(storageOperationEndToEndLatencyMetric)
 }
 
 // OperationCompleteHook returns a hook to call when an operation is completed
@@ -95,3 +105,9 @@ func GetFullQualifiedPluginNameForVolume(pluginName string, spec *volume.Spec) s
 	}
 	return pluginName
 }
+
+// RecordOperationLatencyMetric records the end to end latency for certain operation
+// into metric volume_operation_total_seconds
+func RecordOperationLatencyMetric(plugin, operationName string, secondsTaken float64) {
+	storageOperationEndToEndLatencyMetric.WithLabelValues(plugin, operationName).Observe(secondsTaken)
+}