From b4e4e27c4ba4ef6036c13d742fabd53386bddd67 Mon Sep 17 00:00:00 2001 From: malmert Date: Sun, 3 May 2026 19:13:24 +0200 Subject: [PATCH] try overload fix 2, backend upsert and test script fix --- backend/src/config.py | 5 + backend/src/utilities/database.py | 323 +++++++++++++----------- backend/src/utilities/packet_tracker.py | 60 +++-- tools/network_timing_benchmark.py | 67 ++++- 4 files changed, 281 insertions(+), 174 deletions(-) diff --git a/backend/src/config.py b/backend/src/config.py index 0cef55e..cacf04e 100644 --- a/backend/src/config.py +++ b/backend/src/config.py @@ -43,6 +43,7 @@ class BackendSettings: packet_tracker_retention_seconds: float packet_tracker_min_flush_interval_seconds: float packet_tracker_persist_timeout_seconds: float + packet_tracker_batch_persist_timeout_seconds: float packet_tracker_persist_retry_backoff_seconds: float packet_tracker_persist_retry_backoff_max_seconds: float packet_tracker_error_log_interval_seconds: float @@ -89,6 +90,10 @@ def load_settings() -> BackendSettings: packet_tracker_retention_seconds=_env_float("BACKEND_PACKET_TRACKER_RETENTION_SECONDS", 10.0), packet_tracker_min_flush_interval_seconds=_env_float("BACKEND_PACKET_TRACKER_MIN_FLUSH_INTERVAL_SECONDS", 0.05), packet_tracker_persist_timeout_seconds=_env_float("BACKEND_PACKET_TRACKER_PERSIST_TIMEOUT_SECONDS", 2.0), + packet_tracker_batch_persist_timeout_seconds=_env_float( + "BACKEND_PACKET_TRACKER_BATCH_PERSIST_TIMEOUT_SECONDS", + 10.0, + ), packet_tracker_persist_retry_backoff_seconds=_env_float( "BACKEND_PACKET_TRACKER_PERSIST_RETRY_BACKOFF_SECONDS", 0.25, diff --git a/backend/src/utilities/database.py b/backend/src/utilities/database.py index a6d0ef5..5780363 100644 --- a/backend/src/utilities/database.py +++ b/backend/src/utilities/database.py @@ -263,6 +263,42 @@ class DatabasePool: if self._pool is None: await self.init_pool() + if self._pool is None: + return + + try: + async with self._pool.acquire() as conn: + row = await self._upsert_packet_with_conn(conn, pkt_info) + except Exception: + logger.exception("DB upsert failed") + return + + self._publish_packet_row(pkt_info, row) + + async def upsert_packets(self, packets: List[Dict[str, Any]]) -> None: + """Insert or update packet records using one database connection for the batch.""" + if not packets: + return + if self._pool is None: + await self.init_pool() + if self._pool is None: + return + + published_rows: List[tuple[Dict[str, Any], Optional[Dict[str, Any]]]] = [] + try: + async with self._pool.acquire() as conn: + for pkt_info in packets: + row = await self._upsert_packet_with_conn(conn, pkt_info) + published_rows.append((pkt_info, row)) + except Exception: + logger.exception("DB packet batch upsert failed") + raise + + for pkt_info, row in published_rows: + self._publish_packet_row(pkt_info, row) + + async def _upsert_packet_with_conn(self, conn: asyncpg.Connection, pkt_info: Dict[str, Any]) -> Optional[Dict[str, Any]]: + """Insert or update one packet record using an already-acquired connection.""" _normalize_json_fields(pkt_info) _attach_derived_fields(pkt_info) @@ -270,154 +306,151 @@ class DatabasePool: telemetry_metadata = pkt_info.get("telemetry_metadata") capture_observations = pkt_info.get("capture_observations") - try: - async with self._pool.acquire() as conn: - row = await conn.fetchrow( - """ - INSERT INTO packets ( - timestamp, - correlation_key, - packet_id, - packet_uid, - flow_id, - capture_session_id, - correlation_source, - skb_mark, - capture_iface, - ingress_if, - egress_if, - verdict, - verdict_reason, - verdict_confidence, - ingress_seen_at, - egress_seen_at, - verdict_seen_at, - src_mac, - dst_mac, - eth_type_raw, - vlan_id, - src_ip, - dst_ip, - ip_proto_raw, - src_port, - dst_port, - length, - capture_sources, - app_protocol, - app_category, - app_confidence, - app_hostname, - app_is_encrypted, - app_risk_score, - dpi_metadata, - capture_metadata, - telemetry_metadata, - capture_observations, - raw - ) VALUES( - $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20, - $21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31,$32,$33,$34,$35::jsonb,$36::jsonb,$37::jsonb,$38::jsonb,$39 + row = await conn.fetchrow( + """ + INSERT INTO packets ( + timestamp, + correlation_key, + packet_id, + packet_uid, + flow_id, + capture_session_id, + correlation_source, + skb_mark, + capture_iface, + ingress_if, + egress_if, + verdict, + verdict_reason, + verdict_confidence, + ingress_seen_at, + egress_seen_at, + verdict_seen_at, + src_mac, + dst_mac, + eth_type_raw, + vlan_id, + src_ip, + dst_ip, + ip_proto_raw, + src_port, + dst_port, + length, + capture_sources, + app_protocol, + app_category, + app_confidence, + app_hostname, + app_is_encrypted, + app_risk_score, + dpi_metadata, + capture_metadata, + telemetry_metadata, + capture_observations, + raw + ) VALUES( + $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20, + $21,$22,$23,$24,$25,$26,$27,$28,$29,$30,$31,$32,$33,$34,$35::jsonb,$36::jsonb,$37::jsonb,$38::jsonb,$39 + ) + ON CONFLICT (correlation_key) DO UPDATE SET + updated_at = NOW(), + timestamp = CASE + WHEN packets.timestamp IS NULL THEN EXCLUDED.timestamp + WHEN EXCLUDED.timestamp IS NULL THEN packets.timestamp + ELSE LEAST(packets.timestamp, EXCLUDED.timestamp) + END, + packet_id = COALESCE(EXCLUDED.packet_id, packets.packet_id), + packet_uid = COALESCE(EXCLUDED.packet_uid, packets.packet_uid), + flow_id = COALESCE(EXCLUDED.flow_id, packets.flow_id), + capture_session_id = COALESCE(EXCLUDED.capture_session_id, packets.capture_session_id), + correlation_source = COALESCE(EXCLUDED.correlation_source, packets.correlation_source), + skb_mark = COALESCE(EXCLUDED.skb_mark, packets.skb_mark), + capture_iface = COALESCE(EXCLUDED.capture_iface, packets.capture_iface), + ingress_if = COALESCE(EXCLUDED.ingress_if, packets.ingress_if), + egress_if = COALESCE(EXCLUDED.egress_if, packets.egress_if), + verdict = COALESCE(EXCLUDED.verdict, packets.verdict), + verdict_reason = COALESCE(EXCLUDED.verdict_reason, packets.verdict_reason), + verdict_confidence = COALESCE(EXCLUDED.verdict_confidence, packets.verdict_confidence), + ingress_seen_at = COALESCE(EXCLUDED.ingress_seen_at, packets.ingress_seen_at), + egress_seen_at = COALESCE(EXCLUDED.egress_seen_at, packets.egress_seen_at), + verdict_seen_at = COALESCE(EXCLUDED.verdict_seen_at, packets.verdict_seen_at), + src_mac = COALESCE(EXCLUDED.src_mac, packets.src_mac), + dst_mac = COALESCE(EXCLUDED.dst_mac, packets.dst_mac), + eth_type_raw = COALESCE(EXCLUDED.eth_type_raw, packets.eth_type_raw), + vlan_id = COALESCE(EXCLUDED.vlan_id, packets.vlan_id), + src_ip = COALESCE(EXCLUDED.src_ip, packets.src_ip), + dst_ip = COALESCE(EXCLUDED.dst_ip, packets.dst_ip), + ip_proto_raw = COALESCE(EXCLUDED.ip_proto_raw, packets.ip_proto_raw), + src_port = COALESCE(EXCLUDED.src_port, packets.src_port), + dst_port = COALESCE(EXCLUDED.dst_port, packets.dst_port), + length = COALESCE(EXCLUDED.length, packets.length), + capture_sources = ( + SELECT ARRAY( + SELECT DISTINCT source + FROM unnest( + COALESCE(packets.capture_sources, ARRAY[]::text[]) || + COALESCE(EXCLUDED.capture_sources, ARRAY[]::text[]) + ) AS source ) - ON CONFLICT (correlation_key) DO UPDATE SET - updated_at = NOW(), - timestamp = CASE - WHEN packets.timestamp IS NULL THEN EXCLUDED.timestamp - WHEN EXCLUDED.timestamp IS NULL THEN packets.timestamp - ELSE LEAST(packets.timestamp, EXCLUDED.timestamp) - END, - packet_id = COALESCE(EXCLUDED.packet_id, packets.packet_id), - packet_uid = COALESCE(EXCLUDED.packet_uid, packets.packet_uid), - flow_id = COALESCE(EXCLUDED.flow_id, packets.flow_id), - capture_session_id = COALESCE(EXCLUDED.capture_session_id, packets.capture_session_id), - correlation_source = COALESCE(EXCLUDED.correlation_source, packets.correlation_source), - skb_mark = COALESCE(EXCLUDED.skb_mark, packets.skb_mark), - capture_iface = COALESCE(EXCLUDED.capture_iface, packets.capture_iface), - ingress_if = COALESCE(EXCLUDED.ingress_if, packets.ingress_if), - egress_if = COALESCE(EXCLUDED.egress_if, packets.egress_if), - verdict = COALESCE(EXCLUDED.verdict, packets.verdict), - verdict_reason = COALESCE(EXCLUDED.verdict_reason, packets.verdict_reason), - verdict_confidence = COALESCE(EXCLUDED.verdict_confidence, packets.verdict_confidence), - ingress_seen_at = COALESCE(EXCLUDED.ingress_seen_at, packets.ingress_seen_at), - egress_seen_at = COALESCE(EXCLUDED.egress_seen_at, packets.egress_seen_at), - verdict_seen_at = COALESCE(EXCLUDED.verdict_seen_at, packets.verdict_seen_at), - src_mac = COALESCE(EXCLUDED.src_mac, packets.src_mac), - dst_mac = COALESCE(EXCLUDED.dst_mac, packets.dst_mac), - eth_type_raw = COALESCE(EXCLUDED.eth_type_raw, packets.eth_type_raw), - vlan_id = COALESCE(EXCLUDED.vlan_id, packets.vlan_id), - src_ip = COALESCE(EXCLUDED.src_ip, packets.src_ip), - dst_ip = COALESCE(EXCLUDED.dst_ip, packets.dst_ip), - ip_proto_raw = COALESCE(EXCLUDED.ip_proto_raw, packets.ip_proto_raw), - src_port = COALESCE(EXCLUDED.src_port, packets.src_port), - dst_port = COALESCE(EXCLUDED.dst_port, packets.dst_port), - length = COALESCE(EXCLUDED.length, packets.length), - capture_sources = ( - SELECT ARRAY( - SELECT DISTINCT source - FROM unnest( - COALESCE(packets.capture_sources, ARRAY[]::text[]) || - COALESCE(EXCLUDED.capture_sources, ARRAY[]::text[]) - ) AS source - ) - ), - app_protocol = COALESCE(EXCLUDED.app_protocol, packets.app_protocol), - app_category = COALESCE(EXCLUDED.app_category, packets.app_category), - app_confidence = COALESCE(EXCLUDED.app_confidence, packets.app_confidence), - app_hostname = COALESCE(EXCLUDED.app_hostname, packets.app_hostname), - app_is_encrypted = COALESCE(EXCLUDED.app_is_encrypted, packets.app_is_encrypted), - app_risk_score = COALESCE(EXCLUDED.app_risk_score, packets.app_risk_score), - dpi_metadata = COALESCE(EXCLUDED.dpi_metadata, packets.dpi_metadata), - capture_metadata = COALESCE(EXCLUDED.capture_metadata, packets.capture_metadata), - telemetry_metadata = COALESCE(EXCLUDED.telemetry_metadata, packets.telemetry_metadata), - capture_observations = COALESCE(EXCLUDED.capture_observations, packets.capture_observations), - raw = COALESCE(EXCLUDED.raw, packets.raw) - RETURNING * - """, - pkt_info.get("timestamp"), - pkt_info["correlation_key"], - pkt_info.get("packet_id"), - pkt_info.get("packet_uid"), - pkt_info.get("flow_id"), - pkt_info.get("capture_session_id"), - pkt_info.get("correlation_source"), - pkt_info.get("skb_mark"), - pkt_info.get("capture_iface"), - pkt_info.get("ingress_if"), - pkt_info.get("egress_if"), - pkt_info.get("verdict"), - pkt_info.get("verdict_reason"), - pkt_info.get("verdict_confidence"), - pkt_info.get("ingress_seen_at"), - pkt_info.get("egress_seen_at"), - pkt_info.get("verdict_seen_at"), - pkt_info.get("src_mac"), - pkt_info.get("dst_mac"), - pkt_info.get("eth_type_raw"), - pkt_info.get("vlan_id"), - pkt_info.get("src_ip"), - pkt_info.get("dst_ip"), - pkt_info.get("protocol_raw"), - pkt_info.get("src_port"), - pkt_info.get("dst_port"), - pkt_info.get("length"), - pkt_info.get("capture_sources"), - pkt_info.get("app_protocol"), - pkt_info.get("app_category"), - pkt_info.get("app_confidence"), - pkt_info.get("app_hostname"), - pkt_info.get("app_is_encrypted"), - pkt_info.get("app_risk_score"), - json.dumps(dpi_metadata) if dpi_metadata is not None else None, - json.dumps(pkt_info.get("capture_metadata")) if pkt_info.get("capture_metadata") is not None else None, - json.dumps(telemetry_metadata) if telemetry_metadata is not None else None, - json.dumps(capture_observations) if capture_observations is not None else None, - pkt_info.get("raw"), - ) - except Exception: - logger.exception("DB upsert failed") - return + ), + app_protocol = COALESCE(EXCLUDED.app_protocol, packets.app_protocol), + app_category = COALESCE(EXCLUDED.app_category, packets.app_category), + app_confidence = COALESCE(EXCLUDED.app_confidence, packets.app_confidence), + app_hostname = COALESCE(EXCLUDED.app_hostname, packets.app_hostname), + app_is_encrypted = COALESCE(EXCLUDED.app_is_encrypted, packets.app_is_encrypted), + app_risk_score = COALESCE(EXCLUDED.app_risk_score, packets.app_risk_score), + dpi_metadata = COALESCE(EXCLUDED.dpi_metadata, packets.dpi_metadata), + capture_metadata = COALESCE(EXCLUDED.capture_metadata, packets.capture_metadata), + telemetry_metadata = COALESCE(EXCLUDED.telemetry_metadata, packets.telemetry_metadata), + capture_observations = COALESCE(EXCLUDED.capture_observations, packets.capture_observations), + raw = COALESCE(EXCLUDED.raw, packets.raw) + RETURNING * + """, + pkt_info.get("timestamp"), + pkt_info["correlation_key"], + pkt_info.get("packet_id"), + pkt_info.get("packet_uid"), + pkt_info.get("flow_id"), + pkt_info.get("capture_session_id"), + pkt_info.get("correlation_source"), + pkt_info.get("skb_mark"), + pkt_info.get("capture_iface"), + pkt_info.get("ingress_if"), + pkt_info.get("egress_if"), + pkt_info.get("verdict"), + pkt_info.get("verdict_reason"), + pkt_info.get("verdict_confidence"), + pkt_info.get("ingress_seen_at"), + pkt_info.get("egress_seen_at"), + pkt_info.get("verdict_seen_at"), + pkt_info.get("src_mac"), + pkt_info.get("dst_mac"), + pkt_info.get("eth_type_raw"), + pkt_info.get("vlan_id"), + pkt_info.get("src_ip"), + pkt_info.get("dst_ip"), + pkt_info.get("protocol_raw"), + pkt_info.get("src_port"), + pkt_info.get("dst_port"), + pkt_info.get("length"), + pkt_info.get("capture_sources"), + pkt_info.get("app_protocol"), + pkt_info.get("app_category"), + pkt_info.get("app_confidence"), + pkt_info.get("app_hostname"), + pkt_info.get("app_is_encrypted"), + pkt_info.get("app_risk_score"), + json.dumps(dpi_metadata) if dpi_metadata is not None else None, + json.dumps(pkt_info.get("capture_metadata")) if pkt_info.get("capture_metadata") is not None else None, + json.dumps(telemetry_metadata) if telemetry_metadata is not None else None, + json.dumps(capture_observations) if capture_observations is not None else None, + pkt_info.get("raw"), + ) + return dict(row) if row else None + def _publish_packet_row(self, pkt_info: Dict[str, Any], row: Optional[Dict[str, Any]]) -> None: if row: - persisted = _serialize_row_for_broadcast(dict(row)) + persisted = _serialize_row_for_broadcast(row) pkt_info.update(persisted) if self.broadcaster: diff --git a/backend/src/utilities/packet_tracker.py b/backend/src/utilities/packet_tracker.py index abb3150..99b22c6 100644 --- a/backend/src/utilities/packet_tracker.py +++ b/backend/src/utilities/packet_tracker.py @@ -178,6 +178,8 @@ class PacketTracker: "persist_failed_total": 0, "persist_timeout_total": 0, "persist_failure_log_suppressed": 0, + "persist_batch_total": 0, + "persist_batch_failed_total": 0, "evicted_persisted_total": 0, "evicted_unpersisted_total": 0, } @@ -689,8 +691,7 @@ class PacketTracker: for correlation_key in expired_keys: self._entries.pop(correlation_key, None) - for entry in due_entries: - self._persist(entry) + self._persist_batch(due_entries) def _persist_backoff_elapsed(self, entry: Dict[str, Any], now_ts: float) -> bool: failures = int(entry.get("persist_failures") or 0) @@ -707,24 +708,37 @@ class PacketTracker: ) return now_ts - float(entry.get("last_persist_attempt_at") or 0.0) >= backoff - def _persist(self, entry: Dict[str, Any]) -> None: - payload = dict(entry["payload"]) + def _persist_batch(self, entries: List[Dict[str, Any]]) -> None: + if not entries: + return + payloads = [dict(entry["payload"]) for entry in entries] web_loop = getattr(shared_objects, "web_loop", None) web_db = getattr(shared_objects, "db", None) if web_loop is None or web_db is None: return + fut: concurrent.futures.Future[Any] try: - fut = asyncio.run_coroutine_threadsafe(web_db.upsert_packet(payload), web_loop) - fut.result(timeout=settings.packet_tracker_persist_timeout_seconds) + if hasattr(web_db, "upsert_packets"): + fut = asyncio.run_coroutine_threadsafe(web_db.upsert_packets(payloads), web_loop) + else: + fut = asyncio.run_coroutine_threadsafe(self._persist_payloads_one_by_one(web_db, payloads), web_loop) + timeout = max( + settings.packet_tracker_persist_timeout_seconds, + settings.packet_tracker_batch_persist_timeout_seconds, + ) + fut.result(timeout=timeout) with self._lock: - current = self._entries.get(entry["correlation_key"]) - if current is not None: - current["persisted"] = True - current["dirty"] = False - current["last_persisted_at"] = time.time() - current["persist_failures"] = 0 + self._stats["persist_batch_total"] += 1 + persisted_at = time.time() + for entry in entries: + current = self._entries.get(entry["correlation_key"]) + if current is not None: + current["persisted"] = True + current["dirty"] = False + current["last_persisted_at"] = persisted_at + current["persist_failures"] = 0 except Exception as exc: try: fut.cancel() @@ -733,26 +747,32 @@ class PacketTracker: is_timeout = isinstance(exc, (TimeoutError, concurrent.futures.TimeoutError, asyncio.TimeoutError)) with self._lock: - self._stats["persist_failed_total"] += 1 + self._stats["persist_batch_failed_total"] += 1 + self._stats["persist_failed_total"] += len(entries) if is_timeout: - self._stats["persist_timeout_total"] += 1 - current = self._entries.get(entry["correlation_key"]) - if current is not None: - current["dirty"] = True - current["persist_failures"] = int(current.get("persist_failures") or 0) + 1 + self._stats["persist_timeout_total"] += len(entries) + for entry in entries: + current = self._entries.get(entry["correlation_key"]) + if current is not None: + current["dirty"] = True + current["persist_failures"] = int(current.get("persist_failures") or 0) + 1 now_ts = time.time() if now_ts - self._last_persist_error_log_at >= settings.packet_tracker_error_log_interval_seconds: self._last_persist_error_log_at = now_ts logger.warning( - "Packet persistence is overloaded; failed to persist %s (%s). Further errors are rate-limited.", - entry["correlation_key"], + "Packet persistence is overloaded; failed to persist batch of %s packets (%s). Further errors are rate-limited.", + len(entries), type(exc).__name__, ) else: with self._lock: self._stats["persist_failure_log_suppressed"] += 1 + async def _persist_payloads_one_by_one(self, web_db: Any, payloads: List[Dict[str, Any]]) -> None: + for payload in payloads: + await web_db.upsert_packet(payload) + def _record_stats(self, payload: Dict[str, Any]) -> None: capture_sources = set(payload.get("capture_sources") or []) self._stats["persisted_total"] += 1 diff --git a/tools/network_timing_benchmark.py b/tools/network_timing_benchmark.py index 5e683ab..6a2904d 100755 --- a/tools/network_timing_benchmark.py +++ b/tools/network_timing_benchmark.py @@ -24,6 +24,7 @@ import os import platform import re import shutil +import signal import socket import statistics import subprocess @@ -264,33 +265,42 @@ def run_command( start_monotonic = time.perf_counter() start_wall = now_iso() sampler.start() + process: subprocess.Popen[str] | None = None try: - completed = subprocess.run( + process = subprocess.Popen( command, cwd=str(cwd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, - timeout=timeout, - check=False, + start_new_session=True, ) + stdout, stderr = process.communicate(timeout=timeout) result = CommandRun( args=command, - returncode=completed.returncode, - stdout=completed.stdout or "", - stderr=completed.stderr or "", + returncode=process.returncode, + stdout=stdout or "", + stderr=stderr or "", ) except subprocess.TimeoutExpired as exc: - stdout = exc.stdout.decode("utf-8", errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "") - stderr = exc.stderr.decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "") + if process is not None: + terminate_process_group(process) + stdout, stderr = process.communicate() + else: + stdout = exc.stdout.decode("utf-8", errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "") + stderr = exc.stderr.decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "") result = CommandRun( args=command, - returncode=None, + returncode=process.returncode if process is not None else None, stdout=stdout, stderr=stderr, timed_out=True, error=f"command timed out after {timeout} seconds", ) + except KeyboardInterrupt: + if process is not None: + terminate_process_group(process) + raise except OSError as exc: result = CommandRun( args=command, @@ -337,6 +347,22 @@ def run_command( return result, meta +def terminate_process_group(process: subprocess.Popen[str], grace_seconds: float = 2.0) -> None: + """Terminate a command and any children it spawned.""" + if process.poll() is not None: + return + try: + os.killpg(process.pid, signal.SIGTERM) + process.wait(timeout=grace_seconds) + except subprocess.TimeoutExpired: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + except ProcessLookupError: + pass + + def parse_ping_output(text: str, sent_fallback: int) -> dict[str, Any]: rtts: list[float] = [] for line in text.splitlines(): @@ -1031,6 +1057,12 @@ def parse_csv_strings(value: str) -> list[str]: return items +def cooldown(seconds: float) -> None: + if seconds <= 0: + return + time.sleep(seconds) + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Measure latency, jitter, throughput, loss, CPU, and interface counters.", @@ -1066,6 +1098,12 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--mtr-count", type=int, default=100) parser.add_argument("--no-archive", action="store_true") parser.add_argument("--command-timeout-extra", type=float, default=15.0) + parser.add_argument( + "--cooldown-seconds", + type=float, + default=0.0, + help="Sleep after each test so local queues, CPU, and the backend can drain before the next test.", + ) return parser @@ -1083,6 +1121,8 @@ def main() -> int: parser.error("--flent-duration must be >= 1") if args.sockperf_duration < 1: parser.error("--sockperf-duration must be >= 1") + if args.cooldown_seconds < 0: + parser.error("--cooldown-seconds must be >= 0") for protocol in args.sockperf_protocols: if protocol not in {"tcp", "udp"}: parser.error("--sockperf-protocols may only contain tcp and/or udp") @@ -1095,14 +1135,23 @@ def main() -> int: tests: list[dict[str, Any]] = [] tests.extend(run_ping_tests(args, out_dir)) + cooldown(args.cooldown_seconds) tests.append(run_arping(args, out_dir)) + cooldown(args.cooldown_seconds) tests.append(run_iperf_tcp(args, out_dir)) + cooldown(args.cooldown_seconds) tests.append(run_iperf_tcp_reverse(args, out_dir)) + cooldown(args.cooldown_seconds) tests.append(run_iperf_udp(args, out_dir)) + cooldown(args.cooldown_seconds) tests.append(run_iperf_udp_reverse(args, out_dir)) + cooldown(args.cooldown_seconds) tests.extend(run_sockperf(args, out_dir)) + cooldown(args.cooldown_seconds) tests.extend(run_flent(args, out_dir)) + cooldown(args.cooldown_seconds) tests.append(run_traceroute(args, out_dir)) + cooldown(args.cooldown_seconds) tests.append(run_mtr(args, out_dir)) summary = {