Skip to content

Commit 7f4ee65

Browse files
pascal260303pascal260303
committed
ft-orchestration run csv creation / download and replication in parallel
1 parent 8332ad2 commit 7f4ee65

4 files changed

Lines changed: 87 additions & 43 deletions

File tree

‎tools/ft-orchestration/src/collector/jq.py‎

Lines changed: 21 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,11 @@
5151
"MSG_LENGTH": np.uint64,
5252
}
5353

54+
CONN_USER = None
55+
CONN_HOST = None
56+
CONN_KWARGS = None
57+
CMD_CSV = None
58+
5459

5560
class JQ(CollectorOutputReaderInterface):
5661
CONFIG_FILE = "config.xml"
@@ -67,6 +72,12 @@ def __init__(self, executor: Executor, file: str):
6772
assert_tool_is_installed("ipfixcol2", executor)
6873
assert_tool_is_installed("jq", executor)
6974

75+
global CONN_USER, CONN_HOST, CONN_KWARGS, CMD_CSV
76+
connection: fabric.Connection = executor._connection
77+
CONN_USER = connection.user
78+
CONN_HOST = connection.host
79+
CONN_KWARGS = connection.connect_kwargs
80+
7081
command = Tool(f"ls {file} -1", executor=executor, failure_verbosity="silent")
7182
stdout, _ = command.run()
7283
if command.returncode() != 0:
@@ -104,6 +115,8 @@ def __init__(self, executor: Executor, file: str):
104115
In the csv output the columns `iana:sourceIPv4Address` and `iana:sourceIPv6Address` are merged to `iana:sourceIPAddress`\\
105116
The same is done for `iana:destinationIPAddress`
106117
"""
118+
CMD_CSV = self._cmd_csv
119+
107120
self._process = None
108121
self._buf = None
109122
self._idx = 0
@@ -281,7 +294,8 @@ def __next__(self):
281294

282295
raise StopIteration
283296

284-
def save_csv(self, csv_file: str):
297+
@staticmethod
298+
def save_csv(csv_file: str):
285299
"""Convert flows from FDS format to CSV file.
286300
Used for significant amount of flows in performance testing.
287301
@@ -291,19 +305,13 @@ def save_csv(self, csv_file: str):
291305
Path to CSV file. Local file, CSV will be downloaded when collector running on remote.
292306
"""
293307
header = CSV_HEADER_TO_ANALYZER_HEADER.values()
294-
conn: fabric.Connection = self._executor._connection
295-
user = conn.user
296-
host = conn.host
297-
kwargs = conn.connect_kwargs
298-
299-
if "key_filename" in kwargs:
300-
key_filename = kwargs["key_filename"][0]
301-
ssh_cmd = (
302-
f"ssh -i {key_filename} {user}@{host} '{self._cmd_csv}' >> {csv_file}"
303-
)
308+
309+
if "key_filename" in CONN_KWARGS:
310+
key_filename = CONN_KWARGS["key_filename"][0]
311+
ssh_cmd = f"ssh -i {key_filename} {CONN_USER}@{CONN_HOST} '{CMD_CSV}' >> {csv_file}"
304312
else:
305-
password = kwargs["password"]
306-
ssh_cmd = f"sshpass -p {password} ssh {user}@{host} '{self._cmd_csv}' >> {csv_file}"
313+
password = CONN_KWARGS["password"]
314+
ssh_cmd = f"sshpass -p {password} ssh {CONN_USER}@{CONN_HOST} '{CMD_CSV}' >> {csv_file}"
307315

308316
logging.getLogger().info(
309317
"Preparing CSV output by calling ipfixcol2 + jq command..."

‎tools/ft-orchestration/tests/simulation/test_simulation_general.py‎

Lines changed: 22 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,11 @@
88
Allows to modify most of the configuration options.
99
"""
1010

11+
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
1112
import ipaddress
1213
import logging
1314
import os
15+
import sys
1416
from typing import Optional
1517

1618
import pytest
@@ -41,6 +43,8 @@
4143
from src.generator.interface import GeneratorStats, MultiplierSpeed, Replicator
4244
from src.probe.probe_builder import ProbeBuilder
4345

46+
EXECUTOR_CLS = ThreadPoolExecutor if sys.gettrace() is not None else ProcessPoolExecutor
47+
4448
PROJECT_ROOT = get_project_root()
4549
SIMULATION_TESTS_DIR = os.path.join(PROJECT_ROOT, "testing/simulation")
4650
select_topologies(["replicator"])
@@ -348,17 +352,26 @@ def finalizer_download_logs():
348352
collector_instance.stop()
349353

350354
flows_file = os.path.join(tmp_dir, "flows.csv")
351-
collector_instance.get_reader().save_csv(flows_file)
352355

353-
replicated_ref = flow_replicator.replicate(
354-
input_file=ref_file,
355-
loops=scenario.test.loops,
356-
generator_stats=stats,
357-
speed_multiplier=speed.speed if isinstance(speed, MultiplierSpeed) else 1.0,
358-
)
356+
# get flows.csv and replicated reference file in parallel
357+
with EXECUTOR_CLS() as parallel_executor:
358+
flows_file_future = parallel_executor.submit(
359+
collector_instance.get_reader().save_csv, flows_file
360+
)
361+
362+
replicated_ref_future = parallel_executor.submit(
363+
flow_replicator.replicate,
364+
input_file=ref_file,
365+
loops=scenario.test.loops,
366+
generator_stats=stats,
367+
speed_multiplier=speed.speed if isinstance(speed, MultiplierSpeed) else 1.0,
368+
)
369+
370+
if not probe_instance.host_statistics.local_file:
371+
probe_instance.host_statistics.get_csv(tmp_dir)
359372

360-
if not probe_instance.host_statistics.local_file:
361-
probe_instance.host_statistics.get_csv(tmp_dir)
373+
flows_file_future.result()
374+
replicated_ref = replicated_ref_future.result()
362375

363376
stats_report, precise_report = validate(
364377
analysis=scenario.test.analysis,

‎tools/ft-orchestration/tests/simulation/test_simulation_overload.py‎

Lines changed: 22 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,11 @@
88
and evaluating the number of processed packets before and after the flooding period.
99
"""
1010

11+
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
1112
import ipaddress
1213
import logging
1314
import os
15+
import sys
1416

1517
import pytest
1618
from ftanalyzer.models.sm_data_types import SMRule, SMSubnetSegment
@@ -31,6 +33,8 @@
3133
from src.generator.interface import MultiplierSpeed, Replicator
3234
from src.probe.probe_builder import ProbeBuilder
3335

36+
EXECUTOR_CLS = ThreadPoolExecutor if sys.gettrace() is not None else ProcessPoolExecutor
37+
3438
PROJECT_ROOT = get_project_root()
3539
SIMULATION_TESTS_DIR = os.path.join(PROJECT_ROOT, "testing/simulation")
3640
DEFAULT_REPLICATOR_PREFIX = 8
@@ -296,17 +300,26 @@ def finalizer_download_logs():
296300
collector_instance.stop()
297301

298302
flows_file = os.path.join(tmp_dir, "flows.csv")
299-
collector_instance.get_reader().save_csv(flows_file)
300303

301-
replicated_ref = flow_replicator.replicate(
302-
input_file=ref_file,
303-
loops=LOOPS,
304-
generator_stats=stats,
305-
speed_multiplier=speed.speed if isinstance(speed, MultiplierSpeed) else 1.0,
306-
)
304+
# get flows.csv and replicated reference file in parallel
305+
with EXECUTOR_CLS() as parallel_executor:
306+
flows_file_future = parallel_executor.submit(
307+
collector_instance.get_reader().save_csv, flows_file
308+
)
309+
310+
replicated_ref_future = parallel_executor.submit(
311+
flow_replicator.replicate,
312+
input_file=ref_file,
313+
loops=LOOPS,
314+
generator_stats=stats,
315+
speed_multiplier=speed.speed if isinstance(speed, MultiplierSpeed) else 1.0,
316+
)
317+
318+
if not probe_instance.host_statistics.local_file:
319+
probe_instance.host_statistics.get_csv(tmp_dir)
307320

308-
if not probe_instance.host_statistics.local_file:
309-
probe_instance.host_statistics.get_csv(tmp_dir)
321+
flows_file_future.result()
322+
replicated_ref = replicated_ref_future.result()
310323

311324
model = StatisticalModel(
312325
flows_file,

‎tools/ft-orchestration/tests/simulation/test_simulation_threshold.py‎

Lines changed: 22 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,13 @@
88
can be replayed so that the number of lost packets and bytes is below specified threshold.
99
"""
1010

11+
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
1112
import gc
1213
import ipaddress
1314
import logging
1415
import os
1516
import shutil
16-
import time
17+
import sys
1718

1819
import pytest
1920
from ftanalyzer.models.sm_data_types import SMRule
@@ -34,6 +35,8 @@
3435
from src.generator.interface import GeneratorStats, MbpsSpeed, Replicator
3536
from src.probe.probe_builder import ProbeBuilder
3637

38+
EXECUTOR_CLS = ThreadPoolExecutor if sys.gettrace() is not None else ProcessPoolExecutor
39+
3740
PROJECT_ROOT = get_project_root()
3841
SIMULATION_TESTS_DIR = os.path.join(PROJECT_ROOT, "testing/simulation")
3942
select_topologies(["replicator"])
@@ -245,21 +248,28 @@ def run_single_test(loops: int, speed: MbpsSpeed) -> tuple[bool, StatisticalRepo
245248
probe_instance.stop()
246249
collector_instance.stop()
247250

248-
collector_instance.get_reader().save_csv(flows_file)
249-
logging.getLogger().debug("Start applying flow replicator...")
250-
start = time.time()
251-
replicated_ref = flow_replicator.replicate(
252-
input_file=ref_file, loops=loops, generator_stats=stats
253-
)
254-
end = time.time()
255-
logging.getLogger().debug("Flows replicated in %.2f seconds.", (end - start))
251+
# get flows.csv and replicated reference file in parallel
252+
with EXECUTOR_CLS() as parallel_executor:
253+
flows_file_future = parallel_executor.submit(
254+
collector_instance.get_reader().save_csv, flows_file
255+
)
256+
logging.getLogger().debug("Start applying flow replicator...")
257+
replicated_ref_future = parallel_executor.submit(
258+
flow_replicator.replicate,
259+
input_file=ref_file,
260+
loops=loops,
261+
generator_stats=stats,
262+
)
263+
264+
if not probe_instance.host_statistics.local_file:
265+
probe_instance.host_statistics.get_csv(tmp_dir)
266+
267+
flows_file_future.result()
268+
replicated_ref = replicated_ref_future.result()
256269

257270
flow_replicator = None
258271
gc.collect()
259272

260-
if not probe_instance.host_statistics.local_file:
261-
probe_instance.host_statistics.get_csv(tmp_dir)
262-
263273
ret, report = validate(
264274
analysis=scenario.test.analysis,
265275
flows_file=flows_file,

0 commit comments

Comments
 (0)