e2e: validate storing of csi.volume.owner attribute

The csi.volume.owner should get stored when the csi-provisioner sidecar
passes additional metadata. This option is now enabled by default, so
the owner (Kubernetes Namespace) of RBD images is expected to be
available.

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

e2e/rbd.go

@@ -260,6 +260,15 @@ var _ = Describe("RBD", func() {
 				}
 			})
 
+			By("create a PVC and validate owner", func() {
+				err := validateImageOwner(pvcPath, f)
+				if err != nil {
+					e2elog.Failf("failed to validate owner of pvc with error %v", err)
+				}
+				// validate created backend rbd images
+				validateRBDImageCount(f, 0)
+			})
+
 			By("create a PVC and bind it to an app", func() {
 				err := validatePVCAndAppBinding(pvcPath, appPath, f)
 				if err != nil {
@@ -1109,8 +1118,15 @@ var _ = Describe("RBD", func() {
 
 				updateConfigMap("e2e-ns")
 
+				err := validateImageOwner(pvcPath, f)
+				if err != nil {
+					e2elog.Failf("failed to validate owner of pvc with error %v", err)
+				}
+				// validate created backend rbd images
+				validateRBDImageCount(f, 0)
+
 				// Create a PVC and bind it to an app within the namesapce
-				err := validatePVCAndAppBinding(pvcPath, appPath, f)
+				err = validatePVCAndAppBinding(pvcPath, appPath, f)
 				if err != nil {
 					e2elog.Failf("failed to validate pvc and application binding with error %v", err)
 				}

e2e/rbd_helper.go

@@ -190,6 +190,50 @@ func getImageMeta(rbdImageSpec, metaKey string, f *framework.Framework) (string,
 	return strings.TrimSpace(stdOut), nil
 }
 
+// validateImageOwner checks the "csi.volume.owner" key on the image journal
+// and verifies that the owner is set to the namespace where the PVC is
+// created.
+func validateImageOwner(pvcPath string, f *framework.Framework) error {
+	const ownerKey = "csi.volume.owner"
+
+	pvc, err := loadPVC(pvcPath)
+	if err != nil {
+		return err
+	}
+	pvc.Namespace = f.UniqueName
+	pvc.Name = f.UniqueName
+	err = createPVCAndvalidatePV(f.ClientSet, pvc, deployTimeout)
+	if err != nil {
+		return err
+	}
+
+	imageData, err := getImageInfoFromPVC(pvc.Namespace, pvc.Name, f)
+	if err != nil {
+		return err
+	}
+
+	stdOut, stdErr, err := execCommandInToolBoxPod(f,
+		fmt.Sprintf("rados %s getomapval csi.volume.%s %s", rbdOptions(defaultRBDPool), imageData.imageID, ownerKey), rookNamespace)
+	if err != nil {
+		return err
+	}
+	if stdErr != "" {
+		return fmt.Errorf("failed to getomapval %v", stdErr)
+	}
+
+	if radosNamespace != "" {
+		e2elog.Logf("found image journal %s in pool %s namespace %s", "csi.volume."+imageData.imageID, defaultRBDPool, radosNamespace)
+	} else {
+		e2elog.Logf("found image journal %s in pool %s", "csi.volume."+imageData.imageID, defaultRBDPool)
+	}
+
+	if !strings.Contains(stdOut, pvc.Namespace) {
+		return fmt.Errorf("%q does not contain %q: %s", ownerKey, pvc.Namespace, stdOut)
+	}
+
+	return deletePVCAndValidatePV(f.ClientSet, pvc, deployTimeout)
+}
+
 func validateEncryptedPVCAndAppBinding(pvcPath, appPath, kms string, f *framework.Framework) error {
 	pvc, app, err := createPVCAndAppBinding(pvcPath, appPath, f, deployTimeout)
 	if err != nil {