ft-analyzer add active time and in cache time

Author: pascal260303

tools/ft-analyzer/ftanalyzer/counter/__init__.py

@@ -1,3 +1,4 @@
 from .counter import Counter
 from .continuous_counter import ContinuousCounter
 from .time_series_counter import TimeSeriesCounter
+from .discrete_counter import DiscreteCounter

tools/ft-analyzer/ftanalyzer/events/events.py

@@ -32,6 +32,8 @@
     "PACKETS": np.uint64,
     "BYTES": np.uint64,
     "FLOWS": np.uint64,
+    "ACTIVE_TIME": np.uint64,
+    "CACHE_TIME": np.uint64,
 }
 
 STATS_CSV_COLUMN_TYPES = {
@@ -120,26 +122,43 @@ class FlowEndEvent(Event):
     time = 0
     flow_rate: float
     flows: int
-
-    def __init__(self, data_rate, packet_rate, end_time, flow_rate, flows):
+    active_time: np.uint64
+    cache_time: np.uint64
+
+    def __init__(
+        self,
+        data_rate,
+        packet_rate,
+        end_time,
+        flow_rate,
+        flows,
+        active_time,
+        cache_time,
+    ):
         self.data_rate = -data_rate
         self.packet_rate = -packet_rate
         self.time = end_time
         self.flow_rate = -flow_rate
         self.flows = -int(flows)
+        self.active_time = active_time
+        self.cache_time = cache_time
 
 
 class OnePacketFlow(Event):
     bytes: np.uint64
     packets: np.uint64
     time = 0
     flows: np.uint64
+    active_time: np.uint64
+    cache_time: np.uint64
 
-    def __init__(self, bytes, packets, time, flows):
+    def __init__(self, bytes, packets, time, flows, active_time, cache_time):
         self.bytes = bytes
         self.packets = packets
         self.time = time
         self.flows = flows
+        self.active_time = active_time
+        self.cache_time = cache_time
 
 
 class ExportEvent(Event):
@@ -244,6 +263,10 @@ def create_event_queue(
         temp_end = tempfile.NamedTemporaryFile(
             mode="w", prefix="end_time", suffix=".csv", dir=out_dir, delete=False
         )
+        temp_export = tempfile.NamedTemporaryFile(
+            mode="w", prefix="export_time", suffix=".csv", dir=out_dir, delete=False
+        )
+        tmp_export.append(temp_export)
         tmp_start_time.append(temp_start)
         tmp_end_time.append(temp_end)
         tmp_one_pack.append(temp_one)
@@ -255,10 +278,23 @@ def create_event_queue(
         else:
             end = max(end, math.ceil(chunk["END_TIME"].max() / 1000))
 
+        # Export Events
+        if "EXPORT_TIME" not in chunk.columns:
+            # accurate expected EXPORT_TIME
+            chunk["EXPORT_TIME"] = chunk["END_TIME"] // 1000 + inactive_timeout + 1
+            # approximate SEQ_NUMBER
+            chunk["SEQ_NUMBER"] = chunk["EXPORT_TIME"] % 32
+            # random MSG_LENGTH
+            chunk["MSG_LENGTH"] = random.randint(100, 2048)
+        chunk["ACTIVE_TIME"] = chunk["END_TIME"] - chunk["START_TIME"] + 1
+        chunk["CACHE_TIME"] = (
+            chunk["EXPORT_TIME"] * 1000 - chunk["START_TIME"] + 1
+        ).clip(lower=0)
+
         # One-packet flows
         (
             chunk[chunk["PACKETS"] == 1]
-            .groupby("START_TIME", as_index=False)
+            .groupby(["START_TIME", "ACTIVE_TIME", "CACHE_TIME"], as_index=False)
             .agg(**agg_dict)
             .sort_values("START_TIME")
             .to_csv(
@@ -267,19 +303,6 @@ def create_event_queue(
             )
         )
 
-        # Export Events
-        if "EXPORT_TIME" not in chunk.columns:
-            # accurate expected EXPORT_TIME
-            chunk["EXPORT_TIME"] = chunk["END_TIME"] // 1000 + inactive_timeout + 1
-            # approximate SEQ_NUMBER
-            chunk["SEQ_NUMBER"] = chunk["EXPORT_TIME"] % 32
-            # random MSG_LENGTH
-            chunk["MSG_LENGTH"] = random.randint(100, 2048)
-
-        temp_export = tempfile.NamedTemporaryFile(
-            mode="w", prefix="export_time", suffix=".csv", dir=out_dir, delete=False
-        )
-        tmp_export.append(temp_export)
         (
             chunk.groupby(["EXPORT_TIME", "SEQ_NUMBER"], as_index=False)
             .agg(
@@ -296,7 +319,9 @@ def create_event_queue(
         # Multi-packet flows
         (
             chunk[chunk["PACKETS"] > 1]
-            .groupby(["START_TIME", "END_TIME"], as_index=False)
+            .groupby(
+                ["START_TIME", "END_TIME", "ACTIVE_TIME", "CACHE_TIME"], as_index=False
+            )
             .agg(**agg_dict)
             .sort_values("START_TIME")
             .to_csv(
@@ -307,7 +332,9 @@ def create_event_queue(
 
         (
             chunk[chunk["PACKETS"] > 1]
-            .groupby(["START_TIME", "END_TIME"], as_index=False)
+            .groupby(
+                ["START_TIME", "END_TIME", "ACTIVE_TIME", "CACHE_TIME"], as_index=False
+            )
             .agg(**agg_dict)
             .sort_values("END_TIME")
             .to_csv(
@@ -403,6 +430,8 @@ def read_one_packet_events(path: str) -> Iterator[OnePacketFlow]:
             packets=row["PACKETS"],
             time=row["START_TIME"],
             flows=row["FLOWS"],
+            active_time=row["ACTIVE_TIME"],
+            cache_time=row["CACHE_TIME"],
         )
 
 
@@ -433,4 +462,6 @@ def read_end_events(path: str) -> Iterator[FlowEndEvent]:
             end_time=row["END_TIME"],
             flow_rate=flow_rate,
             flows=row["FLOWS"],
+            active_time=row["ACTIVE_TIME"],
+            cache_time=row["CACHE_TIME"],
         )

tools/ft-analyzer/ftanalyzer/models/statistical_model.py

@@ -23,7 +23,7 @@
 import numpy as np
 import pandas as pd
 from ftanalyzer.common.pandas_multiprocessing import PandasMultiprocessingHelper
-from ftanalyzer.events.events import ExportEvent, FlowStartEvent
+from ftanalyzer.events.events import ExportEvent, FlowStartEvent, FlowEndEvent
 from ftanalyzer.models.sm_data_types import (
     SMException,
     SMMetricType,
@@ -34,7 +34,7 @@
 )
 from ftanalyzer.reports import StatisticalReport
 from src.generator.interface import GeneratorStats
-from ftanalyzer.counter import ContinuousCounter, TimeSeriesCounter
+from ftanalyzer.counter import ContinuousCounter, TimeSeriesCounter, DiscreteCounter
 from ftanalyzer.statistic_object import StatisticObject, SimState
 from ftanalyzer.events import (
     Event,
@@ -665,6 +665,8 @@ def setup_statsitic_objects(
             measure_start_time=start_time_offset + inactive_timeout * 1000,
             measure_end_time=end_time,
         ),
+        "dt_flows_active_time": DiscreteCounter("Flow Duration Active"),
+        "dr_flows_cache_time": DiscreteCounter("Flow Duration in Cache"),
         "tsc_data_rate": TimeSeriesCounter(
             "data rate in Gb/s",
             sim,
@@ -730,6 +732,22 @@ def setup_statsitic_objects(
             measure_start_time=start_time_offset + inactive_timeout * 1000,
             measure_end_time=end_time,
         ),
+        "tsc_flows_active_time": TimeSeriesCounter(
+            "Flow Duration Active",
+            sim,
+            start_time,
+            end_time,
+            measure_start_time=start_time_offset,
+            measure_end_time=end_time_offset,
+        ),
+        "tsc_flows_cache_time": TimeSeriesCounter(
+            "Flow Duration in Cache",
+            sim,
+            start_time,
+            end_time,
+            measure_start_time=start_time_offset,
+            measure_end_time=end_time_offset,
+        ),
     }
 
     metric_mapping: dict[str, List[str]] = {
@@ -744,6 +762,8 @@ def setup_statsitic_objects(
             "tsc_flows_per_export_packet",
             "ct_flows_per_export_packet",
         ],
+        "active_time": ["dt_flows_active_time", "tsc_flows_active_time"],
+        "cache_time": ["dt_flows_cache_time", "tsc_flows_cache_time"],
     }
 
     return (statistic_objects, metric_mapping)
@@ -817,6 +837,15 @@ def _flush_event_data(
             statistic_objects, metric_mapping, **host_stats_event.row._asdict()
         )
 
+    for event in simultaneous_events:
+        if isinstance(event, OnePacketFlow) or isinstance(event, FlowEndEvent):
+            update_statistic_objects(
+                statistic_objects,
+                metric_mapping,
+                active_time=event.active_time,
+                cache_time=event.cache_time,
+            )
+
     return last_export