From b8e1c131853459c3700991c4785aa445863d5d13 Mon Sep 17 00:00:00 2001 From: maximus Date: Wed, 15 Jul 2026 20:02:51 +0330 Subject: [PATCH 1/2] Add performance benchmark and soak validation --- .gitignore | 2 + README.md | 2 + benchmarks/README.md | 54 +++ benchmarks/__init__.py | 1 + benchmarks/benchmark_batch_persistence.py | 104 +++++ benchmarks/benchmark_live_ring_buffer.py | 82 ++++ benchmarks/benchmark_packet_pipeline.py | 67 ++++ benchmarks/benchmark_report.py | 323 ++++++++++++++++ benchmarks/benchmark_websocket_aggregator.py | 96 +++++ benchmarks/benchmark_worker_pool.py | 82 ++++ benchmarks/soak_test_pipeline.py | 375 +++++++++++++++++++ docs/ARCHITECTURE.md | 6 +- docs/PERFORMANCE_PIPELINE.md | 62 ++- docs/PERFORMANCE_VALIDATION.md | 117 ++++++ tests/test_performance_benchmarks.py | 141 +++++++ tests/test_release_readiness.py | 8 + 16 files changed, 1515 insertions(+), 7 deletions(-) create mode 100644 benchmarks/README.md create mode 100644 benchmarks/__init__.py create mode 100644 benchmarks/benchmark_batch_persistence.py create mode 100644 benchmarks/benchmark_live_ring_buffer.py create mode 100644 benchmarks/benchmark_packet_pipeline.py create mode 100644 benchmarks/benchmark_report.py create mode 100644 benchmarks/benchmark_websocket_aggregator.py create mode 100644 benchmarks/benchmark_worker_pool.py create mode 100644 benchmarks/soak_test_pipeline.py create mode 100644 docs/PERFORMANCE_VALIDATION.md create mode 100644 tests/test_performance_benchmarks.py diff --git a/.gitignore b/.gitignore index a0d7d88..04951c7 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,8 @@ htmlcov/ venv/ .env .runtime/ +benchmark_results_*.json +benchmark_summary_*.md node_modules/ frontend/node_modules/ diff --git a/README.md b/README.md index 4eda697..bdb3cca 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,7 @@ or PCAP artifacts. | Live Ring Buffer | Foundation step | Recent packet, flow, alert, and Expert summaries are held in per-category bounded memory with eviction/query metrics and Ops Snapshot visibility. | Stored and returned records are redacted summaries; raw packet payloads and credentials are excluded. | | Batch Persistence / Storage Backpressure | Foundation step | Redacted packet, alert, and flow records use bounded batches with queue health, write latency, retry, backlog, failure, and Ops Snapshot visibility. | Audit and report exports remain synchronous; this is not a distributed storage engine or the complete performance pipeline. | | WebSocket Event Aggregator | Foundation step | Realtime packet/alert batching, slow-client protection, WebSocket pressure metrics, and Ops Snapshot visibility are tested. | Realtime delivery batching only. | +| Performance Benchmark Suite | Validated foundation | Synthetic stage benchmarks, a bounded full-pipeline soak mode, JSON/Markdown reports, and a CI-safe structural smoke test cover the current pipeline. | No live capture, external network, admin privileges, or production throughput claims; see [Performance Validation](docs/PERFORMANCE_VALIDATION.md). | | Demo and operational QA | Ready | Token-safe demo and Agent script behavior is tested. | Demo launchers and status commands do not print raw tokens. | | Windows release packaging | Validated path | Desktop smoke, version consistency, and release workflow checks run in CI. | Versioned artifacts include SHA256 checksums. | | Linux desktop packaging | Staged | Build workflow exists; native production validation remains pending. | Publish only after native smoke and release QA. | @@ -253,6 +254,7 @@ and surfaced in Ops Snapshot metrics. - [Agent Mode](docs/AGENT_MODE.md) - [Performance Pipeline](docs/PERFORMANCE_PIPELINE.md) +- [Performance Validation](docs/PERFORMANCE_VALIDATION.md) - [Flow Analysis And Protocol Intelligence](docs/FLOW_ANALYSIS.md) - [Deep Packet Inspection](docs/DEEP_PACKET_INSPECTION.md) - [Agent Operational QA Checklist](docs/AGENT_QA_CHECKLIST.md) diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 0000000..f22e403 --- /dev/null +++ b/benchmarks/README.md @@ -0,0 +1,54 @@ +# Performance Benchmarks + +This directory validates NetBotPro's bounded performance pipeline with local, +synthetic packet summaries. The tools do not capture traffic, open network +connections, require Administrator privileges, or store raw payloads. + +## Quick smoke test + +```powershell +python benchmarks/soak_test_pipeline.py ` + --duration-sec 10 ` + --events-per-sec 200 ` + --flows 20 ` + --ci-safe ` + --output .runtime/benchmarks/smoke +``` + +The command writes: + +- `benchmark_results.json`: structured configuration, resource, stage, and + operational-health metrics. +- `benchmark_summary.md`: a compact human-readable interpretation. + +Both outputs pass through central redaction before they are written. + +## Stage benchmarks + +Run a focused benchmark with the same common CLI options: + +```powershell +python benchmarks/benchmark_packet_pipeline.py --duration-sec 10 --events-per-sec 500 +python benchmarks/benchmark_worker_pool.py --duration-sec 10 --events-per-sec 500 --flows 50 +python benchmarks/benchmark_websocket_aggregator.py --duration-sec 10 --events-per-sec 500 +python benchmarks/benchmark_batch_persistence.py --duration-sec 10 --events-per-sec 500 +python benchmarks/benchmark_live_ring_buffer.py --duration-sec 10 --events-per-sec 500 +``` + +Available options include `--duration-sec`, `--events-per-sec`, `--flows`, +`--packet-rate`, `--alert-rate`, `--output`, `--json`, `--markdown`, +`--profile`, `--no-live-capture`, and `--ci-safe`. The `ci-safe` profile caps +duration and event rate. Live capture remains disabled regardless of profile. + +## Longer local soak + +The local profile defaults to five minutes when duration is omitted: + +```powershell +python benchmarks/soak_test_pipeline.py --profile local --events-per-sec 1000 --flows 100 +``` + +Results are machine-dependent. They validate bounded behavior, observable +drops, ordering, report generation, and resource trends; they are not a +production capacity guarantee. Use authorized workload traces separately when +sizing a deployment. diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..f47749c --- /dev/null +++ b/benchmarks/__init__.py @@ -0,0 +1 @@ +"""Safe synthetic performance validation tools for NetBotPro.""" diff --git a/benchmarks/benchmark_batch_persistence.py b/benchmarks/benchmark_batch_persistence.py new file mode 100644 index 0000000..de187f5 --- /dev/null +++ b/benchmarks/benchmark_batch_persistence.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import argparse +import sys +import threading +import time +from pathlib import Path +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from backend.app.services.batch_persistence import BatchPersistenceWriter +from benchmarks.benchmark_report import ( + BenchmarkConfig, + add_common_arguments, + config_from_args, + synthetic_alert, + synthetic_packet, +) + + +def run_persistence_benchmark(config: BenchmarkConfig) -> dict[str, Any]: + config = config.normalized() + event_count = max(1, min(int(config.duration_sec * config.events_per_sec), 50_000)) + written = 0 + batches = 0 + lock = threading.Lock() + + def write_batch(grouped: dict[str, list[dict[str, Any]]]) -> None: + nonlocal written, batches + with lock: + written += sum(len(rows) for rows in grouped.values()) + batches += 1 + + writer = BatchPersistenceWriter( + write_batch, + enabled=True, + queue_max=max(100, min(event_count, 5000)), + overflow_policy="drop_oldest", + retry_max=1, + retry_backoff_ms=1, + batch_sizes={ + "packet_record": 100, + "flow_record": 50, + "alert_record": 25, + }, + flush_ms={ + "packet_record": 20, + "flow_record": 20, + "alert_record": 20, + }, + ) + try: + for sequence in range(event_count): + packet = synthetic_packet(sequence, config.flows) + writer.enqueue("packet_record", packet, source="synthetic_benchmark") + if sequence % 10 == 0: + writer.enqueue( + "flow_record", + {"flow_id": packet["flow_key"], "packets": sequence + 1}, + source="synthetic_benchmark", + ) + if ( + config.alert_rate + and sequence % max(1, config.events_per_sec // config.alert_rate) == 0 + ): + writer.enqueue( + "alert_record", + synthetic_alert(sequence, packet), + source="synthetic_benchmark", + ) + flushed = writer.flush(10.0) + deadline = time.monotonic() + 10.0 + while time.monotonic() < deadline: + metrics = writer.metrics() + if metrics["queue_depth"] == 0: + break + time.sleep(0.01) + metrics = writer.metrics() + finally: + writer.close(10.0) + metrics.update( + { + "sink_events_written_total": written, + "sink_batches_total": batches, + "manual_flush_completed": flushed, + "bounded": metrics["queue_depth"] <= metrics["queue_max"], + "write_latency_ms_avg": metrics["write_latency_avg_ms"], + "write_latency_ms_p95": metrics["write_latency_p95_ms"], + } + ) + return metrics + + +def main() -> int: + parser = argparse.ArgumentParser(description="Benchmark batch persistence.") + add_common_arguments(parser) + print(run_persistence_benchmark(config_from_args(parser.parse_args()))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/benchmark_live_ring_buffer.py b/benchmarks/benchmark_live_ring_buffer.py new file mode 100644 index 0000000..48ef672 --- /dev/null +++ b/benchmarks/benchmark_live_ring_buffer.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import argparse +import sys +import time +from pathlib import Path +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from backend.app.services.live_ring_buffer import DEFAULT_CAPACITIES, LiveRingBuffer +from benchmarks.benchmark_report import ( + BenchmarkConfig, + add_common_arguments, + config_from_args, + percentile, + synthetic_packet, +) + + +def run_live_ring_benchmark(config: BenchmarkConfig) -> dict[str, Any]: + config = config.normalized() + event_count = max(1, min(int(config.duration_sec * config.events_per_sec), 50_000)) + packet_capacity = max(10, min(event_count // 2 or 1, 5000)) + capacities = {category: 10 for category in DEFAULT_CAPACITIES} + capacities["packet"] = packet_capacity + ring = LiveRingBuffer( + capacities=capacities, + default_query_limit=50, + max_query_limit=100, + ) + append_latencies: list[float] = [] + query_latencies: list[float] = [] + started = time.perf_counter() + for sequence in range(event_count): + packet = synthetic_packet(sequence, config.flows) + item_started = time.perf_counter() + ring.append( + "packet", + packet, + flow_key=packet["flow_key"], + source="synthetic_benchmark", + ) + append_latencies.append((time.perf_counter() - item_started) * 1000.0) + if sequence % 100 == 0: + query_started = time.perf_counter() + ring.query("packet", limit=50) + query_latencies.append((time.perf_counter() - query_started) * 1000.0) + ring.query("packet", limit=1000) + elapsed = max(time.perf_counter() - started, 0.000001) + metrics = ring.metrics() + metrics.update( + { + "benchmark_events_total": event_count, + "benchmark_throughput_events_sec": round(event_count / elapsed, 2), + "append_latency_ms_avg": round( + sum(append_latencies) / len(append_latencies), 3 + ), + "append_latency_ms_p95": percentile(append_latencies, 0.95), + "query_latency_ms_avg": ( + round(sum(query_latencies) / len(query_latencies), 3) + if query_latencies + else 0.0 + ), + "query_latency_ms_p95": percentile(query_latencies, 0.95), + "bounded": metrics["categories"]["packet"]["records"] + <= metrics["categories"]["packet"]["capacity"], + } + ) + return metrics + + +def main() -> int: + parser = argparse.ArgumentParser(description="Benchmark Live Ring Buffer.") + add_common_arguments(parser) + print(run_live_ring_benchmark(config_from_args(parser.parse_args()))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/benchmark_packet_pipeline.py b/benchmarks/benchmark_packet_pipeline.py new file mode 100644 index 0000000..c85f73e --- /dev/null +++ b/benchmarks/benchmark_packet_pipeline.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import argparse +import logging +import sys +import time +from pathlib import Path +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from backend.app.services.packet_queue import BoundedPacketQueue +from benchmarks.benchmark_report import ( + BenchmarkConfig, + add_common_arguments, + config_from_args, + percentile, + synthetic_packet, +) + + +def run_packet_queue_benchmark(config: BenchmarkConfig) -> dict[str, Any]: + config = config.normalized() + event_count = max(1, min(int(config.duration_sec * config.packet_rate), 50_000)) + queue_max = max(10, min(event_count // 2 or 1, 5000)) + packet_queue = BoundedPacketQueue(max_size=queue_max, overflow_policy="drop_oldest") + latencies_ms: list[float] = [] + started = time.perf_counter() + queue_logger = logging.getLogger("backend.app.services.packet_queue") + previous_disabled = queue_logger.disabled + queue_logger.disabled = True + try: + for sequence in range(event_count): + item_started = time.perf_counter() + packet_queue.put(synthetic_packet(sequence, config.flows)) + latencies_ms.append((time.perf_counter() - item_started) * 1000.0) + finally: + queue_logger.disabled = previous_disabled + while not packet_queue.empty(): + packet_queue.get(timeout=0.1) + packet_queue.task_done() + elapsed = max(time.perf_counter() - started, 0.000001) + metrics = packet_queue.stats(worker_alive=True) + metrics.update( + { + "benchmark_events_total": event_count, + "benchmark_throughput_events_sec": round(event_count / elapsed, 2), + "intake_latency_ms_avg": round(sum(latencies_ms) / len(latencies_ms), 3), + "intake_latency_ms_p95": percentile(latencies_ms, 0.95), + "intake_latency_ms_max": round(max(latencies_ms), 3), + "bounded": metrics["high_water_mark"] <= metrics["max_size"], + } + ) + return metrics + + +def main() -> int: + parser = argparse.ArgumentParser(description="Benchmark bounded packet intake.") + add_common_arguments(parser) + config = config_from_args(parser.parse_args()) + print(run_packet_queue_benchmark(config)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/benchmark_report.py b/benchmarks/benchmark_report.py new file mode 100644 index 0000000..305f9e0 --- /dev/null +++ b/benchmarks/benchmark_report.py @@ -0,0 +1,323 @@ +from __future__ import annotations + +import argparse +import json +import statistics +import sys +import threading +import time +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable + +import psutil + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from backend.app.services.redaction import redact_sensitive_data + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def percentile(values: Iterable[float], percent: float) -> float: + ordered = sorted(float(value) for value in values) + if not ordered: + return 0.0 + index = min(len(ordered) - 1, max(0, int((len(ordered) - 1) * percent))) + return round(ordered[index], 3) + + +@dataclass(frozen=True) +class BenchmarkConfig: + duration_sec: float = 30.0 + events_per_sec: int = 1000 + flows: int = 100 + packet_rate: int = 900 + alert_rate: int = 100 + output: str = ".runtime/benchmarks/latest" + profile: str = "ci-safe" + ci_safe: bool = True + live_capture: bool = False + + def normalized(self) -> "BenchmarkConfig": + duration_max = 30.0 if self.ci_safe else 3600.0 + events_max = 10_000 if self.ci_safe else 100_000 + return BenchmarkConfig( + duration_sec=max(0.01, min(float(self.duration_sec), duration_max)), + events_per_sec=max(1, min(int(self.events_per_sec), events_max)), + flows=max(1, min(int(self.flows), 100_000)), + packet_rate=max(1, min(int(self.packet_rate), events_max)), + alert_rate=max(0, min(int(self.alert_rate), events_max)), + output=str(self.output or ".runtime/benchmarks/latest"), + profile=str(self.profile or "ci-safe")[:40], + ci_safe=bool(self.ci_safe), + live_capture=False, + ) + + +def synthetic_packet(sequence: int, flows: int) -> dict[str, Any]: + flow_index = sequence % max(1, flows) + local_octet = flow_index % 250 + 1 + remote_octet = flow_index % 200 + 1 + return { + "id": f"benchmark-packet-{sequence}", + "sequence": sequence, + "src": f"10.20.0.{local_octet}", + "dst": f"198.51.100.{remote_octet}", + "sport": 20_000 + flow_index % 40_000, + "dport": 443 if flow_index % 3 else 53, + "proto": "TCP" if flow_index % 3 else "UDP", + "app_protocol": "TLS" if flow_index % 3 else "DNS", + "length": 128 + sequence % 1300, + "flow_key": f"benchmark-flow-{flow_index}", + "source": "synthetic_benchmark", + } + + +def synthetic_alert(sequence: int, packet: dict[str, Any]) -> dict[str, Any]: + return { + "id": f"benchmark-alert-{sequence}", + "packet_id": packet["id"], + "flow_id": packet["flow_key"], + "severity": "medium" if sequence % 10 else "high", + "attack_type": "synthetic_validation_signal", + "source": "synthetic_benchmark", + } + + +class ResourceSampler: + def __init__(self, interval_sec: float = 0.05) -> None: + self._process = psutil.Process() + self._interval_sec = max(0.01, float(interval_sec)) + self._stop = threading.Event() + self._memory: list[int] = [] + self._cpu: list[float] = [] + self._thread: threading.Thread | None = None + self._memory_start = int(self._process.memory_info().rss) + + def start(self) -> None: + self._process.cpu_percent(interval=None) + self._thread = threading.Thread( + target=self._sample, + name="netbotpro-benchmark-resource-sampler", + daemon=True, + ) + self._thread.start() + + def stop(self) -> dict[str, float | int]: + self._stop.set() + if self._thread: + self._thread.join(timeout=1.0) + memory_end = int(self._process.memory_info().rss) + memory_values = [self._memory_start, *self._memory, memory_end] + return { + "memory_start_bytes": self._memory_start, + "memory_peak_bytes": max(memory_values), + "memory_end_bytes": memory_end, + "memory_growth_bytes": memory_end - self._memory_start, + "cpu_percent_avg": ( + round(statistics.mean(self._cpu), 2) if self._cpu else 0.0 + ), + "cpu_percent_peak": round(max(self._cpu), 2) if self._cpu else 0.0, + } + + def _sample(self) -> None: + while not self._stop.wait(self._interval_sec): + self._memory.append(int(self._process.memory_info().rss)) + self._cpu.append(float(self._process.cpu_percent(interval=None))) + + +def report_status(results: dict[str, Any]) -> dict[str, str]: + mapping = { + "Bounded Packet Queue": "packet_queue", + "Flow-aware Worker Pool": "flow_worker_pool", + "WebSocket Event Aggregator": "event_aggregator", + "Batch Persistence": "persistence", + "Live Ring Buffer": "live_ring_buffer", + } + return { + label: "validated" if isinstance(results.get(key), dict) else "needs follow-up" + for label, key in mapping.items() + } + + +def write_reports( + results: dict[str, Any], + output: str | Path, + *, + write_json: bool = True, + write_markdown: bool = True, +) -> dict[str, str]: + destination = Path(output) + destination.mkdir(parents=True, exist_ok=True) + safe_results = redact_sensitive_data(results) + paths: dict[str, str] = {} + if write_json: + json_path = destination / "benchmark_results.json" + json_path.write_text( + json.dumps(safe_results, indent=2, ensure_ascii=True, default=str) + "\n", + encoding="utf-8", + ) + paths["json"] = str(json_path) + if write_markdown: + markdown_path = destination / "benchmark_summary.md" + markdown_path.write_text(render_markdown(safe_results), encoding="utf-8") + paths["markdown"] = str(markdown_path) + return paths + + +def render_markdown(results: dict[str, Any]) -> str: + general = results.get("general") or {} + resources = results.get("resources") or {} + statuses = report_status(results) + lines = [ + "# NetBotPro Synthetic Performance Summary", + "", + "> This report uses local synthetic metadata only. It does not capture or transmit network traffic.", + "", + "## Run Summary", + "", + "| Metric | Result |", + "| --- | ---: |", + f"| Duration | {float(general.get('duration_sec') or 0):.3f} sec |", + f"| Events generated | {int(general.get('events_generated_total') or 0)} |", + f"| Events processed | {int(general.get('events_processed_total') or 0)} |", + f"| Events dropped | {int(general.get('events_dropped_total') or 0)} |", + f"| Events failed | {int(general.get('events_failed_total') or 0)} |", + f"| Throughput | {float(general.get('throughput_events_sec') or 0):.2f} events/sec |", + f"| Peak memory | {int(resources.get('memory_peak_bytes') or 0)} bytes |", + f"| Memory growth | {int(resources.get('memory_growth_bytes') or 0)} bytes |", + f"| Average CPU | {float(resources.get('cpu_percent_avg') or 0):.2f}% |", + "", + "## Performance Foundation Status", + "", + ] + lines.extend(f"- {label}: **{status}**" for label, status in statuses.items()) + lines.extend( + [ + "", + "## Pipeline Metrics", + "", + "| Stage | Health | Processed / Written | Dropped / Evicted | Utilization |", + "| --- | --- | ---: | ---: | ---: |", + _stage_row( + "Packet Queue", + results.get("packet_queue") or {}, + "accepted_total", + "dropped_total", + ), + _stage_row( + "Flow Worker Pool", + results.get("flow_worker_pool") or {}, + "jobs_processed_total", + "jobs_dropped_total", + ), + _stage_row( + "Event Aggregator", + results.get("event_aggregator") or {}, + "events_sent_total", + "events_dropped_total", + ), + _stage_row( + "Batch Persistence", + results.get("persistence") or {}, + "events_written_total", + "events_dropped_total", + ), + _stage_row( + "Live Ring Buffer", + results.get("live_ring_buffer") or {}, + "records_added_total", + "records_evicted_total", + ), + "", + "## Ops Health", + "", + f"- Final state: **{(results.get('ops_health') or {}).get('final_health', 'unknown')}**", + f"- Pressure reasons: {', '.join((results.get('ops_health') or {}).get('pressure_reasons') or []) or 'none'}", + "", + "## Interpretation", + "", + "Results are machine-dependent and are not production capacity promises. Drops and evictions are explicit bounded-pressure behavior. Validate with authorized workload traces before production sizing.", + "", + ] + ) + return "\n".join(lines) + + +def _stage_row( + label: str, metrics: dict[str, Any], processed_key: str, dropped_key: str +) -> str: + return ( + f"| {label} | {metrics.get('health', 'unknown')} | " + f"{int(metrics.get(processed_key) or 0)} | " + f"{int(metrics.get(dropped_key) or 0)} | " + f"{float(metrics.get('utilization_percent') or 0):.2f}% |" + ) + + +def add_common_arguments(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--duration-sec", type=float, default=30.0) + parser.add_argument("--events-per-sec", type=int, default=1000) + parser.add_argument("--flows", type=int, default=100) + parser.add_argument("--packet-rate", type=int, default=900) + parser.add_argument("--alert-rate", type=int, default=100) + parser.add_argument("--output", default=".runtime/benchmarks/latest") + parser.add_argument( + "--json", + action="store_true", + help="Generate JSON output (generated by default).", + ) + parser.add_argument( + "--markdown", + action="store_true", + help="Generate Markdown output (generated by default).", + ) + parser.add_argument( + "--profile", choices=("ci-safe", "local", "heavy"), default="ci-safe" + ) + parser.add_argument("--no-live-capture", action="store_true", default=True) + parser.add_argument("--ci-safe", action="store_true") + + +def config_from_args( + args: argparse.Namespace, *, soak: bool = False +) -> BenchmarkConfig: + ci_safe = bool(args.ci_safe or args.profile == "ci-safe") + duration = args.duration_sec + if soak and args.duration_sec == 30.0 and not ci_safe: + duration = 300.0 + return BenchmarkConfig( + duration_sec=duration, + events_per_sec=args.events_per_sec, + flows=args.flows, + packet_rate=args.packet_rate, + alert_rate=args.alert_rate, + output=args.output, + profile=args.profile, + ci_safe=ci_safe, + live_capture=False, + ).normalized() + + +def config_dict(config: BenchmarkConfig) -> dict[str, Any]: + return asdict(config.normalized()) + + +__all__ = [ + "BenchmarkConfig", + "ResourceSampler", + "add_common_arguments", + "config_dict", + "config_from_args", + "percentile", + "render_markdown", + "synthetic_alert", + "synthetic_packet", + "utc_now", + "write_reports", +] diff --git a/benchmarks/benchmark_websocket_aggregator.py b/benchmarks/benchmark_websocket_aggregator.py new file mode 100644 index 0000000..31380ba --- /dev/null +++ b/benchmarks/benchmark_websocket_aggregator.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import argparse +import sys +import time +from pathlib import Path +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from backend.app.services.event_aggregator import EventAggregator +from benchmarks.benchmark_report import ( + BenchmarkConfig, + add_common_arguments, + config_from_args, + percentile, + synthetic_alert, + synthetic_packet, +) + + +def run_websocket_benchmark(config: BenchmarkConfig) -> dict[str, Any]: + config = config.normalized() + event_count = max(1, min(int(config.duration_sec * config.events_per_sec), 50_000)) + emitted: list[dict[str, Any]] = [] + emit_latencies: list[float] = [] + + def emit(message: dict[str, Any]) -> None: + started = time.perf_counter() + emitted.append(message) + emit_latencies.append((time.perf_counter() - started) * 1000.0) + + aggregator = EventAggregator( + emit, + packet_batch_ms=10_000, + packet_batch_max=100, + alert_batch_ms=10_000, + alert_batch_max=50, + flow_batch_ms=10_000, + flow_batch_max=50, + ) + try: + for sequence in range(event_count): + packet = synthetic_packet(sequence, config.flows) + aggregator.publish("packet:new", packet) + if ( + config.alert_rate + and sequence % max(1, config.events_per_sec // config.alert_rate) == 0 + ): + aggregator.publish("alert:new", synthetic_alert(sequence, packet)) + if sequence % 10 == 0: + aggregator.publish( + "flow:update", + {"flow_id": packet["flow_key"], "packets": sequence + 1}, + ) + aggregator.flush_all() + metrics = aggregator.stats() + finally: + aggregator.close() + metrics.update( + { + "batches_emitted_total": metrics["batches_sent_total"], + "emitted_messages_total": len(emitted), + "send_latency_ms_avg": ( + round(sum(emit_latencies) / len(emit_latencies), 3) + if emit_latencies + else 0.0 + ), + "send_latency_ms_p95": percentile(emit_latencies, 0.95), + "bounded": all( + metrics[key] <= limit + for key, limit in ( + ("pending_packet_events", aggregator.packet_batch_max), + ("pending_alert_events", aggregator.alert_batch_max), + ("pending_flow_events", aggregator.flow_batch_max), + ) + ), + } + ) + return metrics + + +# Keep a stage-oriented name for callers that do not model a real WebSocket client. +run_event_aggregator_benchmark = run_websocket_benchmark + + +def main() -> int: + parser = argparse.ArgumentParser(description="Benchmark websocket batching.") + add_common_arguments(parser) + print(run_websocket_benchmark(config_from_args(parser.parse_args()))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/benchmark_worker_pool.py b/benchmarks/benchmark_worker_pool.py new file mode 100644 index 0000000..d074517 --- /dev/null +++ b/benchmarks/benchmark_worker_pool.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import argparse +import sys +import threading +import time +from collections import defaultdict +from pathlib import Path +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from backend.app.services.flow_worker_pool import FlowWorkerPool +from benchmarks.benchmark_report import ( + BenchmarkConfig, + add_common_arguments, + config_from_args, + synthetic_packet, +) + + +def run_worker_pool_benchmark(config: BenchmarkConfig) -> dict[str, Any]: + config = config.normalized() + event_count = max(1, min(int(config.duration_sec * config.events_per_sec), 50_000)) + sequences: dict[str, list[int]] = defaultdict(list) + worker_threads: set[str] = set() + lock = threading.Lock() + + def process(packet: dict[str, Any]) -> None: + time.sleep(0.0001) + with lock: + sequences[str(packet["flow_key"])].append(int(packet["sequence"])) + worker_threads.add(threading.current_thread().name) + + pool = FlowWorkerPool( + process, + worker_count=min(4, config.flows), + queue_max=max(100, min(event_count, 5000)), + overflow_policy="block_short", + block_timeout_sec=0.2, + shutdown_timeout_sec=10.0, + slow_job_ms=1000.0, + ) + try: + accepted = sum( + 1 + for sequence in range(event_count) + if pool.submit(synthetic_packet(sequence, config.flows)) + ) + drained = pool.wait_until_drained(10.0) + metrics = pool.stats() + finally: + pool.close(10.0) + metrics.update( + { + "benchmark_events_total": event_count, + "benchmark_accepted_total": accepted, + "drained": drained, + "same_flow_order_preserved": all( + values == sorted(values) for values in sequences.values() + ), + "processing_threads_observed": len(worker_threads), + "parallel_workers_observed": len(worker_threads) > 1, + "bounded": metrics["queue_depth_total"] <= metrics["queue_max_total"], + "processing_latency_ms_avg": metrics["avg_processing_latency_ms"], + "processing_latency_ms_p95": metrics["p95_processing_latency_ms"], + "processing_latency_ms_max": metrics["max_processing_latency_ms"], + } + ) + return metrics + + +def main() -> int: + parser = argparse.ArgumentParser(description="Benchmark flow-aware workers.") + add_common_arguments(parser) + print(run_worker_pool_benchmark(config_from_args(parser.parse_args()))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/soak_test_pipeline.py b/benchmarks/soak_test_pipeline.py new file mode 100644 index 0000000..6485854 --- /dev/null +++ b/benchmarks/soak_test_pipeline.py @@ -0,0 +1,375 @@ +from __future__ import annotations + +import argparse +import json +import queue +import sys +import threading +import time +from pathlib import Path +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from backend.app.services.batch_persistence import BatchPersistenceWriter +from backend.app.services.event_aggregator import EventAggregator +from backend.app.services.flow_worker_pool import FlowWorkerPool +from backend.app.services.live_ring_buffer import DEFAULT_CAPACITIES, LiveRingBuffer +from backend.app.services.packet_queue import BoundedPacketQueue +from benchmarks.benchmark_report import ( + BenchmarkConfig, + ResourceSampler, + add_common_arguments, + config_dict, + config_from_args, + synthetic_alert, + synthetic_packet, + utc_now, + write_reports, +) + + +def run_soak_test(config: BenchmarkConfig) -> dict[str, Any]: + config = config.normalized() + target_packet_rate = min(config.events_per_sec, config.packet_rate) + target_alert_rate = min(config.alert_rate, target_packet_rate) + packet_queue = BoundedPacketQueue(max_size=2000, overflow_policy="drop_oldest") + capacities = {category: value for category, value in DEFAULT_CAPACITIES.items()} + if config.ci_safe: + capacities.update( + { + "packet": 500, + "flow": 200, + "alert": 100, + "expert_info": 100, + "protocol_metadata": 100, + "agent_status": 50, + "ops_event": 50, + } + ) + ring = LiveRingBuffer( + capacities=capacities, + default_query_limit=50, + max_query_limit=250, + ) + + emitted: list[dict[str, Any]] = [] + emit_lock = threading.Lock() + + def emit(message: dict[str, Any]) -> None: + with emit_lock: + if len(emitted) >= 1000: + del emitted[:100] + emitted.append(message) + + aggregator = EventAggregator( + emit, + packet_batch_ms=100, + packet_batch_max=100, + alert_batch_ms=100, + alert_batch_max=50, + flow_batch_ms=200, + flow_batch_max=100, + ) + + sink_events_written = 0 + sink_batches = 0 + sink_lock = threading.Lock() + + def write_batch(grouped: dict[str, list[dict[str, Any]]]) -> None: + nonlocal sink_events_written, sink_batches + with sink_lock: + sink_events_written += sum(len(rows) for rows in grouped.values()) + sink_batches += 1 + + persistence = BatchPersistenceWriter( + write_batch, + enabled=True, + queue_max=2000, + overflow_policy="drop_oldest", + retry_max=1, + retry_backoff_ms=1, + batch_sizes={ + "packet_record": 100, + "flow_record": 50, + "alert_record": 25, + }, + flush_ms={ + "packet_record": 100, + "flow_record": 200, + "alert_record": 100, + }, + ) + + processed_total = 0 + alert_total = 0 + counter_lock = threading.Lock() + + def process(packet: dict[str, Any]) -> None: + nonlocal processed_total, alert_total + flow_key = str(packet["flow_key"]) + ring.append("packet", packet, flow_key=flow_key, source="synthetic_benchmark") + flow = { + "flow_id": flow_key, + "last_sequence": packet["sequence"], + "packets_count": int(packet["sequence"]) + 1, + "source": "synthetic_benchmark", + } + ring.append("flow", flow, flow_key=flow_key, source="synthetic_benchmark") + persistence.enqueue("packet_record", packet, source="synthetic_benchmark") + if int(packet["sequence"]) % 10 == 0: + persistence.enqueue("flow_record", flow, source="synthetic_benchmark") + aggregator.publish("flow:update", flow) + aggregator.publish("packet:new", packet) + should_alert = ( + target_alert_rate + and int(packet["sequence"]) + % max(1, target_packet_rate // target_alert_rate) + == 0 + ) + if should_alert: + alert = synthetic_alert(int(packet["sequence"]), packet) + ring.append("alert", alert, flow_key=flow_key, source="synthetic_benchmark") + persistence.enqueue("alert_record", alert, source="synthetic_benchmark") + aggregator.publish("alert:new", alert) + with counter_lock: + alert_total += 1 + with counter_lock: + processed_total += 1 + + workers = FlowWorkerPool( + process, + worker_count=min(4, config.flows), + queue_max=2000, + overflow_policy="drop_oldest", + shutdown_timeout_sec=10.0, + error_threshold=100, + slow_job_ms=100.0, + ) + + sampler = ResourceSampler() + start_time = utc_now() + started = time.perf_counter() + deadline = started + config.duration_sec + generated_total = 0 + health_transitions: list[str] = [] + last_health = "" + sampler.start() + try: + interval = 1.0 / target_packet_rate + next_event_at = started + while time.perf_counter() < deadline: + packet = synthetic_packet(generated_total, config.flows) + packet_queue.put(packet) + try: + queued = packet_queue.get(timeout=0.1) + except queue.Empty: + continue + try: + workers.submit(queued.packet) + finally: + packet_queue.task_done() + generated_total += 1 + next_event_at += interval + wait = next_event_at - time.perf_counter() + if wait > 0: + time.sleep(min(wait, 0.05)) + if generated_total % max(1, target_packet_rate) == 0: + health = _pipeline_health( + packet_queue.stats(), + workers.stats(), + aggregator.stats(), + persistence.metrics(), + ring.metrics(), + ) + if health != last_health: + health_transitions.append(health) + last_health = health + packet_queue.wait_until_drained(2.0) + workers.wait_until_drained(10.0) + aggregator.flush_all() + persistence.flush(10.0) + _wait_for_persistence(persistence, 10.0) + finally: + resource_metrics = sampler.stop() + + ring_query_started = time.perf_counter() + ring.query("all", limit=100) + ring_query_latency_ms = (time.perf_counter() - ring_query_started) * 1000.0 + elapsed = max(time.perf_counter() - started, 0.000001) + packet_metrics = packet_queue.stats(worker_alive=True) + worker_metrics = workers.stats() + aggregator_metrics = aggregator.stats() + persistence_metrics = persistence.metrics() + ring_metrics = ring.metrics() + workers.close(10.0) + aggregator.close() + persistence.close(10.0) + + persistence_metrics["sink_events_written_total"] = sink_events_written + persistence_metrics["sink_batches_total"] = sink_batches + persistence_metrics["write_latency_ms_avg"] = persistence_metrics[ + "write_latency_avg_ms" + ] + persistence_metrics["write_latency_ms_p95"] = persistence_metrics[ + "write_latency_p95_ms" + ] + worker_metrics["processing_latency_ms_avg"] = worker_metrics[ + "avg_processing_latency_ms" + ] + worker_metrics["processing_latency_ms_p95"] = worker_metrics[ + "p95_processing_latency_ms" + ] + worker_metrics["processing_latency_ms_max"] = worker_metrics[ + "max_processing_latency_ms" + ] + aggregator_metrics["batches_emitted_total"] = aggregator_metrics[ + "batches_sent_total" + ] + ring_metrics["query_latency_ms_avg"] = round(ring_query_latency_ms, 3) + ring_metrics["query_latency_ms_p95"] = round(ring_query_latency_ms, 3) + aggregator_metrics.setdefault("send_latency_ms_avg", 0.0) + aggregator_metrics.setdefault("send_latency_ms_p95", 0.0) + websocket_metrics = { + "health": aggregator_metrics.get("health", "healthy"), + "slow_clients": 0, + "dropped_for_slow_client_total": 0, + "coalesced_for_slow_client_total": aggregator_metrics.get( + "events_coalesced_total", 0 + ), + "send_latency_ms_avg": aggregator_metrics.get("send_latency_ms_avg", 0.0), + "send_latency_ms_p95": aggregator_metrics.get("send_latency_ms_p95", 0.0), + } + dropped_total = sum( + int(value or 0) + for value in ( + packet_metrics.get("dropped_total"), + worker_metrics.get("jobs_dropped_total"), + worker_metrics.get("jobs_rejected_total"), + aggregator_metrics.get("events_dropped_total"), + persistence_metrics.get("events_dropped_total"), + ) + ) + failed_total = int(worker_metrics.get("jobs_failed_total") or 0) + int( + persistence_metrics.get("events_failed_total") or 0 + ) + final_health = _pipeline_health( + packet_metrics, + worker_metrics, + aggregator_metrics, + persistence_metrics, + ring_metrics, + ) + pressure_reasons = _pressure_reasons( + packet_metrics, + worker_metrics, + aggregator_metrics, + persistence_metrics, + ring_metrics, + ) + return { + "schema_version": 1, + "benchmark_type": "synthetic_full_pipeline_soak", + "safe_synthetic_only": True, + "configuration": config_dict(config), + "general": { + "start_time": start_time, + "end_time": utc_now(), + "duration_sec": round(elapsed, 3), + "target_packet_rate": target_packet_rate, + "target_alert_rate": target_alert_rate, + "events_generated_total": generated_total, + "events_processed_total": processed_total, + "alerts_generated_total": alert_total, + "events_dropped_total": dropped_total, + "events_failed_total": failed_total, + "throughput_events_sec": round(processed_total / elapsed, 2), + "pipeline_processing_latency_ms_p95": worker_metrics[ + "processing_latency_ms_p95" + ], + }, + "resources": resource_metrics, + "packet_queue": packet_metrics, + "flow_worker_pool": worker_metrics, + "event_aggregator": aggregator_metrics, + "websocket": websocket_metrics, + "persistence": persistence_metrics, + "live_ring_buffer": ring_metrics, + "ops_health": { + "final_health": final_health, + "pressure_reasons": pressure_reasons, + "transitions": health_transitions, + }, + "validation": { + "packet_queue_bounded": packet_metrics["high_water_mark"] + <= packet_metrics["max_size"], + "flow_worker_pool_bounded": worker_metrics["queue_depth_total"] + <= worker_metrics["queue_max_total"], + "persistence_bounded": persistence_metrics["queue_depth"] + <= persistence_metrics["queue_max"], + "live_ring_buffer_bounded": all( + values["records"] <= values["capacity"] + for values in ring_metrics["categories"].values() + ), + "reports_redacted": True, + "external_network_used": False, + "admin_privileges_required": False, + "live_capture_used": False, + }, + } + + +def _wait_for_persistence(writer: BatchPersistenceWriter, timeout_sec: float) -> bool: + deadline = time.monotonic() + timeout_sec + while time.monotonic() < deadline: + if writer.metrics()["queue_depth"] == 0: + return True + time.sleep(0.01) + return writer.metrics()["queue_depth"] == 0 + + +def _pipeline_health(*sections: dict[str, Any]) -> str: + levels = [str(section.get("health") or "healthy") for section in sections] + if "critical" in levels: + return "critical" + if "degraded" in levels or "warning" in levels: + return "degraded" + return "healthy" + + +def _pressure_reasons(*sections: dict[str, Any]) -> list[str]: + reasons: list[str] = [] + for section in sections: + for reason in section.get("pressure_reasons") or []: + safe_reason = str(reason) + if safe_reason not in reasons: + reasons.append(safe_reason) + return reasons + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Run a safe synthetic NetBotPro full-pipeline soak test." + ) + add_common_arguments(parser) + args = parser.parse_args() + config = config_from_args(args, soak=True) + results = run_soak_test(config) + paths = write_reports(results, config.output) + print( + json.dumps( + { + "ok": True, + "events_processed_total": results["general"]["events_processed_total"], + "final_health": results["ops_health"]["final_health"], + "output": paths, + }, + ensure_ascii=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index b802bdc..f74a91d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -237,7 +237,11 @@ retaining raw payloads or allowing unbounded memory growth. The Event Aggregator remains the realtime delivery-pressure boundary after the ring buffer, while Batch Persistence remains the storage-pressure boundary. -Benchmark/soak validation remains a future step. +The synthetic benchmark suite exercises each boundary independently and as a +full local pipeline. Its CI-safe smoke test checks bounded queues, visible +drops, report generation, and structural health without real capture, +Administrator privileges, or external network access. Machine-dependent +throughput is documented separately and is not treated as a capacity promise. ## WebSocket Event Aggregator diff --git a/docs/PERFORMANCE_PIPELINE.md b/docs/PERFORMANCE_PIPELINE.md index 32408bc..3de87ba 100644 --- a/docs/PERFORMANCE_PIPELINE.md +++ b/docs/PERFORMANCE_PIPELINE.md @@ -516,15 +516,67 @@ safe error type, per-category utilization, and fixed pressure reason enums. High utilization, frequent evictions, or capped queries degrade health; an internal read/write error is critical and contributes to overall Ops health. +## Benchmark / Soak Tests + +The safe synthetic suite under `benchmarks/` validates the five bounded stages +individually and through an integrated local soak path: + +```text +Synthetic packet summaries +-> Bounded Packet Intake Queue +-> Flow-aware Worker Pool +-> Live Ring Buffer +-> WebSocket Event Aggregator +-> Batch Persistence +``` + +The soak runner records generated, processed, dropped, and failed totals; +throughput; process RSS and CPU samples; queue pressure; stage latency; ring +evictions; and final Ops health. JSON and Markdown outputs pass through central +redaction. The CI-safe profile caps duration and load and asserts structural +properties instead of machine-specific throughput thresholds. + +Quick validation: + +```powershell +python benchmarks/soak_test_pipeline.py ` + --duration-sec 10 ` + --events-per-sec 200 ` + --flows 20 ` + --ci-safe ` + --output .runtime/benchmarks/smoke +``` + +Synthetic benchmarks do not prove production capacity, packet-capture driver +performance, disk durability under host failure, or behavior on real traffic. +They do make bounded pressure, drops, latency, and memory trends repeatable. +See `docs/PERFORMANCE_VALIDATION.md` for the recorded validation run. + +### Tuning Guidance + +- Increase intake or persistence queue sizes only after checking sustained + utilization and memory headroom. +- Prefer more flow workers only when CPU capacity exists and multiple flows are + active; same-flow work remains serialized by design. +- Adjust WebSocket batch windows before increasing client queue capacity. +- Size Live Ring categories for the investigation window actually needed; + eviction is expected bounded behavior, not data corruption. +- Validate every production setting with an authorized workload representative + of the deployment. + ## Current Limitations This is not the complete performance engine yet. Not implemented yet: -- Benchmark / Soak Tests; - Optional ClickHouse or external metrics backend. +The benchmark suite is synthetic and local-only. Real capture-driver load, +real PCAP replay, multi-host clients, long production soaks, and machine sizing +remain deployment validation work. This is still not the complete performance +engine or a production capacity certification. + This step also does not add command/control, remote shell, file collection, Agent raw packet forwarding, Agent raw payload forwarding, Agent PCAP forwarding, TLS decryption, MITM, credential collection, auto-block/IPS @@ -532,11 +584,9 @@ behavior, or AI autonomous actions. ## Next Planned Steps -1. Benchmark and Soak Tests -2. Performance Validation Report -3. Service Attribution / Destination Intelligence -4. Incident / Correlation Engine -5. Read-only AI Analyst +1. Service Attribution / Destination Intelligence +2. Incident / Correlation Engine +3. Read-only AI Analyst ### Recorded Product Direction diff --git a/docs/PERFORMANCE_VALIDATION.md b/docs/PERFORMANCE_VALIDATION.md new file mode 100644 index 0000000..9fcb989 --- /dev/null +++ b/docs/PERFORMANCE_VALIDATION.md @@ -0,0 +1,117 @@ +# Performance Validation + +## Status + +This report records Step 7 validation of NetBotPro's existing performance +foundation. It is a safe synthetic benchmark, not a production capacity claim. + +| Component | Status | +| --- | --- | +| Bounded Packet Queue | Validated | +| Flow-aware Worker Pool | Validated | +| WebSocket Event Aggregator | Validated | +| Batch Persistence | Validated | +| Live Ring Buffer | Validated with expected bounded eviction | + +## Purpose and System Under Test + +The suite verifies that pressure is bounded and observable from packet intake +through flow-aware processing, recent-memory storage, realtime aggregation, and +batched persistence. Each stage has a focused benchmark, and +`benchmarks/soak_test_pipeline.py` exercises the integrated path. + +```text +Synthetic packet/flow/alert summaries +-> Bounded Packet Intake Queue +-> Flow-aware Worker Pool +-> Live Ring Buffer +-> WebSocket Event Aggregator +-> Batch Persistence +-> Ops health and redacted reports +``` + +## Methodology and Safety + +The July 15, 2026 reference smoke used deterministic synthetic metadata on a +local Windows developer machine. It did not capture traffic, open external +network connections, require Administrator privileges, scan hosts, or store +raw payloads. Report data passed through central redaction before JSON and +Markdown were written. + +Configuration: + +```text +duration: 10 seconds +events per second: 200 +flow keys: 20 +profile: ci-safe +live capture: disabled +``` + +The CI test uses a shorter version and checks structure, bounded state, output +generation, and redaction. It deliberately avoids strict throughput limits so +slower CI operating systems are not treated as failures. + +## Reference Results + +| Metric | Result | +| --- | ---: | +| Measured duration | 10.001 sec | +| Events generated / processed | 2,000 / 2,000 | +| Alerts generated | 1,000 | +| Pipeline throughput | 199.98 events/sec | +| Dropped / failed events | 0 / 0 | +| Packet queue high-water / capacity | 1 / 2,000 | +| Worker average / p95 / max processing latency | 0.33 / 0.65 / 2.78 ms | +| Aggregator events received / sent | 3,200 / 3,200 | +| Persistence events received / written | 3,200 / 3,200 | +| Persistence batches | 126 | +| Ring records added / evicted | 5,000 / 4,200 | +| Memory start / peak / end | 25.62 / 29.73 / 29.68 MiB | +| CPU average / peak | 9.12% / 49.60% | + +These values describe one synthetic run on one machine and may vary. The +important validation result is that every queue remained within capacity, +processing completed without failure, drops remained observable, and the +output stayed bounded. + +## Observations + +No intake, worker, aggregator, or persistence drops occurred in the reference +run. The final Ops state was `degraded` only because the deliberately small +CI-safe Ring Buffer reached category limits and evicted old records. Pressure +reasons were `live_ring_high_utilization` and +`live_ring_frequent_evictions`. This is expected bounded behavior and confirms +that retention pressure is visible instead of causing unbounded growth. + +Process RSS grew by about 4.07 MiB during this short warm-up run and stabilized +below the observed peak at shutdown. A longer, deployment-specific soak is +still required before making memory-retention conclusions. + +## Tuning Recommendations + +| Workload signal | First tuning action | +| --- | --- | +| Intake drops or sustained queue utilization | Reduce capture pressure or increase `NETBOT_PACKET_QUEUE_MAX_SIZE` within memory limits. | +| Worker backlog across multiple flows | Increase `NETBOT_FLOW_WORKER_COUNT` only when CPU headroom exists. | +| Slow realtime clients | Tune WebSocket batch windows and slow-client policy before raising queue limits. | +| Persistence backlog | Tune batch size/flush interval, then queue capacity; inspect write latency first. | +| Frequent Ring Buffer eviction | Increase only the affected category capacity or accept a shorter live-history window. | + +## Limitations + +- No real packet capture, PCAP replay, driver pressure, or privileged interface + was tested. +- No external WebSocket clients or storage service was used. +- The persistence sink was local and safe; host crash durability was not + measured. +- The reference run is short and synthetic, so it is not a benchmark comparison + or production sizing certificate. +- Optional ClickHouse or an external metrics backend is not implemented. + +## Next Product Phase + +The performance foundation is ready to close Step 7. The next recommended +phase is conservative Service Attribution / Destination Intelligence. It must +preserve the existing local-first, redaction, Agent telemetry-only, and Remote +Sensor security boundaries. diff --git a/tests/test_performance_benchmarks.py b/tests/test_performance_benchmarks.py new file mode 100644 index 0000000..d56331f --- /dev/null +++ b/tests/test_performance_benchmarks.py @@ -0,0 +1,141 @@ +import json +import tempfile +import time +import unittest +from pathlib import Path + +from benchmarks.benchmark_batch_persistence import run_persistence_benchmark +from benchmarks.benchmark_live_ring_buffer import run_live_ring_benchmark +from benchmarks.benchmark_packet_pipeline import run_packet_queue_benchmark +from benchmarks.benchmark_report import ( + BenchmarkConfig, + ResourceSampler, + synthetic_alert, + synthetic_packet, + write_reports, +) +from benchmarks.benchmark_websocket_aggregator import ( + run_event_aggregator_benchmark, +) +from benchmarks.benchmark_worker_pool import run_worker_pool_benchmark +from benchmarks.soak_test_pipeline import run_soak_test + + +class PerformanceBenchmarkTests(unittest.TestCase): + def setUp(self): + self.config = BenchmarkConfig( + duration_sec=0.05, + events_per_sec=100, + flows=4, + packet_rate=90, + alert_rate=10, + ci_safe=True, + ) + + def test_synthetic_generators_are_safe_and_deterministic(self): + packet = synthetic_packet(7, 4) + alert = synthetic_alert(7, packet) + + self.assertEqual(packet, synthetic_packet(7, 4)) + self.assertEqual(packet["source"], "synthetic_benchmark") + self.assertEqual(alert["flow_id"], packet["flow_key"]) + serialized = json.dumps({"packet": packet, "alert": alert}).lower() + for forbidden in ["authorization", "cookie", "password", "bearer "]: + self.assertNotIn(forbidden, serialized) + + def test_resource_sampler_produces_structural_metrics(self): + sampler = ResourceSampler(interval_sec=0.01) + sampler.start() + time.sleep(0.03) + metrics = sampler.stop() + + self.assertGreater(metrics["memory_peak_bytes"], 0) + self.assertIn("cpu_percent_avg", metrics) + self.assertIn("memory_growth_bytes", metrics) + + def test_stage_benchmarks_produce_bounded_metrics(self): + scenarios = { + "packet_queue": run_packet_queue_benchmark, + "flow_worker_pool": run_worker_pool_benchmark, + "event_aggregator": run_event_aggregator_benchmark, + "persistence": run_persistence_benchmark, + "live_ring_buffer": run_live_ring_benchmark, + } + + for name, scenario in scenarios.items(): + with self.subTest(stage=name): + metrics = scenario(self.config) + self.assertTrue(metrics["bounded"]) + self.assertIn("health", metrics) + + def test_ci_safe_soak_is_local_bounded_and_complete(self): + started = time.perf_counter() + results = run_soak_test(self.config) + elapsed = time.perf_counter() - started + + self.assertGreater(results["general"]["events_processed_total"], 0) + self.assertGreaterEqual(results["general"]["duration_sec"], 0.04) + self.assertLess(results["general"]["duration_sec"], 1.0) + self.assertLess(elapsed, 2.0) + for section in [ + "packet_queue", + "flow_worker_pool", + "event_aggregator", + "websocket", + "persistence", + "live_ring_buffer", + "ops_health", + ]: + self.assertIn(section, results) + for key in [ + "packet_queue_bounded", + "flow_worker_pool_bounded", + "persistence_bounded", + "live_ring_buffer_bounded", + "reports_redacted", + ]: + self.assertTrue(results["validation"][key]) + self.assertFalse(results["validation"]["external_network_used"]) + self.assertFalse(results["validation"]["admin_privileges_required"]) + self.assertFalse(results["validation"]["live_capture_used"]) + + def test_report_generator_writes_redacted_json_and_markdown(self): + results = run_soak_test(self.config) + results["validation"]["unsafe_fixture"] = { + "Authorization": "Bearer benchmark-secret-token", + "password": "benchmark-password", + } + + with tempfile.TemporaryDirectory() as temporary_directory: + paths = write_reports(results, temporary_directory) + json_path = Path(paths["json"]) + markdown_path = Path(paths["markdown"]) + combined = json_path.read_text(encoding="utf-8") + markdown_path.read_text( + encoding="utf-8" + ) + + self.assertTrue(json_path.name.endswith(".json")) + self.assertTrue(markdown_path.name.endswith(".md")) + self.assertNotIn("benchmark-secret-token", combined) + self.assertNotIn("benchmark-password", combined) + self.assertIn("[REDACTED]", combined) + self.assertIn("packet_queue", combined) + self.assertIn("NetBotPro Synthetic Performance Summary", combined) + + def test_benchmark_sources_do_not_use_capture_or_network_clients(self): + root = Path(__file__).resolve().parents[1] / "benchmarks" + source = "\n".join( + path.read_text(encoding="utf-8") for path in root.glob("*.py") + ).lower() + for forbidden in [ + "import socket", + "import requests", + "from scapy", + "systemcapture", + "subprocess", + ]: + self.assertNotIn(forbidden, source) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_release_readiness.py b/tests/test_release_readiness.py index d4660f9..9fc88b0 100644 --- a/tests/test_release_readiness.py +++ b/tests/test_release_readiness.py @@ -117,6 +117,14 @@ def test_performance_pipeline_docs_are_linked_and_scoped(self): self.assertIn("Live Ring Buffer", architecture) self.assertIn("NETBOT_LIVE_RING_PACKET_MAX", performance) self.assertIn("GET /api/live/recent", performance) + self.assertIn("Benchmark / Soak Tests", performance) + self.assertIn("Performance Benchmark Suite", readme) + self.assertIn("docs/PERFORMANCE_VALIDATION.md", readme) + self.assertTrue((REPO_ROOT / "benchmarks/README.md").is_file()) + validation = self._read("docs/PERFORMANCE_VALIDATION.md").lower() + self.assertIn("safe synthetic benchmark", validation) + self.assertIn("not a production capacity claim", validation) + self.assertIn("live capture: disabled", validation) self.assertIn("Queue pressure metrics", readme) self.assertIn("Ops Snapshot packet queue visibility", readme) self.assertIn("NETBOT_PACKET_QUEUE_MAX_SIZE", readme) From f7a1b685432fc00354d497c5890fc7e22ff46cec Mon Sep 17 00:00:00 2001 From: maximus Date: Wed, 15 Jul 2026 20:06:59 +0330 Subject: [PATCH 2/2] Stabilize event aggregator timer test --- tests/test_event_aggregator.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/tests/test_event_aggregator.py b/tests/test_event_aggregator.py index 902483a..dbebdce 100644 --- a/tests/test_event_aggregator.py +++ b/tests/test_event_aggregator.py @@ -79,10 +79,13 @@ def test_batch_interval_triggers_flush(self): ) aggregator.publish("packet:new", {"id": "pkt-1"}) - time.sleep(0.08) + deadline = time.monotonic() + 1.0 + while not emitted and time.monotonic() < deadline: + time.sleep(0.01) self.assertEqual(len(emitted), 1) self.assertEqual(emitted[0]["type"], "packet_batch") + aggregator.close() def test_metrics_track_drops_and_health(self): aggregator = EventAggregator(lambda _message: None) @@ -118,7 +121,9 @@ def _fill_slow_client(self, policy: str): queue = bus.subscribe() for index in range(5): - bus._publish_direct({"version": 1, "type": "manual", "payload": {"i": index}}) + bus._publish_direct( + {"version": 1, "type": "manual", "payload": {"i": index}} + ) return bus, queue