[#248] add metrics methods

src/frostfs_testlib/steps/metrics.py

@@ -0,0 +1,45 @@
+import re
+
+from frostfs_testlib import reporter
+from frostfs_testlib.testing.test_control import wait_for_success
+from frostfs_testlib.storage.cluster import ClusterNode
+
+
+@reporter.step("Check metrics result")
+@wait_for_success(interval=10)
+def check_metrics_counter(
+    cluster_nodes: list[ClusterNode],
+    operator: str = "==",
+    counter_exp: int = 0,
+    parse_from_command: bool = False,
+    **metrics_greps: str,
+):
+    counter_act = 0
+    for cluster_node in cluster_nodes:
+        counter_act += get_metrics_value(cluster_node, parse_from_command, **metrics_greps)
+    assert eval(
+        f"{counter_act} {operator} {counter_exp}"
+    ), f"Expected: {counter_exp}, Actual: {counter_act} in node: {cluster_node}"
+
+
+@reporter.step("Get metrics value from node: {node}")
+def get_metrics_value(node: ClusterNode, parse_from_command: bool = False, **metrics_greps: str):
+    try:
+        command_result = node.metrics.storage.get_metrics_search_by_greps(**metrics_greps)
+        if parse_from_command:
+            metrics_counter = calc_metrics_count_from_stdout(command_result.stdout, **metrics_greps)
+        else:
+            metrics_counter = calc_metrics_count_from_stdout(command_result.stdout)
+    except RuntimeError as e:
+        metrics_counter = 0
+
+    return metrics_counter
+
+
+@reporter.step("Parse metrics count and calc sum of result")
+def calc_metrics_count_from_stdout(metric_result_stdout: str, command: str = None):
+    if command:
+        result = re.findall(rf"{command}\s*([\d.e+-]+)", metric_result_stdout)
+    else:
+        result = re.findall(r"}\s*([\d.e+-]+)", metric_result_stdout)
+    return sum(map(lambda x: int(float(x)), result))

src/frostfs_testlib/storage/dataclasses/metrics.py

@@ -16,11 +16,6 @@ class StorageMetrics:
         self.host = host
         self.metrics_endpoint = metrics_endpoint
 
-    def get_metric_container(self, metric: str, cid: str) -> CommandResult:
-        shell = self.host.get_shell()
-        result = shell.exec(f"curl -s {self.metrics_endpoint} | grep {metric} |grep {cid}")
-        return result
-
     def get_metrics_search_by_greps(self, **greps) -> CommandResult:
         """
         Get a metrics, search by: cid, metric_type, shard_id etc.
@@ -34,3 +29,8 @@ class StorageMetrics:
         additional_greps = " |grep ".join([grep_command for grep_command in greps.values()])
         result = shell.exec(f"curl -s {self.metrics_endpoint} | grep {additional_greps}")
         return result
+    
+    def get_all_metrics(self) -> CommandResult:
+        shell = self.host.get_shell()
+        result = shell.exec(f"curl -s {self.metrics_endpoint}")
+        return result