rbd: set cluster Name as metadata on the image

This change helps read the cluster name from the cmdline args,
the provisioner will set the same on the RBD images.

Fixes: #2973
Signed-off-by: Prasanna Kumar Kalever <prasanna.kalever@redhat.com>

cmd/cephcsi.go

@@ -68,6 +68,7 @@ func init() {
 	flag.StringVar(&conf.NodeID, "nodeid", "", "node id")
 	flag.StringVar(&conf.PluginPath, "pluginpath", defaultPluginPath, "plugin path")
 	flag.StringVar(&conf.StagingPath, "stagingpath", defaultStagingPath, "staging path")
+	flag.StringVar(&conf.ClusterName, "clustername", "", "name of the cluster")
 	flag.StringVar(&conf.InstanceID, "instanceid", "", "Unique ID distinguishing this instance of Ceph CSI among other"+
 		" instances, when sharing Ceph clusters across CSI instances for provisioning")
 	flag.IntVar(&conf.PidLimit, "pidlimit", 0, "the PID limit to configure through cgroups")
@@ -247,8 +248,9 @@ func main() {
 
 	case controllerType:
 		cfg := controller.Config{
-			DriverName: dname,
-			Namespace:  conf.DriverNamespace,
+			DriverName:  dname,
+			Namespace:   conf.DriverNamespace,
+			ClusterName: conf.ClusterName,
 		}
 		// initialize all controllers before starting.
 		initControllers()

internal/controller/controller.go

@@ -35,8 +35,9 @@ type Manager interface {
 
 // Config holds the drivername and namespace name.
 type Config struct {
-	DriverName string
-	Namespace  string
+	DriverName  string
+	Namespace   string
+	ClusterName string
 }
 
 // ControllerList holds the list of managers need to be started.

internal/controller/persistentvolume/persistentvolume.go

@@ -184,6 +184,7 @@ func (r ReconcilePersistentVolume) reconcilePV(ctx context.Context, obj runtime.
 		volumeHandler,
 		requestName,
 		pvcNamespace,
+		r.config.ClusterName,
 		cr)
 	if err != nil {
 		log.ErrorLogMsg("failed to regenerate journal %s", err)

internal/rbd/controllerserver.go

@@ -51,6 +51,9 @@ type ControllerServer struct {
 
 	// A map storing all volumes/snapshots with ongoing operations.
 	OperationLocks *util.OperationLock
+
+	// Cluster name
+	ClusterName string
 }
 
 func (cs *ControllerServer) validateVolumeReq(ctx context.Context, req *csi.CreateVolumeRequest) error {
@@ -132,6 +135,8 @@ func (cs *ControllerServer) parseVolCreateRequest(
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
+	rbdVol.ClusterName = cs.ClusterName
+
 	// if the KMS is of type VaultToken, additional metadata is needed
 	// depending on the tenant, the KMS can be configured with other
 	// options
@@ -1090,6 +1095,7 @@ func (cs *ControllerServer) CreateSnapshot(
 
 	// Update the metadata on snapshot not on the original image
 	rbdVol.RbdImageName = rbdSnap.RbdSnapName
+	rbdVol.ClusterName = cs.ClusterName
 
 	err = rbdVol.unsetAllMetadata(k8s.GetVolumeMetadataKeys())
 	if err != nil {

internal/rbd/driver/driver.go

@@ -161,6 +161,7 @@ func (r *Driver) Run(conf *util.Config) {
 
 	if conf.IsControllerServer {
 		r.cs = NewControllerServer(r.cd)
+		r.cs.ClusterName = conf.ClusterName
 		r.rs = NewReplicationServer(r.cs)
 	}
 	if !conf.IsControllerServer && !conf.IsNodeServer {

internal/rbd/rbd_journal.go

@@ -541,7 +541,8 @@ func RegenerateJournal(
 	claimName,
 	volumeID,
 	requestName,
-	owner string,
+	owner,
+	clusterName string,
 	cr *util.Credentials,
 ) (string, error) {
 	ctx := context.Background()
@@ -555,6 +556,7 @@ func RegenerateJournal(
 
 	rbdVol = &rbdVolume{}
 	rbdVol.VolID = volumeID
+	rbdVol.ClusterName = clusterName
 
 	err = vi.DecomposeCSIID(rbdVol.VolID)
 	if err != nil {

internal/rbd/rbd_util.go

@@ -78,6 +78,9 @@ const (
 
 	// krbd attribute file to check supported features.
 	krbdSupportedFeaturesFile = "/sys/bus/rbd/supported_features"
+
+	// clusterNameKey cluster Key, set on RBD image.
+	clusterNameKey = "csi.ceph.com/cluster/name"
 )
 
 // rbdImage contains common attributes and methods for the rbdVolume and
@@ -121,6 +124,9 @@ type rbdImage struct {
 	// Primary represent if the image is primary or not.
 	Primary bool
 
+	// Cluster name
+	ClusterName string
+
 	// encryption provides access to optional VolumeEncryption functions
 	encryption *util.VolumeEncryption
 	// Owner is the creator (tenant, Kubernetes Namespace) of the volume
@@ -2001,6 +2007,14 @@ func (rv *rbdVolume) setAllMetadata(parameters map[string]string) error {
 		}
 	}
 
+	if rv.ClusterName != "" {
+		err := rv.SetMetadata(clusterNameKey, rv.ClusterName)
+		if err != nil {
+			return fmt.Errorf("failed to set metadata key %q, value %q on image: %w",
+				clusterNameKey, rv.ClusterName, err)
+		}
+	}
+
 	return nil
 }
 

internal/util/util.go

@@ -121,6 +121,9 @@ type Config struct {
 
 	// CSI-Addons endpoint
 	CSIAddonsEndpoint string
+
+	// Cluster name
+	ClusterName string
 }
 
 // ValidateDriverName validates the driver name.