Merge pull request #9484 from leseb/subvolgr-csi-glue

csi: change rook-ceph-csi-config to expose clusterID for subvolume

pkg/operator/ceph/cluster/cluster.go

@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"os/exec"
 	"path"
-	"sync"
 	"syscall"
 
 	"github.com/pkg/errors"
@@ -67,7 +66,7 @@ type clusterHealth struct {
 	internalCancel context.CancelFunc
 }
 
-func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync.Mutex, ownerInfo *k8sutil.OwnerInfo) *cluster {
+func newCluster(c *cephv1.CephCluster, context *clusterd.Context, ownerInfo *k8sutil.OwnerInfo) *cluster {
 	return &cluster{
 		// at this phase of the cluster creation process, the identity components of the cluster are
 		// not yet established. we reserve this struct which is filled in as soon as the cluster's
@@ -79,7 +78,7 @@ func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync
 		namespacedName:     types.NamespacedName{Namespace: c.Namespace, Name: c.Name},
 		monitoringRoutines: make(map[string]*clusterHealth),
 		ownerInfo:          ownerInfo,
-		mons:               mon.New(context, c.Namespace, c.Spec, ownerInfo, csiMutex),
+		mons:               mon.New(context, c.Namespace, c.Spec, ownerInfo),
 	}
 }
 

pkg/operator/ceph/cluster/cluster_external.go

@@ -111,7 +111,7 @@ func (c *ClusterController) configureExternalCephCluster(cluster *cluster) error
 	}
 
 	// Save CSI configmap
-	err = csi.SaveClusterConfig(c.context.Clientset, c.namespacedName.Namespace, cluster.ClusterInfo, c.csiConfigMutex)
+	err = csi.SaveClusterConfig(c.context.Clientset, c.namespacedName.Namespace, cluster.ClusterInfo, &csi.CsiClusterConfigEntry{Monitors: csi.MonEndpoints(cluster.ClusterInfo.Monitors)})
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi cluster config")
 	}

pkg/operator/ceph/cluster/controller.go

@@ -21,7 +21,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"sync"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -85,7 +84,6 @@ type ClusterController struct {
 	context        *clusterd.Context
 	rookImage      string
 	clusterMap     map[string]*cluster
-	csiConfigMutex *sync.Mutex
 	osdChecker     *osd.OSDHealthMonitor
 	client         client.Client
 	namespacedName types.NamespacedName
@@ -327,10 +325,9 @@ func (r *ReconcileCephCluster) reconcileDelete(cephCluster *cephv1.CephCluster)
 // NewClusterController create controller for watching cluster custom resources created
 func NewClusterController(context *clusterd.Context, rookImage string) *ClusterController {
 	return &ClusterController{
-		context:        context,
-		rookImage:      rookImage,
-		clusterMap:     make(map[string]*cluster),
-		csiConfigMutex: &sync.Mutex{},
+		context:    context,
+		rookImage:  rookImage,
+		clusterMap: make(map[string]*cluster),
 	}
 }
 
@@ -343,7 +340,7 @@ func (c *ClusterController) reconcileCephCluster(clusterObj *cephv1.CephCluster,
 	cluster, ok := c.clusterMap[clusterObj.Namespace]
 	if !ok {
 		// It's a new cluster so let's populate the struct
-		cluster = newCluster(clusterObj, c.context, c.csiConfigMutex, ownerInfo)
+		cluster = newCluster(clusterObj, c.context, ownerInfo)
 	}
 	cluster.namespacedName = c.namespacedName
 

pkg/operator/ceph/cluster/mon/drain_test.go

@@ -18,7 +18,6 @@ package mon
 
 import (
 	"context"
-	"sync"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -48,7 +47,7 @@ func createFakeCluster(t *testing.T, cephClusterObj *cephv1.CephCluster, k8sVers
 
 	cl := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects().Build()
 	clientset := test.New(t, 3)
-	c := New(&clusterd.Context{Client: cl, Clientset: clientset}, mockNamespace, cephClusterObj.Spec, ownerInfo, &sync.Mutex{})
+	c := New(&clusterd.Context{Client: cl, Clientset: clientset}, mockNamespace, cephClusterObj.Spec, ownerInfo)
 	test.SetFakeKubernetesVersion(clientset, k8sVersion)
 	return c
 }

pkg/operator/ceph/cluster/mon/health_test.go

@@ -22,7 +22,6 @@ import (
 	"io/ioutil"
 	"os"
 	"reflect"
-	"sync"
 	"testing"
 	"time"
 
@@ -66,7 +65,7 @@ func TestCheckHealth(t *testing.T) {
 		Executor:  executor,
 	}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
 	// clusterInfo is nil so we return err
 	err := c.checkHealth(ctx)
 	assert.NotNil(t, err)
@@ -155,7 +154,7 @@ func TestCheckHealth(t *testing.T) {
 }
 
 func TestSkipMonFailover(t *testing.T) {
-	c := New(&clusterd.Context{}, "ns", cephv1.ClusterSpec{}, nil, nil)
+	c := New(&clusterd.Context{}, "ns", cephv1.ClusterSpec{}, nil)
 	c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
 	monName := "arb"
 
@@ -201,7 +200,7 @@ func TestEvictMonOnSameNode(t *testing.T) {
 	}
 	context := &clusterd.Context{Clientset: clientset, ConfigDir: configDir, Executor: executor}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0}, "myversion")
 	c.maxMonID = 2
 	c.waitForStart = false
@@ -256,7 +255,7 @@ func TestScaleMonDeployment(t *testing.T) {
 	clientset := test.New(t, 1)
 	context := &clusterd.Context{Clientset: clientset}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0, AllowMultiplePerNode: true}, "myversion")
 
 	name := "a"
@@ -307,7 +306,7 @@ func TestCheckHealthNotFound(t *testing.T) {
 		Executor:  executor,
 	}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
 	setCommonMonProperties(c, 2, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 	c.waitForStart = false
 	defer os.RemoveAll(c.context.ConfigDir)
@@ -370,7 +369,7 @@ func TestAddRemoveMons(t *testing.T) {
 		Executor:  executor,
 	}
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
+	c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo)
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
 	c.maxMonID = 0 // "a" is max mon id
 	c.waitForStart = false

pkg/operator/ceph/cluster/mon/mon.go

@@ -120,7 +120,6 @@ type Cluster struct {
 	monTimeoutList     map[string]time.Time
 	mapping            *Mapping
 	ownerInfo          *k8sutil.OwnerInfo
-	csiConfigMutex     *sync.Mutex
 	isUpgrade          bool
 	arbiterMon         string
 }
@@ -164,7 +163,7 @@ type SchedulingResult struct {
 }
 
 // New creates an instance of a mon cluster
-func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, ownerInfo *k8sutil.OwnerInfo, csiConfigMutex *sync.Mutex) *Cluster {
+func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, ownerInfo *k8sutil.OwnerInfo) *Cluster {
 	return &Cluster{
 		context:        context,
 		spec:           spec,
@@ -175,8 +174,7 @@ func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, o
 		mapping: &Mapping{
 			Schedule: map[string]*MonScheduleInfo{},
 		},
-		ownerInfo:      ownerInfo,
-		csiConfigMutex: csiConfigMutex,
+		ownerInfo: ownerInfo,
 	}
 }
 
@@ -1042,7 +1040,7 @@ func (c *Cluster) saveMonConfig() error {
 		return errors.Wrap(err, "failed to write connection config for new mons")
 	}
 
-	if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, c.csiConfigMutex); err != nil {
+	if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, &csi.CsiClusterConfigEntry{Monitors: csi.MonEndpoints(c.ClusterInfo.Monitors)}); err != nil {
 		return errors.Wrap(err, "failed to update csi cluster config")
 	}
 

pkg/operator/ceph/cluster/mon/mon_test.go

@@ -25,7 +25,6 @@ import (
 	"reflect"
 	"strconv"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -273,7 +272,7 @@ func validateStart(t *testing.T, c *Cluster) {
 func TestPersistMons(t *testing.T) {
 	clientset := test.New(t, 1)
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{Annotations: cephv1.AnnotationsSpec{cephv1.KeyClusterMetadata: cephv1.Annotations{"key": "value"}}}, ownerInfo, &sync.Mutex{})
+	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{Annotations: cephv1.AnnotationsSpec{cephv1.KeyClusterMetadata: cephv1.Annotations{"key": "value"}}}, ownerInfo)
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
 	// Persist mon a
@@ -302,7 +301,7 @@ func TestSaveMonEndpoints(t *testing.T) {
 	configDir, _ := ioutil.TempDir("", "")
 	defer os.RemoveAll(configDir)
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
+	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo)
 	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
 	// create the initial config map
@@ -350,7 +349,7 @@ func TestMaxMonID(t *testing.T) {
 	configDir, _ := ioutil.TempDir("", "")
 	defer os.RemoveAll(configDir)
 	ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
-	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{})
+	c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo)
 	c.ClusterInfo = clienttest.CreateTestClusterInfo(1)
 
 	// when the configmap is not found, the maxMonID is -1

pkg/operator/ceph/cluster/mon/node_test.go

@@ -19,7 +19,6 @@ package mon
 import (
 	"context"
 	"strings"
-	"sync"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -37,7 +36,7 @@ import (
 func TestNodeAffinity(t *testing.T) {
 	ctx := context.TODO()
 	clientset := test.New(t, 4)
-	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{})
+	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{})
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
 	c.spec.Placement = map[cephv1.KeyType]cephv1.Placement{}
@@ -167,7 +166,7 @@ func TestPodMemory(t *testing.T) {
 
 func TestHostNetwork(t *testing.T) {
 	clientset := test.New(t, 3)
-	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{})
+	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{})
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
 
 	c.spec.Network.HostNetwork = true

pkg/operator/ceph/cluster/mon/service_test.go

@@ -18,7 +18,6 @@ package mon
 
 import (
 	"context"
-	"sync"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -33,7 +32,7 @@ import (
 func TestCreateService(t *testing.T) {
 	ctx := context.TODO()
 	clientset := test.New(t, 1)
-	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{})
+	c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{})
 	c.ClusterInfo = client.AdminTestClusterInfo("rook-ceph")
 	m := &monConfig{ResourceName: "rook-ceph-mon-b", DaemonName: "b"}
 	clusterIP, err := c.createService(m)

pkg/operator/ceph/cluster/mon/spec_test.go

@@ -17,7 +17,6 @@ limitations under the License.
 package mon
 
 import (
-	"sync"
 	"testing"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -48,7 +47,6 @@ func testPodSpec(t *testing.T, monID string, pvc bool) {
 		"ns",
 		cephv1.ClusterSpec{},
 		ownerInfo,
-		&sync.Mutex{},
 	)
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "rook/rook:myversion")
 	c.spec.CephVersion = cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:myceph"}
@@ -98,7 +96,6 @@ func TestDeploymentPVCSpec(t *testing.T) {
 		"ns",
 		cephv1.ClusterSpec{},
 		ownerInfo,
-		&sync.Mutex{},
 	)
 	setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "rook/rook:myversion")
 	c.spec.CephVersion = cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:myceph"}
@@ -158,7 +155,6 @@ func testRequiredDuringScheduling(t *testing.T, hostNetwork, allowMultiplePerNod
 		"ns",
 		cephv1.ClusterSpec{},
 		&k8sutil.OwnerInfo{},
-		&sync.Mutex{},
 	)
 
 	c.spec.Network.HostNetwork = hostNetwork

pkg/operator/ceph/csi/cluster_config.go

@@ -33,15 +33,22 @@ import (
 )
 
 var (
-	logger = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-csi")
+	logger      = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-csi")
+	configMutex sync.Mutex
 )
 
-type csiClusterConfigEntry struct {
-	ClusterID string   `json:"clusterID"`
-	Monitors  []string `json:"monitors"`
+type CsiClusterConfigEntry struct {
+	ClusterID      string         `json:"clusterID"`
+	Monitors       []string       `json:"monitors"`
+	CephFS         *CsiCephFSSpec `json:"cephFS,omitempty"`
+	RadosNamespace string         `json:"radosNamespace,omitempty"`
 }
 
-type csiClusterConfig []csiClusterConfigEntry
+type CsiCephFSSpec struct {
+	SubvolumeGroup string `json:"subvolumeGroup,omitempty"`
+}
+
+type csiClusterConfig []CsiClusterConfigEntry
 
 // FormatCsiClusterConfig returns a json-formatted string containing
 // the cluster-to-mon mapping required to configure ceph csi.
@@ -79,7 +86,7 @@ func formatCsiClusterConfig(cc csiClusterConfig) (string, error) {
 	return string(ccJson), nil
 }
 
-func monEndpoints(mons map[string]*cephclient.MonInfo) []string {
+func MonEndpoints(mons map[string]*cephclient.MonInfo) []string {
 	endpoints := make([]string, 0)
 	for _, m := range mons {
 		endpoints = append(endpoints, m.Endpoint)
@@ -89,32 +96,59 @@ func monEndpoints(mons map[string]*cephclient.MonInfo) []string {
 
 // updateCsiClusterConfig returns a json-formatted string containing
 // the cluster-to-mon mapping required to configure ceph csi.
-func updateCsiClusterConfig(
-	curr, clusterKey string, mons map[string]*cephclient.MonInfo) (string, error) {
-
+func updateCsiClusterConfig(curr, clusterKey string, newCsiClusterConfigEntry *CsiClusterConfigEntry) (string, error) {
 	var (
 		cc     csiClusterConfig
-		centry csiClusterConfigEntry
+		centry CsiClusterConfigEntry
 		found  bool
 	)
+
 	cc, err := parseCsiClusterConfig(curr)
 	if err != nil {
 		return "", errors.Wrap(err, "failed to parse current csi cluster config")
 	}
 
+	// Regardless of which controllers call updateCsiClusterConfig(), the values will be preserved since
+	// a lock is acquired for the update operation. So concurrent updates (rare event) will block and
+	// wait for the other update to complete. Monitors and Subvolumegroup will be updated
+	// independently and won't collide.
 	for i, centry := range cc {
 		if centry.ClusterID == clusterKey {
-			centry.Monitors = monEndpoints(mons)
+			// If the new entry is nil, this means the entry is being deleted so remove it from the list
+			if newCsiClusterConfigEntry == nil {
+				cc = append(cc[:i], cc[i+1:]...)
+				found = true
+				break
+			}
+			centry.Monitors = newCsiClusterConfigEntry.Monitors
+			if newCsiClusterConfigEntry.CephFS != nil && newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" {
+				centry.CephFS = newCsiClusterConfigEntry.CephFS
+			}
+			if newCsiClusterConfigEntry.RadosNamespace != "" {
+				centry.RadosNamespace = newCsiClusterConfigEntry.RadosNamespace
+			}
 			found = true
 			cc[i] = centry
 			break
 		}
 	}
 	if !found {
-		centry.ClusterID = clusterKey
-		centry.Monitors = monEndpoints(mons)
-		cc = append(cc, centry)
+		// If it's the first time we create the cluster, the entry does not exist, so the removal
+		// will fail with a dangling pointer
+		if newCsiClusterConfigEntry != nil {
+			centry.ClusterID = clusterKey
+			centry.Monitors = newCsiClusterConfigEntry.Monitors
+			// Add a condition not to fill with empty values
+			if newCsiClusterConfigEntry.CephFS != nil && newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" {
+				centry.CephFS = newCsiClusterConfigEntry.CephFS
+			}
+			if newCsiClusterConfigEntry.RadosNamespace != "" {
+				centry.RadosNamespace = newCsiClusterConfigEntry.RadosNamespace
+			}
+			cc = append(cc, centry)
+		}
 	}
+
 	return formatCsiClusterConfig(cc)
 }
 
@@ -155,24 +189,24 @@ func CreateCsiConfigMap(namespace string, clientset kubernetes.Interface, ownerI
 // value that is provided to ceph-csi uses in the storage class.
 // The locker l is typically a mutex and is used to prevent the config
 // map from being updated for multiple clusters simultaneously.
-func SaveClusterConfig(
-	clientset kubernetes.Interface, clusterNamespace string,
-	clusterInfo *cephclient.ClusterInfo, l sync.Locker) error {
+func SaveClusterConfig(clientset kubernetes.Interface, clusterNamespace string, clusterInfo *cephclient.ClusterInfo, newCsiClusterConfigEntry *CsiClusterConfigEntry) error {
 	if !CSIEnabled() {
+		logger.Debug("skipping csi config update because csi is not enabled")
 		return nil
 	}
-	l.Lock()
-	defer l.Unlock()
+
+	configMutex.Lock()
+	defer configMutex.Unlock()
+
 	// csi is deployed into the same namespace as the operator
 	csiNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
 	if csiNamespace == "" {
-		return errors.Errorf("namespace value missing for %s", k8sutil.PodNamespaceEnvVar)
+		return errors.Errorf("namespace value missing for %q", k8sutil.PodNamespaceEnvVar)
 	}
-	logger.Debugf("Using %+v for CSI ConfigMap Namespace", csiNamespace)
+	logger.Debugf("using %q for csi configmap namespace", csiNamespace)
 
 	// fetch current ConfigMap contents
-	configMap, err := clientset.CoreV1().ConfigMaps(csiNamespace).Get(clusterInfo.Context,
-		ConfigName, metav1.GetOptions{})
+	configMap, err := clientset.CoreV1().ConfigMaps(csiNamespace).Get(clusterInfo.Context, ConfigName, metav1.GetOptions{})
 	if err != nil {
 		return errors.Wrap(err, "failed to fetch current csi config map")
 	}
@@ -182,8 +216,8 @@ func SaveClusterConfig(
 	if currData == "" {
 		currData = "[]"
 	}
-	newData, err := updateCsiClusterConfig(
-		currData, clusterNamespace, clusterInfo.Monitors)
+
+	newData, err := updateCsiClusterConfig(currData, clusterNamespace, newCsiClusterConfigEntry)
 	if err != nil {
 		return errors.Wrap(err, "failed to update csi config map data")
 	}
@@ -191,7 +225,7 @@ func SaveClusterConfig(
 
 	// update ConfigMap with new contents
 	if _, err := clientset.CoreV1().ConfigMaps(csiNamespace).Update(clusterInfo.Context, configMap, metav1.UpdateOptions{}); err != nil {
-		return errors.Wrapf(err, "failed to update csi config map")
+		return errors.Wrap(err, "failed to update csi config map")
 	}
 
 	return nil

pkg/operator/ceph/csi/cluster_config_test.go

@@ -19,71 +19,142 @@ package csi
 import (
 	"testing"
 
-	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestUpdateCsiClusterConfig(t *testing.T) {
-	// initialize an empty list & add a simple mons list
-	mons := map[string]*cephclient.MonInfo{
-		"foo": {Name: "foo", Endpoint: "1.2.3.4:5000"},
+	csiClusterConfigEntry := CsiClusterConfigEntry{
+		Monitors: []string{"1.2.3.4:5000"},
 	}
-	s, err := updateCsiClusterConfig("[]", "alpha", mons)
-	assert.NoError(t, err)
-	assert.Equal(t, s,
-		`[{"clusterID":"alpha","monitors":["1.2.3.4:5000"]}]`)
-
-	// add a 2nd mon to the current cluster
-	mons["bar"] = &cephclient.MonInfo{
-		Name: "bar", Endpoint: "10.11.12.13:5000"}
-	s, err = updateCsiClusterConfig(s, "alpha", mons)
-	assert.NoError(t, err)
-	cc, err := parseCsiClusterConfig(s)
-	assert.NoError(t, err)
-	assert.Equal(t, len(cc), 1)
-	assert.Equal(t, cc[0].ClusterID, "alpha")
-	assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
-	assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
-	assert.Equal(t, len(cc[0].Monitors), 2)
-
-	// add a 2nd cluster with 3 mons
-	mons2 := map[string]*cephclient.MonInfo{
-		"flim": {Name: "flim", Endpoint: "20.1.1.1:5000"},
-		"flam": {Name: "flam", Endpoint: "20.1.1.2:5000"},
-		"blam": {Name: "blam", Endpoint: "20.1.1.3:5000"},
+	csiClusterConfigEntry2 := CsiClusterConfigEntry{
+		Monitors: []string{"20.1.1.1:5000", "20.1.1.2:5000", "20.1.1.3:5000"},
 	}
-	s, err = updateCsiClusterConfig(s, "beta", mons2)
-	assert.NoError(t, err)
-	cc, err = parseCsiClusterConfig(s)
-	assert.NoError(t, err)
-	assert.Equal(t, len(cc), 2)
-	assert.Equal(t, cc[0].ClusterID, "alpha")
-	assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
-	assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
-	assert.Equal(t, len(cc[0].Monitors), 2)
-	assert.Equal(t, cc[1].ClusterID, "beta")
-	assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000")
-	assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000")
-	assert.Contains(t, cc[1].Monitors, "20.1.1.3:5000")
-	assert.Equal(t, len(cc[1].Monitors), 3)
+	csiClusterConfigEntry3 := CsiClusterConfigEntry{
+		Monitors: []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"},
+		CephFS: &CsiCephFSSpec{
+			SubvolumeGroup: "mygroup",
+		},
+	}
+	var s string
+	var err error
 
-	// remove a mon from the 2nd cluster
-	delete(mons2, "blam")
-	s, err = updateCsiClusterConfig(s, "beta", mons2)
-	assert.NoError(t, err)
-	cc, err = parseCsiClusterConfig(s)
-	assert.NoError(t, err)
-	assert.Equal(t, len(cc), 2)
-	assert.Equal(t, cc[0].ClusterID, "alpha")
-	assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
-	assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
-	assert.Equal(t, len(cc[0].Monitors), 2)
-	assert.Equal(t, cc[1].ClusterID, "beta")
-	assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000")
-	assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000")
-	assert.Equal(t, len(cc[1].Monitors), 2)
+	t.Run("add a simple mons list", func(t *testing.T) {
+		s, err = updateCsiClusterConfig("[]", "alpha", &csiClusterConfigEntry)
+		assert.NoError(t, err)
+		assert.Equal(t, s,
+			`[{"clusterID":"alpha","monitors":["1.2.3.4:5000"]}]`)
+	})
 
-	// does it return error on garbage input?
-	_, err = updateCsiClusterConfig("qqq", "beta", mons2)
-	assert.Error(t, err)
+	t.Run("add a 2nd mon to the current cluster", func(t *testing.T) {
+		csiClusterConfigEntry.Monitors = append(csiClusterConfigEntry.Monitors, "10.11.12.13:5000")
+		s, err = updateCsiClusterConfig(s, "alpha", &csiClusterConfigEntry)
+		assert.NoError(t, err)
+		cc, err := parseCsiClusterConfig(s)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(cc))
+		assert.Equal(t, "alpha", cc[0].ClusterID)
+		assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
+		assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
+		assert.Equal(t, 2, len(cc[0].Monitors))
+	})
+
+	t.Run("add a 2nd cluster with 3 mons", func(t *testing.T) {
+		s, err = updateCsiClusterConfig(s, "beta", &csiClusterConfigEntry2)
+		assert.NoError(t, err)
+		cc, err := parseCsiClusterConfig(s)
+		assert.NoError(t, err)
+		assert.Equal(t, 2, len(cc))
+		assert.Equal(t, "alpha", cc[0].ClusterID)
+		assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
+		assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
+		assert.Equal(t, 2, len(cc[0].Monitors))
+		assert.Equal(t, "beta", cc[1].ClusterID)
+		assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000")
+		assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000")
+		assert.Contains(t, cc[1].Monitors, "20.1.1.3:5000")
+		assert.Equal(t, 3, len(cc[1].Monitors))
+	})
+
+	t.Run("remove a mon from the 2nd cluster", func(t *testing.T) {
+		i := 2
+		// Remove last element of the slice
+		csiClusterConfigEntry2.Monitors = append(csiClusterConfigEntry2.Monitors[:i], csiClusterConfigEntry2.Monitors[i+1:]...)
+		s, err = updateCsiClusterConfig(s, "beta", &csiClusterConfigEntry2)
+		assert.NoError(t, err)
+		cc, err := parseCsiClusterConfig(s)
+		assert.NoError(t, err)
+		assert.Equal(t, 2, len(cc))
+		assert.Equal(t, "alpha", cc[0].ClusterID)
+		assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000")
+		assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000")
+		assert.Equal(t, 2, len(cc[0].Monitors))
+		assert.Equal(t, "beta", cc[1].ClusterID)
+		assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000")
+		assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000")
+		assert.Equal(t, 2, len(cc[1].Monitors))
+	})
+
+	t.Run("add a 3rd cluster with subvolumegroup", func(t *testing.T) {
+		s, err = updateCsiClusterConfig(s, "baba", &csiClusterConfigEntry3)
+		assert.NoError(t, err)
+		cc, err := parseCsiClusterConfig(s)
+		assert.NoError(t, err)
+		assert.Equal(t, 3, len(cc))
+		assert.Equal(t, "alpha", cc[0].ClusterID)
+		assert.Equal(t, 2, len(cc[0].Monitors))
+		assert.Equal(t, "beta", cc[1].ClusterID)
+		assert.Equal(t, "baba", cc[2].ClusterID)
+		assert.Equal(t, "10.1.1.1:5000", cc[2].Monitors[0])
+		assert.Equal(t, 3, len(cc[2].Monitors))
+		assert.Equal(t, "mygroup", cc[2].CephFS.SubvolumeGroup)
+
+	})
+
+	t.Run("add a 4th mon to the 3rd cluster and subvolumegroup is preserved", func(t *testing.T) {
+		csiClusterConfigEntry3.Monitors = append(csiClusterConfigEntry3.Monitors, "10.11.12.13:5000")
+		s, err = updateCsiClusterConfig(s, "baba", &csiClusterConfigEntry3)
+		assert.NoError(t, err)
+		cc, err := parseCsiClusterConfig(s)
+		assert.NoError(t, err)
+		assert.Equal(t, 3, len(cc))
+		assert.Equal(t, 4, len(cc[2].Monitors))
+		assert.Equal(t, "mygroup", cc[2].CephFS.SubvolumeGroup)
+	})
+
+	t.Run("remove subvolumegroup", func(t *testing.T) {
+		csiClusterConfigEntry3.CephFS.SubvolumeGroup = ""
+		s, err = updateCsiClusterConfig(s, "baba", nil)
+		assert.NoError(t, err)
+		cc, err := parseCsiClusterConfig(s)
+		assert.NoError(t, err)
+		assert.Equal(t, 2, len(cc)) // we only have 2 clusters now
+	})
+
+	t.Run("add subvolumegroup and mons after", func(t *testing.T) {
+		csiClusterConfigEntry4 := CsiClusterConfigEntry{
+			CephFS: &CsiCephFSSpec{
+				SubvolumeGroup: "mygroup2",
+			},
+		}
+		s, err = updateCsiClusterConfig(s, "quatre", &csiClusterConfigEntry4)
+		assert.NoError(t, err)
+		cc, err := parseCsiClusterConfig(s)
+		assert.NoError(t, err)
+		assert.Equal(t, 3, len(cc), cc)
+		assert.Equal(t, 0, len(cc[2].Monitors))
+		assert.Equal(t, "mygroup2", cc[2].CephFS.SubvolumeGroup, cc)
+
+		csiClusterConfigEntry4.Monitors = []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"}
+		s, err = updateCsiClusterConfig(s, "quatre", &csiClusterConfigEntry4)
+		assert.NoError(t, err)
+		cc, err = parseCsiClusterConfig(s)
+		assert.NoError(t, err)
+		assert.Equal(t, "mygroup2", cc[2].CephFS.SubvolumeGroup)
+		assert.Equal(t, 3, len(cc[2].Monitors))
+	})
+
+	t.Run("does it return error on garbage input?", func(t *testing.T) {
+		_, err = updateCsiClusterConfig("qqq", "beta", &csiClusterConfigEntry2)
+		assert.Error(t, err)
+	})
 }

pkg/operator/ceph/csi/predicate.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	"github.com/rook/rook/pkg/operator/ceph/controller"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -47,7 +46,7 @@ func predicateController(ctx context.Context, c client.Client, opNamespace strin
 			// If a Ceph Cluster is created we want to reconcile the csi driver
 			if cephCluster, ok := e.Object.(*cephv1.CephCluster); ok {
 				// If there are more than one ceph cluster in the same namespace do not reconcile
-				if controller.DuplicateCephClusters(ctx, c, e.Object, false) {
+				if opcontroller.DuplicateCephClusters(ctx, c, e.Object, false) {
 					return false
 				}
 

pkg/operator/ceph/file/subvolumegroup/controller.go

@@ -40,7 +40,9 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
+	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -175,9 +177,18 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 		logger.Debugf("deleting subvolume group %q", cephFilesystemSubVolumeGroup.Name)
 		err := r.deleteSubVolumeGroup(cephFilesystemSubVolumeGroup)
 		if err != nil {
+			if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
+				logger.Info(opcontroller.OperatorNotInitializedMessage)
+				return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
+			}
 			return reconcile.Result{}, errors.Wrapf(err, "failed to delete ceph ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
 		}
 
+		err = csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephFilesystemSubVolumeGroup), r.clusterInfo, nil)
+		if err != nil {
+			return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
+		}
+
 		// Remove finalizer
 		err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephFilesystemSubVolumeGroup)
 		if err != nil {
@@ -218,6 +229,20 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ
 		return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
 	}
 
+	// Update CSI config map
+	// If the mon endpoints change, the mon health check go routine will take care of updating the
+	// config map, so no special care is needed in this controller
+	csiClusterConfigEntry := csi.CsiClusterConfigEntry{
+		Monitors: csi.MonEndpoints(r.clusterInfo.Monitors),
+		CephFS: &csi.CsiCephFSSpec{
+			SubvolumeGroup: cephFilesystemSubVolumeGroup.Name,
+		},
+	}
+	err = csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephFilesystemSubVolumeGroup), r.clusterInfo, &csiClusterConfigEntry)
+	if err != nil {
+		return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config")
+	}
+
 	// Success! Let's update the status
 	r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionReady)
 
@@ -241,12 +266,17 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) createOrUpdateSubVolumeGroup(cep
 // Delete the ceph filesystem subvolume group
 func (r *ReconcileCephFilesystemSubVolumeGroup) deleteSubVolumeGroup(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup) error {
 	logger.Infof("deleting ceph filesystem subvolume group object %q", cephFilesystemSubVolumeGroup.Name)
-	if err := cephclient.DeleteCephFSSubVolumeGroup(r.context, r.clusterInfo, cephFilesystemSubVolumeGroup.Name, cephFilesystemSubVolumeGroup.Spec.FilesystemName); err != nil {
+	if err := cephclient.DeleteCephFSSubVolumeGroup(r.context, r.clusterInfo, cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name); err != nil {
 		code, ok := exec.ExitStatus(err)
+		// If the subvolumegroup does not exit, we should not return an error
+		if ok && code == int(syscall.ENOENT) {
+			logger.Debugf("ceph filesystem subvolume group %q do not exist", cephFilesystemSubVolumeGroup.Name)
+			return nil
+		}
 		// If the subvolumegroup has subvolumes the command will fail with:
 		// Error ENOTEMPTY: error in rmdir /volumes/csi
-		if ok && code != int(syscall.ENOTEMPTY) {
-			return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q, remove the subvolumes first.", cephFilesystemSubVolumeGroup.Name)
+		if ok && (code == int(syscall.ENOTEMPTY)) {
+			return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q, remove the subvolumes first", cephFilesystemSubVolumeGroup.Name)
 		}
 
 		return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name)
@@ -272,9 +302,15 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateStatus(client client.Clien
 	}
 
 	cephFilesystemSubVolumeGroup.Status.Phase = status
+	cephFilesystemSubVolumeGroup.Status.Info = map[string]string{"clusterID": buildClusterID(cephFilesystemSubVolumeGroup)}
 	if err := reporting.UpdateStatus(client, cephFilesystemSubVolumeGroup); err != nil {
 		logger.Errorf("failed to set ceph ceph filesystem subvolume group %q status to %q. %v", name, status, err)
 		return
 	}
 	logger.Debugf("ceph ceph filesystem subvolume group %q status updated to %q", name, status)
 }
+
+func buildClusterID(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup) string {
+	clusterID := fmt.Sprintf("%s-%s-file-%s", cephFilesystemSubVolumeGroup.Namespace, cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name)
+	return k8sutil.Hash(clusterID)
+}

pkg/operator/ceph/file/subvolumegroup/controller_test.go

@@ -26,6 +26,7 @@ import (
 	rookclient "github.com/rook/rook/pkg/client/clientset/versioned/fake"
 	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
 	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	testop "github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
@@ -206,15 +207,39 @@ func TestCephClientController(t *testing.T) {
 			client:           cl,
 			scheme:           s,
 			context:          c,
-			opManagerContext: context.TODO(),
+			opManagerContext: ctx,
 		}
 
+		// Enable CSI
+		csi.EnableRBD = true
+		os.Setenv("POD_NAMESPACE", namespace)
+		// Create CSI config map
+		ownerRef := &metav1.OwnerReference{}
+		ownerInfo := k8sutil.NewOwnerInfoWithOwnerRef(ownerRef, "")
+		err = csi.CreateCsiConfigMap(namespace, c.Clientset, ownerInfo)
+		assert.NoError(t, err)
+
 		res, err := r.Reconcile(ctx, req)
 		assert.NoError(t, err)
 		assert.False(t, res.Requeue)
 
-		err = r.client.Get(context.TODO(), req.NamespacedName, cephFilesystemSubVolumeGroup)
+		err = r.client.Get(ctx, req.NamespacedName, cephFilesystemSubVolumeGroup)
 		assert.NoError(t, err)
 		assert.Equal(t, cephv1.ConditionReady, cephFilesystemSubVolumeGroup.Status.Phase)
+		assert.NotEmpty(t, cephFilesystemSubVolumeGroup.Status.Info["clusterID"])
+
+		// test that csi configmap is created
+		cm, err := c.Clientset.CoreV1().ConfigMaps(namespace).Get(ctx, csi.ConfigName, metav1.GetOptions{})
+		assert.NoError(t, err)
+		assert.NotEmpty(t, cm.Data[csi.ConfigKey])
+		assert.Contains(t, cm.Data[csi.ConfigKey], "clusterID")
+		assert.Contains(t, cm.Data[csi.ConfigKey], "group-a")
 	})
 }
+
+func Test_buildClusterID(t *testing.T) {
+	longName := "foooooooooooooooooooooooooooooooooooooooooooo"
+	cephFilesystemSubVolumeGroup := &cephv1.CephFilesystemSubVolumeGroup{ObjectMeta: metav1.ObjectMeta{Namespace: "rook-ceph", Name: longName}, Spec: cephv1.CephFilesystemSubVolumeGroupSpec{FilesystemName: "myfs"}}
+	clusterID := buildClusterID(cephFilesystemSubVolumeGroup)
+	assert.Equal(t, "29e92135b7e8c014079b9f9f3566777d", clusterID)
+}

pkg/operator/ceph/nfs/nfs.go

@@ -18,7 +18,6 @@ limitations under the License.
 package nfs
 
 import (
-	"crypto/sha256"
 	"fmt"
 
 	"github.com/banzaicloud/k8s-objectmatcher/patch"
@@ -215,13 +214,7 @@ func (r *ReconcileCephNFS) createConfigMap(n *cephv1.CephNFS, name string) (stri
 		}
 	}
 
-	h := sha256.New()
-	if _, err := h.Write([]byte(fmt.Sprintf("%v", configMap.Data))); err != nil {
-		return "", "", errors.Wrapf(err, "failed to get hash of ganesha config map")
-	}
-	configHash := fmt.Sprintf("%x", h.Sum(nil))
-
-	return configMap.Name, configHash, nil
+	return configMap.Name, k8sutil.Hash(fmt.Sprintf("%v", configMap.Data)), nil
 }
 
 // Down scale the ganesha server