From 0ce144d2fbb14e73e9a8ee5e4d01d9c9bf26f74e Mon Sep 17 00:00:00 2001 From: Josh Knapp Date: Fri, 9 Oct 2026 10:13:12 -0700 Subject: [PATCH] fix(api): stop silent clients from wedging the gunicorn thread pool On 2026-10-09 (sdbees) both :8000 API workers had 7 of 8 gthread pool threads blocked in tcp_recvmsg -> sk_wait_data; /health and /api/blocked-ips hung 40s+ while the container reported healthy. The app makes no in-process outbound socket reads (every ACME/HAProxy interaction is an already-bounded subprocess). The blocked reads were gunicorn gthread reading the CLIENT socket: TConn.init() calls setblocking(True), i.e. timeout None, before parsing the request. Any peer that sends a few bytes and goes quiet (a TLS ClientHello on this plain-HTTP port, scanners; :8000 is published on 0.0.0.0) pins a pool thread forever. gunicorn --timeout does not catch it (gthread only watches the main thread's heartbeat), --max-requests never triggers on a worker that finishes nothing, and socket.setdefaulttimeout() is reset by setblocking(True). - haproxy_gunicorn_worker.TimeoutThreadWorker (used for :8000 and :8080): * per-connection socket timeout HAPROXY_MGR_CLIENT_TIMEOUT (20s), re-applied on keep-alive reuse; bounds socket I/O, not handling time * watchdog: if every pool thread has been in the same request for more than HAPROXY_MGR_WEDGE_SECONDS (1200s, > certbot renew's 900s), dump all thread stacks and exit so the master respawns the worker * client timeouts logged as one INFO line, not a traceback - haproxy_manager: socket.setdefaulttimeout (HAPROXY_MGR_SOCKET_DEFAULT_TIMEOUT, 60s) as defence in depth for any future in-process client; /health now reports the answering worker's busy/total threads and oldest request age - scripts/healthcheck.sh: probes :8000/health (4s) AND :80; image HEALTHCHECK uses it (WHP's --health-cmd override must too) - scripts/test-gunicorn-socket-timeouts.py (12 tests, real gunicorn), added to the Docker build gate Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 4 +- Dockerfile | 11 +- VERSION | 2 +- haproxy_gunicorn_worker.py | 209 +++++++++++++ haproxy_manager.py | 34 ++- scripts/healthcheck.sh | 31 ++ scripts/start-up.sh | 17 +- scripts/test-gunicorn-socket-timeouts.py | 356 +++++++++++++++++++++++ 8 files changed, 654 insertions(+), 10 deletions(-) create mode 100644 haproxy_gunicorn_worker.py create mode 100755 scripts/healthcheck.sh create mode 100644 scripts/test-gunicorn-socket-timeouts.py diff --git a/CLAUDE.md b/CLAUDE.md index f4c43cf..b1e28b6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,6 +10,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co - **Stick-table contract**: `python3 scripts/test-stick-table-contract.py` - offline; holds the templates' `store` clauses, `STICK_TABLE_FIELD_CONTRACT`, and every consumer to each other. Run it after touching any `stick-table` line. - **Runtime-map contract**: `python3 scripts/test-runtime-map-contract.py` - offline; asserts the runtime map commands are `@1`-prefixed, reference the map by FILE PATH (never `#`), carry the value `1`, and that every captured rejection is classified as a failure. Run it after touching any `add map`/`del map`/`clear map` path. - **Certificate destruction safety**: `python3 scripts/test-cert-write-safety.py` - offline; asserts a live `.pem` is never truncated, removed, or its certbot lineage deleted while any configured domain still references it (one bundle serves many names, so `ssl_cert_path` is routinely shared). Run it after touching any `os.remove`/`certbot delete`/PEM-write path. +- **API thread-pool safety**: `python3 scripts/test-gunicorn-socket-timeouts.py` - runs REAL gunicorn (~30s); asserts a silent/half-written client cannot pin a gthread pool thread (per-connection socket timeout, `HAPROXY_MGR_CLIENT_TIMEOUT`), slow legit requests are not cut off, the wedge watchdog (`HAPROXY_MGR_WEDGE_SECONDS`) replaces a fully stuck worker, and an in-process call to a stalled upstream times out. Run it after touching `haproxy_gunicorn_worker.py`, start-up.sh's gunicorn lines, or upgrading gunicorn (the worker overrides `ThreadWorker.enqueue_req`). - **Manual Testing**: Run `curl` commands against `http://localhost:8000` endpoints as shown in README.md ### Reading stick tables (and why it is easy to get silently wrong) @@ -166,6 +167,7 @@ re-reads it on reload, so a failed runtime command must degrade to - Exposes ports 80 (HTTP), 443 (HTTPS), and 8000 (management API/UI) - Stats page on port 8404 (localhost only inside container) - Management interface on port 8000 should be firewall-protected in production -- Dockerfile HEALTHCHECK verifies both port 8000 (Flask API) and port 80 (HAProxy), with `start-period=60s` and `timeout=10s` +- Dockerfile HEALTHCHECK runs `scripts/healthcheck.sh`: port 8000 `/health` (4s max) AND port 80 (HAProxy). WHP's `lib-edge-container.sh` passes its own `--health-cmd`, which REPLACES the image's; it must call `/haproxy/scripts/healthcheck.sh` or the fleet never probes :8000 +- gunicorn uses `haproxy_gunicorn_worker.TimeoutThreadWorker` (gthread + client socket timeout + wedge watchdog). Stock gthread leaves client sockets blocking with NO timeout; `--timeout` only watches the main thread's heartbeat - Supports deployment on servers with git directory at `/root/whp` and web file sync via rsync to `/docker/whp/web/` - HAProxy is version 3.0.11 \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index f66654f..cf4674e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -44,6 +44,7 @@ WORKDIR /haproxy COPY ./templates /haproxy/templates COPY requirements.txt /haproxy/ COPY haproxy_manager.py /haproxy/ +COPY haproxy_gunicorn_worker.py /haproxy/ COPY scripts /haproxy/scripts COPY trusted_ips.list /etc/haproxy/trusted_ips.list COPY trusted_ips.map /etc/haproxy/trusted_ips.map @@ -92,7 +93,8 @@ RUN python3 /haproxy/scripts/test-wpadmin-gate.py \ && python3 /haproxy/scripts/test-config-rollback.py \ && python3 /haproxy/scripts/test-cert-write-safety.py \ && python3 /haproxy/scripts/test-cert-scripts.py \ - && python3 /haproxy/scripts/validate-rendered-config.py + && python3 /haproxy/scripts/validate-rendered-config.py \ + && python3 /haproxy/scripts/test-gunicorn-socket-timeouts.py # Create log directories RUN mkdir -p /var/log && touch /var/log/haproxy-manager.log /var/log/haproxy-manager-errors.log RUN chmod 755 /var/log/haproxy-manager.log /var/log/haproxy-manager-errors.log @@ -105,7 +107,10 @@ RUN mkdir -p /var/spool/cron/crontabs && \ # 443/udp carries HTTP/3 (QUIC). EXPOSE is documentation only — the container # must still be run with `-p 443:443/udp` for the UDP listener to be reachable. EXPOSE 80 443 443/udp 8000 -# Add health check +# Health check: management API (:8000, short timeout) AND HAProxy (:80). +# See scripts/healthcheck.sh. NOTE: WHP's lib-edge-container.sh passes its own +# --health-cmd, which REPLACES this one on deployed hosts; it must be pointed +# at /haproxy/scripts/healthcheck.sh for the fleet to get the :8000 probe. HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \ - CMD curl -sf --max-time 5 http://localhost:8000/health && curl -s --max-time 5 -o /dev/null http://localhost/ || exit 1 + CMD /haproxy/scripts/healthcheck.sh CMD ["/haproxy/scripts/start-up.sh"] \ No newline at end of file diff --git a/VERSION b/VERSION index 3715deb..fd74b1f 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2026.09.1 +2026.10.1 diff --git a/haproxy_gunicorn_worker.py b/haproxy_gunicorn_worker.py new file mode 100644 index 0000000..e497ad1 --- /dev/null +++ b/haproxy_gunicorn_worker.py @@ -0,0 +1,209 @@ +"""gunicorn gthread worker that cannot be wedged by a silent client. + +Incident 2026-10-09 (sdbees), and very likely the 2026-07-07 whp01 one too +------------------------------------------------------------------------- +Both :8000 API workers had 7 of 8 pool threads parked in the kernel in +``tcp_recvmsg -> sk_wait_data``: a blocking socket read with no timeout. +/health and /api/blocked-ips hung for 40s+, the container stayed "healthy". + +haproxy_manager.py makes NO in-process outbound network reads (no requests, +urllib, http.client, raw socket or ssl client anywhere; every ACME/HAProxy +interaction is a subprocess, already bounded since 2026.07.3). The reads that +blocked were gunicorn's own reads of the CLIENT socket: + + gunicorn/workers/gthread.py TConn.init(): self.sock.setblocking(True) + +Once the main loop sees a connection become readable it hands it to a pool +thread and turns the socket fully blocking -- timeout None -- before parsing +the request. Any peer that sends a few bytes and then waits (a TLS ClientHello +sent to this plain-HTTP port, a port scanner, a half-written request, a body +shorter than its Content-Length) pins that pool thread forever. :8000 is +published on 0.0.0.0 by the deploy (`-p 8000:8000`, which bypasses firewalld), +and on sdbees internet scanners were measured connecting to it. Each such +connection leaks one thread; 4 per worker exhausts the pool. + +gunicorn's --timeout does not catch this: for gthread it only checks that the +worker's MAIN thread keeps heart-beating, and the main thread keeps polling +happily while every pool thread is stuck. --max-requests does not help either: +a worker whose threads never finish a request never reaches the limit. +socket.setdefaulttimeout() does not help: setblocking(True) explicitly resets +the timeout to None, overriding any default. + +What this worker changes (and nothing else) +------------------------------------------- +1. Every client connection gets an explicit socket timeout + (HAPROXY_MGR_CLIENT_TIMEOUT, default 20s) at the exact point gthread makes it + blocking. It bounds each recv/send on the client socket -- request line, + headers, the body the app reads, and response writes. It does NOT bound how + long the app takes to handle a request (a long certbot run is a subprocess + with its own timeout and touches the client socket only once it is done). + A stalled peer now costs one thread for at most that long, then the + connection is closed and the thread is returned to the pool. + +2. A per-worker watchdog thread (HAPROXY_MGR_WEDGE_SECONDS, default 1200s; 0 + disables). If EVERY pool thread has been inside the same request for longer + than that -- i.e. the pool has not turned over at all for 20 minutes, longer + than the slowest legitimate request (certbot renew, 900s) -- the worker logs + every thread's Python stack to the error log and exits. The gunicorn master + respawns it in about a second, without touching HAProxy or the edge. + This is a backstop for blocking sources we have not found yet; the fix for + the known one is (1). + +3. pool_status() exposes busy/total threads and the oldest in-flight age so + /health can report saturation. +""" + +import errno +import logging +import os +import sys +import threading +import time +import traceback + +from gunicorn.workers.gthread import ThreadWorker + + +def _env_float(name, default): + try: + return float(os.environ.get(name, default)) + except (TypeError, ValueError): + return float(default) + + +CLIENT_TIMEOUT = _env_float('HAPROXY_MGR_CLIENT_TIMEOUT', 20) +WEDGE_SECONDS = _env_float('HAPROXY_MGR_WEDGE_SECONDS', 1200) +WATCHDOG_INTERVAL = _env_float('HAPROXY_MGR_WATCHDOG_INTERVAL', 15) + +# Exit status for a watchdog-killed worker. Must NOT be 3 (WORKER_BOOT_ERROR) +# or 4 (APP_LOAD_ERROR): on those the gunicorn master halts the whole server +# instead of respawning the worker. +WEDGED_EXIT_CODE = 1 + +_current_worker = None + + +def pool_status(): + """Busy/total threads of the worker serving the caller, or None when the + app is not running under this worker class (e.g. tests, dev server).""" + w = _current_worker + if w is None: + return None + return w.pool_status() + + +class _QuietClientTimeout(logging.Filter): + """gthread logs any OSError without an errno it recognises as + log.exception("Socket error processing request.") -- a full traceback. + A client timeout is now an expected, bounded event (scanners hit this port + constantly), so reduce it to one line instead of flooding the error log.""" + + def filter(self, record): + exc = record.exc_info[1] if record.exc_info else None + if isinstance(exc, TimeoutError) and \ + str(record.msg).startswith('Socket error processing request'): + record.msg = ('Closed client connection idle for more than %ss ' + '(HAPROXY_MGR_CLIENT_TIMEOUT)' % CLIENT_TIMEOUT) + record.args = () + record.exc_info = None + record.exc_text = None + record.levelno = logging.INFO + record.levelname = 'INFO' + return True + + +class TimeoutThreadWorker(ThreadWorker): + """gthread worker whose client sockets always carry a read/write timeout.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._inflight = {} # thread ident -> monotonic start time + self._inflight_lock = threading.Lock() + + # -- (1) client socket timeout ------------------------------------------ + def enqueue_req(self, conn): + # Same three steps as ThreadWorker.enqueue_req, with the timeout set + # right after conn.init() -- init() is what calls setblocking(True) + # (== settimeout(None)), and it runs again on every keep-alive reuse, + # so the timeout must be re-applied here each time, not at accept(). + conn.init() + if CLIENT_TIMEOUT > 0: + conn.sock.settimeout(CLIENT_TIMEOUT) + fs = self.tpool.submit(self.handle, conn) + self._wrap_future(fs, conn) + + # -- busy tracking -------------------------------------------------------- + def handle(self, conn): + ident = threading.get_ident() + with self._inflight_lock: + self._inflight[ident] = time.monotonic() + try: + return super().handle(conn) + finally: + with self._inflight_lock: + self._inflight.pop(ident, None) + + def pool_status(self): + now = time.monotonic() + with self._inflight_lock: + starts = list(self._inflight.values()) + return { + 'pid': os.getpid(), + 'threads': self.cfg.threads, + 'busy': len(starts), + 'oldest_busy_seconds': round(now - min(starts), 1) if starts else 0.0, + 'client_timeout_seconds': CLIENT_TIMEOUT, + } + + # -- (2) watchdog --------------------------------------------------------- + def _wedged_for(self): + """Seconds the pool has been completely wedged (every thread busy in + its current request), measured from the YOUNGEST in-flight request. + 0 when at least one thread is free.""" + with self._inflight_lock: + starts = list(self._inflight.values()) + if len(starts) < self.cfg.threads: + return 0.0 + return time.monotonic() - max(starts) + + def _dump_stacks(self): + frames = sys._current_frames() + names = {t.ident: t.name for t in threading.enumerate()} + out = [] + for ident, frame in frames.items(): + out.append('--- thread %s (%s)\n%s' % ( + ident, names.get(ident, '?'), ''.join(traceback.format_stack(frame)))) + return '\n'.join(out) + + def _watchdog(self): + while self.alive: + time.sleep(WATCHDOG_INTERVAL) + wedged = self._wedged_for() + if wedged > WEDGE_SECONDS: + self.log.critical( + 'WEDGED WORKER (pid %s): all %s threads busy in the same request ' + 'for %.0fs > HAPROXY_MGR_WEDGE_SECONDS=%s. Exiting so the master ' + 'respawns it. Thread stacks:\n%s', + os.getpid(), self.cfg.threads, wedged, WEDGE_SECONDS, + self._dump_stacks()) + for h in logging.getLogger('gunicorn.error').handlers: + try: + h.flush() + except Exception: + pass + # os._exit, not SIGTERM: a graceful exit joins the (stuck, + # non-daemon) pool threads at interpreter shutdown and would + # hang until the master's --timeout escalates to SIGKILL. + os._exit(WEDGED_EXIT_CODE) + + def init_process(self): + global _current_worker + _current_worker = self + try: + self.log.error_log.addFilter(_QuietClientTimeout()) + except AttributeError: + pass + if WEDGE_SECONDS > 0: + threading.Thread(target=self._watchdog, name='wedge-watchdog', + daemon=True).start() + super().init_process() # runs the event loop; does not return diff --git a/haproxy_manager.py b/haproxy_manager.py index 2e023bf..8d0bb00 100644 --- a/haproxy_manager.py +++ b/haproxy_manager.py @@ -51,6 +51,28 @@ def _bounded_subprocess_run(*args, **kwargs): subprocess.run = _bounded_subprocess_run +# --------------------------------------------------------------------------- +# Default socket timeout (incident 2026-10-09) -- defence in depth only +# --------------------------------------------------------------------------- +# This module makes no outbound network calls today (ACME, HAProxy's admin +# socket etc. all go through bounded subprocesses above). This default makes +# sure any socket created in-process later -- a urllib/http.client call added +# to some handler -- cannot block a worker thread forever by omission. +# +# It does NOT protect the API's CLIENT connections: gunicorn's gthread worker +# calls sock.setblocking(True) on each one, which resets the timeout to None +# and overrides this default. That is what actually wedged the pool on +# 2026-10-09; it is fixed by haproxy_gunicorn_worker.TimeoutThreadWorker. +DEFAULT_SOCKET_TIMEOUT = float(os.environ.get('HAPROXY_MGR_SOCKET_DEFAULT_TIMEOUT', '60')) +if DEFAULT_SOCKET_TIMEOUT > 0: + socket.setdefaulttimeout(DEFAULT_SOCKET_TIMEOUT) + +try: + from haproxy_gunicorn_worker import pool_status as _gunicorn_pool_status +except Exception: # dev server / tests without gunicorn + def _gunicorn_pool_status(): + return None + app = Flask(__name__) # Default page server (port 8080) — served to HAProxy clients whose request hit @@ -741,11 +763,19 @@ def health_check(): cursor.execute('SELECT 1') cursor.fetchone() - return jsonify({ + body = { 'status': 'healthy', 'haproxy_status': 'running' if haproxy_running else 'stopped', 'database': 'connected' - }), 200 + } + # Pool saturation of the worker that answered ('busy' includes this + # probe). Leaking threads show as busy/oldest_busy_seconds climbing; + # a fully wedged worker cannot answer at all, which is what the + # healthcheck's short --max-time catches. + pool = _gunicorn_pool_status() + if pool is not None: + body['worker'] = pool + return jsonify(body), 200 except Exception as e: return jsonify({ 'status': 'unhealthy', diff --git a/scripts/healthcheck.sh b/scripts/healthcheck.sh new file mode 100755 index 0000000..642f3a5 --- /dev/null +++ b/scripts/healthcheck.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Container liveness probe: BOTH planes must answer quickly. +# +# 1. Management API (:8000) -- GET /health with a short --max-time. On +# 2026-10-09 (sdbees) the API workers' thread pools were wedged for long +# enough that /health hung 40s+, while the container reported "healthy" +# because the deployed --health-cmd only probed :80. A hung API is a dead +# control plane (every WHP site update times out), so it must turn the +# container unhealthy. +# 2. HAProxy data plane (:80) -- any HTTP answer at all (the default page, +# a redirect, a 4xx/5xx) proves HAProxy accepted and answered. Only a +# connect/timeout failure fails it, which is why there is no -f here. +# +# 127.0.0.1, not localhost: the API binds IPv4 only, and resolving localhost +# to ::1 first costs a failed connect on every probe. +# +# WHP's lib-edge-container.sh overrides the image HEALTHCHECK with its own +# --health-cmd; point that at this script so the fleet gets the same probe. +set -u +API_TIMEOUT="${HAPROXY_MGR_HEALTH_API_TIMEOUT:-4}" +EDGE_TIMEOUT="${HAPROXY_MGR_HEALTH_EDGE_TIMEOUT:-5}" + +if ! curl -sf -o /dev/null --max-time "$API_TIMEOUT" http://127.0.0.1:8000/health; then + echo "management API :8000 /health failed or took longer than ${API_TIMEOUT}s" + exit 1 +fi +if ! curl -s -o /dev/null --max-time "$EDGE_TIMEOUT" http://127.0.0.1/; then + echo "HAProxy :80 did not answer within ${EDGE_TIMEOUT}s" + exit 1 +fi +exit 0 diff --git a/scripts/start-up.sh b/scripts/start-up.sh index 02494d2..a700b3b 100755 --- a/scripts/start-up.sh +++ b/scripts/start-up.sh @@ -77,7 +77,7 @@ HAPROXY_SUPERVISOR_INTERVAL="${HAPROXY_SUPERVISOR_INTERVAL:-15}" ) & # Phase 2: WSGI servers -# Tunable via env: HAPROXY_MGR_API_WORKERS (default 1), HAPROXY_MGR_API_TIMEOUT +# Tunable via env: HAPROXY_MGR_API_WORKERS (default 2), HAPROXY_MGR_API_TIMEOUT # (default 120 — API can do slow ACME calls), HAPROXY_MGR_MAX_REQUESTS (default # 1000 — worker recycle frequency). # @@ -91,12 +91,23 @@ API_TIMEOUT="${HAPROXY_MGR_API_TIMEOUT:-120}" MAX_REQ="${HAPROXY_MGR_MAX_REQUESTS:-1000}" MAX_REQ_JITTER="${HAPROXY_MGR_MAX_REQUESTS_JITTER:-100}" +# Worker class (incident 2026-10-09): stock gthread makes every client socket +# fully blocking with NO timeout, so any peer that sends a few bytes and goes +# quiet (TLS hello on this plain-HTTP port, scanners -- :8000 is published on +# 0.0.0.0) pins a pool thread forever; 4 of them wedge a worker. This subclass +# sets a per-connection socket timeout (HAPROXY_MGR_CLIENT_TIMEOUT, default +# 20s) and runs a watchdog that exits a worker whose threads have ALL been +# stuck for HAPROXY_MGR_WEDGE_SECONDS (default 1200) so the master respawns it. +# See haproxy_gunicorn_worker.py. --timeout below does NOT cover either case: +# for gthread it only watches the main thread's heartbeat. +WORKER_CLASS="haproxy_gunicorn_worker.TimeoutThreadWorker" + # Default page server on :8080. Stays in the background. # --threads 4 lets one worker handle bursts of blocked-IP/default-page hits # without forking. --max-requests recycles the worker to bound memory drift. gunicorn \ --bind 0.0.0.0:8080 \ - --workers 1 --threads 4 --worker-class gthread \ + --workers 1 --threads 4 --worker-class "${WORKER_CLASS}" \ --max-requests "${MAX_REQ}" --max-requests-jitter "${MAX_REQ_JITTER}" \ --timeout 30 \ --access-logfile - --error-logfile - --log-level info \ @@ -108,7 +119,7 @@ gunicorn \ # up). Longer --timeout because cert issuance hits ACME and can take a while. exec gunicorn \ --bind 0.0.0.0:8000 \ - --workers "${API_WORKERS}" --threads 4 --worker-class gthread \ + --workers "${API_WORKERS}" --threads 4 --worker-class "${WORKER_CLASS}" \ --max-requests "${MAX_REQ}" --max-requests-jitter "${MAX_REQ_JITTER}" \ --timeout "${API_TIMEOUT}" \ --access-logfile - --error-logfile - --log-level info \ diff --git a/scripts/test-gunicorn-socket-timeouts.py b/scripts/test-gunicorn-socket-timeouts.py new file mode 100644 index 0000000..438b6c4 --- /dev/null +++ b/scripts/test-gunicorn-socket-timeouts.py @@ -0,0 +1,356 @@ +#!/usr/bin/env python3 +"""Regression tests: a silent peer must not be able to wedge the API's threads. + +Why this file exists +-------------------- +On 2026-10-09 (sdbees) both :8000 API workers had 7 of 8 gthread pool threads +blocked in the kernel in tcp_recvmsg -> sk_wait_data, /health hung 40s+, and the +container stayed "healthy". The app makes no in-process outbound socket reads; +the blocked reads were gunicorn gthread reading the CLIENT socket, which it +makes fully blocking with no timeout (TConn.init -> setblocking(True)). + +These tests run REAL gunicorn processes (no mocks) against tiny WSGI apps and +pin: + + * the failure mode itself, on the stock gthread worker (control); + * that haproxy_gunicorn_worker.TimeoutThreadWorker frees the thread when a + client stalls before / during the request line, mid-body, or on a reused + keep-alive connection; + * that a legitimately SLOW request (handling time > client timeout) is not + cut off -- the timeout bounds socket I/O, not request duration; + * the watchdog: a worker whose every thread is stuck is replaced; + * the timeout is logged as one line, not a traceback per scanner; + * haproxy_manager's socket.setdefaulttimeout: an in-process client call to a + stalled upstream (accepts, never replies) times out instead of hanging; + * why that default alone is not enough (setblocking(True) clears it). + +Running +------- + python3 scripts/test-gunicorn-socket-timeouts.py + +stdlib unittest, like the other scripts/test-*.py; needs only the app's own +requirements (gunicorn, Flask...). Runs in the Docker build gate. ~25s. +""" + +import os +import socket +import subprocess +import sys +import tempfile +import textwrap +import threading +import time +import unittest +import urllib.request + +REPO_DIR = os.path.abspath( + os.environ.get('HAPROXY_MANAGER_DIR', + os.path.join(os.path.dirname(os.path.abspath(__file__)), '..'))) + +try: + import gunicorn # noqa: F401 +except ImportError: + sys.stderr.write('SKIP: gunicorn not installed (pip install -r requirements.txt)\n') + raise SystemExit(77) + +WSGI_APP = textwrap.dedent('''\ + import os, time + + def app(environ, start_response): + path = environ.get('PATH_INFO', '/') + if path == '/sleep': + time.sleep(float(environ.get('QUERY_STRING') or 3)) + elif path == '/hang': + time.sleep(3600) + elif path == '/body': + n = int(environ.get('CONTENT_LENGTH') or 0) + environ['wsgi.input'].read(n) + body = ('pid=%d' % os.getpid()).encode() + start_response('200 OK', [('Content-Type', 'text/plain'), + ('Content-Length', str(len(body)))]) + return [body] +''') + + +def free_port(): + s = socket.socket() + s.bind(('127.0.0.1', 0)) + port = s.getsockname()[1] + s.close() + return port + + +def http_get(port, path='/', timeout=5): + with urllib.request.urlopen('http://127.0.0.1:%d%s' % (port, path), + timeout=timeout) as r: + return r.read().decode() + + +class Gunicorn: + """A real gunicorn master on a free port, with its output captured.""" + + def __init__(self, worker_class, threads=1, env=None): + self.tmp = tempfile.mkdtemp(prefix='gunicorn-timeout-test-') + with open(os.path.join(self.tmp, 'tapp.py'), 'w') as fh: + fh.write(WSGI_APP) + self.port = free_port() + self.log_path = os.path.join(self.tmp, 'gunicorn.log') + e = dict(os.environ) + e.update(env or {}) + self.log = open(self.log_path, 'w') + self.proc = subprocess.Popen( + [sys.executable, '-m', 'gunicorn', + '--bind', '127.0.0.1:%d' % self.port, + '--workers', '1', '--threads', str(threads), + '--worker-class', worker_class, + '--timeout', '120', '--graceful-timeout', '2', + '--log-level', 'info', '--error-logfile', '-', + '--pythonpath', '%s,%s' % (REPO_DIR, self.tmp), + 'tapp:app'], + stdout=self.log, stderr=subprocess.STDOUT, env=e) + deadline = time.time() + 15 + while time.time() < deadline: + try: + http_get(self.port, '/', timeout=1) + return + except Exception: + time.sleep(0.2) + self.stop() + raise RuntimeError('gunicorn did not come up:\n' + self.output()) + + def output(self): + self.log.flush() + with open(self.log_path) as fh: + return fh.read() + + def stop(self): + if self.proc.poll() is None: + self.proc.terminate() + try: + self.proc.wait(5) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc.wait() + self.log.close() + + +def stall_client(port, payload=b'\x16\x03\x01\x02\x00\x01\x00\x01\xfc\x03\x03'): + """Connect and send a few bytes that are not a complete HTTP request (by + default the start of a TLS ClientHello -- what an HTTPS scanner sends to + this plain-HTTP port), then go silent and keep the connection open.""" + s = socket.create_connection(('127.0.0.1', port)) + if payload: + s.sendall(payload) + return s + + +def wait_closed_by_server(sock, within): + """True if the server closed (EOF/RST) the connection within `within` s.""" + sock.settimeout(within) + try: + while True: + data = sock.recv(4096) + if not data: + return True + except (ConnectionResetError, BrokenPipeError): + return True + except socket.timeout: + return False + + +STOCK = 'gthread' +FIXED = 'haproxy_gunicorn_worker.TimeoutThreadWorker' +FAST = {'HAPROXY_MGR_CLIENT_TIMEOUT': '1'} + + +class StockGthreadControl(unittest.TestCase): + """Proves the failure mode is real, so the fixed-worker tests mean something.""" + + def test_stock_gthread_is_wedged_by_one_silent_client(self): + g = Gunicorn(STOCK, threads=1) + self.addCleanup(g.stop) + s = stall_client(g.port) + self.addCleanup(s.close) + time.sleep(0.3) + with self.assertRaises(Exception, msg='stock gthread should be wedged'): + http_get(g.port, '/', timeout=3) + + +class TimeoutWorkerTests(unittest.TestCase): + + def start(self, threads=1, env=None): + g = Gunicorn(FIXED, threads=threads, env=env or FAST) + self.addCleanup(g.stop) + return g + + def test_silent_client_is_dropped_and_thread_freed(self): + g = self.start() + s = stall_client(g.port) + self.addCleanup(s.close) + t0 = time.time() + # Only one thread: this request can only be served once the stalled + # connection has released it. + self.assertTrue(http_get(g.port, '/', timeout=5).startswith('pid=')) + self.assertLess(time.time() - t0, 4) + self.assertTrue(wait_closed_by_server(s, 3), + 'server should have closed the stalled connection') + + def test_connect_and_send_nothing_does_not_hold_a_thread(self): + # Never readable -> never handed to a pool thread (main loop only). + g = self.start() + s = stall_client(g.port, payload=b'') + self.addCleanup(s.close) + self.assertTrue(http_get(g.port, '/', timeout=3).startswith('pid=')) + + def test_half_written_headers(self): + g = self.start() + s = stall_client(g.port, payload=b'GET /health HTTP/1.1\r\nHost: x\r\n') + self.addCleanup(s.close) + self.assertTrue(http_get(g.port, '/', timeout=5).startswith('pid=')) + self.assertTrue(wait_closed_by_server(s, 3)) + + def test_body_shorter_than_content_length(self): + g = self.start() + s = stall_client(g.port, payload=( + b'POST /body HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\n0123456789')) + self.addCleanup(s.close) + self.assertTrue(http_get(g.port, '/', timeout=5).startswith('pid=')) + + def test_timeout_reapplied_on_keepalive_reuse(self): + g = self.start() + s = socket.create_connection(('127.0.0.1', g.port)) + self.addCleanup(s.close) + s.sendall(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n') + s.settimeout(3) + self.assertIn(b'200 OK', s.recv(4096)) + # Same connection, now stall mid-request. gthread calls conn.init() + # (setblocking(True)) again for the reused connection. + s.sendall(b'GET / HTT') + time.sleep(0.3) + self.assertTrue(http_get(g.port, '/', timeout=5).startswith('pid=')) + self.assertTrue(wait_closed_by_server(s, 3)) + + def test_slow_legit_request_is_not_cut_off(self): + # Handling takes 2.5s, client timeout is 1s: must still succeed. + g = self.start() + self.assertTrue(http_get(g.port, '/sleep?2.5', timeout=10).startswith('pid=')) + + def test_timeout_logged_as_one_line_not_traceback(self): + g = self.start() + s = stall_client(g.port) + self.addCleanup(s.close) + http_get(g.port, '/', timeout=5) + time.sleep(0.3) + out = g.output() + self.assertIn('Closed client connection idle', out) + self.assertNotIn('Traceback', out) + + def test_watchdog_replaces_fully_wedged_worker(self): + g = self.start(env={'HAPROXY_MGR_CLIENT_TIMEOUT': '1', + 'HAPROXY_MGR_WEDGE_SECONDS': '2', + 'HAPROXY_MGR_WATCHDOG_INTERVAL': '0.5'}) + pid_before = http_get(g.port, '/') + # Wedge the only thread inside the app (not a socket read). + hung = threading.Thread(target=lambda: self._swallow(g.port), daemon=True) + hung.start() + time.sleep(0.5) + deadline = time.time() + 12 + pid_after = None + while time.time() < deadline: + try: + pid_after = http_get(g.port, '/', timeout=2) + if pid_after != pid_before: + break + except Exception: + pass + time.sleep(0.3) + self.assertIsNotNone(pid_after) + self.assertNotEqual(pid_before, pid_after, 'worker was not replaced') + out = g.output() + self.assertIn('WEDGED WORKER', out) + self.assertIn('tapp.py', out) # the stuck thread's stack was dumped + self.assertIsNone(g.proc.poll(), 'master must keep running') + + def test_watchdog_leaves_partially_busy_pool_alone(self): + # 2 threads, one hung forever: the pool still serves, so no restart. + g = self.start(threads=2, env={'HAPROXY_MGR_CLIENT_TIMEOUT': '1', + 'HAPROXY_MGR_WEDGE_SECONDS': '1', + 'HAPROXY_MGR_WATCHDOG_INTERVAL': '0.3'}) + pid_before = http_get(g.port, '/') + threading.Thread(target=lambda: self._swallow(g.port), daemon=True).start() + time.sleep(3) + self.assertEqual(pid_before, http_get(g.port, '/', timeout=3)) + self.assertNotIn('WEDGED WORKER', g.output()) + + @staticmethod + def _swallow(port): + try: + http_get(port, '/hang', timeout=60) + except Exception: + pass + + +class StalledUpstreamDefaultTimeout(unittest.TestCase): + """haproxy_manager sets socket.setdefaulttimeout() at import. An in-process + client call to an upstream that accepts and never replies must time out.""" + + def test_in_process_client_to_stalled_upstream_times_out(self): + srv = socket.socket() + srv.bind(('127.0.0.1', 0)) + srv.listen(5) + self.addCleanup(srv.close) + port = srv.getsockname()[1] + accepted = [] + threading.Thread(target=lambda: accepted.append(srv.accept()), + daemon=True).start() + # Import the real module in a child (it changes process-global socket + # state and opens its log files at import), then make a client call + # with NO explicit timeout -- what a careless future handler would do. + probe = textwrap.dedent('''\ + import logging, os, sys, tempfile, time, urllib.request + d = tempfile.mkdtemp() + real = logging.FileHandler + logging.FileHandler = lambda fn, *a, **k: real(os.path.join(d, os.path.basename(fn)), *a, **k) + os.chdir(sys.argv[1]); sys.path.insert(0, sys.argv[1]) + import haproxy_manager + logging.FileHandler = real + import socket + print('default', socket.getdefaulttimeout()) + t0 = time.time() + try: + urllib.request.urlopen('http://127.0.0.1:%s/' % sys.argv[2]).read() + print('result returned') + except Exception as e: + print('result', type(e).__name__, round(time.time() - t0, 1)) + ''') + env = dict(os.environ, HAPROXY_MGR_SOCKET_DEFAULT_TIMEOUT='1.5') + r = subprocess.run([sys.executable, '-c', probe, REPO_DIR, str(port)], + capture_output=True, text=True, timeout=30, env=env) + if 'No module named' in r.stderr and 'flask' in r.stderr.lower(): + self.skipTest('Flask not installed') + self.assertIn('default 1.5', r.stdout, r.stderr) + line = [l for l in r.stdout.splitlines() if l.startswith('result')][0] + self.assertTrue(line.startswith('result TimeoutError '), line) + elapsed = float(line.split()[-1]) + self.assertLess(elapsed, 5) + self.assertTrue(accepted, 'upstream never saw the connection') + accepted[0][0].close() + + def test_default_timeout_alone_does_not_cover_gthread_client_sockets(self): + # Documents why the custom worker exists: gthread's setblocking(True) + # resets any timeout -- including the process default -- to None. + old = socket.getdefaulttimeout() + socket.setdefaulttimeout(60) + try: + a, b = socket.socketpair() + finally: + socket.setdefaulttimeout(old) + self.addCleanup(a.close) + self.addCleanup(b.close) + self.assertEqual(a.gettimeout(), 60) + a.setblocking(True) + self.assertIsNone(a.gettimeout()) + + +if __name__ == '__main__': + unittest.main(verbosity=2)