diff --git a/backend/app/api/fleet.py b/backend/app/api/fleet.py index 79dc1ce..b769981 100644 --- a/backend/app/api/fleet.py +++ b/backend/app/api/fleet.py @@ -1,6 +1,7 @@ """Fleet summary and consolidation endpoints.""" from __future__ import annotations +import logging from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta from typing import Any @@ -16,6 +17,8 @@ from ..models import Alert, FailureEvent, PoolLoadSample, SyncLog, Worker from ..sync.taskcluster import ALL_WORKER_POOLS, HW_WORKER_POOLS +log = logging.getLogger(__name__) + # Job-source sampling is served through the SWR cache so page loads never block on # live Taskcluster fan-out; a task's project/user is immutable, so it's cached too. _POOL_SOURCES_TTL = 90 @@ -763,9 +766,11 @@ def android_pools() -> dict[str, Any]: _ANDROID_PROVISIONER: dict[str, str] = {wt: prov for prov, wt in ANDROID_WORKER_POOLS} -@router.get("/android-pool-sources") -def android_pool_sources(pool: str) -> dict[str, Any]: - """Sample running tasks for an Android pool via the TC workers endpoint (no DB).""" +def _compute_android_pool_sources(pool: str) -> dict[str, Any]: + """Sample running tasks for an Android pool via the TC workers endpoint (no DB). + + Runs from a background thread (SWR refresh), so it must not touch a + request-scoped session — it uses none.""" provisioner = _ANDROID_PROVISIONER.get(pool) if not provisioner: return {"pool": pool, "sample_size": 0, "by_project": {}, "by_user": {}} @@ -828,6 +833,18 @@ def _fetch(task_id: str) -> dict[str, str]: } +@router.get("/android-pool-sources") +def android_pool_sources(pool: str) -> dict[str, Any]: + """Job-source breakdown for one Android pool. + + Cached: the uncached version made one live Taskcluster call per pool per page + load (limit=1000 workers, then one task fetch per running task).""" + return cache.swr( + f"android-pool-sources:{pool}", _POOL_SOURCES_TTL, + lambda: _compute_android_pool_sources(pool), + ) + + # ── Android per-device health (creds-free) ─────────────────────────────────── # Android devices (Bitbar / Lambda) have no rows in the Worker table and we don't # yet have Bitbar/Lambda API access. But the Taskcluster queue is fully public, and @@ -1080,6 +1097,59 @@ def pool_sources(pool: str) -> dict[str, Any]: return cache.swr(f"pool-sources:{pool}", _POOL_SOURCES_TTL, lambda: _compute_pool_sources(pool)) +# One request per pool was half the dashboard's entire Cloud Armor budget: a single +# Pools page load fired 35 /pool-sources + 14 /android-pool-sources, and the limit is +# per-IP for the whole origin. Both are individually cached, so the cost was purely +# request COUNT — which is exactly what the rate limiter charges for. These batch +# variants collapse each loop into one request with identical payloads. +_MAX_BATCH_POOLS = 120 + + +def _parse_pools(pools: str) -> list[str]: + """Comma-separated pool names → de-duplicated, order-preserving, bounded list.""" + seen: dict[str, None] = {} + for raw in pools.split(","): + name = raw.strip() + if name: + seen.setdefault(name) + return list(seen)[:_MAX_BATCH_POOLS] + + +def _sources_batch(pools: str, key_prefix: str, compute: Any) -> dict[str, Any]: + names = _parse_pools(pools) + if not names: + return {"sources": {}} + # Bounded parallelism: a warm batch is pure cache reads, but a cold one would + # otherwise sample Taskcluster serially. The per-pool compute functions each + # open their own DB session (or none), so they are safe off-thread. + out: dict[str, Any] = {} + with ThreadPoolExecutor(max_workers=min(8, len(names))) as ex: + futures = { + ex.submit(cache.swr, f"{key_prefix}:{n}", _POOL_SOURCES_TTL, lambda n=n: compute(n)): n + for n in names + } + for fut in as_completed(futures): + name = futures[fut] + try: + out[name] = fut.result() + except Exception: + # One bad pool must not fail the batch; the card just stays empty. + log.warning("pool-sources batch: %s failed for %s", key_prefix, name, exc_info=True) + return {"sources": out} + + +@router.get("/pool-sources-batch") +def pool_sources_batch(pools: str) -> dict[str, Any]: + """Job-source breakdown for many pools in one request. `pools` is comma-separated.""" + return _sources_batch(pools, "pool-sources", _compute_pool_sources) + + +@router.get("/android-pool-sources-batch") +def android_pool_sources_batch(pools: str) -> dict[str, Any]: + """Android job-source breakdown for many pools in one request.""" + return _sources_batch(pools, "android-pool-sources", _compute_android_pool_sources) + + def warm_pool_sources() -> int: """Background job: refresh the job-source cache for every pool with running tasks, so the first page load already has warm data. Returns pools warmed.""" diff --git a/backend/tests/test_pool_sources_batch.py b/backend/tests/test_pool_sources_batch.py new file mode 100644 index 0000000..3ccdf76 --- /dev/null +++ b/backend/tests/test_pool_sources_batch.py @@ -0,0 +1,155 @@ +"""Batched job-source endpoints: one request per page, not one per pool. + +These exist because the per-pool loops were measured, on 2026-09-17, as 49 of the +100 requests in a single minute that tripped Cloud Armor and replaced the +dashboard with a bare "429 Too Many Requests" page: + + /api/fleet/pool-sources 35 + /api/fleet/android-pool-sources 14 + +Both per-pool computations were already cached server-side, so the cost was purely +request COUNT -- which is exactly what a per-IP rate limiter charges for. The batch +variants must therefore collapse N requests into 1 while returning byte-identical +per-pool payloads, so no card changes behaviour. + +Dependency-light on purpose (no DB, no Taskcluster): the compute functions are +monkeypatched, since what is under test is batching, parsing and fan-out -- not +the sampling itself. +""" +from __future__ import annotations + +import pytest +from app import cache +from app.api import fleet + + +@pytest.fixture(autouse=True) +def _clear_cache(): + """The SWR cache is module-global; a leaked key would make these order-dependent.""" + cache._cache.clear() + cache._refreshing.clear() + yield + cache._cache.clear() + cache._refreshing.clear() + + +# ── Parsing ───────────────────────────────────────────────────────────────────── + +def test_parses_and_preserves_order(): + assert fleet._parse_pools("b,a,c") == ["b", "a", "c"] + + +def test_tolerates_whitespace_and_empty_segments(): + assert fleet._parse_pools(" a , ,b ,") == ["a", "b"] + + +def test_deduplicates_repeated_pools(): + """A repeated name must not become a repeated compute -- that is the whole point.""" + assert fleet._parse_pools("a,b,a,b,a") == ["a", "b"] + + +def test_empty_input_yields_no_pools(): + assert fleet._parse_pools("") == [] + assert fleet._parse_pools(" , , ") == [] + + +def test_batch_size_is_bounded(): + """An unbounded batch would just move the fan-out server-side.""" + names = ",".join(f"pool-{i}" for i in range(fleet._MAX_BATCH_POOLS + 50)) + assert len(fleet._parse_pools(names)) == fleet._MAX_BATCH_POOLS + + +# ── Batching ──────────────────────────────────────────────────────────────────── + +def test_batch_returns_one_entry_per_pool(monkeypatch): + calls: list[str] = [] + + def fake(pool: str) -> dict: + calls.append(pool) + return {"pool": pool, "sample_size": 7, "by_project": {}, "by_user": {}} + + monkeypatch.setattr(fleet, "_compute_pool_sources", fake) + out = fleet.pool_sources_batch("alpha,beta,gamma") + + assert sorted(out["sources"]) == ["alpha", "beta", "gamma"] + assert sorted(calls) == ["alpha", "beta", "gamma"] + assert out["sources"]["beta"]["sample_size"] == 7 + + +def test_batch_payload_is_identical_to_the_single_pool_route(monkeypatch): + """The frontend swapped N single calls for one batch; the per-pool value must not drift.""" + payload = {"pool": "alpha", "sample_size": 3, + "by_project": {"autoland": 3}, "by_user": {"someone": 3}} + monkeypatch.setattr(fleet, "_compute_pool_sources", lambda pool: dict(payload, pool=pool)) + + single = fleet.pool_sources("alpha") + cache._cache.clear() + batched = fleet.pool_sources_batch("alpha")["sources"]["alpha"] + + assert batched == single + + +def test_batch_is_served_from_the_existing_cache(monkeypatch): + """A warm batch must do zero computation -- these keys share the single route's cache.""" + calls: list[str] = [] + monkeypatch.setattr(fleet, "_compute_pool_sources", + lambda pool: (calls.append(pool), {"pool": pool})[1]) + + fleet.pool_sources_batch("alpha,beta") + assert sorted(calls) == ["alpha", "beta"] + + calls.clear() + fleet.pool_sources_batch("alpha,beta") + assert calls == [] + + +def test_one_failing_pool_does_not_fail_the_batch(monkeypatch): + """Previously each pool was its own request, so one failure cost one card.""" + def flaky(pool: str) -> dict: + if pool == "bad": + raise RuntimeError("taskcluster said no") + return {"pool": pool, "sample_size": 1, "by_project": {}, "by_user": {}} + + monkeypatch.setattr(fleet, "_compute_pool_sources", flaky) + out = fleet.pool_sources_batch("good,bad,alsogood") + + assert sorted(out["sources"]) == ["alsogood", "good"] + assert "bad" not in out["sources"] + + +def test_empty_batch_short_circuits(monkeypatch): + called = False + + def fake(pool: str) -> dict: + nonlocal called + called = True + return {} + + monkeypatch.setattr(fleet, "_compute_pool_sources", fake) + assert fleet.pool_sources_batch("") == {"sources": {}} + assert called is False + + +def test_android_batch_uses_the_android_compute(monkeypatch): + """The two batches must not share a cache namespace, or pools with the same + name in both would collide.""" + monkeypatch.setattr(fleet, "_compute_pool_sources", + lambda pool: {"pool": pool, "which": "hardware"}) + monkeypatch.setattr(fleet, "_compute_android_pool_sources", + lambda pool: {"pool": pool, "which": "android"}) + + assert fleet.pool_sources_batch("p")["sources"]["p"]["which"] == "hardware" + assert fleet.android_pool_sources_batch("p")["sources"]["p"]["which"] == "android" + + +def test_android_single_route_is_now_cached(monkeypatch): + """It was the one uncached path: one live TC workers call (limit=1000) per pool, + per page load, 14 times on the Pools page.""" + calls: list[str] = [] + monkeypatch.setattr(fleet, "_compute_android_pool_sources", + lambda pool: (calls.append(pool), {"pool": pool})[1]) + + fleet.android_pool_sources("gecko-t-lambda-test-1") + fleet.android_pool_sources("gecko-t-lambda-test-1") + + assert calls == ["gecko-t-lambda-test-1"] diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 2925ec4..eae32b3 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -2,6 +2,39 @@ const BASE = "/api"; +/** + * GET requests in flight, keyed by full URL. Several components mount together and + * ask for the same endpoint on one page load (Layout + Overview both want + * /fleet/summary; Workers + CommandPalette both want /fleet/pools). Cloud Armor + * rate-limits per source IP across the whole origin, and corp NAT means a shared + * budget, so a duplicate request is never free. Identical concurrent GETs share + * one response; nothing is cached past settlement, so this changes no semantics. + */ +const inflight = new Map>(); + +/** Retry once per step on the statuses that mean "try again", not "you're wrong". */ +const RETRY_STATUSES = new Set([429, 502, 503, 504]); +const RETRY_BACKOFF_MS = [400, 1200]; + +const sleep = (ms: number) => new Promise(r => setTimeout(r, ms)); + +async function getOnce(url: string): Promise { + for (let attempt = 0; ; attempt++) { + const res = await fetch(url); + if (res.ok) return res.json() as Promise; + if (!RETRY_STATUSES.has(res.status) || attempt >= RETRY_BACKOFF_MS.length) { + throw new Error(`${res.status} ${res.statusText} — ${new URL(url).pathname}`); + } + // Honour Retry-After when the server sends one (seconds or HTTP-date). + const hdr = res.headers.get("Retry-After"); + const hinted = hdr ? (/^\d+$/.test(hdr) ? Number(hdr) * 1000 : Date.parse(hdr) - Date.now()) : NaN; + const wait = Number.isFinite(hinted) && hinted > 0 + ? Math.min(hinted, 5_000) + : RETRY_BACKOFF_MS[attempt] * (0.5 + Math.random()); // jitter: don't resynchronise a burst + await sleep(wait); + } +} + async function get(path: string, params?: Record): Promise { const url = new URL(path, window.location.origin); url.pathname = BASE + path; @@ -10,9 +43,13 @@ async function get(path: string, params?: Record; + + const p = getOnce(key).finally(() => inflight.delete(key)); + inflight.set(key, p); + return p; } async function post(path: string, body?: unknown): Promise { @@ -173,6 +210,11 @@ export interface PoolSources { by_user: Record; } +/** Many pools' job-source breakdowns in one response, keyed by pool name. */ +export interface PoolSourcesBatch { + sources: Record; +} + export interface CloudPool { id: string; name: string; @@ -649,12 +691,16 @@ export const api = { pools: () => get("/fleet/pools"), pendingCounts: () => get("/fleet/pending-counts"), poolSources: (pool: string) => get("/fleet/pool-sources", { pool }), + poolSourcesBatch: (pools: string[]) => + get("/fleet/pool-sources-batch", { pools: pools.join(",") }), failures: (days = 7, platform?: string) => get("/fleet/failures", { days, platform }), loadHistory: (hours = 48, includeSeries = false) => get("/fleet/load-history", { hours, include_series: includeSeries || undefined }), cloudPools: () => get("/fleet/cloud-pools"), androidPools: () => get("/fleet/android-pools"), androidPoolSources: (pool: string) => get("/fleet/android-pool-sources", { pool }), + androidPoolSourcesBatch: (pools: string[]) => + get("/fleet/android-pool-sources-batch", { pools: pools.join(",") }), androidDevices: () => get("/fleet/android-devices"), showcase: () => get("/fleet/showcase"), }, diff --git a/frontend/src/pages/Overview.tsx b/frontend/src/pages/Overview.tsx index 283d872..dc1f760 100644 --- a/frontend/src/pages/Overview.tsx +++ b/frontend/src/pages/Overview.tsx @@ -172,9 +172,11 @@ function MonitoredPools({ load }: { load: LoadHistory | null }) { // fetch them only when the pinned set changes rather than on every poll. const monitoredKey = monitored.join(","); useEffect(() => { - for (const name of monitoredKey.split(",").filter(Boolean)) { - api.fleet.poolSources(name).then(s => setSources(prev => ({ ...prev, [name]: s }))).catch(() => {}); - } + const names = monitoredKey.split(",").filter(Boolean); + if (!names.length) return; + api.fleet.poolSourcesBatch(names) + .then(d => setSources(prev => ({ ...prev, ...d.sources }))) + .catch(() => {}); }, [monitoredKey]); const byName = new Map((load?.pools ?? []).map(p => [p.pool, p] as const)); diff --git a/frontend/src/pages/Pools.tsx b/frontend/src/pages/Pools.tsx index 0c96107..577afa8 100644 --- a/frontend/src/pages/Pools.tsx +++ b/frontend/src/pages/Pools.tsx @@ -979,22 +979,21 @@ export function Pools() { }, 300_000); useEffect(() => { - for (const poolName of pinnedPools) { - api.fleet.poolSources(poolName) - .then(s => setSources(prev => ({ ...prev, [poolName]: s }))) - .catch(() => {}); - } + if (!pinnedPools.length) return; + api.fleet.poolSourcesBatch(pinnedPools) + .then(d => setSources(prev => ({ ...prev, ...d.sources }))) + .catch(() => {}); }, [pinnedPools]); // Keyed on the name set (not array identity) so polling refreshes above don't // re-trigger the expensive per-pool task sampling. const linuxWindowsNames = pools.filter(p => isLinuxPool(p.name) || isWindowsPool(p.name)).map(p => p.name).sort().join(","); useEffect(() => { - for (const poolName of linuxWindowsNames.split(",").filter(Boolean)) { - api.fleet.poolSources(poolName) - .then(s => setSources(prev => ({ ...prev, [poolName]: s }))) - .catch(() => {}); - } + const names = linuxWindowsNames.split(",").filter(Boolean); + if (!names.length) return; + api.fleet.poolSourcesBatch(names) + .then(d => setSources(prev => ({ ...prev, ...d.sources }))) + .catch(() => {}); }, [linuxWindowsNames]); // Per-device Android health is a heavier server-side fan-out to the public TC queue, @@ -1010,11 +1009,11 @@ export function Pools() { const androidNames = androidPoolData.map(p => p.name).sort().join(","); useEffect(() => { - for (const poolName of androidNames.split(",").filter(Boolean)) { - api.fleet.androidPoolSources(poolName) - .then(s => setSources(prev => ({ ...prev, [poolName]: s }))) - .catch(() => {}); - } + const names = androidNames.split(",").filter(Boolean); + if (!names.length) return; + api.fleet.androidPoolSourcesBatch(names) + .then(d => setSources(prev => ({ ...prev, ...d.sources }))) + .catch(() => {}); }, [androidNames]); if (error) return
{error}
; diff --git a/terraform/lb.tf b/terraform/lb.tf index 596f227..24b9bf7 100644 --- a/terraform/lb.tf +++ b/terraform/lb.tf @@ -26,7 +26,54 @@ resource "google_compute_region_network_endpoint_group" "hangar" { resource "google_compute_security_policy" "hangar" { name = "hangar-armor" - # Rate limit: 100 requests/min per IP + # Rate limiting. IAP already restricts this origin to @mozilla.com, so these + # limits are DoS hygiene, not access control — sized to never deny a human. + # + # Two rules, because the app shell and the API have very different shapes and + # very different failure modes: + # + # - Denying an /api/* call degrades a card inside a loaded page, and the + # client retries it (frontend/src/api.ts). + # - Denying the document or a hashed asset replaces the whole dashboard with + # Cloud Armor's bare "429 Too Many Requests" page. That is what an operator + # actually reported, twice, on 2026-09-17 — the document was denied after a + # burst of API calls had already spent the budget. + # + # Sizing: one Pools page load was measured at ~100 requests, of which 49 were + # two unbatched per-pool loops (now batched — see fleet.pool_sources_batch). + # A heavy load is ~50 requests post-fix, so 1200/min is ~24 page loads per + # minute per IP. That headroom matters because enforce_on_key = "IP" and corp + # VPN/NAT means many operators can share one egress address and therefore one + # budget — the reason this gets worse, not better, as the audience grows. + # + # NOTE: Cloud Armor enforces the FIRST matching rule and stops. These rules + # match every request, so the OWASP rules at priority 2000+ below are currently + # unreachable (verified in LB logs: every request reports enforcedSecurityPolicy + # priority 1000). Fixing that means moving them above these — do it as its own + # change, with preview = true first to measure false positives. + rule { + action = "throttle" + priority = 900 + match { + expr { + expression = "request.path.startsWith('/api/')" + } + } + rate_limit_options { + conform_action = "allow" + exceed_action = "deny(429)" + rate_limit_threshold { + count = 1200 + interval_sec = 60 + } + enforce_on_key = "IP" + } + description = "Rate limit API calls per IP" + } + + # Everything else: the document, hashed assets, favicon. Cheap and cacheable, so + # the only client that reaches this ceiling is a scanner. Deliberately generous — + # a human must never be shown a bare 429 instead of the dashboard. rule { action = "throttle" priority = 1000 @@ -40,12 +87,12 @@ resource "google_compute_security_policy" "hangar" { conform_action = "allow" exceed_action = "deny(429)" rate_limit_threshold { - count = 100 + count = 3000 interval_sec = 60 } enforce_on_key = "IP" } - description = "Rate limit per IP" + description = "Rate limit the app shell per IP (generous; scanner backstop)" } # OWASP Top 10 pre-configured rules