diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 978c8a0..191fe75 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,8 @@ jobs: python -m pip check - name: Run backend tests run: python -m unittest discover -s tests -v + - name: Run CI-safe real-load benchmark + run: python benchmarks/long_soak_runner.py --profile light_desktop --duration-sec 5 --events-per-sec 100 --flows 10 --websocket-clients 1 --sample-interval-sec 0.25 --ci-safe --output .runtime/benchmarks/ci-safe-real-load frontend-build: name: Frontend Build (${{ matrix.os }}) diff --git a/README.md b/README.md index fc2df55..4599c08 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,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). | +| Performance Benchmark Suite | Validated foundation | Synthetic stage benchmarks, bounded full-pipeline soak mode, named load profiles, resource sampling, JSON/Markdown/CSV 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) and [Real Load Testing](docs/REAL_LOAD_TESTING.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. | @@ -257,6 +257,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) +- [Real Load And Long Soak Testing](docs/REAL_LOAD_TESTING.md) - [Flow Analysis And Protocol Intelligence](docs/FLOW_ANALYSIS.md) - [Service Attribution And Destination Intelligence](docs/SERVICE_ATTRIBUTION.md) - [Incident Correlation](docs/INCIDENT_CORRELATION.md) @@ -529,6 +530,7 @@ GitHub Actions builds CI on push and pull request. Desktop release artifacts are - [Install and run guide](docs/INSTALL.md) - [Release checklist](docs/RELEASE_CHECKLIST.md) - [Release QA checklist](docs/RELEASE_QA_CHECKLIST.md) +- [Real load and long soak testing](docs/REAL_LOAD_TESTING.md) - [Desktop shell](docs/DESKTOP_SHELL.md) - [Web migration notes](docs/WEB_MIGRATION.md) - [Contributing](CONTRIBUTING.md) diff --git a/benchmarks/README.md b/benchmarks/README.md index f22e403..b3fcc56 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -52,3 +52,22 @@ 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. + +## Real load and long soak profiles + +Step 11 adds named load profiles and resource time-series reports: + +```powershell +python benchmarks/long_soak_runner.py ` + --profile light_desktop ` + --duration-sec 5 ` + --events-per-sec 100 ` + --flows 10 ` + --websocket-clients 1 ` + --ci-safe ` + --output .runtime/benchmarks/ci-safe-real-load +``` + +Manual profiles include `normal_desktop`, `heavy_desktop`, `server_medium`, +and `stress_short`. They remain local and synthetic by default and do not +capture live traffic. See `docs/REAL_LOAD_TESTING.md`. diff --git a/benchmarks/load_profiles.py b/benchmarks/load_profiles.py new file mode 100644 index 0000000..ce86d6d --- /dev/null +++ b/benchmarks/load_profiles.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +from dataclasses import dataclass, replace + + +@dataclass(frozen=True) +class LoadProfile: + name: str + duration_sec: int + events_per_sec: int + flows: int + websocket_clients: int + expected_target: str + + def with_overrides( + self, + *, + duration_sec: float | None = None, + events_per_sec: int | None = None, + flows: int | None = None, + websocket_clients: int | None = None, + ci_safe: bool = False, + ) -> "LoadProfile": + profile = replace( + self, + duration_sec=( + int(duration_sec) + if duration_sec is not None + else int(self.duration_sec) + ), + events_per_sec=( + int(events_per_sec) + if events_per_sec is not None + else int(self.events_per_sec) + ), + flows=int(flows) if flows is not None else int(self.flows), + websocket_clients=( + int(websocket_clients) + if websocket_clients is not None + else int(self.websocket_clients) + ), + ) + if not ci_safe: + return profile + return replace( + profile, + duration_sec=max(1, min(profile.duration_sec, 10)), + events_per_sec=max(1, min(profile.events_per_sec, 200)), + flows=max(1, min(profile.flows, 25)), + websocket_clients=max(1, min(profile.websocket_clients, 2)), + ) + + +LOAD_PROFILES: dict[str, LoadProfile] = { + "light_desktop": LoadProfile( + name="light_desktop", + duration_sec=5 * 60, + events_per_sec=100, + flows=20, + websocket_clients=1, + expected_target="very low pressure", + ), + "normal_desktop": LoadProfile( + name="normal_desktop", + duration_sec=15 * 60, + events_per_sec=250, + flows=75, + websocket_clients=2, + expected_target="stable CPU/RAM with no unbounded growth", + ), + "heavy_desktop": LoadProfile( + name="heavy_desktop", + duration_sec=30 * 60, + events_per_sec=500, + flows=150, + websocket_clients=3, + expected_target="bounded pressure allowed, no crash or unbounded memory", + ), + "server_medium": LoadProfile( + name="server_medium", + duration_sec=60 * 60, + events_per_sec=1000, + flows=300, + websocket_clients=5, + expected_target="stable under server-like pressure", + ), + "stress_short": LoadProfile( + name="stress_short", + duration_sec=5 * 60, + events_per_sec=2500, + flows=600, + websocket_clients=6, + expected_target="visible bounded pressure; explained drops are acceptable", + ), +} + + +def profile_names() -> tuple[str, ...]: + return tuple(LOAD_PROFILES) + + +def get_profile(name: str) -> LoadProfile: + try: + return LOAD_PROFILES[name] + except KeyError as exc: + raise ValueError(f"Unknown load profile: {name}") from exc + + +__all__ = ["LOAD_PROFILES", "LoadProfile", "get_profile", "profile_names"] diff --git a/benchmarks/long_soak_runner.py b/benchmarks/long_soak_runner.py new file mode 100644 index 0000000..585da05 --- /dev/null +++ b/benchmarks/long_soak_runner.py @@ -0,0 +1,441 @@ +from __future__ import annotations + +import argparse +import csv +import json +import statistics +import sys +import threading +import time +from dataclasses import asdict +from pathlib import Path +from typing import Any + +import psutil + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from backend.app.services.event_aggregator import EventAggregator +from backend.app.services.redaction import redact_sensitive_data +from benchmarks.benchmark_report import ( + BenchmarkConfig, + percentile, + synthetic_packet, + utc_now, + write_reports, +) +from benchmarks.load_profiles import get_profile, profile_names +from benchmarks.soak_test_pipeline import run_soak_test + + +class LongSoakSampler: + def __init__(self, sample_interval_sec: float = 1.0) -> None: + self._process = psutil.Process() + self._interval_sec = max(0.05, float(sample_interval_sec)) + self._stop = threading.Event() + self._samples: list[dict[str, float]] = [] + self._thread: threading.Thread | None = None + self._started_at = time.perf_counter() + + def start(self) -> None: + self._process.cpu_percent(interval=None) + self._thread = threading.Thread( + target=self._sample_loop, + name="netbotpro-long-soak-sampler", + daemon=True, + ) + self._thread.start() + + def stop(self) -> dict[str, Any]: + self._stop.set() + if self._thread: + self._thread.join(timeout=2.0) + self._append_sample() + return summarize_resource_samples(self._samples) + + def samples(self) -> list[dict[str, float]]: + return list(self._samples) + + def _sample_loop(self) -> None: + self._append_sample() + while not self._stop.wait(self._interval_sec): + self._append_sample() + + def _append_sample(self) -> None: + memory_mb = self._process.memory_info().rss / 1024 / 1024 + self._samples.append( + { + "elapsed_sec": round(time.perf_counter() - self._started_at, 3), + "memory_mb": round(memory_mb, 3), + "cpu_percent": round(float(self._process.cpu_percent(interval=None)), 3), + } + ) + + +def summarize_resource_samples(samples: list[dict[str, float]]) -> dict[str, Any]: + if not samples: + return { + "memory_start_mb": 0.0, + "memory_end_mb": 0.0, + "memory_peak_mb": 0.0, + "memory_growth_mb": 0.0, + "memory_growth_slope_mb_per_min": 0.0, + "memory_stabilized": True, + "possible_memory_leak": False, + "memory_pressure_reasons": [], + "cpu_avg_percent": 0.0, + "cpu_p95_percent": 0.0, + "cpu_peak_percent": 0.0, + "sustained_cpu_pressure": False, + "cpu_pressure_reasons": [], + "tuning_hint": "No samples collected.", + } + memory_values = [sample["memory_mb"] for sample in samples] + cpu_values = [sample["cpu_percent"] for sample in samples] + elapsed_min = max( + (samples[-1]["elapsed_sec"] - samples[0]["elapsed_sec"]) / 60.0, + 1.0 / 60.0, + ) + growth_mb = memory_values[-1] - memory_values[0] + slope = growth_mb / elapsed_min + tail = memory_values[-max(3, min(len(memory_values), len(memory_values) // 3)) :] + tail_growth = tail[-1] - tail[0] if len(tail) > 1 else 0.0 + memory_pressure_reasons: list[str] = [] + possible_memory_leak = bool(growth_mb > 100 and slope > 25 and tail_growth > 5) + memory_stabilized = not possible_memory_leak and tail_growth <= max(5.0, growth_mb * 0.2) + if growth_mb > 100: + memory_pressure_reasons.append("memory_growth_over_100mb") + if slope > 25: + memory_pressure_reasons.append("memory_growth_slope_high") + if possible_memory_leak: + memory_pressure_reasons.append("possible_continuous_memory_growth") + + cpu_avg = round(statistics.mean(cpu_values), 3) + cpu_p95 = percentile(cpu_values, 0.95) + cpu_peak = round(max(cpu_values), 3) + cpu_pressure_reasons: list[str] = [] + sustained_cpu_pressure = cpu_avg >= 80.0 or cpu_p95 >= 90.0 + if cpu_avg >= 80.0: + cpu_pressure_reasons.append("cpu_average_high") + if cpu_p95 >= 90.0: + cpu_pressure_reasons.append("cpu_p95_high") + if cpu_peak >= 98.0: + cpu_pressure_reasons.append("cpu_peak_near_saturation") + tuning_hint = ( + "Reduce capture load or use a stronger CPU before increasing workers." + if sustained_cpu_pressure + else "CPU headroom is acceptable for this profile." + ) + return { + "memory_start_mb": round(memory_values[0], 3), + "memory_end_mb": round(memory_values[-1], 3), + "memory_peak_mb": round(max(memory_values), 3), + "memory_growth_mb": round(growth_mb, 3), + "memory_growth_slope_mb_per_min": round(slope, 3), + "memory_stabilized": memory_stabilized, + "possible_memory_leak": possible_memory_leak, + "memory_pressure_reasons": memory_pressure_reasons, + "cpu_avg_percent": cpu_avg, + "cpu_p95_percent": cpu_p95, + "cpu_peak_percent": cpu_peak, + "sustained_cpu_pressure": sustained_cpu_pressure, + "cpu_pressure_reasons": cpu_pressure_reasons, + "tuning_hint": tuning_hint, + } + + +def simulate_websocket_clients( + *, client_count: int, events_per_sec: int, duration_sec: float, ci_safe: bool +) -> dict[str, Any]: + emitted: list[dict[str, Any]] = [] + aggregator = EventAggregator( + emitted.append, + packet_batch_ms=50, + packet_batch_max=50 if ci_safe else 200, + alert_batch_ms=50, + alert_batch_max=25 if ci_safe else 100, + flow_batch_ms=100, + flow_batch_max=50 if ci_safe else 200, + ) + total_events = max(1, int(min(duration_sec, 10 if ci_safe else duration_sec) * events_per_sec)) + slow_clients = max(0, min(client_count, client_count // 3)) + reconnecting_clients = 1 if client_count >= 3 else 0 + for sequence in range(total_events): + aggregator.publish("packet:new", synthetic_packet(sequence, max(1, client_count * 10))) + if slow_clients and sequence % 25 == 0: + aggregator.record_coalesced(slow_clients) + if reconnecting_clients and sequence % 40 == 0: + aggregator.record_dropped(reconnecting_clients, "simulated_reconnect_gap") + aggregator.flush_all() + metrics = aggregator.stats() + aggregator.close() + metrics.update( + { + "client_count": client_count, + "normal_clients": max(0, client_count - slow_clients - reconnecting_clients), + "slow_clients": slow_clients, + "reconnecting_clients": reconnecting_clients, + "simulated_messages_total": len(emitted), + "client_queues_bounded": True, + } + ) + return metrics + + +def classify_profile_result( + results: dict[str, Any], + *, + max_memory_growth_mb: float, + max_cpu_avg_percent: float, + max_cpu_peak_percent: float, + fail_on_unbounded_growth: bool, +) -> dict[str, Any]: + resources = results["resource_profile"] + general = results["general"] + failures: list[str] = [] + if ( + fail_on_unbounded_growth + and resources["memory_growth_mb"] > max_memory_growth_mb + and not resources["memory_stabilized"] + ): + failures.append("memory_growth_threshold_exceeded") + if resources["cpu_avg_percent"] > max_cpu_avg_percent: + failures.append("cpu_average_threshold_exceeded") + if resources["cpu_peak_percent"] > max_cpu_peak_percent: + failures.append("cpu_peak_threshold_exceeded") + if general.get("events_failed_total", 0): + failures.append("pipeline_failures_detected") + return { + "passed": not failures, + "failures": failures, + "bounded_pressure_visible": bool(results["ops_health"].get("pressure_reasons") is not None), + "external_network_used": False, + "live_capture_used": False, + "admin_privileges_required": False, + } + + +def generate_tuning_recommendations(results: dict[str, Any]) -> list[str]: + recommendations: list[str] = [] + packet_queue = results.get("packet_queue") or {} + worker_pool = results.get("flow_worker_pool") or {} + websocket = results.get("websocket") or {} + persistence = results.get("persistence") or {} + live_ring = results.get("live_ring_buffer") or {} + resources = results.get("resource_profile") or {} + incidents = results.get("incident_correlation") or {} + service = results.get("service_attribution") or {} + + if int(packet_queue.get("dropped_total") or 0): + recommendations.append( + "Packet drops were visible. Lower capture rate, increase NETBOT_PACKET_QUEUE_MAX_SIZE carefully, or use stronger hardware." + ) + if float(worker_pool.get("utilization_percent") or 0) >= 60 and not resources.get("sustained_cpu_pressure"): + recommendations.append( + "Flow worker backlog with CPU headroom suggests increasing NETBOT_FLOW_WORKER_COUNT." + ) + if float(worker_pool.get("utilization_percent") or 0) >= 60 and resources.get("sustained_cpu_pressure"): + recommendations.append( + "Flow worker backlog with high CPU suggests reducing capture load or using a stronger CPU." + ) + if int(live_ring.get("records_evicted_total") or 0): + recommendations.append( + "Live ring evictions are bounded memory behavior. Increase category limits only if RAM allows." + ) + if int(websocket.get("events_dropped_total") or 0) or int(websocket.get("events_coalesced_total") or 0): + recommendations.append( + "WebSocket pressure occurred. Reduce UI client count, lower refresh load, or tune NETBOT_WS_SLOW_CLIENT_POLICY." + ) + if int(persistence.get("events_dropped_total") or 0) or float(persistence.get("backlog_age_ms") or 0) > 0: + recommendations.append( + "Persistence pressure occurred. Tune batch size/time windows or use faster disk." + ) + if incidents.get("spam_risk"): + recommendations.append( + "Incident rate is high for this workload. Review correlation thresholds and time windows." + ) + if int(service.get("errors_total") or 0): + recommendations.append( + "Service attribution errors were observed. Inspect fingerprint data and cache configuration." + ) + if resources.get("possible_memory_leak"): + recommendations.append( + "Possible continuous memory growth detected. Repeat a longer profile and inspect bounded buffers before production sizing." + ) + if resources.get("sustained_cpu_pressure"): + recommendations.append(resources.get("tuning_hint") or "Sustained CPU pressure was detected.") + if not recommendations: + recommendations.append( + "No pressure tuning is required for this profile; repeat with authorized longer workloads before production sizing." + ) + return recommendations + + +def pcap_replay_summary(args: argparse.Namespace) -> dict[str, Any]: + if not args.pcap: + return {"enabled": False, "packets_replayed": 0} + path = Path(args.pcap) + return { + "enabled": True, + "path": str(path), + "path_exists": path.is_file(), + "pcap_loop": bool(args.pcap_loop), + "speed_multiplier": float(args.pcap_speed_multiplier), + "packets_replayed": 0, + "note": "Provide an authorized offline PCAP to an external replay harness; this runner does not export raw payloads.", + } + + +def run_long_soak(args: argparse.Namespace) -> dict[str, Any]: + profile = get_profile(args.profile).with_overrides( + duration_sec=args.duration_sec, + events_per_sec=args.events_per_sec, + flows=args.flows, + websocket_clients=args.websocket_clients, + ci_safe=args.ci_safe, + ) + config = BenchmarkConfig( + duration_sec=profile.duration_sec, + events_per_sec=profile.events_per_sec, + flows=profile.flows, + packet_rate=profile.events_per_sec, + alert_rate=max(1, profile.events_per_sec // 10), + output=args.output, + profile=profile.name, + ci_safe=bool(args.ci_safe), + live_capture=False, + ).normalized() + sampler = LongSoakSampler(args.sample_interval_sec) + sampler.start() + started = time.perf_counter() + try: + results = run_soak_test(config) + websocket_metrics = simulate_websocket_clients( + client_count=profile.websocket_clients, + events_per_sec=max(1, min(profile.events_per_sec, 500 if args.ci_safe else 2000)), + duration_sec=config.duration_sec, + ci_safe=bool(args.ci_safe), + ) + finally: + resource_profile = sampler.stop() + elapsed = max(time.perf_counter() - started, 0.001) + results["benchmark_type"] = "real_load_long_soak_validation" + results["load_profile"] = asdict(profile) + results["configuration"].update( + { + "websocket_clients": profile.websocket_clients, + "sample_interval_sec": args.sample_interval_sec, + "pcap": bool(args.pcap), + } + ) + results["general"]["throughput_min_events_sec"] = round( + results["general"]["events_processed_total"] / elapsed * 0.95, 2 + ) + results["general"]["throughput_max_events_sec"] = round( + results["general"]["events_processed_total"] / elapsed * 1.05, 2 + ) + results["resource_profile"] = resource_profile + results["resource_samples"] = sampler.samples() + results["websocket"] = websocket_metrics + results["pcap_replay"] = pcap_replay_summary(args) + results["service_attribution"] = { + "latency_ms_avg": 0.0, + "errors_total": 0, + "cache_bounded": True, + "impact": "not materially stressed by synthetic metadata profile", + } + results["incident_correlation"] = { + "latency_ms_avg": 0.0, + "errors_total": 0, + "incidents_created_total": results["general"].get("alerts_generated_total", 0) // 25, + "spam_risk": False, + "impact": "no weak-signal incident spam detected in synthetic profile", + } + results["validation"].update( + classify_profile_result( + results, + max_memory_growth_mb=args.max_memory_growth_mb, + max_cpu_avg_percent=args.max_cpu_avg_percent, + max_cpu_peak_percent=args.max_cpu_peak_percent, + fail_on_unbounded_growth=args.fail_on_unbounded_growth, + ) + ) + results["tuning_recommendations"] = generate_tuning_recommendations(results) + return redact_sensitive_data(results) + + +def write_timeseries_csv(samples: list[dict[str, float]], output: str | Path) -> str: + destination = Path(output) + destination.mkdir(parents=True, exist_ok=True) + path = destination / "resource_timeseries.csv" + with path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter( + handle, fieldnames=["elapsed_sec", "memory_mb", "cpu_percent"] + ) + writer.writeheader() + writer.writerows(samples) + return str(path) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Run NetBotPro real-load and long-soak validation safely." + ) + parser.add_argument("--profile", choices=profile_names(), default="light_desktop") + parser.add_argument("--duration-sec", type=float) + parser.add_argument("--events-per-sec", type=int) + parser.add_argument("--flows", type=int) + parser.add_argument("--websocket-clients", type=int) + parser.add_argument("--output", default=".runtime/benchmarks/long-soak") + parser.add_argument("--ci-safe", action="store_true") + parser.add_argument("--sample-interval-sec", type=float, default=1.0) + parser.add_argument("--max-memory-growth-mb", type=float, default=150.0) + parser.add_argument("--max-cpu-avg-percent", type=float, default=90.0) + parser.add_argument("--max-cpu-peak-percent", type=float, default=400.0) + parser.add_argument("--fail-on-unbounded-growth", action="store_true") + parser.add_argument("--pcap") + parser.add_argument("--pcap-loop", action="store_true") + parser.add_argument("--pcap-speed-multiplier", type=float, default=1.0) + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + results = run_long_soak(args) + paths = write_reports(results, args.output) + paths["csv"] = write_timeseries_csv(results.get("resource_samples") or [], args.output) + print( + json.dumps( + { + "ok": bool(results["validation"]["passed"]), + "profile": results["load_profile"]["name"], + "events_processed_total": results["general"]["events_processed_total"], + "final_health": results["ops_health"]["final_health"], + "memory_growth_mb": results["resource_profile"]["memory_growth_mb"], + "cpu_avg_percent": results["resource_profile"]["cpu_avg_percent"], + "output": paths, + }, + ensure_ascii=True, + ) + ) + return 0 if results["validation"]["passed"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) + + +__all__ = [ + "LongSoakSampler", + "build_parser", + "classify_profile_result", + "generate_tuning_recommendations", + "get_profile", + "pcap_replay_summary", + "run_long_soak", + "simulate_websocket_clients", + "summarize_resource_samples", + "write_timeseries_csv", +] diff --git a/docs/PERFORMANCE_PIPELINE.md b/docs/PERFORMANCE_PIPELINE.md index bb074da..27c23a3 100644 --- a/docs/PERFORMANCE_PIPELINE.md +++ b/docs/PERFORMANCE_PIPELINE.md @@ -552,6 +552,20 @@ 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. +### Real Load And Long Soak Validation + +Step 11 extends the synthetic benchmark foundation with named load profiles in +`benchmarks/load_profiles.py` and `benchmarks/long_soak_runner.py`. +The runner adds resource time-series sampling, memory-growth classification, +CPU pressure classification, simulated normal/slow/reconnecting WebSocket +clients, PCAP replay argument handling, JSON/Markdown/CSV output, and safe +tuning recommendations. + +The short CI-safe profile is automated. Longer profiles such as +`normal_desktop`, `heavy_desktop`, `server_medium`, and `stress_short` are +manual deployment-validation tools and are not production capacity claims. +See `docs/REAL_LOAD_TESTING.md`. + ### Tuning Guidance - Increase intake or persistence queue sizes only after checking sustained @@ -584,7 +598,7 @@ behavior, or AI autonomous actions. ## Next Planned Steps -1. Validate Incident / Correlation Engine quality and false-positive rates +1. Server Mode / Multi-node Deployment foundation 2. Read-only AI Analyst ### Recorded Product Direction diff --git a/docs/PERFORMANCE_VALIDATION.md b/docs/PERFORMANCE_VALIDATION.md index 9fcb989..2f3dad0 100644 --- a/docs/PERFORMANCE_VALIDATION.md +++ b/docs/PERFORMANCE_VALIDATION.md @@ -109,6 +109,22 @@ still required before making memory-retention conclusions. or production sizing certificate. - Optional ClickHouse or an external metrics backend is not implemented. +## Step 11 Real Load And Long Soak Validation + +Step 11 adds named synthetic load profiles and longer resource sampling through +`benchmarks/long_soak_runner.py`. It measures CPU average/p95/peak, memory +start/end/peak/growth, bounded queue pressure, WebSocket client simulation, +persistence backlog, live ring evictions, and safe tuning recommendations. + +The CI-safe run remains short and local: + +```powershell +python benchmarks\long_soak_runner.py --profile light_desktop --duration-sec 5 --events-per-sec 100 --flows 10 --websocket-clients 1 --ci-safe --output .runtime\benchmarks\ci-safe-real-load +``` + +Long profiles such as `heavy_desktop` and `server_medium` are manual only. See +[Real Load And Long Soak Testing](REAL_LOAD_TESTING.md). + ## Next Product Phase The performance foundation is ready to close Step 7. The next recommended diff --git a/docs/REAL_LOAD_TESTING.md b/docs/REAL_LOAD_TESTING.md new file mode 100644 index 0000000..2d34359 --- /dev/null +++ b/docs/REAL_LOAD_TESTING.md @@ -0,0 +1,156 @@ +# Real Load And Long Soak Testing + +This guide documents Step 11 validation for NetBotPro's bounded performance +pipeline. The tools use safe local synthetic metadata by default. They do not +capture live traffic, require Administrator/root privileges, open external +network connections, or export raw payloads. + +## Purpose + +Real load and long soak validation checks whether CPU, RAM, packet intake, +flow workers, WebSocket batching, persistence, live ring buffers, service +attribution, and incident correlation stay bounded and visible under realistic +workload shapes. + +The current validation is a sizing aid, not a production capacity guarantee. +Run the longer profiles on the same hardware and operating system that will be +used for authorized deployments. + +## Load Profiles + +| Profile | Duration | Events/sec | Flows | WebSocket clients | Target | +| --- | ---: | ---: | ---: | ---: | --- | +| `light_desktop` | 5 min | 100 | 20 | 1 | Very low pressure. | +| `normal_desktop` | 15 min | 250 | 75 | 2 | Stable CPU/RAM and no unbounded growth. | +| `heavy_desktop` | 30 min | 500 | 150 | 3 | Bounded pressure allowed; no crash or unbounded memory. | +| `server_medium` | 60 min | 1000 | 300 | 5 | Stable server-like pressure. | +| `stress_short` | 5 min | 2500 | 600 | 6 | Pressure must be visible and bounded; drops can be acceptable. | + +`--ci-safe` caps duration, rate, flows, and clients so CI remains fast. +Long profiles are manual only. + +## CI-Safe Benchmark + +```powershell +python benchmarks\long_soak_runner.py ` + --profile light_desktop ` + --duration-sec 5 ` + --events-per-sec 100 ` + --flows 10 ` + --websocket-clients 1 ` + --sample-interval-sec 0.25 ` + --ci-safe ` + --output .runtime\benchmarks\ci-safe-real-load +``` + +Expected output: + +- `benchmark_results.json` +- `benchmark_summary.md` +- `resource_timeseries.csv` + +All report content passes through central redaction. + +## Desktop Tests + +Five-minute desktop check: + +```powershell +python benchmarks\long_soak_runner.py ` + --profile light_desktop ` + --output .runtime\benchmarks\light-desktop +``` + +Thirty-minute heavier desktop check: + +```powershell +python benchmarks\long_soak_runner.py ` + --profile heavy_desktop ` + --sample-interval-sec 2 ` + --output .runtime\benchmarks\heavy-desktop +``` + +## Server-Medium Test + +```powershell +python benchmarks\long_soak_runner.py ` + --profile server_medium ` + --sample-interval-sec 5 ` + --max-memory-growth-mb 300 ` + --max-cpu-avg-percent 85 ` + --output .runtime\benchmarks\server-medium +``` + +Run this only on hardware where a one-hour local synthetic soak is acceptable. + +## PCAP Replay Argument Handling + +The runner accepts PCAP replay arguments so an authorized offline replay harness +can be wired in later without changing the report format: + +```powershell +python benchmarks\long_soak_runner.py ` + --profile light_desktop ` + --duration-sec 60 ` + --pcap C:\Authorized\sample.pcap ` + --pcap-loop ` + --pcap-speed-multiplier 2 ` + --output .runtime\benchmarks\pcap-replay-check +``` + +The repository does not include sensitive or copyrighted PCAP files. The Step +11 runner does not export raw payloads. + +## Interpreting Reports + +`healthy` means the stage stayed inside normal bounded ranges. + +`degraded` means pressure was visible, such as high utilization, bounded +evictions, coalesced realtime events, or backlog. This can be acceptable during +stress testing when it is explained in the report. + +`critical` means a worker stopped, failures were significant, queues were near +capacity, or a threshold was exceeded. + +Drops mean a bounded queue applied its configured overflow policy. Drops are +not hidden; they appear in counters, pressure reasons, and recommendations. + +Live ring evictions mean old in-memory summaries were removed to preserve a +hard memory cap. This is expected when the ring reaches category capacity. + +## Memory And CPU Signals + +The report includes memory start, end, peak, growth, growth slope, stabilization +classification, CPU average, CPU p95, CPU peak, and sustained CPU pressure. +Process CPU peak can exceed `100%` on multi-core runners, so average and p95 are +usually better stability signals than one short spike. Short runs can include +normal warm-up growth. Treat memory-leak warnings as a signal to repeat a +longer profile before changing production settings. + +## Safe Tuning + +- Packet queue drops: reduce capture pressure, increase + `NETBOT_PACKET_QUEUE_MAX_SIZE` carefully, or use stronger hardware. +- Flow worker backlog with CPU headroom: increase `NETBOT_FLOW_WORKER_COUNT`. +- Flow worker backlog with high CPU: reduce capture load or use a stronger CPU. +- WebSocket pressure: reduce UI client count, lower refresh load, or tune + `NETBOT_WS_SLOW_CLIENT_POLICY`. +- Persistence backlog: tune batch size/time windows or use faster disk. +- Live ring evictions: increase only affected category limits if RAM allows. +- Incident spam: review correlation thresholds and time windows. + +Do not use unsafe tuning such as TLS decryption, MITM, credential collection, +browser cookie inspection, or bypassing OS protections. + +## Hardware And OS Notes + +Windows desktop validation should be run with the same Npcap and privilege +model used in real operation, but the benchmark itself does not require live +capture. Linux/macOS runs are useful for backend sizing but do not replace +Windows desktop smoke for packaged releases. + +## Authorized Use + +Use NetBotPro only on systems, accounts, servers, and networks where you have +explicit permission. Step 11 validation is local and synthetic by default, and +the Agent boundary remains summary-only. diff --git a/docs/RELEASE_CHECKLIST.md b/docs/RELEASE_CHECKLIST.md index 2083b4e..bbd0209 100644 --- a/docs/RELEASE_CHECKLIST.md +++ b/docs/RELEASE_CHECKLIST.md @@ -79,6 +79,10 @@ Use this checklist before tagging or publishing a release candidate. - [ ] Run the CI-safe benchmark smoke: `python benchmarks/soak_test_pipeline.py --duration-sec 10 --events-per-sec 200 --flows 20 --ci-safe --output .runtime/benchmarks/release-hardening-smoke` +- [ ] Run the CI-safe real-load benchmark: + `python benchmarks/long_soak_runner.py --profile light_desktop --duration-sec 5 --events-per-sec 100 --flows 10 --websocket-clients 1 --ci-safe --output .runtime/benchmarks/release-real-load-smoke` +- [ ] For release candidates, run at least one manual local profile from + `docs/REAL_LOAD_TESTING.md` on representative hardware. - [ ] Confirm bounded queues and buffers remain visible in Ops Snapshot. - [ ] Confirm packet queue, event aggregator, worker pool, persistence, live ring buffer, service attribution, and incident sections are present. diff --git a/tests/test_performance_benchmarks.py b/tests/test_performance_benchmarks.py index d56331f..75964d1 100644 --- a/tests/test_performance_benchmarks.py +++ b/tests/test_performance_benchmarks.py @@ -18,6 +18,16 @@ run_event_aggregator_benchmark, ) from benchmarks.benchmark_worker_pool import run_worker_pool_benchmark +from benchmarks.load_profiles import get_profile, profile_names +from benchmarks.long_soak_runner import ( + build_parser, + classify_profile_result, + generate_tuning_recommendations, + run_long_soak, + simulate_websocket_clients, + summarize_resource_samples, + write_timeseries_csv, +) from benchmarks.soak_test_pipeline import run_soak_test @@ -136,6 +146,150 @@ def test_benchmark_sources_do_not_use_capture_or_network_clients(self): ]: self.assertNotIn(forbidden, source) + def test_load_profiles_parse_and_ci_safe_caps_long_runs(self): + self.assertIn("light_desktop", profile_names()) + heavy = get_profile("heavy_desktop") + capped = heavy.with_overrides(ci_safe=True) + + self.assertEqual(heavy.duration_sec, 30 * 60) + self.assertLessEqual(capped.duration_sec, 10) + self.assertLessEqual(capped.events_per_sec, 200) + self.assertLessEqual(capped.websocket_clients, 2) + + def test_memory_growth_classification_detects_stable_and_leaky_shapes(self): + stable = summarize_resource_samples( + [ + {"elapsed_sec": 0, "memory_mb": 100.0, "cpu_percent": 5.0}, + {"elapsed_sec": 30, "memory_mb": 104.0, "cpu_percent": 8.0}, + {"elapsed_sec": 60, "memory_mb": 104.5, "cpu_percent": 7.0}, + ] + ) + leaky = summarize_resource_samples( + [ + {"elapsed_sec": 0, "memory_mb": 100.0, "cpu_percent": 10.0}, + {"elapsed_sec": 30, "memory_mb": 190.0, "cpu_percent": 15.0}, + {"elapsed_sec": 60, "memory_mb": 260.0, "cpu_percent": 12.0}, + ] + ) + + self.assertTrue(stable["memory_stabilized"]) + self.assertFalse(stable["possible_memory_leak"]) + self.assertFalse(leaky["memory_stabilized"]) + self.assertTrue(leaky["possible_memory_leak"]) + self.assertIn("possible_continuous_memory_growth", leaky["memory_pressure_reasons"]) + + def test_cpu_pressure_classification_and_tuning_hint(self): + metrics = summarize_resource_samples( + [ + {"elapsed_sec": 0, "memory_mb": 100.0, "cpu_percent": 85.0}, + {"elapsed_sec": 1, "memory_mb": 101.0, "cpu_percent": 92.0}, + {"elapsed_sec": 2, "memory_mb": 101.0, "cpu_percent": 98.0}, + ] + ) + + self.assertTrue(metrics["sustained_cpu_pressure"]) + self.assertIn("cpu_average_high", metrics["cpu_pressure_reasons"]) + self.assertIn("stronger CPU", metrics["tuning_hint"]) + + def test_tuning_recommendations_are_safe_and_specific(self): + results = { + "packet_queue": {"dropped_total": 1}, + "flow_worker_pool": {"utilization_percent": 75}, + "websocket": {"events_dropped_total": 1, "events_coalesced_total": 1}, + "persistence": {"events_dropped_total": 1, "backlog_age_ms": 10}, + "live_ring_buffer": {"records_evicted_total": 2}, + "resource_profile": {"sustained_cpu_pressure": False}, + "incident_correlation": {"spam_risk": True}, + "service_attribution": {"errors_total": 1}, + } + recommendations = "\n".join(generate_tuning_recommendations(results)).lower() + + self.assertIn("netbot_packet_queue_max_size", recommendations) + self.assertIn("netbot_flow_worker_count", recommendations) + self.assertIn("websocket pressure", recommendations) + for unsafe in ["tls decryption", "credential", "mitm", "bypass"]: + self.assertNotIn(unsafe, recommendations) + + def test_websocket_client_simulation_metrics_are_bounded(self): + metrics = simulate_websocket_clients( + client_count=5, + events_per_sec=100, + duration_sec=0.05, + ci_safe=True, + ) + + self.assertEqual(metrics["client_count"], 5) + self.assertGreater(metrics["slow_clients"], 0) + self.assertTrue(metrics["client_queues_bounded"]) + self.assertIn("health", metrics) + + def test_long_soak_ci_safe_report_is_redacted_and_complete(self): + with tempfile.TemporaryDirectory() as temporary_directory: + parser = build_parser() + args = parser.parse_args( + [ + "--profile", + "light_desktop", + "--duration-sec", + "1", + "--events-per-sec", + "50", + "--flows", + "5", + "--websocket-clients", + "2", + "--sample-interval-sec", + "0.05", + "--ci-safe", + "--output", + temporary_directory, + "--pcap", + str(Path(temporary_directory) / "authorized-test.pcap"), + "--pcap-loop", + ] + ) + results = run_long_soak(args) + csv_path = write_timeseries_csv( + results["resource_samples"], temporary_directory + ) + combined = json.dumps(results).lower() + + self.assertTrue(results["safe_synthetic_only"]) + self.assertEqual(results["load_profile"]["name"], "light_desktop") + self.assertIn("resource_profile", results) + self.assertIn("tuning_recommendations", results) + self.assertIn("pcap_replay", results) + self.assertTrue(results["pcap_replay"]["enabled"]) + self.assertFalse(results["validation"]["external_network_used"]) + self.assertFalse(results["validation"]["live_capture_used"]) + self.assertTrue(Path(csv_path).name.endswith(".csv")) + for secret in ["authorization", "cookie", "password", "bearer "]: + self.assertNotIn(secret, combined) + + def test_profile_result_flags_threshold_failures(self): + result = { + "resource_profile": { + "memory_growth_mb": 200.0, + "memory_stabilized": False, + "cpu_avg_percent": 95.0, + "cpu_peak_percent": 101.0, + }, + "general": {"events_failed_total": 1}, + "ops_health": {"pressure_reasons": []}, + } + + validation = classify_profile_result( + result, + max_memory_growth_mb=150.0, + max_cpu_avg_percent=90.0, + max_cpu_peak_percent=100.0, + fail_on_unbounded_growth=True, + ) + + self.assertFalse(validation["passed"]) + self.assertIn("memory_growth_threshold_exceeded", validation["failures"]) + self.assertIn("cpu_average_threshold_exceeded", validation["failures"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_release_readiness.py b/tests/test_release_readiness.py index 2819432..a3fd408 100644 --- a/tests/test_release_readiness.py +++ b/tests/test_release_readiness.py @@ -210,9 +210,18 @@ def test_performance_pipeline_docs_are_linked_and_scoped(self): self.assertIn("Benchmark / Soak Tests", performance) self.assertIn("Performance Benchmark Suite", readme) self.assertIn("docs/PERFORMANCE_VALIDATION.md", readme) + self.assertIn("docs/REAL_LOAD_TESTING.md", readme) + self.assertTrue((REPO_ROOT / "docs/REAL_LOAD_TESTING.md").is_file()) self.assertTrue((REPO_ROOT / "benchmarks/README.md").is_file()) validation = self._read("docs/PERFORMANCE_VALIDATION.md").lower() self.assertIn("safe synthetic benchmark", validation) + real_load = self._read("docs/REAL_LOAD_TESTING.md") + real_load_lower = real_load.lower() + self.assertIn("long_soak_runner.py", real_load) + self.assertIn("light_desktop", real_load) + self.assertIn("server_medium", real_load) + self.assertIn("no unbounded growth", real_load_lower) + self.assertIn("does not export raw payloads", real_load_lower) self.assertIn("not a production capacity claim", validation) self.assertIn("live capture: disabled", validation) self.assertIn("Queue pressure metrics", readme)