fix(storage): reuse keep-alive connections in webhook storage

common/storage/webhook/__init__.py

@@ -1,19 +1,50 @@
+import os
+import threading
 import time
 
 from lib.vcon_redis import VconRedis
 from lib.logging_utils import init_logger
 from lib.metrics import record_histogram
 
 import requests
+from requests.adapters import HTTPAdapter
 
 logger = init_logger(__name__)
 
 default_options = {
     "webhook-urls": [],
     "headers": {},
     "timeout": 30,
+    # Keep-alive connections kept per worker process. Size it to the in-flight
+    # vCons per worker (CONSERVER_VCON_CONCURRENCY); extra connections still work
+    # but are closed after use.
+    "pool-size": 16,
 }
 
+# One Session per process so writes reuse keep-alive connections. A bare
+# requests.post opened (and TIME_WAITed) a new socket per write, which exhausted
+# the pod's ephemeral ports at a few hundred writes/s. Storage writes run on
+# short-lived threads, so the session is shared across threads; urllib3's pool
+# is thread-safe and these POSTs carry no cookies. Rebuilt after a fork so a
+# child never shares its parent's sockets.
+_session = None
+_session_pid = None
+_session_lock = threading.Lock()
+
+
+def _get_session(pool_size):
+    global _session, _session_pid
+    pid = os.getpid()
+    if _session is None or _session_pid != pid:
+        with _session_lock:
+            if _session is None or _session_pid != pid:
+                session = requests.Session()
+                adapter = HTTPAdapter(pool_connections=4, pool_maxsize=pool_size)
+                session.mount("http://", adapter)
+                session.mount("https://", adapter)
+                _session, _session_pid = session, pid
+    return _session
+
 
 def save(vcon_uuid, opts=default_options):
     vcon_redis = VconRedis()
@@ -38,7 +69,8 @@ def save(vcon_uuid, opts=default_options):
             f"webhook storage: posting vcon {vcon_uuid} to webhook url: {url}"
         )
         webhook_start = time.time()
-        resp = requests.post(
+        session = _get_session(opts.get("pool-size", default_options["pool-size"]))
+        resp = session.post(
             url, json=json_dict, headers=headers,
             timeout=opts.get("timeout", default_options["timeout"]),
         )

common/tests/test_webhook_keepalive.py

@@ -0,0 +1,46 @@
+"""Webhook storage reuses one keep-alive connection instead of a new socket per write."""
+
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from unittest.mock import patch
+
+from storage import webhook
+
+
+class _Receiver(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+    client_ports = []
+
+    def do_POST(self):
+        self.rfile.read(int(self.headers["Content-Length"]))
+        _Receiver.client_ports.append(self.client_address[1])
+        self.send_response(201)
+        self.send_header("Content-Length", "2")
+        self.end_headers()
+        self.wfile.write(b"{}")
+
+    def log_message(self, *args):
+        pass
+
+
+def test_sequential_writes_share_one_connection():
+    _Receiver.client_ports = []
+    server = ThreadingHTTPServer(("127.0.0.1", 0), _Receiver)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    url = f"http://127.0.0.1:{server.server_address[1]}/probe"
+    try:
+        with patch.object(webhook, "VconRedis") as bodies:
+            bodies.return_value.get_vcon.return_value.to_dict.return_value = {"synthetic_retry_probe": True}
+            for _ in range(5):
+                webhook.save("synthetic-id", {"webhook-urls": [url]})
+    finally:
+        server.shutdown()
+        server.server_close()
+    assert len(_Receiver.client_ports) == 5
+    assert len(set(_Receiver.client_ports)) == 1
+
+
+def test_session_is_rebuilt_after_fork():
+    first = webhook._get_session(4)
+    with patch.object(webhook.os, "getpid", return_value=-1):
+        assert webhook._get_session(4) is not first

common/tests/test_webhook_storage_failures.py

@@ -21,11 +21,12 @@ def test_http_failure_reaches_storage_dlq(status):
     )
     with (
         patch.object(webhook, "VconRedis") as bodies,
-        patch.object(webhook.requests, "post", return_value=response),
+        patch.object(webhook, "_get_session") as session,
         patch("main.Storage") as storage,
         patch("main.queue") as queue,
     ):
         bodies.return_value.get_vcon.return_value.to_dict.return_value = {"synthetic_retry_probe": True}
+        session.return_value.post.return_value = response
         storage.return_value.save.side_effect = lambda vid: webhook.save(vid, {"webhook-urls": [response.url]})
         request._process_storage("test_webhook")
         queue.enqueue_storage_dlq.assert_called_once_with("test_webhook", "synthetic-id", retention_seconds=604800)
@@ -38,8 +39,9 @@ def test_success_has_bounded_request_timeout(options, expected):
     response._content = b"{}"
     with (
         patch.object(webhook, "VconRedis") as bodies,
-        patch.object(webhook.requests, "post", return_value=response) as post,
+        patch.object(webhook, "_get_session") as session,
     ):
         bodies.return_value.get_vcon.return_value.to_dict.return_value = {"synthetic_retry_probe": True}
+        session.return_value.post.return_value = response
         webhook.save("synthetic-id", {"webhook-urls": ["https://receiver.invalid/probe"], **options})
-        assert post.call_args.kwargs["timeout"] == expected
+        assert session.return_value.post.call_args.kwargs["timeout"] == expected