volume_metrics.go

/*
Copyright 2017 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 storage

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/onsi/ginkgo/v2"
	"github.com/onsi/gomega"

	kubeletmetrics "github.com/sanposhiho/kubernetes/pkg/kubelet/metrics"
	"github.com/sanposhiho/kubernetes/test/e2e/framework"
	e2emetrics "github.com/sanposhiho/kubernetes/test/e2e/framework/metrics"
	e2epod "github.com/sanposhiho/kubernetes/test/e2e/framework/pod"
	e2epv "github.com/sanposhiho/kubernetes/test/e2e/framework/pv"
	e2eskipper "github.com/sanposhiho/kubernetes/test/e2e/framework/skipper"
	"github.com/sanposhiho/kubernetes/test/e2e/storage/testsuites"
	"github.com/sanposhiho/kubernetes/test/e2e/storage/utils"
	v1 "k8s.io/api/core/v1"
	storagev1 "k8s.io/api/storage/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	clientset "k8s.io/client-go/kubernetes"
	"k8s.io/component-base/metrics/testutil"
	"k8s.io/component-helpers/storage/ephemeral"
	admissionapi "k8s.io/pod-security-admission/api"
)

// This test needs to run in serial because other tests could interfere
// with metrics being tested here.
var _ = utils.SIGDescribe("[Serial] Volume metrics", func() {
	var (
		c              clientset.Interface
		ns             string
		pvc            *v1.PersistentVolumeClaim
		pvcBlock       *v1.PersistentVolumeClaim
		metricsGrabber *e2emetrics.Grabber
		invalidSc      *storagev1.StorageClass
		defaultScName  string
		err            error
	)
	f := framework.NewDefaultFramework("pv")
	f.NamespacePodSecurityLevel = admissionapi.LevelBaseline

	ginkgo.BeforeEach(func(ctx context.Context) {
		c = f.ClientSet
		ns = f.Namespace.Name
		var err error

		// The tests below make various assumptions about the cluster
		// and the underlying storage driver and therefore don't pass
		// with other kinds of clusters and drivers.
		e2eskipper.SkipUnlessProviderIs("gce", "gke", "aws")
		e2epv.SkipIfNoDefaultStorageClass(ctx, c)
		defaultScName, err = e2epv.GetDefaultStorageClassName(ctx, c)
		framework.ExpectNoError(err)

		test := testsuites.StorageClassTest{
			Name:      "default",
			Timeouts:  f.Timeouts,
			ClaimSize: "2Gi",
		}

		fsMode := v1.PersistentVolumeFilesystem
		pvc = e2epv.MakePersistentVolumeClaim(e2epv.PersistentVolumeClaimConfig{
			ClaimSize:  test.ClaimSize,
			VolumeMode: &fsMode,
		}, ns)

		// selected providers all support PersistentVolumeBlock
		blockMode := v1.PersistentVolumeBlock
		pvcBlock = e2epv.MakePersistentVolumeClaim(e2epv.PersistentVolumeClaimConfig{
			ClaimSize:  test.ClaimSize,
			VolumeMode: &blockMode,
		}, ns)

		metricsGrabber, err = e2emetrics.NewMetricsGrabber(ctx, c, nil, f.ClientConfig(), true, false, true, false, false, false)

		if err != nil {
			framework.Failf("Error creating metrics grabber : %v", err)
		}
	})

	ginkgo.AfterEach(func(ctx context.Context) {
		newPvc, err := c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Get(ctx, pvc.Name, metav1.GetOptions{})
		if err != nil {
			framework.Logf("Failed to get pvc %s/%s: %v", pvc.Namespace, pvc.Name, err)
		} else {
			e2epv.DeletePersistentVolumeClaim(ctx, c, newPvc.Name, newPvc.Namespace)
			if newPvc.Spec.VolumeName != "" {
				err = e2epv.WaitForPersistentVolumeDeleted(ctx, c, newPvc.Spec.VolumeName, 5*time.Second, 5*time.Minute)
				framework.ExpectNoError(err, "Persistent Volume %v not deleted by dynamic provisioner", newPvc.Spec.VolumeName)
			}
		}

		if invalidSc != nil {
			err := c.StorageV1().StorageClasses().Delete(ctx, invalidSc.Name, metav1.DeleteOptions{})
			framework.ExpectNoError(err, "Error deleting storageclass %v: %v", invalidSc.Name, err)
			invalidSc = nil
		}
	})

	provisioning := func(ctx context.Context, ephemeral bool) {
		if !metricsGrabber.HasControlPlanePods() {
			e2eskipper.Skipf("Environment does not support getting controller-manager metrics - skipping")
		}

		ginkgo.By("Getting plugin name")
		defaultClass, err := c.StorageV1().StorageClasses().Get(ctx, defaultScName, metav1.GetOptions{})
		framework.ExpectNoError(err, "Error getting default storageclass: %v", err)
		pluginName := defaultClass.Provisioner

		controllerMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)

		framework.ExpectNoError(err, "Error getting c-m metrics : %v", err)

		storageOpMetrics := getControllerStorageMetrics(controllerMetrics, pluginName)

		if !ephemeral {
			pvc, err = c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Create(ctx, pvc, metav1.CreateOptions{})
			framework.ExpectNoError(err)
			framework.ExpectNotEqual(pvc, nil)
		}

		pod := makePod(ns, pvc, ephemeral)
		pod, err = c.CoreV1().Pods(ns).Create(ctx, pod, metav1.CreateOptions{})
		framework.ExpectNoError(err)

		err = e2epod.WaitTimeoutForPodRunningInNamespace(ctx, c, pod.Name, pod.Namespace, f.Timeouts.PodStart)
		framework.ExpectNoError(err, "Error starting pod %s", pod.Name)

		framework.ExpectNoError(e2epod.DeletePodWithWait(ctx, c, pod))

		updatedStorageMetrics := waitForDetachAndGrabMetrics(ctx, storageOpMetrics, metricsGrabber, pluginName)

		framework.ExpectNotEqual(len(updatedStorageMetrics.latencyMetrics), 0, "Error fetching c-m updated storage metrics")
		framework.ExpectNotEqual(len(updatedStorageMetrics.statusMetrics), 0, "Error fetching c-m updated storage metrics")

		volumeOperations := []string{"volume_detach", "volume_attach"}

		for _, volumeOp := range volumeOperations {
			verifyMetricCount(storageOpMetrics, updatedStorageMetrics, volumeOp, false)
		}
	}

	provisioningError := func(ctx context.Context, ephemeral bool) {
		if !metricsGrabber.HasControlPlanePods() {
			e2eskipper.Skipf("Environment does not support getting controller-manager metrics - skipping")
		}

		ginkgo.By("Getting default storageclass")
		defaultClass, err := c.StorageV1().StorageClasses().Get(ctx, defaultScName, metav1.GetOptions{})
		framework.ExpectNoError(err, "Error getting default storageclass: %v", err)
		pluginName := defaultClass.Provisioner

		invalidSc = &storagev1.StorageClass{
			ObjectMeta: metav1.ObjectMeta{
				Name: fmt.Sprintf("fail-metrics-invalid-sc-%s", pvc.Namespace),
			},
			Provisioner: defaultClass.Provisioner,
			Parameters: map[string]string{
				"invalidparam": "invalidvalue",
			},
		}
		_, err = c.StorageV1().StorageClasses().Create(ctx, invalidSc, metav1.CreateOptions{})
		framework.ExpectNoError(err, "Error creating new storageclass: %v", err)

		pvc.Spec.StorageClassName = &invalidSc.Name
		if !ephemeral {
			pvc, err = c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Create(ctx, pvc, metav1.CreateOptions{})
			framework.ExpectNoError(err, "failed to create PVC %s/%s", pvc.Namespace, pvc.Name)
			framework.ExpectNotEqual(pvc, nil)
		}

		ginkgo.By("Creating a pod and expecting it to fail")
		pod := makePod(ns, pvc, ephemeral)
		pod, err = c.CoreV1().Pods(ns).Create(ctx, pod, metav1.CreateOptions{})
		framework.ExpectNoError(err, "failed to create Pod %s/%s", pod.Namespace, pod.Name)

		err = e2epod.WaitTimeoutForPodRunningInNamespace(ctx, c, pod.Name, pod.Namespace, f.Timeouts.PodStart)
		framework.ExpectError(err)

		framework.Logf("Deleting pod %q/%q", pod.Namespace, pod.Name)
		framework.ExpectNoError(e2epod.DeletePodWithWait(ctx, c, pod))

		ginkgo.By("Checking failure metrics")
		updatedControllerMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
		framework.ExpectNoError(err, "failed to get controller manager metrics")
		updatedStorageMetrics := getControllerStorageMetrics(updatedControllerMetrics, pluginName)

		framework.ExpectNotEqual(len(updatedStorageMetrics.statusMetrics), 0, "Error fetching c-m updated storage metrics")
	}

	filesystemMode := func(ctx context.Context, isEphemeral bool) {
		if !isEphemeral {
			pvc, err = c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Create(ctx, pvc, metav1.CreateOptions{})
			framework.ExpectNoError(err)
			framework.ExpectNotEqual(pvc, nil)
		}

		pod := makePod(ns, pvc, isEphemeral)
		pod, err = c.CoreV1().Pods(ns).Create(ctx, pod, metav1.CreateOptions{})
		framework.ExpectNoError(err)

		err = e2epod.WaitTimeoutForPodRunningInNamespace(ctx, c, pod.Name, pod.Namespace, f.Timeouts.PodStart)
		framework.ExpectNoError(err, "Error starting pod ", pod.Name)

		pod, err = c.CoreV1().Pods(ns).Get(ctx, pod.Name, metav1.GetOptions{})
		framework.ExpectNoError(err)

		pvcName := pvc.Name
		if isEphemeral {
			pvcName = ephemeral.VolumeClaimName(pod, &pod.Spec.Volumes[0])
		}
		pvcNamespace := pod.Namespace

		// Verify volume stat metrics were collected for the referenced PVC
		volumeStatKeys := []string{
			kubeletmetrics.VolumeStatsUsedBytesKey,
			kubeletmetrics.VolumeStatsCapacityBytesKey,
			kubeletmetrics.VolumeStatsAvailableBytesKey,
			kubeletmetrics.VolumeStatsInodesKey,
			kubeletmetrics.VolumeStatsInodesFreeKey,
			kubeletmetrics.VolumeStatsInodesUsedKey,
		}
		key := volumeStatKeys[0]
		kubeletKeyName := fmt.Sprintf("%s_%s", kubeletmetrics.KubeletSubsystem, key)
		// Poll kubelet metrics waiting for the volume to be picked up
		// by the volume stats collector
		var kubeMetrics e2emetrics.KubeletMetrics
		waitErr := wait.PollWithContext(ctx, 30*time.Second, 5*time.Minute, func(ctx context.Context) (bool, error) {
			framework.Logf("Grabbing Kubelet metrics")
			// Grab kubelet metrics from the node the pod was scheduled on
			var err error
			kubeMetrics, err = metricsGrabber.GrabFromKubelet(ctx, pod.Spec.NodeName)
			if err != nil {
				framework.Logf("Error fetching kubelet metrics")
				return false, err
			}
			if !findVolumeStatMetric(kubeletKeyName, pvcNamespace, pvcName, kubeMetrics) {
				return false, nil
			}
			return true, nil
		})
		framework.ExpectNoError(waitErr, "Unable to find metric %s for PVC %s/%s", kubeletKeyName, pvcNamespace, pvcName)

		for _, key := range volumeStatKeys {
			kubeletKeyName := fmt.Sprintf("%s_%s", kubeletmetrics.KubeletSubsystem, key)
			found := findVolumeStatMetric(kubeletKeyName, pvcNamespace, pvcName, kubeMetrics)
			if !found {
				framework.Failf("PVC %s, Namespace %s not found for %s", pvcName, pvcNamespace, kubeletKeyName)
			}
		}

		framework.Logf("Deleting pod %q/%q", pod.Namespace, pod.Name)
		framework.ExpectNoError(e2epod.DeletePodWithWait(ctx, c, pod))
	}

	blockmode := func(ctx context.Context, isEphemeral bool) {
		if !isEphemeral {
			pvcBlock, err = c.CoreV1().PersistentVolumeClaims(pvcBlock.Namespace).Create(ctx, pvcBlock, metav1.CreateOptions{})
			framework.ExpectNoError(err)
			framework.ExpectNotEqual(pvcBlock, nil)
		}

		pod := makePod(ns, pvcBlock, isEphemeral)
		pod.Spec.Containers[0].VolumeDevices = []v1.VolumeDevice{{
			Name:       pod.Spec.Volumes[0].Name,
			DevicePath: "/mnt/" + pvcBlock.Name,
		}}
		pod.Spec.Containers[0].VolumeMounts = nil
		pod, err = c.CoreV1().Pods(ns).Create(ctx, pod, metav1.CreateOptions{})
		framework.ExpectNoError(err)

		err = e2epod.WaitTimeoutForPodRunningInNamespace(ctx, c, pod.Name, pod.Namespace, f.Timeouts.PodStart)
		framework.ExpectNoError(err, "Error starting pod ", pod.Name)

		pod, err = c.CoreV1().Pods(ns).Get(ctx, pod.Name, metav1.GetOptions{})
		framework.ExpectNoError(err)

		// Verify volume stat metrics were collected for the referenced PVC
		volumeStatKeys := []string{
			// BlockMode PVCs only support capacity (for now)
			kubeletmetrics.VolumeStatsCapacityBytesKey,
		}
		key := volumeStatKeys[0]
		kubeletKeyName := fmt.Sprintf("%s_%s", kubeletmetrics.KubeletSubsystem, key)
		pvcName := pvcBlock.Name
		pvcNamespace := pvcBlock.Namespace
		if isEphemeral {
			pvcName = ephemeral.VolumeClaimName(pod, &pod.Spec.Volumes[0])
			pvcNamespace = pod.Namespace
		}
		// Poll kubelet metrics waiting for the volume to be picked up
		// by the volume stats collector
		var kubeMetrics e2emetrics.KubeletMetrics
		waitErr := wait.Poll(30*time.Second, 5*time.Minute, func() (bool, error) {
			framework.Logf("Grabbing Kubelet metrics")
			// Grab kubelet metrics from the node the pod was scheduled on
			var err error
			kubeMetrics, err = metricsGrabber.GrabFromKubelet(ctx, pod.Spec.NodeName)
			if err != nil {
				framework.Logf("Error fetching kubelet metrics")
				return false, err
			}
			if !findVolumeStatMetric(kubeletKeyName, pvcNamespace, pvcName, kubeMetrics) {
				return false, nil
			}
			return true, nil
		})
		framework.ExpectNoError(waitErr, "Unable to find metric %s for PVC %s/%s", kubeletKeyName, pvcNamespace, pvcName)

		for _, key := range volumeStatKeys {
			kubeletKeyName := fmt.Sprintf("%s_%s", kubeletmetrics.KubeletSubsystem, key)
			found := findVolumeStatMetric(kubeletKeyName, pvcNamespace, pvcName, kubeMetrics)
			if !found {
				framework.Failf("PVC %s, Namespace %s not found for %s", pvcName, pvcNamespace, kubeletKeyName)
			}
		}

		framework.Logf("Deleting pod %q/%q", pod.Namespace, pod.Name)
		framework.ExpectNoError(e2epod.DeletePodWithWait(ctx, c, pod))
	}

	totalTime := func(ctx context.Context, isEphemeral bool) {
		if !isEphemeral {
			pvc, err = c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Create(ctx, pvc, metav1.CreateOptions{})
			framework.ExpectNoError(err)
			framework.ExpectNotEqual(pvc, nil)
		}

		pod := makePod(ns, pvc, isEphemeral)
		pod, err = c.CoreV1().Pods(ns).Create(ctx, pod, metav1.CreateOptions{})
		framework.ExpectNoError(err)

		err = e2epod.WaitTimeoutForPodRunningInNamespace(ctx, c, pod.Name, pod.Namespace, f.Timeouts.PodStart)
		framework.ExpectNoError(err, "Error starting pod ", pod.Name)

		pod, err = c.CoreV1().Pods(ns).Get(ctx, pod.Name, metav1.GetOptions{})
		framework.ExpectNoError(err)

		controllerMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
		if err != nil {
			e2eskipper.Skipf("Could not get controller-manager metrics - skipping")
		}

		metricKey := "volume_operation_total_seconds_count"
		dimensions := []string{"operation_name", "plugin_name"}
		err = testutil.ValidateMetrics(testutil.Metrics(controllerMetrics), metricKey, dimensions...)
		framework.ExpectNoError(err, "Invalid metric in P/V Controller metrics: %q", metricKey)

		framework.Logf("Deleting pod %q/%q", pod.Namespace, pod.Name)
		framework.ExpectNoError(e2epod.DeletePodWithWait(ctx, c, pod))
	}

	volumeManager := func(ctx context.Context, isEphemeral bool) {
		if !isEphemeral {
			pvc, err = c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Create(ctx, pvc, metav1.CreateOptions{})
			framework.ExpectNoError(err)
			framework.ExpectNotEqual(pvc, nil)
		}

		pod := makePod(ns, pvc, isEphemeral)
		pod, err = c.CoreV1().Pods(ns).Create(ctx, pod, metav1.CreateOptions{})
		framework.ExpectNoError(err)

		err = e2epod.WaitTimeoutForPodRunningInNamespace(ctx, c, pod.Name, pod.Namespace, f.Timeouts.PodStart)
		framework.ExpectNoError(err, "Error starting pod ", pod.Name)

		pod, err = c.CoreV1().Pods(ns).Get(ctx, pod.Name, metav1.GetOptions{})
		framework.ExpectNoError(err)

		kubeMetrics, err := metricsGrabber.GrabFromKubelet(ctx, pod.Spec.NodeName)
		framework.ExpectNoError(err)

		// Metrics should have dimensions plugin_name and state available
		totalVolumesKey := "volume_manager_total_volumes"
		dimensions := []string{"state", "plugin_name"}
		err = testutil.ValidateMetrics(testutil.Metrics(kubeMetrics), totalVolumesKey, dimensions...)
		framework.ExpectNoError(err, "Invalid metric in Volume Manager metrics: %q", totalVolumesKey)

		framework.Logf("Deleting pod %q/%q", pod.Namespace, pod.Name)
		framework.ExpectNoError(e2epod.DeletePodWithWait(ctx, c, pod))
	}

	adController := func(ctx context.Context, isEphemeral bool) {
		if !isEphemeral {
			pvc, err = c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Create(ctx, pvc, metav1.CreateOptions{})
			framework.ExpectNoError(err)
			framework.ExpectNotEqual(pvc, nil)
		}

		pod := makePod(ns, pvc, isEphemeral)

		// Get metrics
		controllerMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
		if err != nil {
			e2eskipper.Skipf("Could not get controller-manager metrics - skipping")
		}

		// Create pod
		pod, err = c.CoreV1().Pods(ns).Create(ctx, pod, metav1.CreateOptions{})
		framework.ExpectNoError(err)
		err = e2epod.WaitTimeoutForPodRunningInNamespace(ctx, c, pod.Name, pod.Namespace, f.Timeouts.PodStart)
		framework.ExpectNoError(err, "Error starting pod ", pod.Name)
		pod, err = c.CoreV1().Pods(ns).Get(ctx, pod.Name, metav1.GetOptions{})
		framework.ExpectNoError(err)

		// Get updated metrics
		updatedControllerMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
		if err != nil {
			e2eskipper.Skipf("Could not get controller-manager metrics - skipping")
		}

		// Forced detach metric should be present
		forceDetachKey := "attachdetach_controller_forced_detaches"
		_, ok := updatedControllerMetrics[forceDetachKey]
		if !ok {
			framework.Failf("Key %q not found in A/D Controller metrics", forceDetachKey)
		}

		// Wait and validate
		totalVolumesKey := "attachdetach_controller_total_volumes"
		states := []string{"actual_state_of_world", "desired_state_of_world"}
		dimensions := []string{"state", "plugin_name"}
		waitForADControllerStatesMetrics(ctx, metricsGrabber, totalVolumesKey, dimensions, states)

		// Total number of volumes in both ActualStateofWorld and DesiredStateOfWorld
		// states should be higher or equal than it used to be
		oldStates := getStatesMetrics(totalVolumesKey, testutil.Metrics(controllerMetrics))
		updatedStates := getStatesMetrics(totalVolumesKey, testutil.Metrics(updatedControllerMetrics))
		for _, stateName := range states {
			if _, ok := oldStates[stateName]; !ok {
				continue
			}
			for pluginName, numVolumes := range updatedStates[stateName] {
				oldNumVolumes := oldStates[stateName][pluginName]
				gomega.Expect(numVolumes).To(gomega.BeNumerically(">=", oldNumVolumes),
					"Wrong number of volumes in state %q, plugin %q: wanted >=%d, got %d",
					stateName, pluginName, oldNumVolumes, numVolumes)
			}
		}

		framework.Logf("Deleting pod %q/%q", pod.Namespace, pod.Name)
		framework.ExpectNoError(e2epod.DeletePodWithWait(ctx, c, pod))
	}

	testAll := func(isEphemeral bool) {
		ginkgo.It("should create prometheus metrics for volume provisioning and attach/detach", func(ctx context.Context) {
			provisioning(ctx, isEphemeral)
		})
		// TODO(mauriciopoppe): after CSIMigration is turned on we're no longer reporting
		// the volume_provision metric (removed in #106609), issue to investigate the bug #106773
		ginkgo.It("should create prometheus metrics for volume provisioning errors [Slow]", func(ctx context.Context) {
			provisioningError(ctx, isEphemeral)
		})
		ginkgo.It("should create volume metrics with the correct FilesystemMode PVC ref", func(ctx context.Context) {
			filesystemMode(ctx, isEphemeral)
		})
		ginkgo.It("should create volume metrics with the correct BlockMode PVC ref", func(ctx context.Context) {
			blockmode(ctx, isEphemeral)
		})
		ginkgo.It("should create metrics for total time taken in volume operations in P/V Controller", func(ctx context.Context) {
			totalTime(ctx, isEphemeral)
		})
		ginkgo.It("should create volume metrics in Volume Manager", func(ctx context.Context) {
			volumeManager(ctx, isEphemeral)
		})
		ginkgo.It("should create metrics for total number of volumes in A/D Controller", func(ctx context.Context) {
			adController(ctx, isEphemeral)
		})
	}

	ginkgo.Context("PVC", func() {
		testAll(false)
	})

	ginkgo.Context("Ephemeral", func() {
		testAll(true)
	})

	// Test for pv controller metrics, concretely: bound/unbound pv/pvc count.
	ginkgo.Describe("PVController", func() {
		const (
			classKey      = "storage_class"
			namespaceKey  = "namespace"
			pluginNameKey = "plugin_name"
			volumeModeKey = "volume_mode"

			totalPVKey    = "pv_collector_total_pv_count"
			boundPVKey    = "pv_collector_bound_pv_count"
			unboundPVKey  = "pv_collector_unbound_pv_count"
			boundPVCKey   = "pv_collector_bound_pvc_count"
			unboundPVCKey = "pv_collector_unbound_pvc_count"
		)

		var (
			pv  *v1.PersistentVolume
			pvc *v1.PersistentVolumeClaim

			className = "bound-unbound-count-test-sc"
			pvConfig  = e2epv.PersistentVolumeConfig{
				PVSource: v1.PersistentVolumeSource{
					HostPath: &v1.HostPathVolumeSource{Path: "/data"},
				},
				NamePrefix:       "pv-test-",
				StorageClassName: className,
			}
			pvcConfig = e2epv.PersistentVolumeClaimConfig{StorageClassName: &className}

			e2emetrics = []struct {
				name      string
				dimension string
			}{
				{boundPVKey, classKey},
				{unboundPVKey, classKey},
				{boundPVCKey, namespaceKey},
				{unboundPVCKey, namespaceKey},
			}

			// Original metric values before we create any PV/PVCs. The length should be 4,
			// and the elements should be bound pv count, unbound pv count, bound pvc count,
			// unbound pvc count in turn.
			// We use these values to calculate relative increment of each test.
			originMetricValues []map[string]int64
		)

		// validator used to validate each metric's values, the length of metricValues
		// should be 4, and the elements should be bound pv count, unbound pv count, bound
		// pvc count, unbound pvc count in turn.
		validator := func(ctx context.Context, metricValues []map[string]int64) {
			framework.ExpectEqual(len(metricValues), 4, "Wrong metric size: %d", len(metricValues))

			controllerMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
			framework.ExpectNoError(err, "Error getting c-m metricValues: %v", err)

			for i, metric := range e2emetrics {
				expectValues := metricValues[i]
				if expectValues == nil {
					expectValues = make(map[string]int64)
				}
				// We using relative increment value instead of absolute value to reduce unexpected flakes.
				// Concretely, we expect the difference of the updated values and original values for each
				// test suit are equal to expectValues.
				actualValues := calculateRelativeValues(originMetricValues[i],
					testutil.GetMetricValuesForLabel(testutil.Metrics(controllerMetrics), metric.name, metric.dimension))
				framework.ExpectEqual(actualValues, expectValues, "Wrong pv controller metric %s(%s): wanted %v, got %v",
					metric.name, metric.dimension, expectValues, actualValues)
			}
		}

		ginkgo.BeforeEach(func(ctx context.Context) {
			if !metricsGrabber.HasControlPlanePods() {
				e2eskipper.Skipf("Environment does not support getting controller-manager metrics - skipping")
			}

			pv = e2epv.MakePersistentVolume(pvConfig)
			pvc = e2epv.MakePersistentVolumeClaim(pvcConfig, ns)

			// Initializes all original metric values.
			controllerMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
			framework.ExpectNoError(err, "Error getting c-m metricValues: %v", err)
			for _, metric := range e2emetrics {
				originMetricValues = append(originMetricValues,
					testutil.GetMetricValuesForLabel(testutil.Metrics(controllerMetrics), metric.name, metric.dimension))
			}
		})

		ginkgo.AfterEach(func(ctx context.Context) {
			if err := e2epv.DeletePersistentVolume(ctx, c, pv.Name); err != nil {
				framework.Failf("Error deleting pv: %v", err)
			}
			if err := e2epv.DeletePersistentVolumeClaim(ctx, c, pvc.Name, pvc.Namespace); err != nil {
				framework.Failf("Error deleting pvc: %v", err)
			}

			// Clear original metric values.
			originMetricValues = nil
		})

		ginkgo.It("should create none metrics for pvc controller before creating any PV or PVC", func(ctx context.Context) {
			validator(ctx, []map[string]int64{nil, nil, nil, nil})
		})

		ginkgo.It("should create unbound pv count metrics for pvc controller after creating pv only",
			func(ctx context.Context) {
				var err error
				pv, err = e2epv.CreatePV(ctx, c, f.Timeouts, pv)
				framework.ExpectNoError(err, "Error creating pv: %v", err)
				waitForPVControllerSync(ctx, metricsGrabber, unboundPVKey, classKey)
				validator(ctx, []map[string]int64{nil, {className: 1}, nil, nil})
			})

		ginkgo.It("should create unbound pvc count metrics for pvc controller after creating pvc only",
			func(ctx context.Context) {
				var err error
				pvc, err = e2epv.CreatePVC(ctx, c, ns, pvc)
				framework.ExpectNoError(err, "Error creating pvc: %v", err)
				waitForPVControllerSync(ctx, metricsGrabber, unboundPVCKey, namespaceKey)
				validator(ctx, []map[string]int64{nil, nil, nil, {ns: 1}})
			})

		ginkgo.It("should create bound pv/pvc count metrics for pvc controller after creating both pv and pvc",
			func(ctx context.Context) {
				var err error
				pv, pvc, err = e2epv.CreatePVPVC(ctx, c, f.Timeouts, pvConfig, pvcConfig, ns, true)
				framework.ExpectNoError(err, "Error creating pv pvc: %v", err)
				waitForPVControllerSync(ctx, metricsGrabber, boundPVKey, classKey)
				waitForPVControllerSync(ctx, metricsGrabber, boundPVCKey, namespaceKey)
				validator(ctx, []map[string]int64{{className: 1}, nil, {ns: 1}, nil})

			})
		ginkgo.It("should create total pv count metrics for with plugin and volume mode labels after creating pv",
			func(ctx context.Context) {
				var err error
				dimensions := []string{pluginNameKey, volumeModeKey}
				pv, err = e2epv.CreatePV(ctx, c, f.Timeouts, pv)
				framework.ExpectNoError(err, "Error creating pv: %v", err)
				waitForPVControllerSync(ctx, metricsGrabber, totalPVKey, pluginNameKey)
				controllerMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
				framework.ExpectNoError(err, "Error getting c-m metricValues: %v", err)
				err = testutil.ValidateMetrics(testutil.Metrics(controllerMetrics), totalPVKey, dimensions...)
				framework.ExpectNoError(err, "Invalid metric in Controller Manager metrics: %q", totalPVKey)
			})
	})
})

type storageControllerMetrics struct {
	latencyMetrics map[string]int64
	statusMetrics  map[string]statusMetricCounts
}

type statusMetricCounts struct {
	successCount int64
	failCount    int64
	otherCount   int64
}

func newStorageControllerMetrics() *storageControllerMetrics {
	return &storageControllerMetrics{
		latencyMetrics: make(map[string]int64),
		statusMetrics:  make(map[string]statusMetricCounts),
	}
}

func waitForDetachAndGrabMetrics(ctx context.Context, oldMetrics *storageControllerMetrics, metricsGrabber *e2emetrics.Grabber, pluginName string) *storageControllerMetrics {
	backoff := wait.Backoff{
		Duration: 10 * time.Second,
		Factor:   1.2,
		Steps:    21,
	}

	updatedStorageMetrics := newStorageControllerMetrics()
	oldDetachCount, ok := oldMetrics.latencyMetrics["volume_detach"]
	if !ok {
		oldDetachCount = 0
	}

	verifyMetricFunc := func(ctx context.Context) (bool, error) {
		updatedMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)

		if err != nil {
			framework.Logf("Error fetching controller-manager metrics")
			return false, err
		}

		updatedStorageMetrics = getControllerStorageMetrics(updatedMetrics, pluginName)
		newDetachCount, ok := updatedStorageMetrics.latencyMetrics["volume_detach"]

		// if detach metrics are not yet there, we need to retry
		if !ok {
			return false, nil
		}

		// if old Detach count is more or equal to new detach count, that means detach
		// event has not been observed yet.
		if oldDetachCount >= newDetachCount {
			return false, nil
		}

		return true, nil
	}

	waitErr := wait.ExponentialBackoffWithContext(ctx, backoff, verifyMetricFunc)
	framework.ExpectNoError(waitErr, "Unable to get updated metrics for plugin %s", pluginName)
	return updatedStorageMetrics
}

func verifyMetricCount(oldMetrics, newMetrics *storageControllerMetrics, metricName string, expectFailure bool) {
	oldLatencyCount, ok := oldMetrics.latencyMetrics[metricName]
	// if metric does not exist in oldMap, it probably hasn't been emitted yet.
	if !ok {
		oldLatencyCount = 0
	}

	oldStatusCount := int64(0)
	if oldStatusCounts, ok := oldMetrics.statusMetrics[metricName]; ok {
		if expectFailure {
			oldStatusCount = oldStatusCounts.failCount
		} else {
			oldStatusCount = oldStatusCounts.successCount
		}
	}

	newLatencyCount, ok := newMetrics.latencyMetrics[metricName]
	if !expectFailure {
		if !ok {
			framework.Failf("Error getting updated latency metrics for %s", metricName)
		}
	}
	newStatusCounts, ok := newMetrics.statusMetrics[metricName]
	if !ok {
		framework.Failf("Error getting updated status metrics for %s", metricName)
	}

	newStatusCount := int64(0)
	if expectFailure {
		newStatusCount = newStatusCounts.failCount
	} else {
		newStatusCount = newStatusCounts.successCount
	}

	// It appears that in a busy cluster some spurious detaches are unavoidable
	// even if the test is run serially.  We really just verify if new count
	// is greater than old count
	if !expectFailure {
		gomega.Expect(newLatencyCount).To(gomega.BeNumerically(">", oldLatencyCount), "New latency count %d should be more than old count %d for action %s", newLatencyCount, oldLatencyCount, metricName)
	}
	gomega.Expect(newStatusCount).To(gomega.BeNumerically(">", oldStatusCount), "New status count %d should be more than old count %d for action %s", newStatusCount, oldStatusCount, metricName)
}

func getControllerStorageMetrics(ms e2emetrics.ControllerManagerMetrics, pluginName string) *storageControllerMetrics {
	result := newStorageControllerMetrics()

	for method, samples := range ms {
		switch method {
		// from the base metric name "storage_operation_duration_seconds"
		case "storage_operation_duration_seconds_count":
			for _, sample := range samples {
				count := int64(sample.Value)
				operation := string(sample.Metric["operation_name"])
				// if the volumes were provisioned with a CSI Driver
				// the metric operation name will be prefixed with
				// "kubernetes.io/csi:"
				metricPluginName := string(sample.Metric["volume_plugin"])
				status := string(sample.Metric["status"])
				if strings.Index(metricPluginName, pluginName) < 0 {
					// the metric volume plugin field doesn't match
					// the default storageClass.Provisioner field
					continue
				}

				statusCounts := result.statusMetrics[operation]
				switch status {
				case "success":
					statusCounts.successCount = count
				case "fail-unknown":
					statusCounts.failCount = count
				default:
					statusCounts.otherCount = count
				}
				result.statusMetrics[operation] = statusCounts
				result.latencyMetrics[operation] = count
			}
		}
	}
	return result
}

// Finds the sample in the specified metric from `KubeletMetrics` tagged with
// the specified namespace and pvc name
func findVolumeStatMetric(metricKeyName string, namespace string, pvcName string, kubeletMetrics e2emetrics.KubeletMetrics) bool {
	found := false
	errCount := 0
	framework.Logf("Looking for sample in metric `%s` tagged with namespace `%s`, PVC `%s`", metricKeyName, namespace, pvcName)
	if samples, ok := kubeletMetrics[metricKeyName]; ok {
		for _, sample := range samples {
			framework.Logf("Found sample %s", sample.String())
			samplePVC, ok := sample.Metric["persistentvolumeclaim"]
			if !ok {
				framework.Logf("Error getting pvc for metric %s, sample %s", metricKeyName, sample.String())
				errCount++
			}
			sampleNS, ok := sample.Metric["namespace"]
			if !ok {
				framework.Logf("Error getting namespace for metric %s, sample %s", metricKeyName, sample.String())
				errCount++
			}

			if string(samplePVC) == pvcName && string(sampleNS) == namespace {
				found = true
				break
			}
		}
	}
	framework.ExpectEqual(errCount, 0, "Found invalid samples")
	return found
}

// Wait for the count of a pv controller's metric specified by metricName and dimension bigger than zero.
func waitForPVControllerSync(ctx context.Context, metricsGrabber *e2emetrics.Grabber, metricName, dimension string) {
	backoff := wait.Backoff{
		Duration: 10 * time.Second,
		Factor:   1.2,
		Steps:    21,
	}
	verifyMetricFunc := func(ctx context.Context) (bool, error) {
		updatedMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
		if err != nil {
			framework.Logf("Error fetching controller-manager metrics")
			return false, err
		}
		return len(testutil.GetMetricValuesForLabel(testutil.Metrics(updatedMetrics), metricName, dimension)) > 0, nil
	}
	waitErr := wait.ExponentialBackoffWithContext(ctx, backoff, verifyMetricFunc)
	framework.ExpectNoError(waitErr, "Unable to get pv controller metrics")
}

func calculateRelativeValues(originValues, updatedValues map[string]int64) map[string]int64 {
	relativeValues := make(map[string]int64)
	for key, value := range updatedValues {
		relativeValue := value - originValues[key]
		if relativeValue != 0 {
			relativeValues[key] = relativeValue
		}
	}
	for key, value := range originValues {
		if _, exist := updatedValues[key]; !exist && value > 0 {
			relativeValues[key] = -value
		}
	}
	return relativeValues
}

func getStatesMetrics(metricKey string, givenMetrics testutil.Metrics) map[string]map[string]int64 {
	states := make(map[string]map[string]int64)
	for _, sample := range givenMetrics[metricKey] {
		framework.Logf("Found sample %q", sample.String())
		state := string(sample.Metric["state"])
		pluginName := string(sample.Metric["plugin_name"])
		states[state] = map[string]int64{pluginName: int64(sample.Value)}
	}
	return states
}

func waitForADControllerStatesMetrics(ctx context.Context, metricsGrabber *e2emetrics.Grabber, metricName string, dimensions []string, stateNames []string) {
	backoff := wait.Backoff{
		Duration: 10 * time.Second,
		Factor:   1.2,
		Steps:    21,
	}
	verifyMetricFunc := func(ctx context.Context) (bool, error) {
		updatedMetrics, err := metricsGrabber.GrabFromControllerManager(ctx)
		if err != nil {
			e2eskipper.Skipf("Could not get controller-manager metrics - skipping")
			return false, err
		}
		err = testutil.ValidateMetrics(testutil.Metrics(updatedMetrics), metricName, dimensions...)
		if err != nil {
			return false, fmt.Errorf("could not get valid metrics: %v ", err)
		}
		states := getStatesMetrics(metricName, testutil.Metrics(updatedMetrics))
		for _, name := range stateNames {
			if _, ok := states[name]; !ok {
				return false, fmt.Errorf("could not get state %q from A/D Controller metrics", name)
			}
		}
		return true, nil
	}
	waitErr := wait.ExponentialBackoffWithContext(ctx, backoff, verifyMetricFunc)
	framework.ExpectNoError(waitErr, "Unable to get A/D controller metrics")
}

// makePod creates a pod which either references the PVC or creates it via a
// generic ephemeral volume claim template.
func makePod(ns string, pvc *v1.PersistentVolumeClaim, isEphemeral bool) *v1.Pod {
	claims := []*v1.PersistentVolumeClaim{pvc}
	pod := e2epod.MakePod(ns, nil, claims, false, "")
	if isEphemeral {
		volSrc := pod.Spec.Volumes[0]
		volSrc.PersistentVolumeClaim = nil
		volSrc.Ephemeral = &v1.EphemeralVolumeSource{
			VolumeClaimTemplate: &v1.PersistentVolumeClaimTemplate{
				Spec: pvc.Spec,
			},
		}
		pod.Spec.Volumes[0] = volSrc
	}
	return pod
}