definition of the metrics to be measured and update of the simulation input

Author: GioeleB00

src/app/config/constants.py

@@ -182,3 +182,43 @@ class SystemEdges(StrEnum):
     """
 
     NETWORK_CONNECTION = "network_connection"
+
+# ======================================================================
+# CONSTANTS FOR SAMPLED METRICS
+# ======================================================================
+
+class SampledMetricName(StrEnum):
+  """
+  define the metrics sampled every fixed amount of
+  time to create a time series
+  """
+
+  READY_QUEUE_LEN = "ready_queue_len" #length of the event loop ready q
+  CORE_BUSY = "core_busy"
+  EVENT_LOOP_IO_SLEEP = "event_loop_io_sleep"
+  RAM_IN_USE = "ram_in_use"
+  THROUGHPUT_RPS = "throughput_rps"
+
+# ======================================================================
+# CONSTANTS FOR EVENT METRICS
+# ======================================================================
+
+class EventMetricName(StrEnum):
+  """
+  define the metrics triggered by event with no
+  time series
+  """
+
+  RQS_LATENCY = "rqs_latency"
+  LLM_COST = "llm_cost"
+
+
+# ======================================================================
+# CONSTANTS FOR AGGREGATED METRICS
+# ======================================================================
+
+class AggregatedMetricName(StrEnum):
+  """aggregated metrics to calculate at the end of simulation"""
+
+  LATENCY_STATS = "latency_stats"
+  LLM_STATS = "llm_stats"

src/app/core/helpers.py

@@ -0,0 +1,38 @@
+"""helpers for the simulation"""
+
+from collections.abc import Iterable
+
+from app.config.constants import EventMetricName, SampledMetricName
+
+
+def alloc_sample_metric(
+    enabled_sample_metrics: Iterable[SampledMetricName],
+    ) -> dict[str, list[float | int]]:
+    """
+    After the pydantic validation of the whole input we
+    instantiate a dictionary to collect the sampled metrics the
+    user want to measure
+    """
+    # t is the alignmente parameter for example assume
+    # the snapshot for the sampled metrics are done every 10ms
+    # t = [10,20,30,40....] to each t will correspond a measured
+    # metric corresponding to that time interval
+
+    dict_sampled_metrics: dict[str, list[float | int]] = {"t": []}
+    for key in enabled_sample_metrics:
+        dict_sampled_metrics[key] = []
+    return dict_sampled_metrics
+
+
+def alloc_event_metric(
+    enabled_event_metrics: Iterable[EventMetricName],
+    ) -> dict[str, list[float | int]]:
+    """
+    After the pydantic validation of the whole input we
+    instantiate a dictionary to collect the event metrics the
+    user want to measure
+    """
+    dict_event_metrics: dict[str, list[float | int]]  = {}
+    for key in enabled_event_metrics:
+        dict_event_metrics[key] = []
+    return dict_event_metrics

tests/unit/simulation/test_requests_generator.py

@@ -16,9 +16,9 @@
 from app.schemas.requests_generator_input import RqsGeneratorInput
 from app.schemas.simulation_settings_input import SimulationSettings
 from app.schemas.system_topology_schema.full_system_topology_schema import (
-    TopologyGraph,
     Client,
-    TopologyNodes
+    TopologyGraph,
+    TopologyNodes,
 )
 
 if TYPE_CHECKING:  # Only for static type-checking