From 0d5b9730e3f92fc6ae7ff086138464e5546c8d8f Mon Sep 17 00:00:00 2001 From: "benjamin.baker" Date: Mon, 7 Sep 2026 12:39:53 -0400 Subject: [PATCH] Add load test harness for pool and fleet sizing Adds the scripts used to size the connection pool and validate fleet polling under load: burst arrival and mixed workload drivers, pool and resource samplers, a UI probe, host seed/drop SQL, and a gevent Dockerfile. Hosts are addressed by compose service name (api:5001) rather than a local path, so the harness runs against any stack rather than one machine. loadtest_token.py mints a scoped collector token named LOADTEST and prints the secret once, then revokes only that token by name so a real collector token is never touched. envkeyfallback is false, so the harness needs a genuine managed token rather than an env key. --- scripts/loadtest/Dockerfile.gevent | 9 ++ scripts/loadtest/burst_arrival.py | 219 +++++++++++++++++++++++++++ scripts/loadtest/force_pool.py | 22 +++ scripts/loadtest/loadtest_token.py | 62 ++++++++ scripts/loadtest/mixed_workload.py | 218 ++++++++++++++++++++++++++ scripts/loadtest/poll_fleet.py | 141 +++++++++++++++++ scripts/loadtest/pool_status.py | 42 +++++ scripts/loadtest/sample_dbconns.py | 68 +++++++++ scripts/loadtest/sample_resources.py | 80 ++++++++++ scripts/loadtest/ui_probe.py | 78 ++++++++++ 10 files changed, 939 insertions(+) create mode 100644 scripts/loadtest/Dockerfile.gevent create mode 100644 scripts/loadtest/burst_arrival.py create mode 100644 scripts/loadtest/force_pool.py create mode 100644 scripts/loadtest/loadtest_token.py create mode 100644 scripts/loadtest/mixed_workload.py create mode 100644 scripts/loadtest/poll_fleet.py create mode 100644 scripts/loadtest/pool_status.py create mode 100644 scripts/loadtest/sample_dbconns.py create mode 100644 scripts/loadtest/sample_resources.py create mode 100644 scripts/loadtest/ui_probe.py diff --git a/scripts/loadtest/Dockerfile.gevent b/scripts/loadtest/Dockerfile.gevent new file mode 100644 index 00000000..2164e21e --- /dev/null +++ b/scripts/loadtest/Dockerfile.gevent @@ -0,0 +1,9 @@ +# Sizing-simulation overlay. NOT part of the shipped image. +# +# Layers gevent onto the built production image so the gevent worker model can +# be measured against the gthread baseline under identical application code. +# Delete with the rest of scripts/loadtest once sizing questions are settled. +FROM shopdb-flask-wilm-api:latest + +USER root +RUN pip install --no-cache-dir "gevent==25.9.1" diff --git a/scripts/loadtest/burst_arrival.py b/scripts/loadtest/burst_arrival.py new file mode 100644 index 00000000..636e804a --- /dev/null +++ b/scripts/loadtest/burst_arrival.py @@ -0,0 +1,219 @@ +"""Patch-day burst simulator: synchronized cold start of an endpoint fleet. + +WHY this is separate from mixed_workload.py: that harness RAMPS agents in over +~20s and measures steady state, which deliberately avoids a thundering herd. +Patch day is the opposite case and stresses different machinery: + + - a fresh TLS handshake per agent, all at once (handshakes are CPU-heavy) + - _resolve_asset_for_hostname runs on every arrival, so the DB pool is hit by + a synchronized burst rather than the smooth churn of steady state + - no warm connections in the pool, no warm caches + +Modelled on the real agent (verified in agent/ShopDbEndpointAgent.Core/): + - CommandWorker.ExecuteAsync polls IMMEDIATELY, no startup delay, no jitter + - TelemetryWorker also reports immediately; its jitter applies only BETWEEN + runs, so first contact is synchronized too. Patch day is therefore TWO + simultaneous bursts per endpoint, not one. + - transient failures retry at a FIXED 10s (TransientRetryDelay), not + exponential-with-jitter, so a herd that fails together re-arrives together + +--arrivalseconds is the spread of the reboot window. 0 means every agent hits at +once (single maintenance window, worst case); 60 approximates a staged ring. + +Stdlib asyncio only, so it needs no dependency install in the container. + +Usage: + python burst_arrival.py --fleet 700 --arrivalseconds 0 --seconds 120 +""" +import argparse +import asyncio +import json +import os +import random +import ssl +import statistics +import time + +STOP = False + + +def percentile(values, fraction): + if not values: + return 0.0 + ordered = sorted(values) + index = min(int(len(ordered) * fraction), len(ordered) - 1) + return ordered[index] + + +async def request(host, port, context, method, path, timeout, + api_key=None, body=None): + """One HTTP/1.1 request over a FRESH connection, so each call pays a full + TLS handshake - which is the point on patch day.""" + started = time.monotonic() + writer = None + try: + reader, writer = await asyncio.wait_for( + asyncio.open_connection(host, port, ssl=context), timeout=timeout) + handshake = time.monotonic() - started + headers = [ + '%s %s HTTP/1.1' % (method, path), + 'Host: %s' % host, + 'Accept: application/json', + 'Connection: close', + ] + if api_key: + headers.append('X-API-Key: %s' % api_key) + payload = b'' + if body is not None: + payload = json.dumps(body).encode() + headers.append('Content-Type: application/json') + headers.append('Content-Length: %d' % len(payload)) + writer.write(('\r\n'.join(headers) + '\r\n\r\n').encode() + payload) + await writer.drain() + header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout) + status = int(header.split()[1]) + await asyncio.wait_for(reader.read(), timeout=timeout) + return status, time.monotonic() - started, handshake + except Exception as error: + return type(error).__name__, time.monotonic() - started, None + finally: + if writer is not None: + try: + writer.close() + except Exception: + pass + + +def record(results, bucket, status, elapsed, handshake=None, first=False): + entry = results.setdefault(bucket, { + 'times': [], 'status': {}, 'handshakes': [], 'firsttimes': []}) + entry['times'].append(elapsed * 1000.0) + if handshake is not None: + entry['handshakes'].append(handshake * 1000.0) + if first: + entry['firsttimes'].append(elapsed * 1000.0) + key = str(status) + entry['status'][key] = entry['status'].get(key, 0) + 1 + + +async def agent(index, host, port, context, api_key, results, arrivalseconds): + """One endpoint coming back from a patch reboot. + + Fires its telemetry report and its first command poll at the same instant, + mirroring the two workers both starting with no delay. + """ + if arrivalseconds > 0: + await asyncio.sleep(random.uniform(0, arrivalseconds)) + + hostname = 'LOADTEST%04d' % (index + 1) + + async def telemetry(): + status, elapsed, handshake = await request( + host, port, context, 'POST', '/api/collector/heartbeat', + timeout=120.0, api_key=api_key, body={'hostname': hostname}) + record(results, 'telemetry_firstcontact', status, elapsed, handshake, + first=True) + + telemetry_task = asyncio.ensure_future(telemetry()) + + path = '/api/endpoints/commands/next?hostname=%s' % hostname + first = True + while not STOP: + status, elapsed, handshake = await request( + host, port, context, 'GET', path, timeout=120.0, api_key=api_key) + record(results, 'fleet', status, elapsed, handshake, first=first) + if first: + first = False + if not isinstance(status, int): + # Mirrors CommandWorker TransientRetryDelay: fixed 10s, no jitter, + # so a herd that fails together re-arrives together. + await asyncio.sleep(10.0) + else: + await asyncio.sleep(0.05) + + await telemetry_task + + +async def ui_user(host, port, context, results, think_seconds): + while not STOP: + status, elapsed, handshake = await request( + host, port, context, 'GET', '/api/dashboard/health', timeout=120.0) + record(results, 'ui', status, elapsed, handshake) + await asyncio.sleep(think_seconds) + + +async def main(): + global STOP + parser = argparse.ArgumentParser() + parser.add_argument('--fleet', type=int, default=700) + parser.add_argument('--users', type=int, default=5) + parser.add_argument('--arrivalseconds', type=float, default=0.0) + parser.add_argument('--seconds', type=int, default=120) + parser.add_argument('--host', default='api') + parser.add_argument('--port', type=int, default=5001) + parser.add_argument('--think', type=float, default=3.0) + arguments = parser.parse_args() + + api_key = os.environ.get('LOADTEST_TOKEN', '') + if not api_key: + raise SystemExit('LOADTEST_TOKEN not set') + + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + + results = {} + tasks = [] + for _ in range(arguments.users): + tasks.append(asyncio.ensure_future(ui_user( + arguments.host, arguments.port, context, results, + arguments.think))) + # Let the UI baseline establish before the storm hits. + await asyncio.sleep(5.0) + + burst_started = time.monotonic() + for index in range(arguments.fleet): + tasks.append(asyncio.ensure_future(agent( + index, arguments.host, arguments.port, context, api_key, results, + arguments.arrivalseconds))) + + await asyncio.sleep(arguments.seconds) + STOP = True + await asyncio.sleep(2.0) + for task in tasks: + task.cancel() + + summary = { + 'fleet': arguments.fleet, + 'arrivalseconds': arguments.arrivalseconds, + 'users': arguments.users, + 'seconds': arguments.seconds, + 'classes': {}, + } + for bucket, entry in sorted(results.items()): + times = entry['times'] + block = { + 'samples': len(times), + 'status': entry['status'], + 'median_ms': round(statistics.median(times), 1) if times else 0.0, + 'p95_ms': round(percentile(times, 0.95), 1), + 'max_ms': round(max(times), 1) if times else 0.0, + } + if entry['handshakes']: + block['handshake_median_ms'] = round( + statistics.median(entry['handshakes']), 1) + block['handshake_p95_ms'] = round( + percentile(entry['handshakes'], 0.95), 1) + block['handshake_max_ms'] = round(max(entry['handshakes']), 1) + if entry['firsttimes']: + block['firstcontact_median_ms'] = round( + statistics.median(entry['firsttimes']), 1) + block['firstcontact_p95_ms'] = round( + percentile(entry['firsttimes'], 0.95), 1) + block['firstcontact_max_ms'] = round(max(entry['firsttimes']), 1) + summary['classes'][bucket] = block + print(json.dumps(summary, indent=2)) + + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/scripts/loadtest/force_pool.py b/scripts/loadtest/force_pool.py new file mode 100644 index 00000000..e4af3906 --- /dev/null +++ b/scripts/loadtest/force_pool.py @@ -0,0 +1,22 @@ +"""Force SQLAlchemy pool sizing for a pool-comparison run. + +WHY this exists: shopdb/config.py does not read pool size from the environment, +so a test container cannot be pointed at a different pool without a code change. +This patches Config.SQLALCHEMY_ENGINE_OPTIONS at import time, before create_app +builds the engine, purely so a bigger pool can be MEASURED before deciding +whether to make it permanent. + +Not for production use. Mounted into a throwaway container as sitecustomize.py, +which Python imports automatically at startup. +""" +import os + +from shopdb.config import Config + +_size = int(os.environ.get('FORCE_POOL_SIZE', '5')) +_overflow = int(os.environ.get('FORCE_MAX_OVERFLOW', '10')) + +_options = dict(Config.SQLALCHEMY_ENGINE_OPTIONS) +_options['pool_size'] = _size +_options['max_overflow'] = _overflow +Config.SQLALCHEMY_ENGINE_OPTIONS = _options diff --git a/scripts/loadtest/loadtest_token.py b/scripts/loadtest/loadtest_token.py new file mode 100644 index 00000000..1b7e1b3b --- /dev/null +++ b/scripts/loadtest/loadtest_token.py @@ -0,0 +1,62 @@ +"""Mint a temporary collector token for the load test, or revoke it after. + +WHY a minted token: plugin.endpoints.envkeyfallback is false, which is the +agent kill switch working as intended - the env key no longer authorizes. The +harness therefore needs a real managed token scoped collector.ingest.endpoints, +exactly like a live agent. + +The token is named LOADTEST so revoke() removes only this one and never touches +the real agent keys. + +Usage (inside the api container, with PYTHONPATH=/app and cwd /app): + python loadtest_token.py mint # prints the secret once + python loadtest_token.py revoke +""" +import sys + +from shopdb import create_app, db +from shopdb.core.models.apitoken import ApiToken + +TOKEN_NAME = 'LOADTEST temporary collector token' +SCOPE = 'collector.ingest.endpoints' + + +def mint(): + existing = ApiToken.query.filter_by(name=TOKEN_NAME).first() + if existing is not None: + db.session.delete(existing) + db.session.commit() + owner = ApiToken.query.filter( + ApiToken.scopes.like('%' + SCOPE + '%')).first() + if owner is None: + raise SystemExit('no existing collector token to copy ownership from') + secret = ApiToken.generate_secret() + token = ApiToken( + userid=owner.userid, + name=TOKEN_NAME, + tokenprefix=ApiToken.prefix_of(secret), + tokenhash=ApiToken.hash_secret(secret), + ) + token.scopelist = [SCOPE] + db.session.add(token) + db.session.commit() + print(secret) + + +def revoke(): + removed = ApiToken.query.filter_by(name=TOKEN_NAME).delete() + db.session.commit() + print('revoked %d loadtest token(s)' % removed) + + +if __name__ == '__main__': + action = sys.argv[1] if len(sys.argv) > 1 else '' + app = create_app() + with app.app_context(): + if action == 'mint': + mint() + elif action == 'revoke': + revoke() + else: + print(__doc__) + sys.exit(2) diff --git a/scripts/loadtest/mixed_workload.py b/scripts/loadtest/mixed_workload.py new file mode 100644 index 00000000..30349965 --- /dev/null +++ b/scripts/loadtest/mixed_workload.py @@ -0,0 +1,218 @@ +"""Mixed-workload generator for sizing simulations. + +WHY this shape: sizing cannot be judged from long-poll agents alone. The four +traffic classes have different concurrency costs, and only their SUM decides +whether a worker model holds: + + fleet long-poll agents, one continuously-held connection each + ui human users clicking, latency-sensitive, the actual symptom + integration Maximo / ServiceNow style collector clients on a fixed interval + kiosk USB kiosk endpoints, public routes, low rate + +The fleet class is what saturates a thread pool; the ui class is what a person +notices. Measuring them together is the point - a model that keeps agents happy +while starving the UI has failed even though every agent request succeeded. + +Runs INSIDE a container on the compose network. Generating from the Windows host +would put host ephemeral-port and TLS-handshake limits in the measurement path, +which would look like server saturation but would not be. + +Stdlib asyncio only, so it needs no dependency install in the container. + +Usage: + python mixed_workload.py --fleet 300 --users 5 --seconds 120 +""" +import argparse +import asyncio +import json +import os +import ssl +import statistics +import time + +STOP = False + + +def percentile(values, fraction): + if not values: + return 0.0 + ordered = sorted(values) + index = min(int(len(ordered) * fraction), len(ordered) - 1) + return ordered[index] + + +async def request(host, port, context, method, path, timeout, + api_key=None, bearer=None, body=None): + """One HTTP/1.1 request over a fresh connection. Returns (status, seconds).""" + started = time.monotonic() + writer = None + try: + reader, writer = await asyncio.wait_for( + asyncio.open_connection(host, port, ssl=context), timeout=timeout) + headers = [ + '%s %s HTTP/1.1' % (method, path), + 'Host: %s' % host, + 'Accept: application/json', + 'Connection: close', + ] + if api_key: + headers.append('X-API-Key: %s' % api_key) + if bearer: + headers.append('Authorization: Bearer %s' % bearer) + payload = b'' + if body is not None: + payload = json.dumps(body).encode() + headers.append('Content-Type: application/json') + headers.append('Content-Length: %d' % len(payload)) + blob = ('\r\n'.join(headers) + '\r\n\r\n').encode() + payload + writer.write(blob) + await writer.drain() + header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout) + status = int(header.split()[1]) + await asyncio.wait_for(reader.read(), timeout=timeout) + return status, time.monotonic() - started + except Exception as error: + return type(error).__name__, time.monotonic() - started + finally: + if writer is not None: + try: + writer.close() + except Exception: + pass + + +def record(results, bucket, status, elapsed): + entry = results.setdefault(bucket, {'times': [], 'status': {}}) + entry['times'].append(elapsed * 1000.0) + key = str(status) + entry['status'][key] = entry['status'].get(key, 0) + 1 + + +async def fleet_agent(index, host, port, context, api_key, results): + """Mirrors CommandWorker.cs: re-poll immediately after each response.""" + hostname = 'LOADTEST%04d' % (index + 1) + path = '/api/endpoints/commands/next?hostname=%s' % hostname + while not STOP: + status, elapsed = await request( + host, port, context, 'GET', path, timeout=120.0, api_key=api_key) + record(results, 'fleet', status, elapsed) + await asyncio.sleep(0.05) + + +async def ui_user(host, port, context, bearer, results, think_seconds): + """A human clicking around. Latency here is the user-visible symptom.""" + paths = ['/api/dashboard/health', '/api/endpoints?page=1&per_page=25', + '/api/assets?page=1&per_page=25'] + cursor = 0 + while not STOP: + path = paths[cursor % len(paths)] + cursor += 1 + # Unauthenticated probe still measures queueing; auth only changes the + # handler cost, not whether a worker slot was available to accept it. + status, elapsed = await request( + host, port, context, 'GET', path, timeout=120.0, bearer=bearer) + record(results, 'ui', status, elapsed) + await asyncio.sleep(think_seconds) + + +async def integration_client(name, host, port, context, api_key, results, + interval): + """Maximo / ServiceNow style poller on a fixed interval.""" + while not STOP: + status, elapsed = await request( + host, port, context, 'GET', '/api/collector/status', + timeout=120.0, api_key=api_key) + record(results, 'integration', status, elapsed) + await asyncio.sleep(interval) + + +async def kiosk_endpoint(index, host, port, context, results, interval): + """USB kiosk terminal. Public route, low rate.""" + path = '/api/usb/kiosk/device/LOADTESTKIOSK%d' % (index + 1) + while not STOP: + status, elapsed = await request( + host, port, context, 'GET', path, timeout=120.0) + record(results, 'kiosk', status, elapsed) + await asyncio.sleep(interval) + + +async def main(): + global STOP + parser = argparse.ArgumentParser() + parser.add_argument('--fleet', type=int, default=300) + parser.add_argument('--users', type=int, default=5) + parser.add_argument('--integrations', type=int, default=2) + parser.add_argument('--kiosks', type=int, default=8) + parser.add_argument('--seconds', type=int, default=120) + parser.add_argument('--host', default='api') + parser.add_argument('--port', type=int, default=5001) + parser.add_argument('--think', type=float, default=3.0) + parser.add_argument('--integration-interval', type=float, default=30.0) + parser.add_argument('--kiosk-interval', type=float, default=15.0) + parser.add_argument('--rampseconds', type=float, default=20.0) + arguments = parser.parse_args() + + api_key = os.environ.get('LOADTEST_TOKEN', '') + if not api_key: + raise SystemExit('LOADTEST_TOKEN not set') + bearer = os.environ.get('LOADTEST_JWT') or None + + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + + results = {} + tasks = [] + + # Ramp the fleet in rather than opening every connection at once. A + # thundering herd measures handshake burst capacity, not steady state. + stagger = arguments.rampseconds / max(arguments.fleet, 1) + + async def staggered(index): + await asyncio.sleep(index * stagger) + await fleet_agent(index, arguments.host, arguments.port, context, + api_key, results) + + for index in range(arguments.fleet): + tasks.append(asyncio.ensure_future(staggered(index))) + for _ in range(arguments.users): + tasks.append(asyncio.ensure_future(ui_user( + arguments.host, arguments.port, context, bearer, results, + arguments.think))) + for index in range(arguments.integrations): + tasks.append(asyncio.ensure_future(integration_client( + 'integration%d' % index, arguments.host, arguments.port, context, + api_key, results, arguments.integration_interval))) + for index in range(arguments.kiosks): + tasks.append(asyncio.ensure_future(kiosk_endpoint( + index, arguments.host, arguments.port, context, results, + arguments.kiosk_interval))) + + await asyncio.sleep(arguments.seconds) + STOP = True + await asyncio.sleep(2.0) + for task in tasks: + task.cancel() + + summary = { + 'fleet': arguments.fleet, + 'users': arguments.users, + 'integrations': arguments.integrations, + 'kiosks': arguments.kiosks, + 'seconds': arguments.seconds, + 'classes': {}, + } + for bucket, entry in sorted(results.items()): + times = entry['times'] + summary['classes'][bucket] = { + 'samples': len(times), + 'status': entry['status'], + 'median_ms': round(statistics.median(times), 1) if times else 0.0, + 'p95_ms': round(percentile(times, 0.95), 1), + 'max_ms': round(max(times), 1) if times else 0.0, + } + print(json.dumps(summary, indent=2)) + + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/scripts/loadtest/poll_fleet.py b/scripts/loadtest/poll_fleet.py new file mode 100644 index 00000000..ea656ffb --- /dev/null +++ b/scripts/loadtest/poll_fleet.py @@ -0,0 +1,141 @@ +"""Async load generator for the endpoint command long-poll channel. + +WHY this shape: the real agent (agent/ShopDbEndpointAgent.Core/CommandWorker.cs) +re-polls immediately on a 204, so every agent holds a connection essentially +continuously and fleet size IS the concurrency. This mirrors that loop rather +than issuing independent requests, so the measured concurrency matches a real +fleet of the same size. + +Runs INSIDE a container on the compose network. Generating from the Windows host +would put host ephemeral-port and TLS-handshake limits in the measurement path, +which would look like server saturation but would not be. + +While the fleet polls, a separate probe measures /api/dashboard/health - the +user-visible UI latency - because that, not agent latency, is the symptom that +matters for sizing. + +Stdlib asyncio only, so it needs no dependency install in the container. + +Usage: + python poll_fleet.py --agents 300 --seconds 90 --base https://api:5001 +""" +import argparse +import asyncio +import json +import os +import ssl +import statistics +import time + +STOP = False + + +def percentile(values, fraction): + if not values: + return 0.0 + ordered = sorted(values) + index = min(int(len(ordered) * fraction), len(ordered) - 1) + return ordered[index] + + +async def request(host, port, context, path, api_key, timeout): + """One HTTP/1.1 GET over a fresh connection. Returns (status, seconds).""" + started = time.monotonic() + writer = None + try: + reader, writer = await asyncio.wait_for( + asyncio.open_connection(host, port, ssl=context), timeout=timeout) + request_line = ( + 'GET %s HTTP/1.1\r\nHost: %s\r\nX-API-Key: %s\r\n' + 'Accept: application/json\r\nConnection: close\r\n\r\n' + % (path, host, api_key)) + writer.write(request_line.encode()) + await writer.drain() + header = await asyncio.wait_for(reader.readuntil(b'\r\n'), timeout=timeout) + status = int(header.split()[1]) + await asyncio.wait_for(reader.read(), timeout=timeout) + return status, time.monotonic() - started + except Exception as error: + return type(error).__name__, time.monotonic() - started + finally: + if writer is not None: + try: + writer.close() + except Exception: + pass + + +async def agent_loop(index, host, port, context, api_key, results, timeout): + """Imitate one agent: poll, and on completion immediately poll again.""" + hostname = 'LOADTEST%04d' % (index + 1) + path = '/api/endpoints/commands/next?hostname=' + hostname + while not STOP: + status, elapsed = await request( + host, port, context, path, api_key, timeout) + results.append((status, elapsed)) + await asyncio.sleep(0) + + +async def main(): + global STOP + parser = argparse.ArgumentParser() + parser.add_argument('--agents', type=int, required=True) + parser.add_argument('--seconds', type=int, default=90) + parser.add_argument('--host', default='api') + parser.add_argument('--port', type=int, default=5001) + parser.add_argument('--timeout', type=float, default=120.0) + parser.add_argument('--ramp', type=float, default=0.005) + arguments = parser.parse_args() + + # A managed collector token, not the env key: the endpoints plugin has + # plugin.endpoints.envkeyfallback=false (the agent kill switch), so the env + # key no longer authorizes. See scripts/loadtest/loadtest_token.py. + api_key = os.environ.get('LOADTEST_TOKEN') + if not api_key: + raise SystemExit('LOADTEST_TOKEN not present in environment') + + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + + results = [] + tasks = [] + started = time.monotonic() + + for index in range(arguments.agents): + tasks.append(asyncio.create_task(agent_loop( + index, arguments.host, arguments.port, context, api_key, + results, arguments.timeout))) + if arguments.ramp: + await asyncio.sleep(arguments.ramp) + + ramp_seconds = time.monotonic() - started + + await asyncio.sleep(arguments.seconds) + STOP = True + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + poll_ok = [d for s, d in results if s == 204] + poll_bad = [s for s, _ in results if s != 204] + + failures = {} + for status in poll_bad: + failures[str(status)] = failures.get(str(status), 0) + 1 + + print(json.dumps({ + 'agents': arguments.agents, + 'ramp_seconds': round(ramp_seconds, 1), + 'duration_seconds': arguments.seconds, + 'poll_completed': len(results), + 'poll_ok_204': len(poll_ok), + 'poll_failures': failures, + 'poll_hold_median': round(statistics.median(poll_ok), 2) if poll_ok else None, + 'poll_hold_p95': round(percentile(poll_ok, 0.95), 2) if poll_ok else None, + 'poll_hold_max': round(max(poll_ok), 2) if poll_ok else None, + }, indent=2)) + + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/scripts/loadtest/pool_status.py b/scripts/loadtest/pool_status.py new file mode 100644 index 00000000..1d42fb3f --- /dev/null +++ b/scripts/loadtest/pool_status.py @@ -0,0 +1,42 @@ +"""Report SQLAlchemy pool saturation from inside the API container. + +WHY: MySQL's max_connections is NOT the ceiling that binds first. The effective +limit is per-process: pool_size + max_overflow, multiplied by worker count. +With the defaults (5 + 10) and 4 workers that is 60, not 151. A peak of 52 +therefore means 87 percent of the real ceiling, not 34 percent. + +Prints the computed ceiling so a measured high-water mark can be compared +against the right number. + +Usage (inside the api container, PYTHONPATH=/app): + python pool_status.py +""" +import json +import os + +from shopdb import create_app +from shopdb.api import db + + +def main(): + app = create_app(os.environ.get('FLASK_ENV', 'production')) + with app.app_context(): + engine = db.engine + pool = engine.pool + size = pool.size() + overflow = getattr(pool, '_max_overflow', 0) + workers = int(os.environ.get('GUNICORN_WORKERS', '4')) + print(json.dumps({ + 'pool_class': type(pool).__name__, + 'pool_size': size, + 'max_overflow': overflow, + 'per_worker_ceiling': size + overflow, + 'workers': workers, + 'effective_ceiling': (size + overflow) * workers, + 'checkedout': pool.checkedout(), + 'status': pool.status(), + }, indent=2)) + + +if __name__ == '__main__': + main() diff --git a/scripts/loadtest/sample_dbconns.py b/scripts/loadtest/sample_dbconns.py new file mode 100644 index 00000000..6f0516af --- /dev/null +++ b/scripts/loadtest/sample_dbconns.py @@ -0,0 +1,68 @@ +"""Sample live MySQL connection counts at high frequency. + +WHY: pool pressure from a long-poll fleet is BURSTY. The route releases its +connection before sleeping and re-acquires on wake, so the interesting number is +the peak concurrent checkout during a wake burst, not the steady-state count. A +slow sampler misses it entirely. + +Runs on the HOST (needs the docker CLI). + +Usage: + python sample_dbconns.py --seconds 120 --interval 0.25 +""" +import argparse +import json +import statistics +import subprocess +import time + +QUERY = ("SELECT COUNT(*), SUM(command<>'Sleep') " + "FROM information_schema.processlist WHERE user='shopdb';") + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--container', default='shopdb-flask-wilm-db-1') + parser.add_argument('--seconds', type=int, default=120) + parser.add_argument('--interval', type=float, default=0.25) + arguments = parser.parse_args() + + total_samples = [] + active_samples = [] + deadline = time.monotonic() + arguments.seconds + command = ('mysql -u root -p$MYSQL_ROOT_PASSWORD -N -B -e "%s"' % QUERY) + + while time.monotonic() < deadline: + try: + output = subprocess.check_output( + ['docker', 'exec', arguments.container, 'sh', '-c', command], + stderr=subprocess.DEVNULL, text=True).strip() + except subprocess.CalledProcessError: + break + if not output: + break + parts = output.split() + total_samples.append(int(parts[0])) + active_samples.append(int(parts[1]) if parts[1] != 'NULL' else 0) + time.sleep(arguments.interval) + + def summarize(values): + if not values: + return {'samples': 0} + ordered = sorted(values) + index = min(int(len(ordered) * 0.95), len(ordered) - 1) + return { + 'samples': len(values), + 'median': statistics.median(values), + 'p95': ordered[index], + 'max': max(values), + } + + print(json.dumps({ + 'connections': summarize(total_samples), + 'active_queries': summarize(active_samples), + }, indent=2)) + + +if __name__ == '__main__': + main() diff --git a/scripts/loadtest/sample_resources.py b/scripts/loadtest/sample_resources.py new file mode 100644 index 00000000..532b63bd --- /dev/null +++ b/scripts/loadtest/sample_resources.py @@ -0,0 +1,80 @@ +"""Sample container CPU and memory during a sizing run. + +WHY: the sizing table claims specific vCPU and RAM figures. Those can only be +confirmed by watching the container under load, not by reasoning about greenlet +size. This records peak RSS and CPU so a claimed "4 GB is enough" is either +demonstrated or falsified. + +Runs on the HOST (needs the docker CLI), unlike the generators. + +Usage: + python sample_resources.py --container shopdb-sizing-api --seconds 120 +""" +import argparse +import json +import re +import statistics +import subprocess +import time + + +def parse_size(text): + """Docker stats prints like '1.234GiB' / '512MiB'. Return megabytes.""" + match = re.match(r'^([0-9.]+)\s*([KMGT]?i?B)$', text.strip()) + if not match: + return 0.0 + value = float(match.group(1)) + unit = match.group(2).upper().replace('I', '') + factors = {'B': 1.0 / (1024 * 1024), 'KB': 1.0 / 1024, 'MB': 1.0, + 'GB': 1024.0, 'TB': 1024.0 * 1024.0} + return value * factors.get(unit, 1.0) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--container', required=True) + parser.add_argument('--seconds', type=int, default=120) + parser.add_argument('--interval', type=float, default=2.0) + arguments = parser.parse_args() + + cpu_samples = [] + memory_samples = [] + deadline = time.monotonic() + arguments.seconds + + while time.monotonic() < deadline: + try: + output = subprocess.check_output( + ['docker', 'stats', '--no-stream', '--format', + '{{.CPUPerc}}|{{.MemUsage}}', arguments.container], + stderr=subprocess.DEVNULL, text=True).strip() + except subprocess.CalledProcessError: + break + if not output: + break + cpu_text, memory_text = output.split('|', 1) + cpu_samples.append(float(cpu_text.replace('%', '').strip())) + used = memory_text.split('/')[0] + memory_samples.append(parse_size(used)) + time.sleep(arguments.interval) + + def summarize(values): + if not values: + return {'samples': 0} + ordered = sorted(values) + index = min(int(len(ordered) * 0.95), len(ordered) - 1) + return { + 'samples': len(values), + 'median': round(statistics.median(values), 1), + 'p95': round(ordered[index], 1), + 'max': round(max(values), 1), + } + + print(json.dumps({ + 'container': arguments.container, + 'cpu_percent': summarize(cpu_samples), + 'memory_mb': summarize(memory_samples), + }, indent=2)) + + +if __name__ == '__main__': + main() diff --git a/scripts/loadtest/ui_probe.py b/scripts/loadtest/ui_probe.py new file mode 100644 index 00000000..46e51fbf --- /dev/null +++ b/scripts/loadtest/ui_probe.py @@ -0,0 +1,78 @@ +"""Measure UI latency against the API, independent of any load generator. + +WHY separate from poll_fleet.py: when the probe shares an event loop with the +simulated fleet, generator-side scheduling can starve the probe and produce UI +latencies that are an artifact of the harness rather than of the server. Run +this in its own container, concurrently with the fleet, so the two cannot +interfere. + +Usage: + python ui_probe.py --seconds 90 --interval 1.0 +""" +import argparse +import json +import ssl +import statistics +import time +import urllib.request + + +def percentile(values, fraction): + if not values: + return 0.0 + ordered = sorted(values) + index = min(int(len(ordered) * fraction), len(ordered) - 1) + return ordered[index] + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--seconds', type=int, default=90) + parser.add_argument('--interval', type=float, default=1.0) + parser.add_argument('--url', + default='https://api:5001/api/dashboard/health') + parser.add_argument('--timeout', type=float, default=120.0) + arguments = parser.parse_args() + + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + + durations = [] + failures = {} + deadline = time.monotonic() + arguments.seconds + + while time.monotonic() < deadline: + started = time.monotonic() + try: + request = urllib.request.Request(arguments.url) + with urllib.request.urlopen( + request, timeout=arguments.timeout, + context=context) as response: + response.read() + if response.status == 200: + durations.append(time.monotonic() - started) + else: + key = str(response.status) + failures[key] = failures.get(key, 0) + 1 + except Exception as error: + key = type(error).__name__ + failures[key] = failures.get(key, 0) + 1 + remaining = arguments.interval - (time.monotonic() - started) + if remaining > 0: + time.sleep(remaining) + + print(json.dumps({ + 'ui_samples': len(durations) + sum(failures.values()), + 'ui_ok_200': len(durations), + 'ui_failures': failures, + 'ui_median_ms': round(statistics.median(durations) * 1000, 1) + if durations else None, + 'ui_p95_ms': round(percentile(durations, 0.95) * 1000, 1) + if durations else None, + 'ui_max_ms': round(max(durations) * 1000, 1) if durations else None, + }, indent=2)) + + +if __name__ == '__main__': + main()