Initial version of defining the interfaces to accept metrics

tests/common/telemetry/examples.py

@@ -0,0 +1,77 @@
+from reporter_factory import TelemetryReporterFactory
+from metrics import (
+    GaugeMetric,
+    METRIC_LABEL_TESTBED,
+    METRIC_LABEL_TEST_BUILD,
+    METRIC_LABEL_TEST_CASE,
+    METRIC_LABEL_TEST_FILE,
+    METRIC_LABEL_TEST_JOBID,
+    METRIC_LABEL_DEVICE_ID,
+    METRIC_LABEL_DEVICE_PSU_ID,
+    METRIC_LABEL_DEVICE_PSU_MODEL,
+    METRIC_LABEL_DEVICE_PSU_SERIAL
+)
+
+
+def main():
+    """
+
+    PSU    Model            Serial             HW Rev    Voltage (V)    Current (A)    Power (W)  Status    LED
+    -----  ---------------  ---------------  --------  -------------  -------------  -----------  --------  -----
+    PSU 1  PWR-ABCD         1Z011010112349Q        01          12.09          18.38       222.00  OK        green
+    PSU 2  PWR-ABCD         1Z011010156787X        01          12.01          17.72       214.00  OK        green
+
+    """
+    common_labels = {
+        METRIC_LABEL_TESTBED: "TB-XYZ",
+        METRIC_LABEL_TEST_BUILD: "2024.1103",
+        METRIC_LABEL_TEST_CASE: "mock-case",
+        METRIC_LABEL_TEST_FILE: "mock-test.py",
+        METRIC_LABEL_TEST_JOBID: "2024_1225_0621"
+    }
+
+    # Create a MetricReporterFactory and build a MetricReporter
+    reporter = TelemetryReporterFactory.create_periodic_metrics_reporter(common_labels)
+
+    metric_labels = {METRIC_LABEL_DEVICE_ID: "switch-A"}
+
+    # Create a metric
+    voltage = GaugeMetric(name="Voltage",
+                          description="Power supply unit voltage reading",
+                          unit="V",
+                          reporter=reporter)
+
+    # Create a metric
+    current = GaugeMetric(name="Current",
+                          description="Power supply unit current reading",
+                          unit="A",
+                          reporter=reporter)
+
+    # Create a metric
+    power = GaugeMetric(name="Power",
+                        description="Power supply unit power reading",
+                        unit="W",
+                        reporter=reporter)
+
+    # Pass metrics to the reporter
+    metric_labels[METRIC_LABEL_DEVICE_PSU_ID] = "PSU 1"
+    metric_labels[METRIC_LABEL_DEVICE_PSU_MODEL] = "PWR-ABCD"
+    metric_labels[METRIC_LABEL_DEVICE_PSU_SERIAL] = "1Z011010112349Q"
+    voltage.record(metric_labels, 12.09)
+    current.record(metric_labels, 18.38)
+    power.record(metric_labels, 222.00)
+
+    # Pass metrics to the reporter
+    metric_labels[METRIC_LABEL_DEVICE_PSU_ID] = "PSU 2"
+    metric_labels[METRIC_LABEL_DEVICE_PSU_MODEL] = "PWR-ABCD"
+    metric_labels[METRIC_LABEL_DEVICE_PSU_SERIAL] = "1Z011010156787X"
+    voltage.record(metric_labels, 12.01)
+    current.record(metric_labels, 17.72)
+    power.record(metric_labels, 214.00)
+
+    # Report all metrics at a specific timestamp
+    reporter.report()
+
+
+if __name__ == '__main__':
+    main()

tests/common/telemetry/metrics.py

@@ -0,0 +1,118 @@
+"""
+This file defines the classes receiving metrics from snappi tests and processing them.
+"""
+import time
+
+from copy import deepcopy
+from typing import Dict, Final, Union
+
+# Only certain labels are allowed
+METRIC_LABEL_TESTBED: Final[str] = "test.testbed"                   # testbed name/ID
+METRIC_LABEL_TEST_BUILD: Final[str] = "test.os.version"             # Software/build version
+METRIC_LABEL_TEST_CASE: Final[str] = "test.testcase"                # test case name/ID
+METRIC_LABEL_TEST_FILE: Final[str] = "test.file"                    # test file name
+METRIC_LABEL_TEST_JOBID: Final[str] = "test.job.id"                 # test job ID
+METRIC_LABEL_DEVICE_ID: Final[str] = "device.id"                    # device refers to the level of switch
+METRIC_LABEL_DEVICE_PORT_ID: Final[str] = "device.port.id"
+METRIC_LABEL_DEVICE_PSU_ID: Final[str] = "device.psu.id"
+METRIC_LABEL_DEVICE_PSU_MODEL: Final[str] = "device.psu.model"
+METRIC_LABEL_DEVICE_PSU_SERIAL: Final[str] = "device.psu.serial"
+METRIC_LABEL_DEVICE_PSU_HW_REV: Final[str] = "device.psu.hw_rev"    # hardware revision
+METRIC_LABEL_DEVICE_QUEUE_ID: Final[str] = "device.queue.id"
+METRIC_LABEL_DEVICE_QUEUE_CAST: Final[str] = "device.queue.cast"    # unicast or multicast
+METRIC_LABEL_DEVICE_SENSOR_ID: Final[str] = "device.sensor.id"
+METRIC_LABEL_DEVICE_PG_ID: Final[str] = "device.pg.id"              # priority group
+METRIC_LABEL_DEVICE_BUFFER_POOL_ID: Final[str] = "device.buffer_pool.id"
+
+
+class PeriodicMetricsReporter:
+    def __init__(self, common_labels: Dict[str, str]):
+        # Will be replaced with a real initializer such as OpenTelemetry
+        self.common_labels = deepcopy(common_labels)
+        self.metrics = []
+
+    def stash_record(self, new_metric: 'Metric', labels: Dict[str, str], value: Union[int, float]):
+        # add a new periodic metric
+        copied_labels = deepcopy(labels)
+        self.metrics.append({"labels": copied_labels, "value": value})
+
+    def report(self, timestamp=time.time_ns()):
+        """
+        Report metrics at a given timestamp.
+        The input timestamp must be UNIX Epoch time in nanoseconds since 00:00:00 UTC on 1 January 1970
+
+        # save the metrics in a local variable and release the metrics in the object
+        stashed_metrics = self.metrics
+        self.metrics = []
+
+        """
+        pass
+
+
+class FinalMetricsReporter:
+    def __init__(self, common_labels: Dict[str, str]):
+        # Will be replaced with a real initializer such as Kusto
+        self.common_labels = deepcopy(common_labels)
+        self.metrics = []
+
+    def stash_record(self, new_metric: 'Metric', labels: Dict[str, str], value: Union[int, float]):
+        # add a new final metric
+        copied_labels = deepcopy(labels)
+        self.metrics.append({"labels": copied_labels, "value": value})
+
+    def report(self, timestamp=time.time_ns()):
+        """
+        Report metrics at a given timestamp.
+        The input timestamp must be UNIX Epoch time in nanoseconds since 00:00:00 UTC on 1 January 1970
+
+        # save the metrics in a local variable and release the metrics in the object
+        stashed_metrics = self.metrics
+        self.metrics = []
+
+        """
+        pass
+
+
+class Metric:
+    def __init__(self,
+                 name: str,
+                 description: str,
+                 unit: str,
+                 reporter: PeriodicMetricsReporter):
+        """
+        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 (PeriodicMetricsReporter): object of PeriodicMetricsReporter
+        """
+        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})")
+
+
+class GaugeMetric(Metric):
+    def __init__(self,
+                 name: str,
+                 description: str,
+                 unit: str,
+                 reporter: PeriodicMetricsReporter):
+        # Initialize the base class
+        super().__init__(name, description, unit, reporter)
+
+    def record(self, metric_labels: Dict[str, str], value: Union[int, float]):
+        # Save the metric into the reporter
+        self.reporter.stash_record(self, metric_labels, value)
+
+    def __repr__(self):
+        return (f"GaugeMetric(name={self.name!r}, "
+                f"description={self.description!r}, "
+                f"unit={self.unit!r}, "
+                f"reporter={self.reporter})")

tests/common/telemetry/reporter_factory.py

@@ -0,0 +1,13 @@
+from typing import Dict
+from metrics import PeriodicMetricsReporter, FinalMetricsReporter
+
+
+class TelemetryReporterFactory:
+    def __init__(self):
+        return
+
+    def create_periodic_metrics_reporter(common_labels: Dict[str, str]):
+        return (PeriodicMetricsReporter(common_labels))
+
+    def create_final_metrics_reporter(common_labels: Dict[str, str]):
+        return (FinalMetricsReporter(common_labels))