util: enhance tracevol.py script to work with volumesnapshots

This patch adds the functionality to map the k8s volumesnapshots
to the cephfs/rbd snapshots.

This patch also adds a wrapper around oc/kubectl called `kube_client`
which will help get rid of code duplication.

Closes: ceph#3344

Signed-off-by: Niraj Yadav <[email protected]>

troubleshooting/tools/tracevol.py

@@ -42,6 +42,7 @@
 import re
 import prettytable
 PARSER = argparse.ArgumentParser()
+ARGS=None # just so that the linter is happy, ARGS is global already.
 
 # -p pvc-test -k /home/.kube/config -n default -rn rook-ceph
 PARSER.add_argument("-p", "--pvcname", default="", help="PVC name")
@@ -562,6 +563,111 @@ def get_fsname_from_pvdata(arg, pvdata):
         sys.exit()
     return fsname
 
+def kube_client(*commands):
+    """
+    Executes a kubectl/oc command with the provided arguments and returns the JSON output.
+    Note: Do not pass `-o json` as an argument, this helper adds it automatically.
+
+    Example:
+        kube_client("get", "pvcs")
+
+    Args:
+        *commands: Additional command-line arguments to pass to the Kubernetes or OpenShift command.
+    """
+    arg = ARGS
+    cmd = [arg.command]
+    if arg.kubeconfig != "":
+        cmd += ["--kubeconfig", arg.kubeconfig]
+    cmd += ["--namespace", arg.namespace]
+    cmd += filter(lambda x: x not in ('-o', 'json'), commands)
+    cmd += ["-o", "json"]
+
+    with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) as out:
+        stdout, stderr = out.communicate()
+
+    if stderr is not None:
+        if arg.debug:
+            print(f"failed to execute command {commands} due to {stderr}")
+        sys.exit()
+    try:
+        data = json.loads(stdout)
+    except ValueError as err:
+        print(err, stdout)
+        sys.exit()
+
+    return data
+
+def parse_volume_snapshots(volsnapshots):
+    """
+    Parses a list of volume snapshots and prints a table with their details.
+    """
+    cephfs_snap_tbl = prettytable.PrettyTable()
+    cephfs_snap_tbl.title = "CephFS Volume Snapshots"
+    cephfs_snap_tbl.field_names = ["Name", "PVC", "Subvolume Name", "Snapshot ID"]
+
+    rbd_snap_tbl = prettytable.PrettyTable()
+    rbd_snap_tbl.title = "RBD Volume Snapshots"
+    rbd_snap_tbl.field_names = ["Name", "PVC", "Image Name", "Snapshot ID"]
+
+    for snapshot in volsnapshots['items']:
+        snap_name = snapshot['metadata']['name']
+        pvc_name = snapshot['spec']['source']['persistentVolumeClaimName']
+
+        # Fetch the PVC and its PV, useful later
+        pvc = kube_client("get", "pvc", pvc_name)
+        pv_data = kube_client("get", "pv", pvc['spec']['volumeName'])
+
+        # Find the type of the PVC (RBD or CephFS)
+        csi_attrs = pv_data['spec'].get('csi')
+
+        # Make sure the PVC is a CSI volume
+        if not csi_attrs:
+            if ARGS.debug:
+                print(f"Skipping snapshot {snap_name} as it is not a CSI volume")
+            continue
+        fs_type = "CephFS" if "fsName" in csi_attrs['volumeAttributes'] else "RBD"
+
+        # Get the vol name inside ceph
+        vol_handle = pv_data['spec']['csi']['volumeHandle']
+        img_uuid = get_image_uuid(vol_handle)
+        prefix = get_volname_prefix(ARGS, pv_data)
+        img_id = prefix + img_uuid
+
+        # Fetch the volume snapshot class, it is used to determine the snapshot prefix
+        snap_class_name = snapshot['spec']['volumeSnapshotClassName']
+        snap_class = kube_client("get", "volumesnapshotclass", snap_class_name)
+
+        # Get the snapshot ID, we can inspect volumesnapshotcontent to get the snapshot handle
+        # Alternatively, we can use the toolbox pod to fetch these values from ceph directly
+        snap_content_name = snapshot['status']['boundVolumeSnapshotContentName']
+        snap_content = kube_client("get", "volumesnapshotcontent", snap_content_name)
+
+        # The snapshot prefix can be optionally defined in the volumesnapshotclass parameters 
+        snap_handle = snap_content['status']['snapshotHandle']
+        snap_uuid = get_image_uuid(snap_handle)
+        snap_prefix = snap_class['parameters'].get('snapshotNamePrefix') or 'csi-snap-'
+        snap_id = snap_prefix + snap_uuid
+
+        row = [snap_name, pvc_name, img_id, snap_id]
+        if fs_type == "CephFS":
+            cephfs_snap_tbl.add_row(row)
+        else:
+            rbd_snap_tbl.add_row(row)
+
+    print(rbd_snap_tbl)
+    print(cephfs_snap_tbl)
+
+def list_volume_snapshots():
+    """
+    Retrieve and parse the list of volume snapshots from the cluster.
+    """
+    if ARGS.namespace:
+        volumesnapshots = kube_client("get", "volumesnapshots", "-n", ARGS.namespace)
+    else:
+        volumesnapshots = kube_client("get", "volumesnapshots")
+
+    parse_volume_snapshots(volumesnapshots)
+
 if __name__ == "__main__":
     ARGS = PARSER.parse_args()
     if ARGS.command not in ["kubectl", "oc"]:
@@ -571,3 +677,4 @@ def get_fsname_from_pvdata(arg, pvdata):
         print("python version less than 3 is not supported.")
         sys.exit(1)
     list_pvc_vol_name_mapping(ARGS)
+    list_volume_snapshots()