fix(manager): send snapshot details to Argus Results instead of logging

To improve usability of the snapshots preparer tool, generated backup
snapshot details will be sent to Argus Results, where they can be taken.

It eliminates the need to browse the logfile looking for details.

mgmt_cli_test.py

@@ -34,8 +34,8 @@
 
 from argus.client.generic_result import Status
 from sdcm import mgmt
-from sdcm.argus_results import send_manager_benchmark_results_to_argus, submit_results_to_argus, \
-    ManagerBackupReadResult, ManagerBackupBenchmarkResult
+from sdcm.argus_results import (send_manager_benchmark_results_to_argus, send_manager_snapshot_details_to_argus,
+                                submit_results_to_argus, ManagerBackupReadResult, ManagerBackupBenchmarkResult)
 from sdcm.mgmt import ScyllaManagerError, TaskStatus, HostStatus, HostSsl, HostRestStatus
 from sdcm.mgmt.cli import ScyllaManagerTool, RestoreTask
 from sdcm.mgmt.common import reconfigure_scylla_manager, get_persistent_snapshots
@@ -1437,15 +1437,22 @@ def test_prepare_backup_snapshot(self):
             # from ["'AWS_US_EAST_1:s3:scylla-cloud-backup-8072-7216-v5dn53'"] to scylla-cloud-backup-8072-7216-v5dn53
             original_bucket_name = location_list[0].split(":")[-1].rstrip("'")
             bucket_name = original_bucket_name + "-manager-tests"
-
             self.copy_backup_snapshot_bucket(source=original_bucket_name, destination=bucket_name)
-
-        self.log.info("Log snapshot details")
-        self.log.info(
-            f"Snapshot tag: {backup_task.get_snapshot_tag()}\n"
-            f"Keyspace name: {ks_name}\n"
-            f"Bucket: {location_list}\n"
-            f"Cluster id: {mgr_cluster.id}\n"
+        else:
+            bucket_name = location_list[0]
+
+        self.log.info("Send snapshot details to Argus")
+        snapshot_details = {
+            "tag": backup_task.get_snapshot_tag(),
+            "size": backup_size,
+            "bucket": bucket_name,
+            "ks_name": ks_name,
+            "scylla_version": self.params.get("scylla_version"),
+            "cluster_id": mgr_cluster.id,
+        }
+        send_manager_snapshot_details_to_argus(
+            argus_client=self.test_config.argus_client(),
+            snapshot_details=snapshot_details,
         )
 
 

sdcm/argus_results.py

@@ -135,6 +135,20 @@ class Meta:
         ]
 
 
+class ManagerSnapshotDetails(GenericResultTable):
+    class Meta:
+        name = "Snapshot details"
+        description = "Manager snapshots (pre-created for next utilization in restore tests) details"
+        Columns = [
+            ColumnMetadata(name="tag", unit="", type=ResultType.TEXT),
+            ColumnMetadata(name="size", unit="GB", type=ResultType.INTEGER),
+            ColumnMetadata(name="bucket", unit="", type=ResultType.TEXT),
+            ColumnMetadata(name="ks_name", unit="", type=ResultType.TEXT),
+            ColumnMetadata(name="cluster_id", unit="", type=ResultType.TEXT),
+            ColumnMetadata(name="scylla_version", unit="", type=ResultType.TEXT),
+        ]
+
+
 workload_to_table = {
     "mixed": LatencyCalculatorMixedResult,
     "write": LatencyCalculatorWriteResult,
@@ -257,3 +271,10 @@ def send_manager_benchmark_results_to_argus(argus_client: ArgusClient, result: d
     for key, value in result.items():
         result_table.add_result(column=key, row=row_name, value=value, status=Status.UNSET)
     submit_results_to_argus(argus_client, result_table)
+
+
+def send_manager_snapshot_details_to_argus(argus_client: ArgusClient, snapshot_details: dict) -> None:
+    result_table = ManagerSnapshotDetails()
+    for key, value in snapshot_details.items():
+        result_table.add_result(column=key, row="#1", value=value, status=Status.UNSET)
+    submit_results_to_argus(argus_client, result_table)