perf: single-flight /api/stats; fast-fail health probe
CI/CD Pipeline - Northern Thailand Ping River Monitor / Test Suite (3.11) (push) Failing after 25s
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 13s
Documentation / Generate API Documentation (push) Successful in 10s
Documentation / Build Sphinx Documentation (push) Successful in 16s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Cleanup (push) Successful in 1s
Documentation / Documentation Summary (push) Successful in 4s
Documentation / Validate Documentation (push) Failing after 8s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Test Suite (3.11) (push) Failing after 25s
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 13s
Documentation / Generate API Documentation (push) Successful in 10s
Documentation / Build Sphinx Documentation (push) Successful in 16s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Cleanup (push) Successful in 1s
Documentation / Documentation Summary (push) Successful in 4s
Documentation / Validate Documentation (push) Failing after 8s
The on-box rerun after the inline fast path showed the cached endpoints healthy (latest p50 72-160ms, HII ~100ms) but /api/stats at 81% timeouts and /health at 33%: stats had a cache but NO single-flight, so every concurrent miss ran the heavy whole-DB counts (~1.7M rows) in parallel, re-jamming Postgres and the executor — which also dragged uncached history windows into 60s timeouts. /api/stats now computes through _ttl_cached_stale (one computation per 5min TTL, stale served on failure). The health API probe fails fast (5s instead of 30s) and its cache TTL rises to 30s, so a slow upstream can no longer pin executor threads longer than the cache lifetime.
This commit is contained in:
+1
-1
@@ -92,7 +92,7 @@ class Config:
|
|||||||
LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45"))
|
LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45"))
|
||||||
|
|
||||||
# TTL for /health check results (includes an external RID-API probe)
|
# TTL for /health check results (includes an external RID-API probe)
|
||||||
HEALTH_CACHE_TTL_SECONDS = int(os.getenv("HEALTH_CACHE_TTL_SECONDS", "10"))
|
HEALTH_CACHE_TTL_SECONDS = int(os.getenv("HEALTH_CACHE_TTL_SECONDS", "30"))
|
||||||
|
|
||||||
# Web server worker processes. Above 1, uvicorn forks workers and a
|
# Web server worker processes. Above 1, uvicorn forks workers and a
|
||||||
# localhost lock port elects a single background-collection leader.
|
# localhost lock port elects a single background-collection leader.
|
||||||
|
|||||||
@@ -140,6 +140,10 @@ class APIHealthCheck(HealthCheck):
|
|||||||
|
|
||||||
def __init__(self, api_url: str, session, name: str = "api"):
|
def __init__(self, api_url: str, session, name: str = "api"):
|
||||||
super().__init__(name)
|
super().__init__(name)
|
||||||
|
# A liveness probe should fail fast: the default 30s timeout meant a
|
||||||
|
# slow upstream pinned executor threads for longer than the /health
|
||||||
|
# cache TTL, so the pool never drained under load.
|
||||||
|
self.timeout_seconds = 5
|
||||||
self.api_url = api_url
|
self.api_url = api_url
|
||||||
self.session = session
|
self.session = session
|
||||||
|
|
||||||
|
|||||||
+65
-53
@@ -64,6 +64,7 @@ FORECAST_TTL = 900 # 15 minutes
|
|||||||
|
|
||||||
DB_STATS_CACHE: Dict[str, tuple] = {}
|
DB_STATS_CACHE: Dict[str, tuple] = {}
|
||||||
DB_STATS_CACHE_LOCK = Lock()
|
DB_STATS_CACHE_LOCK = Lock()
|
||||||
|
_DB_STATS_COMPUTE_LOCK = Lock()
|
||||||
DB_STATS_TTL = 300 # 5 minutes
|
DB_STATS_TTL = 300 # 5 minutes
|
||||||
|
|
||||||
# Admin API protection. Read/dashboard endpoints stay public; anything that
|
# Admin API protection. Read/dashboard endpoints stay public; anything that
|
||||||
@@ -1042,20 +1043,9 @@ async def get_database_stats():
|
|||||||
if not scraper or not scraper.db_adapter:
|
if not scraper or not scraper.db_adapter:
|
||||||
raise HTTPException(status_code=503, detail="Database not available")
|
raise HTTPException(status_code=503, detail="Database not available")
|
||||||
|
|
||||||
now = time.monotonic()
|
cached = _cache_fresh(DB_STATS_CACHE, DB_STATS_CACHE_LOCK, "all", DB_STATS_TTL)
|
||||||
with DB_STATS_CACHE_LOCK:
|
if cached is not None:
|
||||||
cached = DB_STATS_CACHE.get("all")
|
return cached
|
||||||
if cached and now - cached[0] < DB_STATS_TTL:
|
|
||||||
return cached[1]
|
|
||||||
|
|
||||||
try:
|
|
||||||
stats = await asyncio.to_thread(scraper.db_adapter.get_database_stats)
|
|
||||||
except Exception as e:
|
|
||||||
logger.error(f"Error fetching database stats: {e}")
|
|
||||||
raise HTTPException(status_code=500, detail=str(e))
|
|
||||||
|
|
||||||
if stats is None:
|
|
||||||
raise HTTPException(status_code=503, detail="Database statistics unavailable")
|
|
||||||
|
|
||||||
def hii_totals():
|
def hii_totals():
|
||||||
engine = _hii_engine()
|
engine = _hii_engine()
|
||||||
@@ -1079,47 +1069,69 @@ async def get_database_stats():
|
|||||||
)
|
)
|
||||||
).one()
|
).one()
|
||||||
|
|
||||||
hii = None
|
def compute():
|
||||||
|
# Heavy: full-table counts and coverage over ~1.7M rows. Runs at most
|
||||||
|
# once per TTL thanks to the single-flight; concurrent misses wait for
|
||||||
|
# this one result instead of piling identical queries onto Postgres.
|
||||||
|
stats = scraper.db_adapter.get_database_stats()
|
||||||
|
if stats is None:
|
||||||
|
raise RuntimeError("Database statistics unavailable")
|
||||||
|
|
||||||
|
hii = None
|
||||||
|
try:
|
||||||
|
hii = hii_totals()
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f"HII stats unavailable: {e}")
|
||||||
|
|
||||||
|
def as_dt(value):
|
||||||
|
if isinstance(value, str):
|
||||||
|
return datetime.fromisoformat(value)
|
||||||
|
return value
|
||||||
|
|
||||||
|
first_ts = stats["first_timestamp"]
|
||||||
|
last_ts = stats["last_timestamp"]
|
||||||
|
rain_n = wl_n = hii_stations = 0
|
||||||
|
if hii is not None:
|
||||||
|
rain_n, wl_n = hii.rain_n or 0, hii.wl_n or 0
|
||||||
|
hii_stations = (hii.rain_s or 0) + (hii.wl_s or 0)
|
||||||
|
for lo in (as_dt(hii.rain_lo), as_dt(hii.wl_lo)):
|
||||||
|
if lo is not None and lo < first_ts:
|
||||||
|
first_ts = lo
|
||||||
|
for hi in (as_dt(hii.rain_hi), as_dt(hii.wl_hi)):
|
||||||
|
if hi is not None and hi > last_ts:
|
||||||
|
last_ts = hi
|
||||||
|
|
||||||
|
return {
|
||||||
|
# Whole-DB totals (RID + HII feeds); breakdown fields alongside
|
||||||
|
"total_measurements": stats["total_measurements"] + rain_n + wl_n,
|
||||||
|
"rid_measurements": stats["total_measurements"],
|
||||||
|
"hii_rainfall_measurements": rain_n,
|
||||||
|
"hii_waterlevel_measurements": wl_n,
|
||||||
|
"station_count": stats["station_count"] + hii_stations,
|
||||||
|
"rid_station_count": stats["station_count"],
|
||||||
|
"hii_station_count": hii_stations,
|
||||||
|
"first_timestamp": first_ts.isoformat(),
|
||||||
|
"last_timestamp": last_ts.isoformat(),
|
||||||
|
"days_spanned": (last_ts.date() - first_ts.date()).days + 1,
|
||||||
|
"coverage_percent": stats["coverage_percent"],
|
||||||
|
}
|
||||||
|
|
||||||
try:
|
try:
|
||||||
hii = await asyncio.to_thread(hii_totals)
|
data, _ = await asyncio.to_thread(
|
||||||
|
_ttl_cached_stale,
|
||||||
|
DB_STATS_CACHE,
|
||||||
|
DB_STATS_CACHE_LOCK,
|
||||||
|
_DB_STATS_COMPUTE_LOCK,
|
||||||
|
"all",
|
||||||
|
DB_STATS_TTL,
|
||||||
|
compute,
|
||||||
|
)
|
||||||
|
return data
|
||||||
|
except RuntimeError as e:
|
||||||
|
raise HTTPException(status_code=503, detail=str(e))
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"HII stats unavailable: {e}")
|
logger.error(f"Error fetching database stats: {e}")
|
||||||
|
raise HTTPException(status_code=500, detail=str(e))
|
||||||
def as_dt(value):
|
|
||||||
if isinstance(value, str):
|
|
||||||
return datetime.fromisoformat(value)
|
|
||||||
return value
|
|
||||||
|
|
||||||
first_ts = stats["first_timestamp"]
|
|
||||||
last_ts = stats["last_timestamp"]
|
|
||||||
rain_n = wl_n = hii_stations = 0
|
|
||||||
if hii is not None:
|
|
||||||
rain_n, wl_n = hii.rain_n or 0, hii.wl_n or 0
|
|
||||||
hii_stations = (hii.rain_s or 0) + (hii.wl_s or 0)
|
|
||||||
for lo in (as_dt(hii.rain_lo), as_dt(hii.wl_lo)):
|
|
||||||
if lo is not None and lo < first_ts:
|
|
||||||
first_ts = lo
|
|
||||||
for hi in (as_dt(hii.rain_hi), as_dt(hii.wl_hi)):
|
|
||||||
if hi is not None and hi > last_ts:
|
|
||||||
last_ts = hi
|
|
||||||
|
|
||||||
data = {
|
|
||||||
# Whole-DB totals (RID + HII feeds); breakdown fields alongside
|
|
||||||
"total_measurements": stats["total_measurements"] + rain_n + wl_n,
|
|
||||||
"rid_measurements": stats["total_measurements"],
|
|
||||||
"hii_rainfall_measurements": rain_n,
|
|
||||||
"hii_waterlevel_measurements": wl_n,
|
|
||||||
"station_count": stats["station_count"] + hii_stations,
|
|
||||||
"rid_station_count": stats["station_count"],
|
|
||||||
"hii_station_count": hii_stations,
|
|
||||||
"first_timestamp": first_ts.isoformat(),
|
|
||||||
"last_timestamp": last_ts.isoformat(),
|
|
||||||
"days_spanned": (last_ts.date() - first_ts.date()).days + 1,
|
|
||||||
"coverage_percent": stats["coverage_percent"],
|
|
||||||
}
|
|
||||||
with DB_STATS_CACHE_LOCK:
|
|
||||||
DB_STATS_CACHE["all"] = (now, data)
|
|
||||||
return data
|
|
||||||
|
|
||||||
|
|
||||||
@app.post("/scrape/trigger", dependencies=[Depends(require_admin_key)])
|
@app.post("/scrape/trigger", dependencies=[Depends(require_admin_key)])
|
||||||
|
|||||||
Reference in New Issue
Block a user