Merge pull request #409 from ggriffiths/snapshot_controller_metrics

Add snapshot controller metrics

Author: k8s-ci-robot

cmd/snapshot-controller/main.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"sync"
 	"time"
 
 	"k8s.io/client-go/kubernetes"
@@ -32,6 +33,7 @@ import (
 
 	"github.com/kubernetes-csi/csi-lib-utils/leaderelection"
 	controller "github.com/kubernetes-csi/external-snapshotter/v3/pkg/common-controller"
+	"github.com/kubernetes-csi/external-snapshotter/v3/pkg/metrics"
 
 	clientset "github.com/kubernetes-csi/external-snapshotter/client/v3/clientset/versioned"
 	snapshotscheme "github.com/kubernetes-csi/external-snapshotter/client/v3/clientset/versioned/scheme"
@@ -48,6 +50,9 @@ var (
 
 	leaderElection          = flag.Bool("leader-election", false, "Enables leader election.")
 	leaderElectionNamespace = flag.String("leader-election-namespace", "", "The namespace where the leader election resource exists. Defaults to the pod namespace if not set.")
+
+	httpEndpoint = flag.String("http-endpoint", "", "The TCP network address where the HTTP server for diagnostics, including metrics, will listen (example: :8080). The default is empty string, which means the server is disabled.")
+	metricsPath  = flag.String("metrics-path", "/metrics", "The HTTP path where prometheus metrics will be exposed. Default is `/metrics`.")
 )
 
 var (
@@ -87,6 +92,28 @@ func main() {
 	factory := informers.NewSharedInformerFactory(snapClient, *resyncPeriod)
 	coreFactory := coreinformers.NewSharedInformerFactory(kubeClient, *resyncPeriod)
 
+	// Create and register metrics manager
+	metricsManager := metrics.NewMetricsManager()
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	if *httpEndpoint != "" {
+		srv, err := metricsManager.StartMetricsEndpoint(*metricsPath, *httpEndpoint, promklog{}, wg)
+		if err != nil {
+			klog.Errorf("Failed to start metrics server: %s", err.Error())
+			os.Exit(1)
+		}
+		defer func() {
+			err := srv.Shutdown(context.Background())
+			if err != nil {
+				klog.Errorf("Failed to shutdown metrics server: %s", err.Error())
+			}
+
+			klog.Infof("Metrics server successfully shutdown")
+			wg.Done()
+		}()
+		klog.Infof("Metrics server successfully started on %s, %s", *httpEndpoint, *metricsPath)
+	}
+
 	// Add Snapshot types to the default Kubernetes so events can be logged for them
 	snapshotscheme.AddToScheme(scheme.Scheme)
 
@@ -99,6 +126,7 @@ func main() {
 		factory.Snapshot().V1beta1().VolumeSnapshotContents(),
 		factory.Snapshot().V1beta1().VolumeSnapshotClasses(),
 		coreFactory.Core().V1().PersistentVolumeClaims(),
+		metricsManager,
 		*resyncPeriod,
 	)
 
@@ -142,3 +170,9 @@ func buildConfig(kubeconfig string) (*rest.Config, error) {
 	}
 	return rest.InClusterConfig()
 }
+
+type promklog struct{}
+
+func (pl promklog) Println(v ...interface{}) {
+	klog.Error(v...)
+}

pkg/common-controller/framework_test.go

@@ -34,6 +34,7 @@ import (
 	snapshotscheme "github.com/kubernetes-csi/external-snapshotter/client/v3/clientset/versioned/scheme"
 	informers "github.com/kubernetes-csi/external-snapshotter/client/v3/informers/externalversions"
 	storagelisters "github.com/kubernetes-csi/external-snapshotter/client/v3/listers/volumesnapshot/v1beta1"
+	"github.com/kubernetes-csi/external-snapshotter/v3/pkg/metrics"
 	"github.com/kubernetes-csi/external-snapshotter/v3/pkg/utils"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -734,6 +735,10 @@ func newTestController(kubeClient kubernetes.Interface, clientset clientset.Inte
 	}
 
 	coreFactory := coreinformers.NewSharedInformerFactory(kubeClient, utils.NoResyncPeriodFunc())
+	metricsManager := metrics.NewMetricsManager()
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	metricsManager.StartMetricsEndpoint("/metrics", "localhost:0", nil, wg)
 
 	ctrl := NewCSISnapshotCommonController(
 		clientset,
@@ -742,6 +747,7 @@ func newTestController(kubeClient kubernetes.Interface, clientset clientset.Inte
 		informerFactory.Snapshot().V1beta1().VolumeSnapshotContents(),
 		informerFactory.Snapshot().V1beta1().VolumeSnapshotClasses(),
 		coreFactory.Core().V1().PersistentVolumeClaims(),
+		metricsManager,
 		60*time.Second,
 	)
 

pkg/common-controller/snapshot_controller.go

@@ -32,6 +32,7 @@ import (
 	klog "k8s.io/klog/v2"
 
 	crdv1 "github.com/kubernetes-csi/external-snapshotter/client/v3/apis/volumesnapshot/v1beta1"
+	"github.com/kubernetes-csi/external-snapshotter/v3/pkg/metrics"
 	"github.com/kubernetes-csi/external-snapshotter/v3/pkg/utils"
 )
 
@@ -235,6 +236,20 @@ func (ctrl *csiSnapshotCommonController) syncSnapshot(snapshot *crdv1.VolumeSnap
 // 2. Call checkandRemoveSnapshotFinalizersAndCheckandDeleteContent() with information obtained from step 1. This function name is very long but the name suggests what it does. It determines whether to remove finalizers on snapshot and whether to delete content.
 func (ctrl *csiSnapshotCommonController) processSnapshotWithDeletionTimestamp(snapshot *crdv1.VolumeSnapshot) error {
 	klog.V(5).Infof("processSnapshotWithDeletionTimestamp VolumeSnapshot[%s]: %s", utils.SnapshotKey(snapshot), utils.GetSnapshotStatusForLogging(snapshot))
+	driverName, err := ctrl.getSnapshotDriverName(snapshot)
+	if err != nil {
+		klog.Errorf("failed to getSnapshotDriverName while recording metrics for snapshot %q: %v", utils.SnapshotKey(snapshot), err)
+	}
+
+	snapshotProvisionType := metrics.DynamicSnapshotType
+	if snapshot.Spec.Source.VolumeSnapshotContentName != nil {
+		snapshotProvisionType = metrics.PreProvisionedSnapshotType
+	}
+
+	// Processing delete, start operation metric
+	deleteOperationKey := metrics.NewOperationKey(metrics.DeleteSnapshotOperationName, snapshot.UID)
+	deleteOperationValue := metrics.NewOperationValue(driverName, snapshotProvisionType)
+	ctrl.metricsManager.OperationStart(deleteOperationKey, deleteOperationValue)
 
 	var contentName string
 	if snapshot.Status != nil && snapshot.Status.BoundVolumeSnapshotContentName != nil {
@@ -269,6 +284,7 @@ func (ctrl *csiSnapshotCommonController) processSnapshotWithDeletionTimestamp(sn
 	}
 
 	klog.V(5).Infof("processSnapshotWithDeletionTimestamp[%s]: delete snapshot content and remove finalizer from snapshot if needed", utils.SnapshotKey(snapshot))
+
 	return ctrl.checkandRemoveSnapshotFinalizersAndCheckandDeleteContent(snapshot, content, deleteContent)
 }
 
@@ -388,6 +404,7 @@ func (ctrl *csiSnapshotCommonController) syncReadySnapshot(snapshot *crdv1.Volum
 		// snapshot is bound but content is not pointing to the snapshot
 		return ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotMisbound", "VolumeSnapshotContent is not bound to the VolumeSnapshot correctly")
 	}
+
 	// everything is verified, return
 	return nil
 }
@@ -396,20 +413,45 @@ func (ctrl *csiSnapshotCommonController) syncReadySnapshot(snapshot *crdv1.Volum
 func (ctrl *csiSnapshotCommonController) syncUnreadySnapshot(snapshot *crdv1.VolumeSnapshot) error {
 	uniqueSnapshotName := utils.SnapshotKey(snapshot)
 	klog.V(5).Infof("syncUnreadySnapshot %s", uniqueSnapshotName)
+	driverName, err := ctrl.getSnapshotDriverName(snapshot)
+	if err != nil {
+		klog.Errorf("failed to getSnapshotDriverName while recording metrics for snapshot %q: %s", utils.SnapshotKey(snapshot), err)
+	}
+
+	snapshotProvisionType := metrics.DynamicSnapshotType
+	if snapshot.Spec.Source.VolumeSnapshotContentName != nil {
+		snapshotProvisionType = metrics.PreProvisionedSnapshotType
+	}
+
+	// Start metrics operations
+	if !utils.IsSnapshotCreated(snapshot) {
+		// Only start CreateSnapshot operation if the snapshot has not been cut
+		ctrl.metricsManager.OperationStart(
+			metrics.NewOperationKey(metrics.CreateSnapshotOperationName, snapshot.UID),
+			metrics.NewOperationValue(driverName, snapshotProvisionType),
+		)
+	}
+	ctrl.metricsManager.OperationStart(
+		metrics.NewOperationKey(metrics.CreateSnapshotAndReadyOperationName, snapshot.UID),
+		metrics.NewOperationValue(driverName, snapshotProvisionType),
+	)
 
 	// Pre-provisioned snapshot
 	if snapshot.Spec.Source.VolumeSnapshotContentName != nil {
 		content, err := ctrl.getPreprovisionedContentFromStore(snapshot)
 		if err != nil {
 			return err
 		}
+
 		// if no content found yet, update status and return
 		if content == nil {
 			// can not find the desired VolumeSnapshotContent from cache store
 			ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotContentMissing", "VolumeSnapshotContent is missing")
 			klog.V(4).Infof("syncUnreadySnapshot[%s]: snapshot content %q requested but not found, will try again", utils.SnapshotKey(snapshot), *snapshot.Spec.Source.VolumeSnapshotContentName)
+
 			return fmt.Errorf("snapshot %s requests an non-existing content %s", utils.SnapshotKey(snapshot), *snapshot.Spec.Source.VolumeSnapshotContentName)
 		}
+
 		// Set VolumeSnapshotRef UID
 		newContent, err := ctrl.checkandBindSnapshotContent(snapshot, content)
 		if err != nil {
@@ -426,8 +468,10 @@ func (ctrl *csiSnapshotCommonController) syncUnreadySnapshot(snapshot *crdv1.Vol
 			ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotStatusUpdateFailed", fmt.Sprintf("Snapshot status update failed, %v", err))
 			return err
 		}
+
 		return nil
 	}
+
 	// snapshot.Spec.Source.VolumeSnapshotContentName == nil - dynamically creating snapshot
 	klog.V(5).Infof("getDynamicallyProvisionedContentFromStore for snapshot %s", uniqueSnapshotName)
 	contentObj, err := ctrl.getDynamicallyProvisionedContentFromStore(snapshot)
@@ -1094,10 +1138,30 @@ func (ctrl *csiSnapshotCommonController) updateSnapshotStatus(snapshot *crdv1.Vo
 	if updated {
 		snapshotClone := snapshotObj.DeepCopy()
 		snapshotClone.Status = newStatus
+
+		// We need to record metrics before updating the status due to a bug causing cache entries after a failed UpdateStatus call.
+		// Must meet the following criteria to emit a successful CreateSnapshot status
+		// 1. Previous status was nil OR Previous status had a nil CreationTime
+		// 2. New status must be non-nil with a non-nil CreationTime
+		driverName := content.Spec.Driver
+		createOperationKey := metrics.NewOperationKey(metrics.CreateSnapshotOperationName, snapshot.UID)
+		if !utils.IsSnapshotCreated(snapshotObj) && utils.IsSnapshotCreated(snapshotClone) {
+			ctrl.metricsManager.RecordMetrics(createOperationKey, metrics.NewSnapshotOperationStatus(metrics.SnapshotStatusTypeSuccess), driverName)
+		}
+
+		// Must meet the following criteria to emit a successful CreateSnapshotAndReady status
+		// 1. Previous status was nil OR Previous status had a nil ReadyToUse OR Previous status had a false ReadyToUse
+		// 2. New status must be non-nil with a ReadyToUse as true
+		if !utils.IsSnapshotReady(snapshotObj) && utils.IsSnapshotReady(snapshotClone) {
+			createAndReadyOperation := metrics.NewOperationKey(metrics.CreateSnapshotAndReadyOperationName, snapshot.UID)
+			ctrl.metricsManager.RecordMetrics(createAndReadyOperation, metrics.NewSnapshotOperationStatus(metrics.SnapshotStatusTypeSuccess), driverName)
+		}
+
 		newSnapshotObj, err := ctrl.clientset.SnapshotV1beta1().VolumeSnapshots(snapshotClone.Namespace).UpdateStatus(context.TODO(), snapshotClone, metav1.UpdateOptions{})
 		if err != nil {
 			return nil, newControllerUpdateError(utils.SnapshotKey(snapshot), err.Error())
 		}
+
 		return newSnapshotObj, nil
 	}
 
@@ -1177,6 +1241,45 @@ func (ctrl *csiSnapshotCommonController) getSnapshotClass(className string) (*cr
 	return class, nil
 }
 
+// getSnapshotDriverName is a helper function to get snapshot driver from the VolumeSnapshot.
+// We try to get the driverName in multiple ways, as snapshot controller metrics depend on the correct driverName.
+func (ctrl *csiSnapshotCommonController) getSnapshotDriverName(vs *crdv1.VolumeSnapshot) (string, error) {
+	klog.V(5).Infof("getSnapshotDriverName: VolumeSnapshot[%s]", vs.Name)
+	var driverName string
+
+	// Pre-Provisioned snapshots have contentName as source
+	var contentName string
+	if vs.Spec.Source.VolumeSnapshotContentName != nil {
+		contentName = *vs.Spec.Source.VolumeSnapshotContentName
+	}
+
+	// Get Driver name from SnapshotContent if we found a contentName
+	if contentName != "" {
+		content, err := ctrl.contentLister.Get(contentName)
+		if err != nil {
+			klog.Errorf("getSnapshotDriverName: failed to get snapshotContent: %v", contentName)
+		} else {
+			driverName = content.Spec.Driver
+		}
+
+		if driverName != "" {
+			return driverName, nil
+		}
+	}
+
+	// Dynamic snapshots will have a snapshotclass with a driver
+	if vs.Spec.VolumeSnapshotClassName != nil {
+		class, err := ctrl.getSnapshotClass(*vs.Spec.VolumeSnapshotClassName)
+		if err != nil {
+			klog.Errorf("getSnapshotDriverName: failed to get snapshotClass: %v", *vs.Spec.VolumeSnapshotClassName)
+		} else {
+			driverName = class.Driver
+		}
+	}
+
+	return driverName, nil
+}
+
 // SetDefaultSnapshotClass is a helper function to figure out the default snapshot class.
 // For pre-provisioned case, it's an no-op.
 // For dynamic provisioning, it gets the default SnapshotClasses in the system if there is any(could be multiple),

pkg/common-controller/snapshot_controller_base.go

@@ -24,6 +24,7 @@ import (
 	clientset "github.com/kubernetes-csi/external-snapshotter/client/v3/clientset/versioned"
 	storageinformers "github.com/kubernetes-csi/external-snapshotter/client/v3/informers/externalversions/volumesnapshot/v1beta1"
 	storagelisters "github.com/kubernetes-csi/external-snapshotter/client/v3/listers/volumesnapshot/v1beta1"
+	"github.com/kubernetes-csi/external-snapshotter/v3/pkg/metrics"
 	"github.com/kubernetes-csi/external-snapshotter/v3/pkg/utils"
 
 	v1 "k8s.io/api/core/v1"
@@ -60,6 +61,8 @@ type csiSnapshotCommonController struct {
 	snapshotStore cache.Store
 	contentStore  cache.Store
 
+	metricsManager metrics.MetricsManager
+
 	resyncPeriod time.Duration
 }
 
@@ -71,6 +74,7 @@ func NewCSISnapshotCommonController(
 	volumeSnapshotContentInformer storageinformers.VolumeSnapshotContentInformer,
 	volumeSnapshotClassInformer storageinformers.VolumeSnapshotClassInformer,
 	pvcInformer coreinformers.PersistentVolumeClaimInformer,
+	metricsManager metrics.MetricsManager,
 	resyncPeriod time.Duration,
 ) *csiSnapshotCommonController {
 	broadcaster := record.NewBroadcaster()
@@ -80,14 +84,15 @@ func NewCSISnapshotCommonController(
 	eventRecorder = broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: fmt.Sprintf("snapshot-controller")})
 
 	ctrl := &csiSnapshotCommonController{
-		clientset:     clientset,
-		client:        client,
-		eventRecorder: eventRecorder,
-		resyncPeriod:  resyncPeriod,
-		snapshotStore: cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),
-		contentStore:  cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),
-		snapshotQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "snapshot-controller-snapshot"),
-		contentQueue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "snapshot-controller-content"),
+		clientset:      clientset,
+		client:         client,
+		eventRecorder:  eventRecorder,
+		resyncPeriod:   resyncPeriod,
+		snapshotStore:  cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),
+		contentStore:   cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),
+		snapshotQueue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "snapshot-controller-snapshot"),
+		contentQueue:   workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "snapshot-controller-content"),
+		metricsManager: metricsManager,
 	}
 
 	ctrl.pvcLister = pvcInformer.Lister()
@@ -363,6 +368,7 @@ func (ctrl *csiSnapshotCommonController) updateSnapshot(snapshot *crdv1.VolumeSn
 	if !newSnapshot {
 		return nil
 	}
+
 	err = ctrl.syncSnapshot(snapshot)
 	if err != nil {
 		if errors.IsConflict(err) {
@@ -407,6 +413,13 @@ func (ctrl *csiSnapshotCommonController) updateContent(content *crdv1.VolumeSnap
 func (ctrl *csiSnapshotCommonController) deleteSnapshot(snapshot *crdv1.VolumeSnapshot) {
 	_ = ctrl.snapshotStore.Delete(snapshot)
 	klog.V(4).Infof("snapshot %q deleted", utils.SnapshotKey(snapshot))
+	driverName, err := ctrl.getSnapshotDriverName(snapshot)
+	if err != nil {
+		klog.Errorf("failed to getSnapshotDriverName while recording metrics for snapshot %q: %s", utils.SnapshotKey(snapshot), err)
+	} else {
+		deleteOperationKey := metrics.NewOperationKey(metrics.DeleteSnapshotOperationName, snapshot.UID)
+		ctrl.metricsManager.RecordMetrics(deleteOperationKey, metrics.NewSnapshotOperationStatus(metrics.SnapshotStatusTypeSuccess), driverName)
+	}
 
 	snapshotContentName := ""
 	if snapshot.Status != nil && snapshot.Status.BoundVolumeSnapshotContentName != nil {
@@ -416,6 +429,7 @@ func (ctrl *csiSnapshotCommonController) deleteSnapshot(snapshot *crdv1.VolumeSn
 		klog.V(5).Infof("deleteSnapshot[%q]: content not bound", utils.SnapshotKey(snapshot))
 		return
 	}
+
 	// sync the content when its snapshot is deleted.  Explicitly sync'ing the
 	// content here in response to snapshot deletion prevents the content from
 	// waiting until the next sync period for its Release.