Merge pull request #296 from red-hat-storage/sync_us--devel

Syncing latest changes from upstream devel for ceph-csi

charts/ceph-csi-rbd/values.yaml

@@ -27,6 +27,7 @@ serviceAccounts:
 #       - "<MONValue2>"
 #     rbd:
 #       netNamespaceFilePath: "{{ .kubeletDir }}/plugins/{{ .driverName }}/net"
+#       mirrorDaemonCount: 1
 #     readAffinity:
 #       enabled: true
 #       crushLocationLabels:

deploy/csi-config-map-sample.yaml

@@ -19,6 +19,8 @@ kind: ConfigMap
 # NOTE: The given radosNamespace must already exists in the pool.
 # NOTE: Make sure you don't add radosNamespace option to a currently in use
 # configuration as it will cause issues.
+# The "rbd.mirrorDaemonCount" is optional and represents the total number of
+# RBD mirror daemons running on the ceph cluster.
 # The field "cephFS.subvolumeGroup" is optional and defaults to "csi".
 # NOTE: The given subvolumeGroup must already exist in the filesystem.
 # The "cephFS.netNamespaceFilePath" fields are the various network namespace
@@ -64,6 +66,7 @@ data:
         "rbd": {
            "netNamespaceFilePath": "<kubeletRootPath>/plugins/rbd.csi.ceph.com/net",
            "radosNamespace": "<rados-namespace>",
+           "mirrorDaemonCount": 1,
         },
         "monitors": [
           "<MONValue1>",

internal/rbd/rbd_util.go

@@ -553,9 +553,13 @@ func (ri *rbdImage) isInUse() (bool, error) {
 	// because we opened the image, there is at least one watcher
 	defaultWatchers := 1
 	if mirrorInfo.Primary {
+		count, err := util.GetRBDMirrorDaemonCount(util.CsiConfigFile, ri.ClusterID)
+		if err != nil {
+			return false, err
+		}
 		// if rbd mirror daemon is running, a watcher will be added by the rbd
 		// mirror daemon for mirrored images.
-		defaultWatchers++
+		defaultWatchers += count
 	}
 
 	return len(watchers) > defaultWatchers, nil

internal/util/csiconfig.go

@@ -45,6 +45,7 @@ const (
 	"clusterID": "<cluster-id>",
 	"rbd": {
 		"radosNamespace": "<rados-namespace>"
+		"mirrorDaemonCount": 1
 	},
 	"monitors": [
 		"<monitor-value>",
@@ -105,6 +106,22 @@ func GetRadosNamespace(pathToConfig, clusterID string) (string, error) {
 	return cluster.RBD.RadosNamespace, nil
 }
 
+// GetRBDMirrorDaemonCount returns the number of mirror daemon count for the
+// given clusterID.
+func GetRBDMirrorDaemonCount(pathToConfig, clusterID string) (int, error) {
+	cluster, err := readClusterInfo(pathToConfig, clusterID)
+	if err != nil {
+		return 0, err
+	}
+
+	// if it is empty, set the default to 1 which is most common in a cluster.
+	if cluster.RBD.MirrorDaemonCount == 0 {
+		return 1, nil
+	}
+
+	return cluster.RBD.MirrorDaemonCount, nil
+}
+
 // CephFSSubvolumeGroup returns the subvolumeGroup for CephFS volumes. If not set, it returns the default value "csi".
 func CephFSSubvolumeGroup(pathToConfig, clusterID string) (string, error) {
 	cluster, err := readClusterInfo(pathToConfig, clusterID)

internal/util/csiconfig_test.go

@@ -17,11 +17,14 @@ limitations under the License.
 package util
 
 import (
+	"bytes"
 	"encoding/json"
 	"os"
 	"testing"
 
 	cephcsi "github.com/ceph/ceph-csi/api/deploy/kubernetes"
+
+	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -530,3 +533,88 @@ func TestGetCephFSMountOptions(t *testing.T) {
 		})
 	}
 }
+
+func TestGetRBDMirrorDaemonCount(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name      string
+		clusterID string
+		want      int
+	}{
+		{
+			name:      "get rbd mirror daemon count for cluster-1",
+			clusterID: "cluster-1",
+			want:      2,
+		},
+		{
+			name:      "get rbd mirror daemon count for cluster-2",
+			clusterID: "cluster-2",
+			want:      4,
+		},
+		{
+			name:      "when rbd mirror daemon count is empty",
+			clusterID: "cluster-3",
+			want:      1, // default mirror daemon count
+		},
+	}
+
+	csiConfig := []cephcsi.ClusterInfo{
+		{
+			ClusterID: "cluster-1",
+			Monitors:  []string{"ip-1", "ip-2"},
+			RBD: cephcsi.RBD{
+				MirrorDaemonCount: 2,
+			},
+		},
+		{
+			ClusterID: "cluster-2",
+			Monitors:  []string{"ip-3", "ip-4"},
+			RBD: cephcsi.RBD{
+				MirrorDaemonCount: 4,
+			},
+		},
+		{
+			ClusterID: "cluster-3",
+			Monitors:  []string{"ip-5", "ip-6"},
+		},
+	}
+	csiConfigFileContent, err := json.Marshal(csiConfig)
+	if err != nil {
+		t.Errorf("failed to marshal csi config info %v", err)
+	}
+	tmpConfPath := t.TempDir() + "/ceph-csi.json"
+	err = os.WriteFile(tmpConfPath, csiConfigFileContent, 0o600)
+	if err != nil {
+		t.Errorf("failed to write %s file content: %v", CsiConfigFile, err)
+	}
+	for _, tt := range tests {
+		ts := tt
+		t.Run(ts.name, func(t *testing.T) {
+			t.Parallel()
+			var got int
+			got, err = GetRBDMirrorDaemonCount(tmpConfPath, ts.clusterID)
+			if err != nil {
+				t.Errorf("GetRBDMirrorDaemonCount() error = %v", err)
+
+				return
+			}
+			if got != ts.want {
+				t.Errorf("GetRBDMirrorDaemonCount() = %v, want %v", got, ts.want)
+			}
+		})
+	}
+
+	// when mirrorDaemonCount is set as string
+	csiConfigFileContent = bytes.Replace(
+		csiConfigFileContent,
+		[]byte(`"mirrorDaemonCount":2`),
+		[]byte(`"mirrorDaemonCount":"2"`),
+		1)
+	tmpCSIConfPath := t.TempDir() + "/ceph-csi.json"
+	err = os.WriteFile(tmpCSIConfPath, csiConfigFileContent, 0o600)
+	if err != nil {
+		t.Errorf("failed to write %s file content: %v", CsiConfigFile, err)
+	}
+	_, err = GetRBDMirrorDaemonCount(tmpCSIConfPath, "test")
+	require.Error(t, err)
+}