rbd: store csi.storage.k8s.io/pvc/namespace metadata as Owner

The Owner of an RBD image (Kubernetes Namespace, tenant) can be used to
identify additional configuration options. This will be used for
fetching the right Vault Token when encrypting/decrypting volumes.

Signed-off-by: Niels de Vos <ndevos@redhat.com>

internal/cephfs/fsjournal.go

@@ -232,7 +232,7 @@ func reserveVol(ctx context.Context, volOptions *volumeOptions, secret map[strin
 	imageUUID, vid.FsSubvolName, err = j.ReserveName(
 		ctx, volOptions.MetadataPool, util.InvalidPoolID,
 		volOptions.MetadataPool, util.InvalidPoolID, volOptions.RequestName,
-		volOptions.NamePrefix, "", "", volOptions.ReservedID)
+		volOptions.NamePrefix, "", "", volOptions.ReservedID, "")
 	if err != nil {
 		return nil, err
 	}
@@ -269,7 +269,7 @@ func reserveSnap(ctx context.Context, volOptions *volumeOptions, parentSubVolNam
 	imageUUID, vid.FsSnapshotName, err = j.ReserveName(
 		ctx, volOptions.MetadataPool, util.InvalidPoolID,
 		volOptions.MetadataPool, util.InvalidPoolID, snap.RequestName,
-		snap.NamePrefix, parentSubVolName, "", snap.ReservedID)
+		snap.NamePrefix, parentSubVolName, "", snap.ReservedID, "")
 	if err != nil {
 		return nil, err
 	}

internal/journal/voljournal.go

@@ -148,6 +148,9 @@ type Config struct {
 	// encryptKMS in which encryption passphrase was saved, default is no encryption
 	encryptKMSKey string
 
+	// ownerKey is used to identify the owner of the volume, can be used with some KMS configurations
+	ownerKey string
+
 	// commonPrefix is the prefix common to all omap keys for this Config
 	commonPrefix string
 }
@@ -165,6 +168,7 @@ func NewCSIVolumeJournal(suffix string) *Config {
 		namespace:               "",
 		csiImageIDKey:           "csi.imageid",
 		encryptKMSKey:           "csi.volume.encryptKMS",
+		ownerKey:                "csi.volume.owner",
 		commonPrefix:            "csi.",
 	}
 }
@@ -182,6 +186,7 @@ func NewCSISnapshotJournal(suffix string) *Config {
 		namespace:               "",
 		csiImageIDKey:           "csi.imageid",
 		encryptKMSKey:           "csi.volume.encryptKMS",
+		ownerKey:                "csi.volume.owner",
 		commonPrefix:            "csi.",
 	}
 }
@@ -494,6 +499,7 @@ Input arguments:
 	- kmsConf: Name of the key management service used to encrypt the image (optional)
         - volUUID: UUID need to be reserved instead of auto-generating one (this is
           useful for mirroring and metro-DR)
+	- owner: the owner of the volume (optional)
 
 Return values:
 	- string: Contains the UUID that was reserved for the passed in reqName
@@ -503,7 +509,7 @@ Return values:
 func (conn *Connection) ReserveName(ctx context.Context,
 	journalPool string, journalPoolID int64,
 	imagePool string, imagePoolID int64,
-	reqName, namePrefix, parentName, kmsConf, volUUID string) (string, string, error) {
+	reqName, namePrefix, parentName, kmsConf, volUUID, owner string) (string, string, error) {
 	// TODO: Take in-arg as ImageAttributes?
 	var (
 		snapSource bool
@@ -574,6 +580,11 @@ func (conn *Connection) ReserveName(ctx context.Context,
 		omapValues[cj.encryptKMSKey] = kmsConf
 	}
 
+	// if owner is passed, set it in the UUID directory too
+	if owner != "" {
+		omapValues[cj.ownerKey] = owner
+	}
+
 	if journalPool != imagePool && journalPoolID != util.InvalidPoolID {
 		buf64 := make([]byte, 8)
 		binary.BigEndian.PutUint64(buf64, uint64(journalPoolID))
@@ -601,6 +612,7 @@ type ImageAttributes struct {
 	SourceName    string // Contains the parent image name for the passed in UUID, if it is a snapshot
 	ImageName     string // Contains the image or subvolume name for the passed in UUID
 	KmsID         string // Contains encryption KMS, if it is an encrypted image
+	Owner         string // Contains the owner to be used in combination with KmsID (for some KMS)
 	ImageID       string // Contains the image id
 	JournalPoolID int64  // Pool ID of the CSI journal pool, stored in big endian format (on-disk data)
 }
@@ -625,6 +637,7 @@ func (conn *Connection) GetImageAttributes(ctx context.Context, pool, objectUUID
 		cj.csiJournalPool,
 		cj.cephSnapSourceKey,
 		cj.csiImageIDKey,
+		cj.ownerKey,
 	}
 	values, err := getOMapValues(
 		ctx, conn, pool, cj.namespace, cj.cephUUIDDirectoryPrefix+objectUUID,
@@ -639,6 +652,7 @@ func (conn *Connection) GetImageAttributes(ctx context.Context, pool, objectUUID
 	var found bool
 	imageAttributes.RequestName = values[cj.csiNameKey]
 	imageAttributes.KmsID = values[cj.encryptKMSKey]
+	imageAttributes.Owner = values[cj.ownerKey]
 	imageAttributes.ImageID = values[cj.csiImageIDKey]
 
 	// image key was added at a later point, so not all volumes will have this

internal/rbd/rbd_journal.go

@@ -344,7 +344,7 @@ func reserveSnap(ctx context.Context, rbdSnap *rbdSnapshot, rbdVol *rbdVolume, c
 
 	rbdSnap.ReservedID, rbdSnap.RbdSnapName, err = j.ReserveName(
 		ctx, rbdSnap.JournalPool, journalPoolID, rbdSnap.Pool, imagePoolID,
-		rbdSnap.RequestName, rbdSnap.NamePrefix, rbdVol.RbdImageName, "", rbdSnap.ReservedID)
+		rbdSnap.RequestName, rbdSnap.NamePrefix, rbdVol.RbdImageName, "", rbdSnap.ReservedID, rbdVol.Owner)
 	if err != nil {
 		return err
 	}
@@ -422,7 +422,7 @@ func reserveVol(ctx context.Context, rbdVol *rbdVolume, rbdSnap *rbdSnapshot, cr
 
 	rbdVol.ReservedID, rbdVol.RbdImageName, err = j.ReserveName(
 		ctx, rbdVol.JournalPool, journalPoolID, rbdVol.Pool, imagePoolID,
-		rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID, rbdVol.ReservedID)
+		rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID, rbdVol.ReservedID, rbdVol.Owner)
 	if err != nil {
 		return err
 	}
@@ -542,6 +542,7 @@ func RegenerateJournal(imageName, volumeID, pool, journalPool, requestName strin
 	if imageData != nil {
 		rbdVol.ReservedID = imageData.ImageUUID
 		rbdVol.ImageID = imageData.ImageAttributes.ImageID
+		rbdVol.Owner = imageData.ImageAttributes.Owner
 		if rbdVol.ImageID == "" {
 			err = rbdVol.storeImageID(ctx, j)
 			if err != nil {
@@ -559,7 +560,7 @@ func RegenerateJournal(imageName, volumeID, pool, journalPool, requestName strin
 
 	rbdVol.ReservedID, rbdVol.RbdImageName, err = j.ReserveName(
 		ctx, rbdVol.JournalPool, journalPoolID, rbdVol.Pool, imagePoolID,
-		rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID, vi.ObjectUUID)
+		rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID, vi.ObjectUUID, rbdVol.Owner)
 	if err != nil {
 		return err
 	}

internal/rbd/rbd_util.go

@@ -105,7 +105,9 @@ type rbdVolume struct {
 	readOnly            bool
 	Primary             bool
 	KMS                 util.EncryptionKMS
-	CreatedAt           *timestamp.Timestamp
+	// Owner is the creator (tenant, Kubernetes Namespace) of the volume.
+	Owner     string
+	CreatedAt *timestamp.Timestamp
 	// conn is a connection to the Ceph cluster obtained from a ConnPool
 	conn *util.ClusterConnection
 	// an opened IOContext, call .openIoctx() before using
@@ -734,6 +736,7 @@ func genVolFromVolID(ctx context.Context, volumeID string, cr *util.Credentials,
 	rbdVol.RbdImageName = imageAttributes.ImageName
 	rbdVol.ReservedID = vi.ObjectUUID
 	rbdVol.ImageID = imageAttributes.ImageID
+	rbdVol.Owner = imageAttributes.Owner
 
 	if imageAttributes.KmsID != "" {
 		rbdVol.Encrypted = true
@@ -813,6 +816,15 @@ func genVolFromVolumeOptions(ctx context.Context, volOptions, credentials map[st
 		rbdVol.Mounter = rbdDefaultMounter
 	}
 
+	// if the KMS is of type VaultToken, additional metadata is needed
+	// depending on the tenant, the KMS can be configured with other
+	// options
+	// FIXME: this works only on Kubernetes, how do other CO supply metadata?
+	rbdVol.Owner, ok = volOptions["csi.storage.k8s.io/pvc/namespace"]
+	if !ok {
+		util.DebugLog(ctx, "could not detect owner for %s", rbdVol.String())
+	}
+
 	rbdVol.Encrypted = false
 	encrypted, ok = volOptions["encrypted"]
 	if ok {