Revert "Automated cherry pick of #547: Limit the maximum number of registered metric collectors."

Author: amacaskill

cmd/csi_driver/main.go

@@ -47,7 +47,6 @@ 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"
@@ -110,7 +109,7 @@ func main() {
 		}
 
 		if *metricsEndpoint != "" {
-			mm = metrics.NewMetricsManager(*metricsEndpoint, *fuseSocketDir, *maximumNumberOfCollectors, clientset)
+			mm = metrics.NewMetricsManager(*metricsEndpoint, *fuseSocketDir, clientset)
 			mm.InitializeHTTPHandler()
 		}
 	}

deploy/base/node/node.yaml

@@ -53,7 +53,6 @@ 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,8 +114,7 @@ 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 {
@@ -166,7 +165,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 collector.
+	// Register metrics collecter.
 	// 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,11 +26,8 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 	"time"
 
-	"k8s.io/apimachinery/pkg/util/sets"
-
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/cloud_provider/clientset"
 	"github.com/googlecloudplatform/gcs-fuse-csi-driver/pkg/util"
 	"github.com/prometheus/client_golang/prometheus"
@@ -59,21 +56,14 @@ type manager struct {
 	metricsEndpoint string
 	fuseSocketDir   string
 	clientset       clientset.Interface
-
-	maximumNumberOfCollectors   int
-	volumePublishPathRegistered sets.Set[string]
-	mutex                       sync.Mutex
 }
 
-func NewMetricsManager(metricsEndpoint, fuseSocketDir string, maximumNumberOfCollectors int, clientset clientset.Interface) Manager {
+func NewMetricsManager(metricsEndpoint, fuseSocketDir string, clientset clientset.Interface) Manager {
 	mm := &manager{
-		registry:                    prometheus.NewRegistry(),
-		metricsEndpoint:             metricsEndpoint,
-		fuseSocketDir:               fuseSocketDir,
-		clientset:                   clientset,
-		volumePublishPathRegistered: sets.Set[string]{},
-		maximumNumberOfCollectors:   maximumNumberOfCollectors,
-		mutex:                       sync.Mutex{},
+		registry:        prometheus.NewRegistry(),
+		metricsEndpoint: metricsEndpoint,
+		fuseSocketDir:   fuseSocketDir,
+		clientset:       clientset,
 	}
 
 	return mm
@@ -125,43 +115,8 @@ func (mm *manager) RegisterMetricsCollector(targetPath, podNamespace, podName, b
 		"bucket_name":    bucketName,
 		"pod_uid":        podUID,
 	}, mm.clientset)
-
-	// Lock the number of registered collectors while we attempt to register a new collector.
-	mm.mutex.Lock()
-	defer mm.mutex.Unlock()
-
-	if mm.maximumNumberOfCollectors == 0 {
-		klog.Infof("could not register metrics collector: podUID: %s, volume: %s. metrics collector limit is set to zero.", podUID, bucketName)
-
-		return
-	}
-
-	// Check if we need to register collector. We register a collector when the following are met:
-	// 1. There is space on the metrics pipeline for the collector to be registered.
-	// 2. The metrics collector has not previously been registered.
-	if mm.maximumNumberOfCollectors > 0 {
-		// If volume is already registered, do not register again. This flow can get triggered
-		//  since CSI driver has republishVolume capability.
-		if mm.volumePublishPathRegistered.Has(targetPath) {
-			return
-		}
-		// If collector hasn't been registered and there's no space left, log a warning.
-		if mm.volumePublishPathRegistered.Len() >= mm.maximumNumberOfCollectors {
-			klog.V(6).Infof("could not register a metrics collector: podUID: %s, volume: %s. there's already %d collectors registered.", podUID, bucketName, mm.volumePublishPathRegistered.Len())
-
-			return
-		}
-	}
-
-	// Attempt to register new metrics collector and record success.
-	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.volumePublishPathRegistered.Insert(targetPath)
-		klog.Infof("successfully registered a new metrics collector: podUID: %s, volume: %s. there's %d collectors registered.", podUID, bucketName, mm.volumePublishPathRegistered.Len())
+	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)
 	}
 }
 
@@ -171,16 +126,8 @@ 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.volumePublishPathRegistered.Delete(targetPath)
-		klog.Infof("successfully unregistered a metrics collector: podUID: %s, volume: %s. there's %d collectors registered.", podUID, volumeName, mm.volumePublishPathRegistered.Len())
 	}
 }