diff --git a/internal/cephfs/fsjournal.go b/internal/cephfs/fsjournal.go
index 7ba8c8404..4eee49f7e 100644
--- a/internal/cephfs/fsjournal.go
+++ b/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
 	}
diff --git a/internal/journal/voljournal.go b/internal/journal/voljournal.go
index a920a704e..952363e84 100644
--- a/internal/journal/voljournal.go
+++ b/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
diff --git a/internal/rbd/rbd_journal.go b/internal/rbd/rbd_journal.go
index fac61cb91..a39f5c558 100644
--- a/internal/rbd/rbd_journal.go
+++ b/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
 	}
diff --git a/internal/rbd/rbd_util.go b/internal/rbd/rbd_util.go
index f1b0e20ae..17e72eaa0 100644
--- a/internal/rbd/rbd_util.go
+++ b/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 {