mover: fetch ceph-csi-config via k8s API instead of volume mount

Makefile

@@ -72,6 +72,11 @@ IMG ?= $(IMAGE_TAG_BASE):v$(VERSION)
 # MOVER_IMG defines the image:tag used for the mover plugin image.
 MOVER_IMG ?= $(MOVER_IMAGE_TAG_BASE):v$(VERSION)
 
+# CEPH_CSI_CONFIG_NAME is the name of the ceph-csi ConfigMap the operator reads.
+CEPH_CSI_CONFIG_NAME ?= ceph-csi-config
+# CEPH_CSI_CONFIG_NAMESPACE is the namespace of the ceph-csi ConfigMap.
+CEPH_CSI_CONFIG_NAMESPACE ?= rook-ceph
+
 # Mover container --build-arg flags
 MOVER_BUILD_ARGS = \
 	--build-arg CEPH_BASE_IMAGE=$(CEPH_BASE_IMAGE) \
@@ -294,7 +299,7 @@ docker-buildx-mover: ## Build and push docker image for the mover for cross-plat
 build-installer: manifests generate kustomize ## Generate a consolidated YAML with CRDs and deployment.
 	mkdir -p dist
 	cd config/manager && $(KUSTOMIZE) edit set image controller=${IMG}
-	$(KUSTOMIZE) build config/default | sed 's|MOVER_IMAGE_PLACEHOLDER|${MOVER_IMG}|g' > dist/install.yaml
+	$(KUSTOMIZE) build config/default | sed 's|MOVER_IMAGE_PLACEHOLDER|${MOVER_IMG}|g; s|CEPH_CSI_CONFIG_NAME_PLACEHOLDER|${CEPH_CSI_CONFIG_NAME}|g; s|CEPH_CSI_CONFIG_NAMESPACE_PLACEHOLDER|${CEPH_CSI_CONFIG_NAMESPACE}|g' > dist/install.yaml
 
 ##@ Deployment
 
@@ -313,7 +318,7 @@ uninstall: ## Uninstall VolSync CRDs from K8s cluster.
 .PHONY: deploy
 deploy: manifests kustomize ## Deploy controller to the K8s cluster specified in ~/.kube/config.
 	cd config/manager && $(KUSTOMIZE) edit set image controller=${IMG}
-	$(KUSTOMIZE) build config/default | sed 's|MOVER_IMAGE_PLACEHOLDER|${MOVER_IMG}|g' | $(KUBECTL) apply -f -
+	$(KUSTOMIZE) build config/default | sed 's|MOVER_IMAGE_PLACEHOLDER|${MOVER_IMG}|g; s|CEPH_CSI_CONFIG_NAME_PLACEHOLDER|${CEPH_CSI_CONFIG_NAME}|g; s|CEPH_CSI_CONFIG_NAMESPACE_PLACEHOLDER|${CEPH_CSI_CONFIG_NAMESPACE}|g' | $(KUBECTL) apply -f -
 
 .PHONY: undeploy
 undeploy: kustomize ## Undeploy controller from the K8s cluster specified in ~/.kube/config. Call with ignore-not-found=true to ignore resource not found errors during deletion.

config/manager/manager.yaml

@@ -59,6 +59,10 @@ spec:
         env:
         - name: MOVER_IMAGE
           value: MOVER_IMAGE_PLACEHOLDER
+        - name: CEPH_CSI_CONFIG_NAME
+          value: CEPH_CSI_CONFIG_NAME_PLACEHOLDER
+        - name: CEPH_CSI_CONFIG_NAMESPACE
+          value: CEPH_CSI_CONFIG_NAMESPACE_PLACEHOLDER
         ports: []
         securityContext:
           allowPrivilegeEscalation: false
@@ -81,18 +85,10 @@ spec:
         # More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/
         resources:
           limits:
-            cpu: 500m
-            memory: 128Mi
+            cpu: 1000m
+            memory: 512Mi
           requests:
             cpu: 10m
             memory: 64Mi
-        volumeMounts:
-        - name: ceph-csi-config
-          mountPath: /etc/ceph-csi-config
-          readOnly: true
-      volumes:
-      - name: ceph-csi-config
-        configMap:
-          name: ceph-csi-config
       serviceAccountName: controller-manager
       terminationGracePeriodSeconds: 10

config/manifests/bases/ceph-volsync-plugin-operator.clusterserviceversion.yaml

@@ -8,7 +8,22 @@ metadata:
   namespace: placeholder
 spec:
   apiservicedefinitions: {}
-  customresourcedefinitions: {}
+  customresourcedefinitions:
+    required:
+    - description: |-
+        A ReplicationDestination is a VolSync resource that you can use to define the destination of a VolSync replication
+        or synchronization.
+      displayName: Replication Destination
+      kind: ReplicationDestination
+      name: replicationdestinations.volsync.backube
+      version: v1alpha1
+    - description: |-
+        A ReplicationSource is a VolSync resource that you can use to define the source PVC and replication mover type,
+        enabling you to replicate or synchronize PVC data to a remote location.
+      displayName: Replication Source
+      kind: ReplicationSource
+      name: replicationsources.volsync.backube
+      version: v1alpha1
   description: Ceph VolSync Plugin Operator acts on VolSync's Custom Resources but
     uses Ceph's internal mechanisms to perform significantly more efficient data replication.
   displayName: Ceph VolSync Plugin Operator

internal/ceph/config/csiconfig.go

@@ -71,20 +71,14 @@ const (
 	}
 }]
 */
-func readClusterInfo(pathToConfig, clusterID string) (*kubernetes.ClusterInfo, error) {
+// ReadClusterInfoFromData parses raw JSON config data and returns the
+// ClusterInfo for the given clusterID. Use this when the config content
+// is already available (e.g. fetched from a ConfigMap via the k8s API).
+func ReadClusterInfoFromData(data []byte, clusterID string) (*kubernetes.ClusterInfo, error) {
 	var config []kubernetes.ClusterInfo
 
-	// #nosec
-	content, err := os.ReadFile(pathToConfig)
-	if err != nil {
-		err = fmt.Errorf("error fetching configuration for cluster ID %q: %w", clusterID, err)
-
-		return nil, err
-	}
-
-	err = json.Unmarshal(content, &config)
-	if err != nil {
-		return nil, fmt.Errorf("unmarshal failed (%w), raw buffer response: %s", err, string(content))
+	if err := json.Unmarshal(data, &config); err != nil {
+		return nil, fmt.Errorf("unmarshal failed (%w), raw buffer response: %s", err, string(data))
 	}
 
 	for i := range config {
@@ -96,6 +90,16 @@ func readClusterInfo(pathToConfig, clusterID string) (*kubernetes.ClusterInfo, e
 	return nil, fmt.Errorf("%w: %q", ErrConfigNotFound, clusterID)
 }
 
+func readClusterInfo(pathToConfig, clusterID string) (*kubernetes.ClusterInfo, error) {
+	// #nosec
+	content, err := os.ReadFile(pathToConfig)
+	if err != nil {
+		return nil, fmt.Errorf("error fetching configuration for cluster ID %q: %w", clusterID, err)
+	}
+
+	return ReadClusterInfoFromData(content, clusterID)
+}
+
 // Mons returns a comma separated MON list from the csi config
 // for the given clusterID.
 func Mons(pathToConfig, clusterID string) (string, error) {
@@ -186,3 +190,29 @@ func GetCephFSControllerPublishSecretRef(pathToConfig, clusterID string) (string
 
 	return secretRef.Name, secretRef.Namespace, nil
 }
+
+// GetRBDControllerPublishSecretRefFromData returns the secret name and
+// namespace for RBD controller publish operations, parsing from raw JSON data.
+func GetRBDControllerPublishSecretRefFromData(data []byte, clusterID string) (string, string, error) {
+	cluster, err := ReadClusterInfoFromData(data, clusterID)
+	if err != nil {
+		return "", "", err
+	}
+
+	secretRef := cluster.RBD.ControllerPublishSecretRef
+
+	return secretRef.Name, secretRef.Namespace, nil
+}
+
+// GetCephFSControllerPublishSecretRefFromData returns the secret name and
+// namespace for CephFS controller publish operations, parsing from raw JSON data.
+func GetCephFSControllerPublishSecretRefFromData(data []byte, clusterID string) (string, string, error) {
+	cluster, err := ReadClusterInfoFromData(data, clusterID)
+	if err != nil {
+		return "", "", err
+	}
+
+	secretRef := cluster.CephFS.ControllerPublishSecretRef
+
+	return secretRef.Name, secretRef.Namespace, nil
+}

internal/ceph/config/csiconfig_test.go

@@ -139,6 +139,94 @@ func TestCSIConfig(t *testing.T) {
 	}
 }
 
+func TestReadClusterInfoFromData(t *testing.T) {
+	t.Parallel()
+	data := []byte(`[{"clusterID":"c1","monitors":["mon1","mon2"]},{"clusterID":"c2","monitors":["mon3"]}]`)
+
+	info, err := ReadClusterInfoFromData(data, "c1")
+	if err != nil {
+		t.Fatalf("ReadClusterInfoFromData() error: %v", err)
+	}
+	if info.ClusterID != "c1" {
+		t.Errorf("ClusterID = %q, want %q", info.ClusterID, "c1")
+	}
+	if len(info.Monitors) != 2 {
+		t.Errorf("Monitors count = %d, want 2", len(info.Monitors))
+	}
+
+	_, err = ReadClusterInfoFromData(data, "missing")
+	if err == nil {
+		t.Error("expected error for missing clusterID")
+	}
+
+	_, err = ReadClusterInfoFromData([]byte("invalid"), "c1")
+	if err == nil {
+		t.Error("expected error for invalid JSON")
+	}
+}
+
+func TestGetRBDControllerPublishSecretRefFromData(t *testing.T) {
+	t.Parallel()
+	csiConfig := []cephcsi.ClusterInfo{
+		{
+			ClusterID: "cluster-1",
+			RBD: cephcsi.RBD{
+				ControllerPublishSecretRef: corev1.SecretReference{
+					Name: "rbd-secret-1", Namespace: "ceph-csi",
+				},
+			},
+		},
+	}
+	data, err := json.Marshal(csiConfig)
+	if err != nil {
+		t.Fatalf("json.Marshal() error: %v", err)
+	}
+
+	name, ns, err := GetRBDControllerPublishSecretRefFromData(data, "cluster-1")
+	if err != nil {
+		t.Fatalf("GetRBDControllerPublishSecretRefFromData() error: %v", err)
+	}
+	if name != "rbd-secret-1" || ns != "ceph-csi" {
+		t.Errorf("got (%q, %q), want (%q, %q)", name, ns, "rbd-secret-1", "ceph-csi")
+	}
+
+	_, _, err = GetRBDControllerPublishSecretRefFromData(data, "missing")
+	if err == nil {
+		t.Error("expected error for missing clusterID")
+	}
+}
+
+func TestGetCephFSControllerPublishSecretRefFromData(t *testing.T) {
+	t.Parallel()
+	csiConfig := []cephcsi.ClusterInfo{
+		{
+			ClusterID: "cluster-1",
+			CephFS: cephcsi.CephFS{
+				ControllerPublishSecretRef: corev1.SecretReference{
+					Name: "cephfs-secret-1", Namespace: "ceph-csi",
+				},
+			},
+		},
+	}
+	data, err := json.Marshal(csiConfig)
+	if err != nil {
+		t.Fatalf("json.Marshal() error: %v", err)
+	}
+
+	name, ns, err := GetCephFSControllerPublishSecretRefFromData(data, "cluster-1")
+	if err != nil {
+		t.Fatalf("GetCephFSControllerPublishSecretRefFromData() error: %v", err)
+	}
+	if name != "cephfs-secret-1" || ns != "ceph-csi" {
+		t.Errorf("got (%q, %q), want (%q, %q)", name, ns, "cephfs-secret-1", "ceph-csi")
+	}
+
+	_, _, err = GetCephFSControllerPublishSecretRefFromData(data, "missing")
+	if err == nil {
+		t.Error("expected error for missing clusterID")
+	}
+}
+
 func TestGetRBDControllerPublishSecretRef(t *testing.T) {
 	t.Parallel()
 	tests := []struct {