Replace hardcoded CephFS in _list_volumes with actual check

Rather than hardcoding the supported storage classes and provisioners in the `_list_volumes` function it will now check if the provisioner of the PVCs has an  accompanying VolumeSnapshotClass.

backend/kale/common/cephfsutils.py

@@ -188,7 +188,7 @@ def list_pvc_snapshots(label_selector=""):
 
 
 def hydrate_pvc_from_snapshot(new_pvc_name, source_snapshot_name):
-    """Create a new PVC out of a Rok snapshot."""
+    """Create a new PVC out of a volume snapshot."""
     log.info("Creating new PVC '%s' from snapshot %s ..." %
              (new_pvc_name, source_snapshot_name))
     snapshot_info = get_pvc_snapshot(source_snapshot_name)

backend/kale/common/podutils.py

@@ -28,9 +28,6 @@
 ROK_CSI_STORAGE_CLASS = "rok"
 ROK_CSI_STORAGE_PROVISIONER = "rok.arrikto.com"
 
-ROOK_CEPHFS_CSI_STORAGE_CLASS = "rook-cephfs"
-ROOK_CEPHFS_CSI_STORAGE_PROVISIONER = "rook-ceph.cephfs.csi.ceph.com"
-
 NAMESPACE_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
 
 K8S_SIZE_RE = re.compile(r'^([0-9]+)(E|Ei|P|Pi|T|Ti|G|Gi|M|Mi|K|Ki){0,1}$')
@@ -161,6 +158,8 @@ def _get_mount_path(container, volume):
 def _list_volumes(client, namespace, pod_name, container_name):
     pod = client.read_namespaced_pod(pod_name, namespace)
     container = _get_pod_container(pod, container_name)
+    snapshotclass_provisioners = list_snapshotclass_storage_provisioners()
+    provisioner_names = get_snapshotclass_provisioners_names()
 
     rok_volumes = []
     for volume in pod.spec.volumes:
@@ -175,22 +174,23 @@ def _list_volumes(client, namespace, pod_name, container_name):
         pvc = client.read_namespaced_persistent_volume_claim(pvc.claim_name,
                                                              namespace)
         if (pvc.spec.storage_class_name != ROK_CSI_STORAGE_CLASS
-           and pvc.spec.storage_class_name != ROOK_CEPHFS_CSI_STORAGE_CLASS):
+           and pvc.spec.storage_class_name not in provisioner_names):
             msg = ("Found PVC with storage class '%s'. "
-                   "Only storage classes '%s' and '%s' are supported."
+                   "Only storage classes able to take snapshots and "
+                   "'%s' are supported."
                    % (pvc.spec.storage_class_name,
-                      ROK_CSI_STORAGE_CLASS, ROOK_CEPHFS_CSI_STORAGE_CLASS))
+                      ROK_CSI_STORAGE_CLASS))
             raise RuntimeError(msg)
 
         ann = pvc.metadata.annotations
         provisioner = ann.get("volume.beta.kubernetes.io/storage-provisioner",
                               None)
         if (provisioner != ROK_CSI_STORAGE_PROVISIONER
-           and provisioner != ROOK_CEPHFS_CSI_STORAGE_PROVISIONER):
+           and provisioner not in snapshotclass_provisioners):
             msg = ("Found PVC storage provisioner '%s'. "
-                   "Only storage provisioners '%s' and '%s' are supported."
-                   % (provisioner, ROK_CSI_STORAGE_PROVISIONER,
-                      ROOK_CEPHFS_CSI_STORAGE_PROVISIONER))
+                   "Only storage provisioners able to take snapshots and "
+                   "'%s' are supported."
+                   % (provisioner, ROK_CSI_STORAGE_PROVISIONER))
             raise RuntimeError(msg)
 
         mount_path = _get_mount_path(container, volume)
@@ -314,3 +314,44 @@ def compute_component_id(pod):
     component_id = component_name + "@sha256=" + component_spec_digest
     log.info("Computed component ID: %s", component_id)
     return component_id
+
+
+def _get_k8s_storage_client():
+    k8s_config.load_incluster_config()
+    api_client = k8s.ApiClient()
+    return k8s.StorageV1beta1Api(api_client)
+
+
+def get_snapshotclasses(label_selector=""):
+    """List snapshotclasses."""
+    co_client = _get_k8s_custom_objects_client()
+
+    snapshotclasses = co_client.list_cluster_custom_object(
+        group="snapshot.storage.k8s.io",
+        version="v1beta1",
+        plural="volumesnapshotclasses",
+        label_selector=label_selector)
+    return snapshotclasses
+
+
+def list_snapshotclass_storage_provisioners(label_selector=""):
+    """List the storage provisioners of the snapshotclasses."""
+    snapshotclasses = get_snapshotclasses(label_selector)["items"]
+    snapshotclass_provisioners = []
+    for i in snapshotclasses:
+        snapshotclass_provisioners.append(i["driver"])
+    return snapshotclass_provisioners
+
+
+def get_snapshotclass_provisioners_names():
+    """Get the names of snapshotclass storage provisioners."""
+    client = _get_k8s_storage_client()
+    classes = client.list_storage_class().items
+    snapshotclass_provisioners = list_snapshotclass_storage_provisioners()
+    storage_class_names = []
+    for i in classes:
+        name = i.metadata.name
+        provisioner = i.provisioner
+        if provisioner in snapshotclass_provisioners:
+            storage_class_names.append(name)
+    return storage_class_names