Limit the maximum number of registered metric collectors.

Change-Id: I18204ab923b8f02a3fd48f3a6ab34cb45e160b01

Author: hime

cmd/csi_driver/main.go

@@ -47,6 +47,7 @@ var (
 	informerResyncDurationSec = flag.Int("informer-resync-duration-sec", 1800, "informer resync duration in seconds")
 	fuseSocketDir             = flag.String("fuse-socket-dir", "/sockets", "FUSE socket directory")
 	metricsEndpoint           = flag.String("metrics-endpoint", "", "The TCP network address where the Prometheus metrics endpoint will listen (example: `:8080`). The default is empty string, which means that the metrics endpoint is disabled.")
+	maximumNumberOfCollectors = flag.Int("max-metric-collectors", -1, "Maximum number of prometheus metric collectors exporting metrics at a time, less than 0 (e.g -1) means no limit.")
 
 	// These are set at compile time.
 	version = "unknown"
@@ -109,7 +110,7 @@ func main() {
 		}
 
 		if *metricsEndpoint != "" {
-			mm = metrics.NewMetricsManager(*metricsEndpoint, *fuseSocketDir, clientset)
+			mm = metrics.NewMetricsManager(*metricsEndpoint, *fuseSocketDir, *maximumNumberOfCollectors, clientset)
 			mm.InitializeHTTPHandler()
 		}
 	}

deploy/base/node/node.yaml

@@ -53,6 +53,7 @@ spec:
             - --node=true
             - --identity-provider=$(IDENTITY_PROVIDER)
             - --metrics-endpoint=:9920
+            - --max-metric-collectors=10
           ports:
           - containerPort: 9920
             name: metrics

pkg/csi_driver/node.go

@@ -114,7 +114,8 @@ func (s *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublish
 			s.volumeStateStore.Store(targetPath, &util.VolumeState{})
 			vs, _ = s.volumeStateStore.Load(targetPath)
 		}
-
+		// volumeState is safe to access for remaining of function since volumeLock prevents
+		// Node Publish/Unpublish Volume calls from running more than once at a time per volume.
 		if !vs.BucketAccessCheckPassed {
 			storageService, err := s.prepareStorageService(ctx, vc)
 			if err != nil {
@@ -165,7 +166,7 @@ func (s *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublish
 		return nil, status.Error(codes.FailedPrecondition, "failed to find the sidecar container in Pod spec")
 	}
 
-	// Register metrics collecter.
+	// Register metrics collector.
 	// It is idempotent to register the same collector in node republish calls.
 	if s.driver.config.MetricsManager != nil && !disableMetricsCollection {
 		klog.V(6).Infof("NodePublishVolume enabling metrics collector for target path %q", targetPath)

pkg/metrics/metrics.go

@@ -26,6 +26,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/cloud_provider/clientset"
@@ -56,14 +57,20 @@ type manager struct {
 	metricsEndpoint string
 	fuseSocketDir   string
 	clientset       clientset.Interface
+
+	maximumNumberOfCollectors int
+	registeredCollectorsCount int
+	mutex                     sync.Mutex
 }
 
-func NewMetricsManager(metricsEndpoint, fuseSocketDir string, clientset clientset.Interface) Manager {
+func NewMetricsManager(metricsEndpoint, fuseSocketDir string, maximumNumberOfCollectors int, clientset clientset.Interface) Manager {
 	mm := &manager{
-		registry:        prometheus.NewRegistry(),
-		metricsEndpoint: metricsEndpoint,
-		fuseSocketDir:   fuseSocketDir,
-		clientset:       clientset,
+		registry:                  prometheus.NewRegistry(),
+		metricsEndpoint:           metricsEndpoint,
+		fuseSocketDir:             fuseSocketDir,
+		clientset:                 clientset,
+		maximumNumberOfCollectors: maximumNumberOfCollectors,
+		mutex:                     sync.Mutex{},
 	}
 
 	return mm
@@ -115,8 +122,25 @@ func (mm *manager) RegisterMetricsCollector(targetPath, podNamespace, podName, b
 		"bucket_name":    bucketName,
 		"pod_uid":        podUID,
 	}, mm.clientset)
-	if err := mm.registry.Register(c); err != nil && !strings.Contains(err.Error(), prometheus.AlreadyRegisteredError{}.Error()) {
-		klog.Errorf("failed to register metrics collector for pod  %v/%v, volume %q, bucket %q: %v", podNamespace, podName, volumeName, bucketName, err)
+
+	// Lock the number of registered collectors while we attemtp to register a new collector.
+	mm.mutex.Lock()
+	defer mm.mutex.Unlock()
+
+	// We skip registration when we already registered all supported metrics collectors. If a limit is
+	// unset (maximumNumberOfCollectors less than zero), we continue to registration.
+	if mm.maximumNumberOfCollectors >= 0 && mm.registeredCollectorsCount >= mm.maximumNumberOfCollectors {
+		return
+	}
+
+	err = mm.registry.Register(c)
+	if err != nil {
+		if !strings.Contains(err.Error(), prometheus.AlreadyRegisteredError{}.Error()) {
+			klog.Errorf("failed to register metrics collector for pod  %v/%v, volume %q, bucket %q: %v", podNamespace, podName, volumeName, bucketName, err)
+		}
+	} else {
+		mm.registeredCollectorsCount += 1
+		klog.Infof("successfully registered a new metrics collector: podUID: %s, volume: %s. there's %d collectors registerd.", podUID, bucketName, mm.registeredCollectorsCount)
 	}
 }
 
@@ -126,8 +150,16 @@ func (mm *manager) UnregisterMetricsCollector(targetPath string) {
 
 	// metricsCollector uses a hash of pod UID and volume name as an identifier.
 	c := NewMetricsCollector("", "", "", "", podUID, volumeName, nil, nil)
+
+	// Lock the number of registered collectors while we attempt to unregister a collector.
+	mm.mutex.Lock()
+	defer mm.mutex.Unlock()
+
 	if ok := mm.registry.Unregister(c); !ok {
 		klog.Infof("Unregister metrics collector for targetPath %q is not needed since the collector is not registered", targetPath)
+	} else {
+		mm.registeredCollectorsCount -= 1
+		klog.Infof("successfully unregistered a metrics collector: podUID: %s, volume: %s. there's %d collectors registerd.", podUID, volumeName, mm.registeredCollectorsCount)
 	}
 }