Initial version of defining the interfaces to accept metrics

tests/snappi_tests/metrics_utils/examples.py

@@ -0,0 +1,62 @@
+import logging
+import json
+import datetime
+import time
+
+from typing import List, Dict, Union
+from metrics_accepter import Metric, GaugeMetric
+from metrics_reporter import MetricReporterFactory, MetricsReporter
+
+def main():
+    """
+
+    PSU    Model            Serial             HW Rev    Voltage (V)    Current (A)    Power (W)  Status    LED
+    -----  ---------------  ---------------  --------  -------------  -------------  -----------  --------  -----
+    PSU 1  PWR-2422-HV-RED  6A011010142349Q        01          12.09          18.38       222.00  OK        green
+    PSU 2  PWR-2422-HV-RED  6A011010142327X        01          12.10          17.72       214.00  OK        green
+
+    """
+    resource_labels = {
+        "testbed.id": "sonic_stress_testbed",
+        "os.version": "11.2.3",
+        "testcase": "stress_test1",
+        "testrun.id": "202412101217"
+    }
+
+    # Create a MetricReporterFactory and build a MetricReporter
+    factory = MetricReporterFactory()
+    reporter = factory.create_metrics_reporter(resource_labels)
+
+    scope_labels = {
+        "device.id": "str-7060x6-64pe-stress-02",
+        "psu.id": "psu1",
+        "model": "PWR-2422-HV-RED",
+        "serial": "6A011010142349Q"}
+
+    # Create a metric and pass it to the reporter
+    vol = GaugeMetric(name = "Voltage",
+                      description = "PSU voltage reading",
+                      unit = "V",
+                      reporter = reporter)
+    vol.set_gauge_metric(scope_labels, 12.09)
+
+    # Create a metric and pass it to the reporter
+    cur = GaugeMetric(name = "Current",
+                      description = "PSU current reading",
+                      unit = "A",
+                      reporter = reporter)
+    cur.set_gauge_metric(scope_labels, 18.38)
+
+    # Create a metric and pass it to the reporter
+    power = GaugeMetric(name = "Power",
+                        description = "PSU power reading",
+                        unit = "W",
+                        reporter = reporter)
+    power.set_gauge_metric(scope_labels, 222.00)
+
+    # Report all metrics at a specific timestamp
+    reporter.report()
+
+if __name__ == '__main__':
+    main()
+

tests/snappi_tests/metrics_utils/metrics_accepter.py

@@ -0,0 +1,64 @@
+"""
+This file defines the classes accepting metrics and records from snappi tests.
+IMPORTANT: Please use the standard labels:
+testbed.id, os.version, testcase, testrun.id, device.id, psu.id, port.id, sensor.id
+"""
+
+import logging
+import json
+import datetime
+import time
+
+from typing import List, Dict, Union
+from metrics_reporter import MetricReporterFactory, MetricsReporter
+
+class Metric:
+    def __init__(self,
+                 name: str,
+                 description: str,
+                 unit: str,
+                 reporter: MetricReporterFactory):
+        """
+        Args:
+            name (str): metric name (e.g., psu power, sensor temperature, port stats, etc.)
+            description (str): brief description of the metric
+            unit (str): metric unit (e.g., seconds, bytes)
+            reporter (MetricReporterFactory): object of MetricReporterFactory
+        """
+        self.name = name
+        self.description = description
+        self.unit = unit
+        self.reporter = reporter
+
+    def __repr__(self):
+        return (f"Metric(name={self.name!r}, "
+                f"description={self.description!r}, "
+                f"unit={self.unit!r}, "
+                f"reporter={self.reporter!r})")
+
+
+class GaugeMetric(Metric):
+    def __init__(self,
+                 name: str,
+                 description: str,
+                 unit: str,
+                 reporter: MetricReporterFactory):
+        # Initialize the base class
+        super().__init__(name, description, unit, reporter)
+
+    def set_gauge_metric(self, scope_labels: Dict[str, str], value: Union[int, str, float]):
+        # Add scope level labels and set the metric value
+        gauge_metric = {
+            "name": self.name,
+            "description": self.description,
+            "unit": self.unit,
+            **scope_labels,  # Add scope_labels to the dictionary
+            "value": value   # Add the metric value
+        }
+        self.reporter.update_metrics(gauge_metric)
+
+    def __repr__(self):
+        return (f"GaugeMetric(name={self.name!r}, "
+                f"description={self.description!r}, "
+                f"unit={self.unit!r}, "
+                f"reporter={self.reporter!r})")

tests/snappi_tests/metrics_utils/metrics_reporter.py

@@ -0,0 +1,60 @@
+# This file defines the classes reporting metrics and records to the corresponding database.
+
+import logging
+import json
+import datetime
+import time
+from pprint import pprint
+from typing import List, Dict, Union
+
+#from metrics_accepter import Metric, GaugeMetric
+
+class MetricReporterFactory:
+    def __init__(self):
+        self.reporter = None
+
+    def create_metrics_reporter(self, resource_labels: Dict[str, str]):
+        self.reporter = MetricsReporter(resource_labels)
+        return self.reporter
+
+    def create_records_reporter(self, resource_labels: Dict[str, str]):
+        self.reporter = RecordsReporter(resource_labels)
+        return self.reporter
+
+
+class MetricsReporter:
+    def __init__(self, resource_labels: Dict[str, str]):
+        # Temporary code initializing a MetricsReporter
+        # will be replaced with a real initializer such as OpenTelemetry 
+        self.resource_labels = resource_labels
+        self.timestamp = int(time.time() * 1_000_000_000) # epoch time in nanoseconds
+        self.metrics = []
+
+    def update_metrics(self, gauge_metric: Dict[str, Union[int, str, float]]):
+        # add a new metric
+        self.metrics.append(gauge_metric)
+
+    def report(self):
+        """
+        Abstract method to report metrics at a given timestamp.
+        Subclasses must override this method.
+        pprint(self.metrics)
+        """
+        pass
+
+
+class RecordsReporter:
+    def __init__(self, resource_labels: Dict[str, str]):
+        # Temporary code initializing a RecordsReporter
+        # will be replaced with a real initializer such as Kusto
+        self.resource_labels = resource_labels
+        self.timestamp = int(time.time() * 1_000_000_000) # epoch time in nanoseconds
+        self.records = []
+
+    def report(self):
+        """
+        Abstract method to report records at a given timestamp.
+        Subclasses must override this method.
+        """
+        pass
+