#!/usr/bin/env python3 """ Simulates N virtual TMS terminals heartbeating on tms/status/{serial} every ~5s, a configurable fraction of which start with an outdated emv_config/ application version so they should get exactly one auto-push each. Measures connection stability and gap-report -> push-received latency to verify the Phase 1 MQTT-delay fix (terminal-push-delay.md) actually works under load. See README.md for the full runbook and safety notes before running the full 10k pass. Always run --smoke first. Usage: ./simulator.py --smoke # 50 devices, 2 min, sanity check ./simulator.py # full run using config.py/env settings ./simulator.py --devices 2000 --duration 600 """ import argparse import asyncio import json import logging import random import signal import statistics import time from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path import gmqtt from gmqtt.mqtt.constants import MQTTv311 from config import SETTINGS log = logging.getLogger("loadtest") def now_iso(): return datetime.now(timezone.utc).isoformat() # --------------------------------------------------------------------------- # Per-device state # --------------------------------------------------------------------------- @dataclass class Device: serial: str is_gap: bool emv_version: str app_version: str # config_type -> monotonic() when this device first reported that type # as a gap (nil for compliant devices, set lazily for gap devices). gap_since: dict = field(default_factory=dict) # config_type -> list of monotonic() timestamps a push for it arrived push_received_at: dict = field(default_factory=lambda: defaultdict(list)) def mark_gap_reported(self, config_type): self.gap_since.setdefault(config_type, time.monotonic()) def mark_pushed(self, config_type): # Flip to compliant so subsequent heartbeats stop reporting a gap — # mirrors a real device that applied the update. if config_type == "emv_config": self.emv_version = SETTINGS.target_emv_version elif config_type == "application": self.app_version = SETTINGS.target_app_version self.push_received_at[config_type].append(time.monotonic()) def status_payload(self): items = [ {"itemkey": "application", "value": self.app_version, "timestamp": now_iso(), "message": ""}, {"itemkey": "parameter_config", "value": "1.0.0", "timestamp": now_iso(), "message": ""}, {"itemkey": "emv_config", "value": self.emv_version, "timestamp": now_iso(), "message": ""}, {"itemkey": "keys_config", "value": "1.0.0", "timestamp": now_iso(), "message": ""}, {"itemkey": "status", "value": "online", "timestamp": now_iso(), "message": ""}, {"itemkey": "battery", "value": str(random.randint(40, 100)), "timestamp": now_iso(), "message": ""}, ] if self.is_gap and self.emv_version != SETTINGS.target_emv_version: self.mark_gap_reported("emv_config") if self.is_gap and self.app_version != SETTINGS.target_app_version: self.mark_gap_reported("application") return { "oid": self.serial, "sn": self.serial, "uploadTime": now_iso(), "vendor": SETTINGS.vendor, "model": SETTINGS.model, "org.device": items, } # --------------------------------------------------------------------------- # Metrics # --------------------------------------------------------------------------- class Metrics: def __init__(self): self.connects = 0 self.connect_failures = 0 self.disconnects = 0 self.heartbeats_sent = 0 self.publish_errors = 0 self.acks_sent = 0 self.pushes_received = 0 self.duplicate_pushes = 0 self.latencies = defaultdict(list) # config_type -> [seconds] self.start_time = None def record_disconnect(self, worker_id, reason): self.disconnects += 1 log.warning("worker %s disconnected: %s", worker_id, reason) def record_push(self, device, config_type): self.pushes_received += 1 n = len(device.push_received_at[config_type]) if n > 1: self.duplicate_pushes += 1 log.warning("DUPLICATE push #%d for %s/%s", n, device.serial, config_type) since = device.gap_since.get(config_type) if since is not None: self.latencies[config_type].append(device.push_received_at[config_type][-1] - since) def summary(self, devices): never_pushed = [] for d in devices: if not d.is_gap: continue for ct in ("emv_config", "application"): if ct in d.gap_since and not d.push_received_at.get(ct): never_pushed.append(f"{d.serial}/{ct}") def pct(values, p): if not values: return None values = sorted(values) idx = min(len(values) - 1, int(len(values) * p)) return round(values[idx], 3) latency_summary = {} for ct, values in self.latencies.items(): latency_summary[ct] = { "count": len(values), "min": round(min(values), 3) if values else None, "p50": pct(values, 0.50), "p95": pct(values, 0.95), "p99": pct(values, 0.99), "max": round(max(values), 3) if values else None, } return { "connects": self.connects, "connect_failures": self.connect_failures, "disconnects": self.disconnects, "heartbeats_sent": self.heartbeats_sent, "publish_errors": self.publish_errors, "acks_sent": self.acks_sent, "pushes_received": self.pushes_received, "duplicate_pushes": self.duplicate_pushes, "gap_devices_never_pushed": len(never_pushed), "gap_devices_never_pushed_sample": never_pushed[:25], "latency_seconds_by_type": latency_summary, } # --------------------------------------------------------------------------- # Worker: one real MQTT connection carrying N virtual devices # --------------------------------------------------------------------------- class Worker: def __init__(self, worker_id, devices, metrics, device_by_serial): self.worker_id = worker_id self.devices = devices self.metrics = metrics self.device_by_serial = device_by_serial self.client = gmqtt.Client(f"loadtest-worker-{worker_id}") self.client.on_connect = self._on_connect self.client.on_message = self._on_message self.client.on_disconnect = self._on_disconnect self._stop = asyncio.Event() def _on_connect(self, client, flags, rc, properties): self.metrics.connects += 1 for d in self.devices: # Leading slash matches how AutoPushService actually publishes: # "/ota/#{product_key}/#{serial}/update". The '+' covers the # product key segment since it's effectively fixed today but # this shouldn't assume that. client.subscribe(f"/ota/+/{d.serial}/update", qos=1) def _on_disconnect(self, client, packet, exc=None): self.metrics.record_disconnect(self.worker_id, exc or packet) def _on_message(self, client, topic, payload, qos, properties): parts = topic.strip("/").split("/") # ['ota', product_key, serial, 'update'] if len(parts) != 4 or parts[0] != "ota" or parts[3] != "update": return serial = parts[2] device = self.device_by_serial.get(serial) if device is None: return try: command = json.loads(payload) except json.JSONDecodeError: log.error("bad push payload for %s: %r", serial, payload[:200]) return # MQTTCommandBuilder's actual wire format uses "command" (not # "command_type") and "requestId" (not "request_id") — confirmed # against the real payload logged by AutoPushService: # %{"command" => "UPDATE_L3_CONFIG", "requestId" => ..., ...} config_type = { "UPDATE_L3_CONFIG": "emv_config", "UPDATE_APPLICATION": "application", "UPDATE_PARAMS": "parameter", "LOAD_KEYS": "keys_config", }.get(command.get("command")) if config_type in ("emv_config", "application"): device.mark_pushed(config_type) self.metrics.record_push(device, config_type) request_id = command.get("requestId") if request_id is not None: ack_topic = f"ota/ack/{serial}" client.publish(ack_topic, json.dumps({"request_id": request_id, "status": "OK"}), qos=1) self.metrics.acks_sent += 1 async def connect(self): try: # Real devices (Tortoise) speak MQTT 3.1.1 — pin to it rather # than gmqtt's v5 default, so the simulator negotiates the same # protocol the fix was built/tested against. await self.client.connect(SETTINGS.mqtt_host, SETTINGS.mqtt_port, version=MQTTv311, keepalive=60) except Exception as e: # noqa: BLE001 — this is a load-test harness, log and move on self.metrics.connect_failures += 1 log.error("worker %s failed to connect: %s", self.worker_id, e) raise async def heartbeat_loop(self, duration_s): deadline = time.monotonic() + duration_s tasks = [asyncio.create_task(self._device_loop(d, deadline)) for d in self.devices] await asyncio.gather(*tasks) async def _device_loop(self, device, deadline): # Stagger this device's tick within the interval so a worker's 50 # devices don't all publish in the same instant. await asyncio.sleep(random.uniform(0, SETTINGS.heartbeat_interval_s)) while time.monotonic() < deadline: try: self.client.publish(f"tms/status/{device.serial}", json.dumps(device.status_payload()), qos=1) self.metrics.heartbeats_sent += 1 except Exception as e: # noqa: BLE001 self.metrics.publish_errors += 1 log.error("publish failed for %s: %s", device.serial, e) jitter = random.uniform(-SETTINGS.heartbeat_jitter_s, SETTINGS.heartbeat_jitter_s) await asyncio.sleep(max(0.1, SETTINGS.heartbeat_interval_s + jitter)) async def disconnect(self): try: await self.client.disconnect() except Exception: # noqa: BLE001 pass # --------------------------------------------------------------------------- # Orchestration # --------------------------------------------------------------------------- def build_devices(device_count, gap_fraction): devices = [] gap_every = max(1, round(1 / gap_fraction)) if gap_fraction > 0 else 0 for i in range(1, device_count + 1): is_gap = gap_fraction > 0 and (i % gap_every == 0) devices.append( Device( serial=SETTINGS.serial(i), is_gap=is_gap, emv_version=SETTINGS.stale_emv_version if is_gap else SETTINGS.target_emv_version, app_version=SETTINGS.stale_app_version if is_gap else SETTINGS.target_app_version, ) ) return devices async def ramp_connect(workers, rate_per_s): for i, w in enumerate(workers): try: await w.connect() except Exception: continue # already logged; keep ramping the rest if (i + 1) % rate_per_s == 0: await asyncio.sleep(1.0) async def run(device_count, duration_s, devices_per_connection): devices = build_devices(device_count, SETTINGS.gap_fraction) device_by_serial = {d.serial: d for d in devices} gap_count = sum(1 for d in devices if d.is_gap) metrics = Metrics() metrics.start_time = time.time() chunks = [devices[i:i + devices_per_connection] for i in range(0, len(devices), devices_per_connection)] workers = [Worker(i, chunk, metrics, device_by_serial) for i, chunk in enumerate(chunks)] log.info( "Starting load test: %d virtual devices (%d gap devices), %d real connections " "(%d devices/connection), duration=%ds, broker=%s:%d", device_count, gap_count, len(workers), devices_per_connection, duration_s, SETTINGS.mqtt_host, SETTINGS.mqtt_port, ) ramp_start = time.monotonic() await ramp_connect(workers, SETTINGS.connect_rate_per_s) log.info("Connected %d/%d workers in %.1fs", metrics.connects, len(workers), time.monotonic() - ramp_start) progress_task = asyncio.create_task(_progress_reporter(metrics, duration_s)) try: await asyncio.gather(*(w.heartbeat_loop(duration_s) for w in workers)) finally: progress_task.cancel() await asyncio.gather(*(w.disconnect() for w in workers), return_exceptions=True) return metrics, devices async def _progress_reporter(metrics, duration_s): start = time.monotonic() while True: await asyncio.sleep(30) elapsed = time.monotonic() - start log.info( "[%5.0fs/%ds] heartbeats=%d pushes=%d acks=%d disconnects=%d connect_failures=%d publish_errors=%d", elapsed, duration_s, metrics.heartbeats_sent, metrics.pushes_received, metrics.acks_sent, metrics.disconnects, metrics.connect_failures, metrics.publish_errors, ) def write_report(metrics, devices): results_dir = Path(SETTINGS.results_dir) results_dir.mkdir(parents=True, exist_ok=True) stamp = datetime.now().strftime("%Y%m%d-%H%M%S") report = metrics.summary(devices) report["device_count"] = len(devices) report["gap_device_count"] = sum(1 for d in devices if d.is_gap) report["mqtt_host"] = SETTINGS.mqtt_host report["mqtt_port"] = SETTINGS.mqtt_port out = results_dir / f"report-{stamp}.json" out.write_text(json.dumps(report, indent=2)) log.info("=" * 70) log.info("REPORT: %s", out) log.info(json.dumps(report, indent=2)) log.info("=" * 70) return out def parse_args(): p = argparse.ArgumentParser() p.add_argument("--devices", type=int, default=SETTINGS.device_count) p.add_argument("--duration", type=int, default=SETTINGS.duration_s) p.add_argument("--devices-per-connection", type=int, default=SETTINGS.devices_per_connection) p.add_argument("--smoke", action="store_true", help="50 devices, 2 minutes — run this first") p.add_argument("-v", "--verbose", action="store_true") return p.parse_args() def main(): args = parse_args() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) device_count = 50 if args.smoke else args.devices duration_s = 120 if args.smoke else args.duration devices_per_connection = 5 if args.smoke else args.devices_per_connection loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) metrics_holder = {} def _sigint(*_): log.warning("interrupted — stopping and writing partial report") for t in asyncio.all_tasks(loop): t.cancel() loop.add_signal_handler(signal.SIGINT, _sigint) try: metrics, devices = loop.run_until_complete(run(device_count, duration_s, devices_per_connection)) metrics_holder["metrics"] = metrics metrics_holder["devices"] = devices except asyncio.CancelledError: pass finally: if "metrics" in metrics_holder: write_report(metrics_holder["metrics"], metrics_holder["devices"]) loop.close() if __name__ == "__main__": main()