Merge pull request #106 from red-hat-storage/sync_us--main

Syncing latest changes from upstream main for kubernetes-csi-addons

.github/workflows/build-push.yaml

@@ -17,7 +17,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Setup Golang
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: "1.20"
 

.github/workflows/golangci-lint.yaml

@@ -16,7 +16,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - uses: actions/setup-go@v4
+      - uses: actions/setup-go@v5
         with:
           go-version: '1.20'
           cache: false

.github/workflows/kind-deploy.yaml

@@ -21,7 +21,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Golang
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: "1.20"
 

.github/workflows/tag-release.yaml

@@ -18,7 +18,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Setup Golang
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: "1.20"
 
@@ -127,7 +127,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Setup Golang
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: "1.20"
 

.github/workflows/test-build.yaml

@@ -16,7 +16,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Setup Golang
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: "1.20"
 

.github/workflows/test-golang.yaml

@@ -16,7 +16,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Golang
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: "1.20"
 
@@ -44,7 +44,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Golang
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: "1.20"
 
@@ -59,7 +59,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Set up Golang
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v5
         with:
           go-version: "1.20"
 

Makefile

@@ -119,7 +119,7 @@ help: ## Display this help.
 
 .PHONY: manifests
 manifests: controller-gen kustomize ## Generate WebhookConfiguration, ClusterRole and CustomResourceDefinition objects.
-	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="{./apis/...,./cmd/...,./controllers/...,./sidecar/...}" output:crd:artifacts:config=config/crd/bases
+	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="{./apis/...,./cmd/...,./controllers/...,./internal/...,./sidecar/...}" output:crd:artifacts:config=config/crd/bases
 	cd config/manager && $(KUSTOMIZE) edit set image controller=${CONTROLLER_IMG} rbac-proxy=${RBAC_PROXY_IMG}
 	$(KUSTOMIZE) build config/crd > deploy/controller/crds.yaml
 	$(KUSTOMIZE) build config/rbac > deploy/controller/rbac.yaml

config/rbac/role.yaml

@@ -21,6 +21,14 @@ rules:
   - get
   - list
   - watch
+- apiGroups:
+  - coordination.k8s.io
+  resources:
+  - leases
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resources:

controllers/csiaddons/csiaddonsnode_controller.go

@@ -118,7 +118,7 @@ func (r *CSIAddonsNodeReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	}
 
 	logger.Info("Connecting to sidecar")
-	newConn, err := connection.NewConnection(ctx, endPoint, nodeID, driverName)
+	newConn, err := connection.NewConnection(ctx, endPoint, nodeID, driverName, csiAddonsNode.Namespace, csiAddonsNode.Name)
 	if err != nil {
 		logger.Error(err, "Failed to establish connection with sidecar")
 

controllers/csiaddons/networkfence_controller.go

@@ -109,7 +109,7 @@ func (r *NetworkFenceReconciler) Reconcile(ctx context.Context, req ctrl.Request
 
 	logger = logger.WithValues("DriverName", nwFence.Spec.Driver, "CIDRs", nwFence.Spec.Cidrs)
 
-	client, err := r.getNetworkFenceClient(nwFence.Spec.Driver, "")
+	client, err := r.getNetworkFenceClient(ctx, nwFence.Spec.Driver)
 	if err != nil {
 		logger.Error(err, "Failed to get NetworkFenceClient")
 		return ctrl.Result{}, err
@@ -298,26 +298,29 @@ func (nf *NetworkFenceInstance) removeFinalizerFromNetworkFence(ctx context.Cont
 	return nil
 }
 
-// getNetworkFenceClient returns a NetworkFenceClient for the given driver.
-func (r *NetworkFenceReconciler) getNetworkFenceClient(drivername, nodeID string) (proto.NetworkFenceClient, error) {
-	conns := r.Connpool.GetByNodeID(drivername, nodeID)
-
-	// Iterate through the connections and find the one that matches the driver name
-	// provided in the NetworkFence spec; so that corresponding network fence and
-	// unfence operations can be performed.
-	for _, v := range conns {
-		for _, cap := range v.Capabilities {
-			// validate if NETWORK_FENCE capability is supported by the driver.
-			if cap.GetNetworkFence() == nil {
-				continue
-			}
+// getNetworkFenceClient returns a NetworkFenceClient that is the leader for
+// the given driver.
+// The NetworkFenceClient should only run on a CONTROLLER_SERVICE capable
+// CSI-Addons plugin, there can only be one plugin that holds the lease.
+func (r *NetworkFenceReconciler) getNetworkFenceClient(ctx context.Context, drivername string) (proto.NetworkFenceClient, error) {
+	conn, err := r.Connpool.GetLeaderByDriver(ctx, r.Client, drivername)
+	if err != nil {
+		return nil, err
+	}
 
-			// validate of NETWORK_FENCE capability is enabled by the storage driver.
-			if cap.GetNetworkFence().GetType() == identity.Capability_NetworkFence_NETWORK_FENCE {
-				return proto.NewNetworkFenceClient(v.Client), nil
-			}
+	// verify that the CSI-Addons plugin holding the lease supports
+	// NetworkFence, it probably is a bug if it doesn't
+	for _, capability := range conn.Capabilities {
+		// validate if NETWORK_FENCE capability is supported by the driver.
+		if capability.GetNetworkFence() == nil {
+			continue
+		}
+
+		// validate of NETWORK_FENCE capability is enabled by the storage driver.
+		if capability.GetNetworkFence().GetType() == identity.Capability_NetworkFence_NETWORK_FENCE {
+			return proto.NewNetworkFenceClient(conn.Client), nil
 		}
 	}
 
-	return nil, fmt.Errorf("no connections for driver: %s", drivername)
+	return nil, fmt.Errorf("leading CSIAddonsNode %q for driver %q does not support NetworkFence", conn.Name, drivername)
 }

controllers/csiaddons/reclaimspacejob_controller.go

@@ -343,17 +343,38 @@ func (r *ReclaimSpaceJobReconciler) getTargetDetails(
 	return &details, nil
 }
 
-// getRSClientWithCap returns ReclaimSpaceClient given driverName, nodeID and capabilityType.
-func (r *ReclaimSpaceJobReconciler) getRSClientWithCap(
-	driverName, nodeID string,
-	capType identity.Capability_ReclaimSpace_Type) (string, proto.ReclaimSpaceClient) {
+// getLeadingRSClient returns ReclaimSpaceClient that holds the lease given driverName.
+func (r *ReclaimSpaceJobReconciler) getLeadingRSClient(
+	ctx context.Context,
+	client client.Client,
+	driverName string) (proto.ReclaimSpaceClient, error) {
+	conn, err := r.ConnPool.GetLeaderByDriver(ctx, r.Client, driverName)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, cap := range conn.Capabilities {
+		if cap.GetReclaimSpace() == nil {
+			continue
+		}
+		if cap.GetReclaimSpace().Type == identity.Capability_ReclaimSpace_OFFLINE {
+			return proto.NewReclaimSpaceClient(conn.Client), nil
+		}
+	}
+
+	return nil, fmt.Errorf("leading controller for driver %q does not support ReclaimSpace_OFFLINE", driverName)
+}
+
+// getRSClientByNode returns ReclaimSpaceClient given driverName and nodeID.
+func (r *ReclaimSpaceJobReconciler) getRSClientByNode(
+	driverName, nodeID string) (string, proto.ReclaimSpaceClient) {
 	conns := r.ConnPool.GetByNodeID(driverName, nodeID)
 	for k, v := range conns {
 		for _, cap := range v.Capabilities {
 			if cap.GetReclaimSpace() == nil {
 				continue
 			}
-			if cap.GetReclaimSpace().Type == capType {
+			if cap.GetReclaimSpace().Type == identity.Capability_ReclaimSpace_ONLINE {
 				return k, proto.NewReclaimSpaceClient(v.Client)
 			}
 		}
@@ -372,12 +393,11 @@ func (r *ReclaimSpaceJobReconciler) controllerReclaimSpace(
 	ctx context.Context,
 	logger *logr.Logger,
 	target *targetDetails) (bool, *int64, error) {
-	clientName, controllerClient := r.getRSClientWithCap(target.driverName, "", identity.Capability_ReclaimSpace_OFFLINE)
-	if controllerClient == nil {
-		logger.Info("Controller Client not found")
-		return false, nil, nil
+	controllerClient, err := r.getLeadingRSClient(ctx, r.Client, target.driverName)
+	if err != nil {
+		logger.Info("Controller Client not found: %v", err)
+		return false, nil, err
 	}
-	*logger = logger.WithValues("controllerClient", clientName)
 
 	logger.Info("Making controller reclaim space request")
 	req := &proto.ReclaimSpaceRequest{
@@ -407,10 +427,9 @@ func (r *ReclaimSpaceJobReconciler) nodeReclaimSpace(
 	ctx context.Context,
 	logger *logr.Logger,
 	target *targetDetails) (*int64, error) {
-	clientName, nodeClient := r.getRSClientWithCap(
+	clientName, nodeClient := r.getRSClientByNode(
 		target.driverName,
-		target.nodeID,
-		identity.Capability_ReclaimSpace_ONLINE)
+		target.nodeID)
 	if nodeClient == nil {
 		return nil, fmt.Errorf("node Client not found for %q nodeID", target.nodeID)
 	}

controllers/replication.storage/volumereplication_controller.go

@@ -174,7 +174,7 @@ func (r *VolumeReplicationReconciler) Reconcile(ctx context.Context, req ctrl.Re
 		logger.Info("Replication handle", "ReplicationHandleName", replicationHandle)
 	}
 
-	replicationClient, err := r.getReplicationClient(vrcObj.Spec.Provisioner)
+	replicationClient, err := r.getReplicationClient(ctx, vrcObj.Spec.Provisioner)
 	if err != nil {
 		logger.Error(err, "Failed to get ReplicationClient")
 
@@ -453,27 +453,25 @@ func getInfoReconcileInterval(parameters map[string]string, logger logr.Logger)
 	return scheduleTime / 2
 }
 
-func (r *VolumeReplicationReconciler) getReplicationClient(driverName string) (grpcClient.VolumeReplication, error) {
-	conns := r.Connpool.GetByNodeID(driverName, "")
-
-	// Iterate through the connections and find the one that matches the driver name
-	// provided in the VolumeReplication spec; so that corresponding
-	// operations can be performed.
-	for _, v := range conns {
-		for _, cap := range v.Capabilities {
-			// validate if VOLUME_REPLICATION capability is supported by the driver.
-			if cap.GetVolumeReplication() == nil {
-				continue
-			}
+func (r *VolumeReplicationReconciler) getReplicationClient(ctx context.Context, driverName string) (grpcClient.VolumeReplication, error) {
+	conn, err := r.Connpool.GetLeaderByDriver(ctx, r.Client, driverName)
+	if err != nil {
+		return nil, fmt.Errorf("no leader for the ControllerService of driver %q", driverName)
+	}
 
-			// validate of VOLUME_REPLICATION capability is enabled by the storage driver.
-			if cap.GetVolumeReplication().GetType() == identity.Capability_VolumeReplication_VOLUME_REPLICATION {
-				return grpcClient.NewReplicationClient(v.Client, r.Timeout), nil
-			}
+	for _, cap := range conn.Capabilities {
+		// validate if VOLUME_REPLICATION capability is supported by the driver.
+		if cap.GetVolumeReplication() == nil {
+			continue
+		}
+
+		// validate of VOLUME_REPLICATION capability is enabled by the storage driver.
+		if cap.GetVolumeReplication().GetType() == identity.Capability_VolumeReplication_VOLUME_REPLICATION {
+			return grpcClient.NewReplicationClient(conn.Client, r.Timeout), nil
 		}
 	}
 
-	return nil, fmt.Errorf("no connections for driver: %s", driverName)
+	return nil, fmt.Errorf("leading CSIAddonsNode %q for driver %q does not support VolumeReplication", conn.Name, driverName)
 
 }
 

deploy/controller/rbac.yaml

@@ -71,6 +71,14 @@ rules:
   - get
   - list
   - watch
+- apiGroups:
+  - coordination.k8s.io
+  resources:
+  - leases
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resources:

go.mod

@@ -12,11 +12,11 @@ require (
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/stretchr/testify v1.8.4
 	go.uber.org/zap v1.26.0
-	google.golang.org/grpc v1.59.0
+	google.golang.org/grpc v1.60.0
 	google.golang.org/protobuf v1.31.0
-	k8s.io/api v0.28.4
-	k8s.io/apimachinery v0.28.4
-	k8s.io/client-go v0.28.4
+	k8s.io/api v0.29.0
+	k8s.io/apimachinery v0.29.0
+	k8s.io/client-go v0.29.0
 	k8s.io/klog/v2 v2.110.1
 	sigs.k8s.io/controller-runtime v0.16.3
 )
@@ -65,23 +65,23 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
 	golang.org/x/net v0.18.0 // indirect
-	golang.org/x/oauth2 v0.11.0 // indirect
+	golang.org/x/oauth2 v0.13.0 // indirect
 	golang.org/x/sys v0.14.0 // indirect
 	golang.org/x/term v0.14.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
-	google.golang.org/appengine v1.6.7 // indirect
+	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20231106174013-bbf56f31fb17 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/apiextensions-apiserver v0.28.3 // indirect
 	k8s.io/component-base v0.28.3 // indirect
-	k8s.io/kube-openapi v0.0.0-20230717233707-2695361300d9 // indirect
-	k8s.io/utils v0.0.0-20230406110748-d93618cff8a2 // indirect
+	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
+	k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
-	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
+	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
 )