Merge branch 'master' into use-central-flags

Author: DerekFrank

cmd/csi-provisioner/csi-provisioner.go

@@ -60,8 +60,8 @@ import (
 	_ "k8s.io/component-base/metrics/prometheus/workqueue"               // register work queues in the default legacy registry
 	csitrans "k8s.io/csi-translation-lib"
 	"k8s.io/klog/v2"
-	"sigs.k8s.io/sig-storage-lib-external-provisioner/v12/controller"
-	libmetrics "sigs.k8s.io/sig-storage-lib-external-provisioner/v12/controller/metrics"
+	"sigs.k8s.io/sig-storage-lib-external-provisioner/v13/controller"
+	libmetrics "sigs.k8s.io/sig-storage-lib-external-provisioner/v13/controller/metrics"
 
 	"github.com/kubernetes-csi/csi-lib-utils/leaderelection"
 	"github.com/kubernetes-csi/csi-lib-utils/metrics"
@@ -78,8 +78,11 @@ import (
 )
 
 var (
+	master               = flag.String("master", "", "Master URL to build a client config from. Either this or kubeconfig needs to be set if the provisioner is being run out of cluster.")
 	volumeNamePrefix     = flag.String("volume-name-prefix", "pvc", "Prefix to apply to the name of a created volume.")
 	volumeNameUUIDLength = flag.Int("volume-name-uuid-length", -1, "Truncates generated UUID of a created volume to this length. Defaults behavior is to NOT truncate.")
+	retryIntervalStart   = flag.Duration("retry-interval-start", time.Second, "Initial retry interval of failed provisioning or deletion. It doubles with each failure, up to retry-interval-max.")
+	retryIntervalMax     = flag.Duration("retry-interval-max", 5*time.Minute, "Maximum retry interval of failed provisioning or deletion.")
 	workerThreads        = flag.Uint("worker-threads", 100, "Number of provisioner worker threads, in other words nr. of simultaneous CSI calls.")
 	finalizerThreads     = flag.Uint("cloning-protection-threads", 1, "Number of simultaneously running threads, handling cloning finalizer removal")
 	capacityThreads      = flag.Uint("capacity-threads", 1, "Number of simultaneously running threads, handling CSIStorageCapacity objects")
@@ -92,6 +95,9 @@ var (
 
 	defaultFSType = flag.String("default-fstype", "", "The default filesystem type of the volume to provision when fstype is unspecified in the StorageClass. If the default is not set and fstype is unset in the StorageClass, then no fstype will be set")
 
+	kubeAPICapacityQPS   = flag.Float32("kube-api-capacity-qps", 1, "QPS to use for storage capacity updates while communicating with the kubernetes apiserver. Defaults to 1.0.")
+	kubeAPICapacityBurst = flag.Int("kube-api-capacity-burst", 5, "Burst to use for storage capacity updates while communicating with the kubernetes apiserver. Defaults to 5.")
+
 	enableCapacity           = flag.Bool("enable-capacity", false, "This enables producing CSIStorageCapacity objects with capacity information from the driver's GetCapacity call.")
 	capacityImmediateBinding = flag.Bool("capacity-for-immediate-binding", false, "Enables producing capacity information for storage classes with immediate binding. Not needed for the Kubernetes scheduler, maybe useful for other consumers or for debugging.")
 	capacityPollInterval     = flag.Duration("capacity-poll-interval", time.Minute, "How long the external-provisioner waits before checking for storage capacity changes.")
@@ -165,9 +171,9 @@ func main() {
 		standardflags.Configuration.KubeConfig = kubeconfigEnv
 	}
 
-	if standardflags.Configuration.Master != "" || standardflags.Configuration.KubeConfig != "" {
+	if *master != "" || standardflags.Configuration.KubeConfig != "" {
 		klog.Infof("Either master or kubeconfig specified. building kube config from that..")
-		config, err = clientcmd.BuildConfigFromFlags(standardflags.Configuration.Master, standardflags.Configuration.KubeConfig)
+		config, err = clientcmd.BuildConfigFromFlags(*master, standardflags.Configuration.KubeConfig)
 	} else {
 		klog.Infof("Building kube configs for running in cluster...")
 		config, err = rest.InClusterConfig()
@@ -373,8 +379,8 @@ func main() {
 
 	// -------------------------------
 	// PersistentVolumeClaims informer
-	genericRateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[string](standardflags.Configuration.RetryIntervalStart, standardflags.Configuration.RetryIntervalMax)
-	claimRateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[string](standardflags.Configuration.RetryIntervalStart, standardflags.Configuration.RetryIntervalMax)
+	genericRateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[string](*retryIntervalStart, *retryIntervalMax)
+	claimRateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[string](*retryIntervalStart, *retryIntervalMax)
 	claimQueue := workqueue.NewTypedRateLimitingQueueWithConfig(claimRateLimiter, workqueue.TypedRateLimitingQueueConfig[string]{Name: "claims"})
 	claimInformer := factory.Core().V1().PersistentVolumeClaims().Informer()
 
@@ -387,8 +393,7 @@ func main() {
 		controller.Threadiness(int(*workerThreads)),
 		controller.CreateProvisionedPVLimiter(workqueue.DefaultTypedControllerRateLimiter[string]()),
 		controller.ClaimsInformer(claimInformer),
-		controller.NodesLister(nodeLister),
-		controller.RetryIntervalMax(standardflags.Configuration.RetryIntervalMax),
+		controller.RetryIntervalMax(*retryIntervalMax),
 	}
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.HonorPVReclaimPolicy) {
@@ -398,6 +403,10 @@ func main() {
 	if supportsMigrationFromInTreePluginName != "" {
 		provisionerOptions = append(provisionerOptions, controller.AdditionalProvisionerNames([]string{supportsMigrationFromInTreePluginName}))
 	}
+	var pvcNodeStore *ctrl.InMemoryStore
+	if ctrl.SupportsTopology(pluginCapabilities) {
+		pvcNodeStore = ctrl.NewInMemoryStore()
+	}
 
 	// Create the provisioner: it implements the Provisioner interface expected by
 	// the controller
@@ -427,12 +436,15 @@ func main() {
 		nodeDeployment,
 		*controllerPublishReadOnly,
 		*preventVolumeModeConversion,
+		pvcNodeStore,
 	)
 
 	var capacityController *capacity.Controller
 	if *enableCapacity {
 		// Publishing storage capacity information uses its own client
 		// with separate rate limiting.
+		config.QPS = *kubeAPICapacityQPS
+		config.Burst = *kubeAPICapacityBurst
 		clientset, err := kubernetes.NewForConfig(config)
 		if err != nil {
 			klog.Fatalf("Failed to create client: %v", err)
@@ -464,7 +476,7 @@ func main() {
 
 		var topologyInformer topology.Informer
 		if nodeDeployment == nil {
-			topologyRateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[string](standardflags.Configuration.RetryIntervalStart, standardflags.Configuration.RetryIntervalMax)
+			topologyRateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[string](*retryIntervalStart, *retryIntervalMax)
 			topologyInformer = topology.NewNodeTopology(
 				provisionerName,
 				clientset,
@@ -530,7 +542,7 @@ func main() {
 			klog.Fatalf("unexpected error when checking for the V1 CSIStorageCapacity API: %v", err)
 		}
 
-		capacityRateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[capacity.QueueKey](standardflags.Configuration.RetryIntervalStart, standardflags.Configuration.RetryIntervalMax)
+		capacityRateLimiter := workqueue.NewTypedItemExponentialFailureRateLimiter[capacity.QueueKey](*retryIntervalStart, *retryIntervalMax)
 		capacityController = capacity.NewCentralCapacityController(
 			csi.NewControllerClient(grpcClient),
 			provisionerName,

go.mod

@@ -33,7 +33,7 @@ require (
 	github.com/onsi/ginkgo/v2 v2.23.4
 	github.com/onsi/gomega v1.37.0
 	k8s.io/kubernetes v1.34.0
-	sigs.k8s.io/sig-storage-lib-external-provisioner/v12 v12.0.1
+	sigs.k8s.io/sig-storage-lib-external-provisioner/v13 v13.0.0
 )
 
 require (

go.sum

@@ -374,8 +374,8 @@ sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5E
 sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
 sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU=
 sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
-sigs.k8s.io/sig-storage-lib-external-provisioner/v12 v12.0.1 h1:rQcMFPcZ12y82+BA7b29gCWyoI/+/gAQUZu/Cw+8bC0=
-sigs.k8s.io/sig-storage-lib-external-provisioner/v12 v12.0.1/go.mod h1:kPy4hBso6PNhP9PdlTDdBZqxP1RKg7DFFM7lIR1bA8k=
+sigs.k8s.io/sig-storage-lib-external-provisioner/v13 v13.0.0 h1:bqSqBfqtToTDMDz+FEzfqofXAp5ptt6Z7ShR0g05PGA=
+sigs.k8s.io/sig-storage-lib-external-provisioner/v13 v13.0.0/go.mod h1:1xSe5kgJcKbrtNdD5WoytKUoByAGDl3wVHlKP0RZIC8=
 sigs.k8s.io/structured-merge-diff/v6 v6.3.0 h1:jTijUJbW353oVOd9oTlifJqOGEkUw2jB/fXCbTiQEco=
 sigs.k8s.io/structured-merge-diff/v6 v6.3.0/go.mod h1:M3W8sfWvn2HhQDIbGWj3S099YozAsymCo/wrT5ohRUE=
 sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs=

pkg/capacity/provision.go

@@ -20,7 +20,7 @@ import (
 	"context"
 
 	v1 "k8s.io/api/core/v1"
-	"sigs.k8s.io/sig-storage-lib-external-provisioner/v12/controller"
+	"sigs.k8s.io/sig-storage-lib-external-provisioner/v13/controller"
 )
 
 type provisionWrapper struct {

pkg/controller/cache.go

@@ -0,0 +1,132 @@
+package controller
+
+import (
+	"fmt"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// TopologyInfo holds the data for NodeLabels and TopologyKeys
+type TopologyInfo struct {
+	NodeLabels     map[string]string
+	TopologyKeys   []string
+	RequisiteTerms []topologyTerm
+}
+
+// TopologyProvider is an interface that defines the behavior for looking up
+// a TopologyInfo object by its pvc UID.
+type TopologyProvider interface {
+	GetByPvcUID(pvcUID types.UID) (*TopologyInfo, error)
+	// The entry is deleted when provision succeeds or returns a final error.
+	Delete(pvcUID types.UID) error
+
+	// Update methods now perform an "upsert" and don't return errors.
+	UpdateNodeLabels(pvcUID types.UID, newLabels map[string]string)
+	UpdateTopologyKeys(pvcUID types.UID, newKeys []string)
+	UpdateRequisiteTerms(pvcUID types.UID, requisiteTerms []topologyTerm)
+}
+
+// InMemoryStore is a concrete implementation of TopologyProvider.
+// It uses an in-memory map for quick lookups.
+type InMemoryStore struct {
+	// The map key is the object's name.
+	data map[types.UID]*TopologyInfo
+	// Adding a mutex for thread-safe access
+	mutex sync.RWMutex
+}
+
+// NewInMemoryStore creates and initializes a new store.
+func NewInMemoryStore() *InMemoryStore {
+	return &InMemoryStore{
+		data: make(map[types.UID]*TopologyInfo),
+	}
+}
+
+// Delete implements the TopologyProvider interface.
+// It uses the built-in delete() function to remove the item from the map.
+func (s *InMemoryStore) Delete(pvcUID types.UID) error {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	// First, check if the key exists to provide a helpful error.
+	_, found := s.data[pvcUID]
+	if !found {
+		return nil
+	}
+	delete(s.data, pvcUID)
+	return nil
+}
+
+// GetByPvcUID implements the TopologyProvider interface.
+func (s *InMemoryStore) GetByPvcUID(pvcUID types.UID) (*TopologyInfo, error) {
+	if s == nil {
+		return nil, fmt.Errorf("pvcNodeStore is nil")
+	}
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+	info, found := s.data[pvcUID]
+	if !found {
+		return nil, fmt.Errorf("topology object with pvcUID '%s' not found", pvcUID)
+	}
+
+	// Return a deep copy to prevent data races
+	infoCopy := &TopologyInfo{}
+	if info.NodeLabels != nil {
+		infoCopy.NodeLabels = make(map[string]string)
+		for k, v := range info.NodeLabels {
+			infoCopy.NodeLabels[k] = v
+		}
+	}
+
+	if info.TopologyKeys != nil {
+		infoCopy.TopologyKeys = make([]string, len(info.TopologyKeys))
+		copy(infoCopy.TopologyKeys, info.TopologyKeys)
+	}
+
+	if info.RequisiteTerms != nil {
+		infoCopy.RequisiteTerms = make([]topologyTerm, len(info.RequisiteTerms))
+		for i, term := range info.RequisiteTerms {
+			newTerm := make(topologyTerm, len(term))
+			copy(newTerm, term)
+			infoCopy.RequisiteTerms[i] = newTerm
+		}
+	}
+
+	return infoCopy, nil
+}
+
+// UpdateNodeLabels finds an object by pvcUID and replaces its NodeLabels.
+func (s *InMemoryStore) UpdateNodeLabels(pvcUID types.UID, newLabels map[string]string) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	info, found := s.data[pvcUID]
+	if !found {
+		s.data[pvcUID] = &TopologyInfo{NodeLabels: newLabels}
+	} else {
+		info.NodeLabels = newLabels
+	}
+}
+
+// UpdateTopologyKeys finds an object by pvcUID and replaces its TopologyKeys.
+func (s *InMemoryStore) UpdateTopologyKeys(pvcUID types.UID, newKeys []string) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	info, found := s.data[pvcUID]
+	if !found {
+		s.data[pvcUID] = &TopologyInfo{TopologyKeys: newKeys}
+	} else {
+		info.TopologyKeys = newKeys
+	}
+}
+
+// UpdateRequisiteTerms finds an object by pvcUID and replaces its RequisiteTerms.
+func (s *InMemoryStore) UpdateRequisiteTerms(pvcUID types.UID, requisiteTerms []topologyTerm) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	info, found := s.data[pvcUID]
+	if !found {
+		s.data[pvcUID] = &TopologyInfo{RequisiteTerms: requisiteTerms}
+	} else {
+		info.RequisiteTerms = requisiteTerms
+	}
+}