ft-analyzer better active flows calculation and predict export time

Author: pascal260303

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

@@ -4,6 +4,7 @@
 import heapq
 import math
 import os
+import random
 import shutil
 import tempfile
 from typing import Iterator
@@ -191,7 +192,10 @@ def convert_row_types(row):
 
 
 def create_event_queue(
-    flows_path: os.PathLike, hosts_stats_file: os.PathLike, out_dir: str = None
+    flows_path: os.PathLike,
+    hosts_stats_file: os.PathLike,
+    inactive_timeout: int = 30,
+    out_dir: str = None,
 ) -> Iterator[Event]:
     if not out_dir:
         out_dir = tempfile.mkdtemp(prefix="flows_split_")
@@ -264,23 +268,30 @@ def create_event_queue(
         )
 
         # Export Events
-        if "EXPORT_TIME" in chunk.columns:
-            temp_export = tempfile.NamedTemporaryFile(
-                mode="w", prefix="export_time", suffix=".csv", dir=out_dir, delete=False
+        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_NUMER"] = 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(
+                MSG_LENGTH=("MSG_LENGTH", "first"),
+                FLOWS=("MSG_LENGTH", "count"),
             )
-            tmp_export.append(temp_export)
-            (
-                chunk.groupby(["EXPORT_TIME", "SEQ_NUMBER"], as_index=False)
-                .agg(
-                    MSG_LENGTH=("MSG_LENGTH", "first"),
-                    FLOWS=("MSG_LENGTH", "count"),
-                )
-                .sort_values("EXPORT_TIME")
-                .to_csv(
-                    temp_export.name,
-                    index=False,
-                )
+            .sort_values("EXPORT_TIME")
+            .to_csv(
+                temp_export.name,
+                index=False,
             )
+        )
 
         # Multi-packet flows
         (

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
+from ftanalyzer.events.events import ExportEvent, FlowStartEvent
 from ftanalyzer.models.sm_data_types import (
     SMException,
     SMMetricType,
@@ -279,7 +279,9 @@ def _run_sim(
         statistic_objects, metric_to_obj = setup_statsitic_objects(
             sim, start_time, end_time, inactive_timeout
         )
-        event_queue = create_event_queue(flows_file, host_stats, output_dir)
+        event_queue = create_event_queue(
+            flows_file, host_stats, inactive_timeout, output_dir
+        )
         process_events(event_queue, statistic_objects, metric_to_obj, sim)
 
         shutil.rmtree(output_dir, ignore_errors=True)
@@ -764,19 +766,17 @@ def _flush_event_data(
     # aggregate OnePacketFlow events within this window
     one_packet_data_rate = sum(e.bytes for e in one_packet_events) * 8 / duration_s
     one_packet_packet_rate = sum(e.packets for e in one_packet_events) / duration_s
-    one_packet_flows = np.uint64(sum(e.flows for e in one_packet_events))
 
     # Compose final rates
     total_data_rate = one_packet_data_rate + current_data_rate
     total_packet_rate = one_packet_packet_rate + current_packet_rate
-    total_flows = one_packet_flows + current_flows
 
     update_statistic_objects(
         statistic_objects,
         metric_mapping,
         data_rate=total_data_rate,
         packet_rate=total_packet_rate,
-        flow_count=total_flows,
+        flow_count=current_flows,
     )
 
     export_events: List[ExportEvent] = [
@@ -889,14 +889,18 @@ def process_events(
 
         # apply each event's effect to current rates
         for e in simultaneous_events:
-            if isinstance(e, HostStatsEvent) or isinstance(e, ExportEvent):
+            if isinstance(e, HostStatsEvent):
                 pass
+            elif isinstance(e, ExportEvent):
+                current_flows -= e.flows
             elif isinstance(e, OnePacketFlow):
                 one_packet_events.append(e)
+                current_flows += e.flows
             else:
                 current_data_rate += e.data_rate
                 current_packet_rate += e.packet_rate
-                current_flows += e.flows
+                if isinstance(e, FlowStartEvent):
+                    current_flows += e.flows
 
         current_data_rate = max(0, current_data_rate)
         current_packet_rate = max(0, current_packet_rate)