perf: gzip responses; multi-worker serving with single collection leader
Documentation / Validate Documentation (push) Failing after 8s
Documentation / Build Sphinx Documentation (push) Successful in 15s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Test Suite (3.11) (push) Failing after 23s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Build Docker Image (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Integration Test with Services (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Staging (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Production (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Performance Test (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Code Quality (push) Successful in 12s
Documentation / Generate API Documentation (push) Successful in 9s
Documentation / Documentation Summary (push) Successful in 2s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Cleanup (push) Successful in 0s
Documentation / Validate Documentation (push) Failing after 8s
Documentation / Build Sphinx Documentation (push) Successful in 15s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Test Suite (3.11) (push) Failing after 23s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Build Docker Image (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Integration Test with Services (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Staging (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Production (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Performance Test (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Code Quality (push) Successful in 12s
Documentation / Generate API Documentation (push) Successful in 9s
Documentation / Documentation Summary (push) Successful in 2s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Cleanup (push) Successful in 0s
GZipMiddleware (min 500 bytes) compresses the dashboard HTML ~4x and station JSON up to ~100x, end-to-end through the Caddy TLS terminator — production load testing showed the deployment is bandwidth-bound once the response caches hit, so compression is the capacity lever. WEB_WORKERS (default 2) runs uvicorn multi-process via the app import string. Every worker executes the lifespan, so a localhost lock port (COLLECTION_LEADER_PORT, default 8901) elects exactly one background-collection leader per machine — RID/HII polling stays once-per-cycle instead of once-per-worker; the lock releases with the process. Locust clients now send Accept-Encoding so future runs measure compressed transfer, as browsers do.
This commit is contained in:
+18
-13
@@ -20,6 +20,9 @@ import random
|
|||||||
|
|
||||||
from locust import FastHttpUser, between, task
|
from locust import FastHttpUser, between, task
|
||||||
|
|
||||||
|
# Explicit so measurements reflect compressed transfer (browsers always send this)
|
||||||
|
GZIP = {"Accept-Encoding": "gzip, deflate"}
|
||||||
|
|
||||||
|
|
||||||
class DashboardVisitor(FastHttpUser):
|
class DashboardVisitor(FastHttpUser):
|
||||||
"""A browser session: initial page load, then periodic refresh polling."""
|
"""A browser session: initial page load, then periodic refresh polling."""
|
||||||
@@ -29,23 +32,23 @@ class DashboardVisitor(FastHttpUser):
|
|||||||
|
|
||||||
def on_start(self):
|
def on_start(self):
|
||||||
# What one real page load requests
|
# What one real page load requests
|
||||||
self.client.get("/")
|
self.client.get("/", headers=GZIP)
|
||||||
self.client.get("/stations")
|
self.client.get("/stations", headers=GZIP)
|
||||||
self.client.get("/measurements/latest?limit=500")
|
self.client.get("/measurements/latest?limit=500", headers=GZIP)
|
||||||
self.client.get("/api/hii/waterlevel/latest")
|
self.client.get("/api/hii/waterlevel/latest", headers=GZIP)
|
||||||
self.client.get("/api/hii/rainfall/latest")
|
self.client.get("/api/hii/rainfall/latest", headers=GZIP)
|
||||||
|
|
||||||
@task(4)
|
@task(4)
|
||||||
def poll_latest(self):
|
def poll_latest(self):
|
||||||
self.client.get("/measurements/latest?limit=500")
|
self.client.get("/measurements/latest?limit=500", headers=GZIP)
|
||||||
|
|
||||||
@task(2)
|
@task(2)
|
||||||
def poll_forecast(self):
|
def poll_forecast(self):
|
||||||
self.client.get("/forecast")
|
self.client.get("/forecast", headers=GZIP)
|
||||||
|
|
||||||
@task(2)
|
@task(2)
|
||||||
def poll_rain(self):
|
def poll_rain(self):
|
||||||
self.client.get("/api/hii/rainfall/latest")
|
self.client.get("/api/hii/rainfall/latest", headers=GZIP)
|
||||||
|
|
||||||
@task(1)
|
@task(1)
|
||||||
def view_history(self):
|
def view_history(self):
|
||||||
@@ -53,12 +56,13 @@ class DashboardVisitor(FastHttpUser):
|
|||||||
hours = random.choice([24, 168, 720])
|
hours = random.choice([24, 168, 720])
|
||||||
self.client.get(
|
self.client.get(
|
||||||
f"/measurements/history/{station}?hours={hours}",
|
f"/measurements/history/{station}?hours={hours}",
|
||||||
|
headers=GZIP,
|
||||||
name="/measurements/history/[station]",
|
name="/measurements/history/[station]",
|
||||||
)
|
)
|
||||||
|
|
||||||
@task(1)
|
@task(1)
|
||||||
def stats(self):
|
def stats(self):
|
||||||
self.client.get("/api/stats")
|
self.client.get("/api/stats", headers=GZIP)
|
||||||
|
|
||||||
|
|
||||||
class ApiConsumer(FastHttpUser):
|
class ApiConsumer(FastHttpUser):
|
||||||
@@ -69,25 +73,26 @@ class ApiConsumer(FastHttpUser):
|
|||||||
|
|
||||||
@task(3)
|
@task(3)
|
||||||
def latest(self):
|
def latest(self):
|
||||||
self.client.get("/measurements/latest?limit=100")
|
self.client.get("/measurements/latest?limit=100", headers=GZIP)
|
||||||
|
|
||||||
@task(3)
|
@task(3)
|
||||||
def hii_feeds(self):
|
def hii_feeds(self):
|
||||||
self.client.get(random.choice(
|
self.client.get(random.choice(
|
||||||
["/api/hii/waterlevel/latest", "/api/hii/rainfall/latest"]
|
["/api/hii/waterlevel/latest", "/api/hii/rainfall/latest"]
|
||||||
), name="/api/hii/[feed]/latest")
|
), headers=GZIP, name="/api/hii/[feed]/latest")
|
||||||
|
|
||||||
@task(2)
|
@task(2)
|
||||||
def forecast(self):
|
def forecast(self):
|
||||||
self.client.get("/forecast")
|
self.client.get("/forecast", headers=GZIP)
|
||||||
|
|
||||||
@task(2)
|
@task(2)
|
||||||
def heavy_history(self):
|
def heavy_history(self):
|
||||||
self.client.get(
|
self.client.get(
|
||||||
"/measurements/history/P.1?hours=8760",
|
"/measurements/history/P.1?hours=8760",
|
||||||
|
headers=GZIP,
|
||||||
name="/measurements/history/P.1 [heavy]",
|
name="/measurements/history/P.1 [heavy]",
|
||||||
)
|
)
|
||||||
|
|
||||||
@task(1)
|
@task(1)
|
||||||
def health(self):
|
def health(self):
|
||||||
self.client.get("/health")
|
self.client.get("/health", headers=GZIP)
|
||||||
|
|||||||
@@ -91,6 +91,11 @@ class Config:
|
|||||||
# TTL for the /measurements/latest response cache (hottest endpoint)
|
# TTL for the /measurements/latest response cache (hottest endpoint)
|
||||||
LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45"))
|
LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45"))
|
||||||
|
|
||||||
|
# Web server worker processes. Above 1, uvicorn forks workers and a
|
||||||
|
# localhost lock port elects a single background-collection leader.
|
||||||
|
WEB_WORKERS = int(os.getenv("WEB_WORKERS", "2"))
|
||||||
|
COLLECTION_LEADER_PORT = int(os.getenv("COLLECTION_LEADER_PORT", "8901"))
|
||||||
|
|
||||||
# Umami analytics (self-hosted). The website id is public (it ships in the
|
# Umami analytics (self-hosted). The website id is public (it ships in the
|
||||||
# dashboard <script> tag); server-side API tracking posts to /api/send.
|
# dashboard <script> tag); server-side API tracking posts to /api/send.
|
||||||
UMAMI_API_URL = os.getenv("UMAMI_API_URL", "https://stats.buildfor.life/api/send")
|
UMAMI_API_URL = os.getenv("UMAMI_API_URL", "https://stats.buildfor.life/api/send")
|
||||||
|
|||||||
+15
-6
@@ -344,15 +344,24 @@ def run_web_api():
|
|||||||
try:
|
try:
|
||||||
import uvicorn
|
import uvicorn
|
||||||
|
|
||||||
from .web_api import app
|
|
||||||
|
|
||||||
# Validate configuration
|
# Validate configuration
|
||||||
Config.validate_config()
|
Config.validate_config()
|
||||||
|
|
||||||
# Run the server
|
workers = max(1, Config.WEB_WORKERS)
|
||||||
uvicorn.run(
|
if workers > 1:
|
||||||
app, host="0.0.0.0", port=8000, log_config=None # Use our custom logging
|
# Multi-worker needs the app as an import string; a localhost lock
|
||||||
)
|
# port keeps background collection in exactly one worker.
|
||||||
|
uvicorn.run(
|
||||||
|
"src.web_api:app",
|
||||||
|
host="0.0.0.0",
|
||||||
|
port=8000,
|
||||||
|
workers=workers,
|
||||||
|
log_config=None,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
from .web_api import app
|
||||||
|
|
||||||
|
uvicorn.run(app, host="0.0.0.0", port=8000, log_config=None)
|
||||||
|
|
||||||
except ImportError:
|
except ImportError:
|
||||||
logger.error("FastAPI not installed. Run: pip install fastapi uvicorn")
|
logger.error("FastAPI not installed. Run: pip install fastapi uvicorn")
|
||||||
|
|||||||
+42
-2
@@ -24,6 +24,7 @@ from fastapi import (
|
|||||||
Response,
|
Response,
|
||||||
)
|
)
|
||||||
from fastapi.middleware.cors import CORSMiddleware
|
from fastapi.middleware.cors import CORSMiddleware
|
||||||
|
from fastapi.middleware.gzip import GZipMiddleware
|
||||||
from fastapi.responses import FileResponse, HTMLResponse
|
from fastapi.responses import FileResponse, HTMLResponse
|
||||||
from fastapi.staticfiles import StaticFiles
|
from fastapi.staticfiles import StaticFiles
|
||||||
|
|
||||||
@@ -110,6 +111,26 @@ app_state = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _acquire_collection_leadership(port: int):
|
||||||
|
"""Elect one background-collection leader per machine via a localhost bind.
|
||||||
|
|
||||||
|
With multiple uvicorn workers every process runs this lifespan; only the
|
||||||
|
worker holding the lock port runs the scraper/HII loops, so external APIs
|
||||||
|
are polled once per cycle instead of once per worker. The socket is held
|
||||||
|
for the process lifetime and releases automatically if the worker dies.
|
||||||
|
"""
|
||||||
|
import socket
|
||||||
|
|
||||||
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
|
try:
|
||||||
|
sock.bind(("127.0.0.1", port))
|
||||||
|
sock.listen(1)
|
||||||
|
return sock
|
||||||
|
except OSError:
|
||||||
|
sock.close()
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(app: FastAPI):
|
async def lifespan(app: FastAPI):
|
||||||
"""Application lifespan manager"""
|
"""Application lifespan manager"""
|
||||||
@@ -148,8 +169,18 @@ async def lifespan(app: FastAPI):
|
|||||||
health_manager.add_check(MemoryHealthCheck(max_memory_mb=1000))
|
health_manager.add_check(MemoryHealthCheck(max_memory_mb=1000))
|
||||||
app_state["health_manager"] = health_manager
|
app_state["health_manager"] = health_manager
|
||||||
|
|
||||||
# Start background scraping task
|
# Start background scraping in exactly one worker per machine
|
||||||
app_state["scraping_task"] = asyncio.create_task(background_scraping_task())
|
app_state["leader_lock"] = _acquire_collection_leadership(
|
||||||
|
Config.COLLECTION_LEADER_PORT
|
||||||
|
)
|
||||||
|
if app_state["leader_lock"]:
|
||||||
|
app_state["scraping_task"] = asyncio.create_task(background_scraping_task())
|
||||||
|
logger.info("This worker is the background-collection leader")
|
||||||
|
else:
|
||||||
|
logger.info(
|
||||||
|
"Another worker holds collection leadership; "
|
||||||
|
"background scraping disabled in this process"
|
||||||
|
)
|
||||||
|
|
||||||
logger.info("Water Monitor API started successfully")
|
logger.info("Water Monitor API started successfully")
|
||||||
|
|
||||||
@@ -164,6 +195,8 @@ async def lifespan(app: FastAPI):
|
|||||||
await app_state["scraping_task"]
|
await app_state["scraping_task"]
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
pass
|
pass
|
||||||
|
if app_state.get("leader_lock"):
|
||||||
|
app_state["leader_lock"].close()
|
||||||
|
|
||||||
logger.info("Water Monitor API shutdown complete")
|
logger.info("Water Monitor API shutdown complete")
|
||||||
|
|
||||||
@@ -185,6 +218,13 @@ app.mount(
|
|||||||
# Origins come from CORS_ALLOW_ORIGINS (comma-separated). When none are configured
|
# Origins come from CORS_ALLOW_ORIGINS (comma-separated). When none are configured
|
||||||
# we fall back to a wildcard WITHOUT credentials (a safe, spec-valid combination);
|
# we fall back to a wildcard WITHOUT credentials (a safe, spec-valid combination);
|
||||||
# credentials are only enabled when explicit origins are provided.
|
# credentials are only enabled when explicit origins are provided.
|
||||||
|
# Compress large responses end-to-end: the dashboard HTML and station JSON
|
||||||
|
# payloads shrink ~5-6x, which matters both on the internal VPN hop to the
|
||||||
|
# Caddy TLS terminator and on the public leg (Caddy passes Content-Encoding
|
||||||
|
# through). Load testing showed the deployment is bandwidth-bound, not
|
||||||
|
# compute-bound, once the response caches are hit.
|
||||||
|
app.add_middleware(GZipMiddleware, minimum_size=500)
|
||||||
|
|
||||||
_cors_origins = Config.CORS_ALLOW_ORIGINS or ["*"]
|
_cors_origins = Config.CORS_ALLOW_ORIGINS or ["*"]
|
||||||
_cors_allow_credentials = bool(Config.CORS_ALLOW_ORIGINS)
|
_cors_allow_credentials = bool(Config.CORS_ALLOW_ORIGINS)
|
||||||
app.add_middleware(
|
app.add_middleware(
|
||||||
|
|||||||
Reference in New Issue
Block a user