diff --git a/backend/Dockerfile b/backend/Dockerfile index e5ed738..95d28ec 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -13,6 +13,9 @@ RUN apt-get update \ # Install deps first so this layer caches across code-only changes. COPY requirements.txt /tmp/requirements.txt RUN pip install --no-cache-dir -r /tmp/requirements.txt +# Bake DuckDB's httpfs extension so the lake role's S3 reads need no runtime +# extension download (tasks may roll while egress is degraded). +RUN python -c "import duckdb; duckdb.connect().execute('INSTALL httpfs')" COPY . /app/ RUN chmod +x /app/deploy/entrypoint.sh diff --git a/backend/data/climate.py b/backend/data/climate.py index fc5e411..0a148ff 100644 --- a/backend/data/climate.py +++ b/backend/data/climate.py @@ -21,6 +21,7 @@ from core import audit from core import metrics import paths from data import climate_store +from data import era5lake from data import meteostat from data import store @@ -789,21 +790,32 @@ def _load_history(cell: dict) -> tuple[pl.DataFrame, dict]: def _serve_stale(): return _with_doy(stale), {"cached": True, "stale": True, "cache_age_days": None} - # Fetch. NASA POWER is the primary source (keyless, independent of Open-Meteo); - # the gusts it lacks are filled from Meteostat inside _fetch_history_nasa. - # Open-Meteo is the fallback when NASA is unavailable — still guarded by its - # rate-limit cooldown. Both sources must return a plausibly-full span before - # being accepted and cached as a complete record (a short/partial response is - # rejected rather than held indefinitely). + # Fetch. The ERA5 lake is first when configured (true ERA5 with measured + # gusts, served from our own bucket — no third-party API in the path); + # unconfigured or missing-point cells fall through at zero cost. NASA + # POWER is next (keyless, independent of Open-Meteo; its missing gusts + # are filled from Meteostat inside _fetch_history_nasa), then Open-Meteo + # — still guarded by its rate-limit cooldown. Every source must return a + # plausibly-full span before being accepted and cached as a complete + # record (a short/partial response is rejected rather than held + # indefinitely). df = None source = None try: - fetched = _fetch_history_nasa(cell) + fetched = era5lake.fetch_history_lake(cell, start=START_DATE) if fetched.height >= MIN_ARCHIVE_DAYS: df = fetched - source = "nasa-power" - except Exception: # noqa: BLE001 - primary unavailable; fall through to Open-Meteo + source = "era5-lake" + except Exception: # noqa: BLE001 - lake unconfigured/miss; fall through to NASA pass + if df is None: + try: + fetched = _fetch_history_nasa(cell) + if fetched.height >= MIN_ARCHIVE_DAYS: + df = fetched + source = "nasa-power" + except Exception: # noqa: BLE001 - primary unavailable; fall through to Open-Meteo + pass if df is None and time.time() >= _archive_cooldown_until: try: fetched = _fetch_history(cell) diff --git a/backend/data/era5lake.py b/backend/data/era5lake.py new file mode 100644 index 0000000..7914745 --- /dev/null +++ b/backend/data/era5lake.py @@ -0,0 +1,173 @@ +"""The ERA5 lake: layout, grid math, and read clients for the bucket-hosted archive. + +The lake is a set of parquet files in an S3-compatible bucket (Contabo Object +Storage, ``era5-thermograph``), extracted once from the public Earthmover ERA5 +Icechunk archive by ``gen_era5_lake.py`` and read three ways: + +- the **lake service** (``lake_app.py``, prod-only Swarm service) serves point + histories and SQL-style queries with a local disk cache in front; +- the **web/worker backend** asks the lake service first + (``THERMOGRAPH_LAKE_URL``), or reads the bucket directly when only S3 + credentials are configured (beta/LAN, no lake service); +- ad-hoc analysis hits the hive table straight from the bucket. + +Two layouts, one manifest (all under ``era5/``): + +- ``daily/tile=_/year=/month=/part.parquet`` — the analytical + table, hive-partitioned by location/year/month. The location grain is a + 12x12-point tile (~3°x3°), aligned to the Earthmover store's chunk grid; a + per-point x per-month grain would be ~170M objects, which no object store + enjoys. +- ``points/lat=/lon=.parquet`` — the serving projection: one file per + ERA5 grid point holding its full daily history, so the hot path (one cell's + history) is a single GET. +- ``manifest.parquet`` — Iceberg-manifest-style index of every point file: + grid indices, coordinates, row count, date span. Readers prune with it and + the extractor resumes from it. + +Columns everywhere: the finalized history-store schema (date, tmax, tmin, +precip, wind, gust, humid, fmax, fmin, feels, doy — °F/in/mph/%, as +``climate._finalize_approximated`` emits it), so a lake frame drops straight +into ``climate._write_history_backed`` exactly like a NASA fetch does. +""" +import io +import os + +import polars as pl + +# --- ERA5 native grid (0.25°, 721 x 1440, lat 90..-90, lon 0..359.75) --------- +LAT_N = 721 +LON_N = 1440 +STEP = 0.25 +TILE = 12 # matches the Earthmover chunk grid (8736h x 12 x 12) + +PREFIX = "era5" +MANIFEST_KEY = f"{PREFIX}/manifest.parquet" + +MIN_LAKE_DAYS = 10_000 # a plausibly-full 1980+ record, same spirit as + # climate.MIN_ARCHIVE_DAYS (rejects partial files) + + +def to_idx(lat: float, lon: float) -> tuple[int, int]: + """Nearest ERA5 grid indices for a lat/lon. lat_idx runs north->south.""" + i = round((90.0 - lat) / STEP) + j = round((lon % 360.0) / STEP) % LON_N + return min(max(i, 0), LAT_N - 1), j + + +def to_coords(i: int, j: int) -> tuple[float, float]: + """Center lat/lon ([-180, 180) longitude) of grid indices.""" + lon = j * STEP + return 90.0 - i * STEP, lon - 360.0 if lon >= 180.0 else lon + + +def tile_of(i: int, j: int) -> tuple[int, int]: + return i // TILE, j // TILE + + +def point_key(i: int, j: int) -> str: + return f"{PREFIX}/points/lat={i}/lon={j}.parquet" + + +def daily_part_key(ti: int, tj: int, year: int, month: int) -> str: + return f"{PREFIX}/daily/tile={ti}_{tj}/year={year}/month={month:02d}/part.parquet" + + +# --- Config (all optional: unset means "no lake configured") ------------------ + +def lake_url() -> str: + """Lake service base URL (prod: the Swarm VIP), '' when not deployed.""" + return os.environ.get("THERMOGRAPH_LAKE_URL", "").rstrip("/") + + +def s3_config() -> "dict | None": + """Bucket access for direct reads/writes, None unless fully configured. + Contabo is path-style; region is a signing formality there ('default').""" + key = os.environ.get("THERMOGRAPH_LAKE_S3_ACCESS_KEY", "") + secret = os.environ.get("THERMOGRAPH_LAKE_S3_SECRET_KEY", "") + if not (key and secret): + return None + return { + "endpoint": os.environ.get("THERMOGRAPH_LAKE_S3_ENDPOINT", + "https://eu2.contabostorage.com"), + "bucket": os.environ.get("THERMOGRAPH_LAKE_S3_BUCKET", "era5-thermograph"), + "region": os.environ.get("THERMOGRAPH_LAKE_S3_REGION", "default"), + "access_key": key, + "secret_key": secret, + } + + +def local_dir() -> str: + """Filesystem mirror of the lake layout — tests and offline dev reads.""" + return os.environ.get("THERMOGRAPH_LAKE_LOCAL_DIR", "") + + +def storage_options(cfg: dict) -> dict: + """polars/object_store options for scan_parquet against the bucket.""" + return { + "aws_access_key_id": cfg["access_key"], + "aws_secret_access_key": cfg["secret_key"], + "aws_region": cfg["region"], + "endpoint_url": cfg["endpoint"], + # Contabo serves buckets on the path, not as a subdomain. + "aws_virtual_hosted_style_request": "false", + } + + +# --- Point-history reads (the app's hot path) ---------------------------------- + +class LakeUnavailable(Exception): + """No lake configured, point not in the lake, or the read failed.""" + + +def _read_local(i: int, j: int) -> pl.DataFrame: + path = os.path.join(local_dir(), point_key(i, j)) + if not os.path.exists(path): + raise LakeUnavailable(f"no local lake file for point ({i}, {j})") + return pl.read_parquet(path) + + +def _read_service(i: int, j: int) -> pl.DataFrame: + import httpx # noqa: PLC0415 - keep module import light for the extractor venv + + lat, lon = to_coords(i, j) + r = httpx.get(f"{lake_url()}/history", params={"lat": lat, "lon": lon}, + timeout=httpx.Timeout(20.0, connect=3.0)) + if r.status_code == 404: + raise LakeUnavailable(f"point ({i}, {j}) not in the lake") + r.raise_for_status() + return pl.read_parquet(io.BytesIO(r.content)) + + +def _read_bucket(i: int, j: int, cfg: dict) -> pl.DataFrame: + try: + return pl.read_parquet(f"s3://{cfg['bucket']}/{point_key(i, j)}", + storage_options=storage_options(cfg)) + except Exception as e: # noqa: BLE001 - object_store errors are not one type + raise LakeUnavailable(f"bucket read failed for point ({i}, {j}): {e}") from e + + +def fetch_history_lake(cell: dict, start: "str | None" = None) -> pl.DataFrame: + """Finalized daily history for the ERA5 point nearest a grid cell, from the + first configured source: local dir -> lake service -> bucket. The lake + stores each point's whole 1940+ record; pass ``start`` (the caller's + grading window, e.g. climate.START_DATE) to slice — grading semantics stay + the caller's, the deep record stays in the lake. Raises LakeUnavailable + when unconfigured, missing, or implausibly short.""" + import datetime # noqa: PLC0415 + + i, j = to_idx(cell["center_lat"], cell["center_lon"]) + if local_dir(): + df = _read_local(i, j) + elif lake_url(): + df = _read_service(i, j) + else: + cfg = s3_config() + if cfg is None: + raise LakeUnavailable("no lake configured") + df = _read_bucket(i, j, cfg) + if start is not None: + df = df.filter(pl.col("date") >= datetime.date.fromisoformat(start)) + if df.height < MIN_LAKE_DAYS: + raise LakeUnavailable(f"lake point ({i}, {j}) is short: {df.height} rows") + return df.drop([c for c in ("lat_idx", "lon_idx") if c in df.columns]) diff --git a/backend/deploy/entrypoint.sh b/backend/deploy/entrypoint.sh index 9c68987..9b6f399 100755 --- a/backend/deploy/entrypoint.sh +++ b/backend/deploy/entrypoint.sh @@ -62,6 +62,13 @@ run_migrations() { done } +# The lake role serves the ERA5 bucket only — no database, so no migrations: a +# lake replica must come up (and stay up) even when Postgres is down. +if [ "${THERMOGRAPH_ROLE:-}" = "lake" ]; then + echo "==> Starting lake service on 0.0.0.0:${PORT:-8141} with ${WORKERS:-1} worker(s)" + exec uvicorn lake_app:app --host 0.0.0.0 --port "${PORT:-8141}" --workers "${WORKERS:-1}" +fi + # One-shot migrate mode (`entrypoint.sh migrate`, or THERMOGRAPH_MIGRATE_ONLY=1): run # the migration and exit — a one-shot task, so it runs once against the DB rather # than racing per web replica. diff --git a/backend/gen_era5_lake.py b/backend/gen_era5_lake.py new file mode 100644 index 0000000..f058674 --- /dev/null +++ b/backend/gen_era5_lake.py @@ -0,0 +1,218 @@ +"""Extract the ERA5 lake: Earthmover Icechunk archive -> the Thermograph bucket. + +Builds the lake that ``data/era5lake.py`` reads (layout documented there): +per-point serving files, the tile/year/month hive table, and the manifest. +Pulls are tile-aligned to the source's chunk grid (8736h x 12 x 12) — one +46-year pull (~19s measured) decodes all 144 points of a tile, so per-point +pulls would waste 144x the wire. Roughly: ~600 tiles cover the curated +cities (hours), ~3000 cover land (a day or two); both resume from the +manifest, so ctrl-C and re-run is always safe. + + pip install -r requirements-seed.txt 'numcodecs[pcodec]' boto3 + python gen_era5_lake.py --cities --dry-run # one tile, no writes + python gen_era5_lake.py --cities # tiles covering cities.json + python gen_era5_lake.py --tiles 20:30,40:60 # explicit tile-index window + python gen_era5_lake.py --land # every land-containing tile + +Destination: the bucket from THERMOGRAPH_LAKE_S3_* (see era5lake.s3_config), +or a local mirror when THERMOGRAPH_LAKE_LOCAL_DIR is set (tests/rehearsal). +Seed-only deps (icechunk/xarray/zarr) stay out of the app image, same policy +as seed_era5.py. +""" +import argparse +import datetime +import io +import os +import sys +import time + +import polars as pl + +from data import cities +from data import climate +from data import era5lake +from seed_era5 import ERA5_BUCKET, ERA5_GROUP, ERA5_PREFIX, ERA5_REGION, ERA5_VARS, \ + hourly_to_daily + +# The lake keeps each location's WHOLE available record (the store reaches +# back to 1940); the app slices to its own grading window (climate.START_DATE) +# at read time, so the extra decades are analytical surface, not a grading +# change. +START_DATE = "1940-01-01" + + +def _open_store(): + import icechunk # noqa: PLC0415 - seed-only dep + import xarray as xr # noqa: PLC0415 + + storage = icechunk.s3_storage( + bucket=ERA5_BUCKET, prefix=ERA5_PREFIX, region=ERA5_REGION, anonymous=True) + repo = icechunk.Repository.open(storage) + return xr.open_zarr(repo.readonly_session("main").store, + group=ERA5_GROUP, consolidated=False) + + +class _Sink: + """Writes lake objects to the bucket (boto3) or a local mirror dir.""" + + def __init__(self): + self.local = era5lake.local_dir() + self.cfg = era5lake.s3_config() + if not self.local and self.cfg is None: + sys.exit("no destination: set THERMOGRAPH_LAKE_S3_* or " + "THERMOGRAPH_LAKE_LOCAL_DIR") + self._s3 = None + if not self.local: + import boto3 # noqa: PLC0415 - extractor-only dep + self._s3 = boto3.client( + "s3", endpoint_url=self.cfg["endpoint"], + region_name=self.cfg["region"], + aws_access_key_id=self.cfg["access_key"], + aws_secret_access_key=self.cfg["secret_key"]) + + def put(self, key: str, df: pl.DataFrame) -> None: + if self.local: + path = os.path.join(self.local, key) + os.makedirs(os.path.dirname(path), exist_ok=True) + df.write_parquet(path) + return + buf = io.BytesIO() + df.write_parquet(buf) + self._s3.put_object(Bucket=self.cfg["bucket"], Key=key, + Body=buf.getvalue()) + + def read_manifest(self) -> "pl.DataFrame | None": + try: + if self.local: + return pl.read_parquet( + os.path.join(self.local, era5lake.MANIFEST_KEY)) + obj = self._s3.get_object(Bucket=self.cfg["bucket"], + Key=era5lake.MANIFEST_KEY) + return pl.read_parquet(io.BytesIO(obj["Body"].read())) + except Exception: # noqa: BLE001 - a fresh lake has no manifest yet + return None + + +def _tile_frame(ds, ti: int, tj: int, end: str) -> "pl.DataFrame | None": + """Pull one 12x12 tile's hourly series and return the finalized daily frame + for all its points (lat_idx, lon_idx, date, ...). None for all-sea tiles.""" + t = era5lake.TILE + i0, j0 = ti * t, tj * t + lat_sl = slice(i0, min(i0 + t, era5lake.LAT_N)) + lon_sl = slice(j0, j0 + t) + if "lsm" in ds: # land-sea mask: skip all-sea tiles outright + lsm = ds["lsm"].isel(latitude=lat_sl, longitude=lon_sl) + if "valid_time" in lsm.dims: # static field, stored with a time axis + lsm = lsm.isel(valid_time=-1) + if float(lsm.max()) < 0.05: + return None + sub = (ds[list(ERA5_VARS)] + .isel(latitude=lat_sl, longitude=lon_sl) + .sel(valid_time=slice(START_DATE, end)) + .load()) + + frames = [] + for a, lat in enumerate(sub.latitude.values): + for b, lon in enumerate(sub.longitude.values): + cols = {"time": sub.valid_time.values} + for src, dst in ERA5_VARS.items(): + cols[dst] = sub[src].isel(latitude=a, longitude=b).values + daily = climate._finalize_approximated(hourly_to_daily(pl.DataFrame(cols))) + i, j = era5lake.to_idx(float(lat), float(lon)) + frames.append(daily.with_columns( + pl.lit(i, dtype=pl.Int32).alias("lat_idx"), + pl.lit(j, dtype=pl.Int32).alias("lon_idx"))) + return pl.concat(frames) + + +def _write_tile(sink: _Sink, ti: int, tj: int, tile_df: pl.DataFrame) -> pl.DataFrame: + """Write one tile's point files + hive partitions; return its manifest rows.""" + rows = [] + for (i, j), pdf in tile_df.group_by(["lat_idx", "lon_idx"]): + pdf = pdf.sort("date") + sink.put(era5lake.point_key(i, j), pdf.drop(["lat_idx", "lon_idx"])) + lat, lon = era5lake.to_coords(i, j) + rows.append({"lat_idx": i, "lon_idx": j, "lat": lat, "lon": lon, + "tile": f"{ti}_{tj}", "rows": pdf.height, + "date_min": str(pdf["date"].min()), + "date_max": str(pdf["date"].max())}) + parts = tile_df.with_columns( + pl.col("date").dt.year().alias("year"), + pl.col("date").dt.month().alias("month")) + for (year, month), mdf in parts.group_by(["year", "month"]): + sink.put(era5lake.daily_part_key(ti, tj, year, month), + mdf.drop(["year", "month"]).sort(["lat_idx", "lon_idx", "date"])) + return pl.DataFrame(rows) + + +def _city_tiles() -> list: + from data import grid # noqa: PLC0415 + + tiles = set() + for c in cities.all_cities(): + cell = grid.snap(c["lat"], c["lon"]) + tiles.add(era5lake.tile_of( + *era5lake.to_idx(cell["center_lat"], cell["center_lon"]))) + return sorted(tiles) + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--cities", action="store_true") + ap.add_argument("--land", action="store_true") + ap.add_argument("--tiles", help="ti0:ti1,tj0:tj1 tile-index window") + ap.add_argument("--limit", type=int) + ap.add_argument("--overwrite", action="store_true") + ap.add_argument("--dry-run", action="store_true") + args = ap.parse_args() + + if args.cities: + todo = _city_tiles() + elif args.tiles: + (a, b), (c, d) = (p.split(":") for p in args.tiles.split(",")) + todo = [(ti, tj) for ti in range(int(a), int(b)) + for tj in range(int(c), int(d))] + elif args.land: + todo = [(ti, tj) for ti in range(era5lake.LAT_N // era5lake.TILE + 1) + for tj in range(era5lake.LON_N // era5lake.TILE)] + else: + ap.error("pick one of --cities / --tiles / --land") + if args.limit: + todo = todo[:args.limit] + + sink = _Sink() + ds = _open_store() + end = (datetime.date.today() + - datetime.timedelta(days=climate.ARCHIVE_LATENCY_DAYS)).isoformat() + + manifest = sink.read_manifest() + done = set() if manifest is None or args.overwrite else \ + set(manifest["tile"].unique().to_list()) + + written = skipped = empty = 0 + for n, (ti, tj) in enumerate(todo, 1): + if f"{ti}_{tj}" in done: + skipped += 1 + continue + t0 = time.time() + tile_df = _tile_frame(ds, ti, tj, end) + if tile_df is None: + empty += 1 + continue + if args.dry_run: + print(f"[dry-run] tile {ti}_{tj}: {tile_df.height} rows " + f"({time.time() - t0:.1f}s); nothing written") + print(tile_df.head()) + return + rows = _write_tile(sink, ti, tj, tile_df) + manifest = rows if manifest is None else pl.concat( + [manifest.filter(pl.col("tile") != f"{ti}_{tj}"), rows]) + sink.put(era5lake.MANIFEST_KEY, manifest) # after every tile: resumable + written += 1 + print(f"[{n}/{len(todo)}] tile {ti}_{tj}: {rows.height} points " + f"in {time.time() - t0:.1f}s") + print(f"done: tiles written={written} skipped(done)={skipped} sea={empty}") + + +if __name__ == "__main__": + main() diff --git a/backend/lake_app.py b/backend/lake_app.py new file mode 100644 index 0000000..6618b35 --- /dev/null +++ b/backend/lake_app.py @@ -0,0 +1,176 @@ +"""The lake service: low-latency SQL-ish reads over the ERA5 bucket (prod-only). + +A small FastAPI app run as the Swarm ``lake`` service (same backend image, +``THERMOGRAPH_ROLE=lake`` — the entrypoint launches this module instead of the +web app; no database, no migrations). Web replicas call it instead of touching +the bucket themselves, so a cell's history costs one LAN round-trip against a +warm disk cache instead of an S3 fetch, and analytical queries get +partition-pruned scans without shipping S3 credentials to every web task. + +Endpoints: +- ``GET /healthz`` liveness (also reports manifest row count) +- ``GET /history?lat=&lon=`` the hot path: nearest ERA5 point's full daily + history as parquet bytes, disk-cached per point + (a plain single-file read — no engine needed) +- ``POST /query {"sql": ..}`` SELECT-only SQL over two views: ``era5_daily`` + (the hive table, partition-pruned on + tile/year/month predicates) and ``manifest``. + Runs on DuckDB — a real lake engine (full SQL, + parallel scans, hive + row-group pushdown); the + httpfs extension is baked into the image so S3 + needs no runtime extension fetch. + +Scale-out safe: replicas share nothing — each keeps its own cache under +THERMOGRAPH_LAKE_CACHE (a per-task local volume), so 1..2 replicas behind the +Swarm VIP need no coordination. +""" +import os +import re +import threading +import time + +import polars as pl +from fastapi import FastAPI, HTTPException +from fastapi.responses import FileResponse +from pydantic import BaseModel + +from data import era5lake + +CACHE_DIR = os.environ.get("THERMOGRAPH_LAKE_CACHE", "/state/lake-cache") +MANIFEST_TTL = 6 * 3600 # re-pull the manifest at most this often +QUERY_ROW_CAP = 10_000 # /query responses are JSON; keep them bounded + +app = FastAPI(title="thermograph-lake") + +_manifest_lock = threading.Lock() +_manifest: "pl.DataFrame | None" = None +_manifest_at = 0.0 + + +def _source(): + """(mode, cfg) for this process: a local lake dir (tests/dev) or the bucket.""" + if era5lake.local_dir(): + return "local", None + cfg = era5lake.s3_config() + return ("bucket", cfg) if cfg else ("none", None) + + +def _manifest_frame() -> "pl.DataFrame | None": + global _manifest, _manifest_at + with _manifest_lock: + if _manifest is not None and time.time() - _manifest_at < MANIFEST_TTL: + return _manifest + mode, cfg = _source() + try: + if mode == "local": + df = pl.read_parquet( + os.path.join(era5lake.local_dir(), era5lake.MANIFEST_KEY)) + elif mode == "bucket": + df = pl.read_parquet( + f"s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}", + storage_options=era5lake.storage_options(cfg)) + else: + return None + except Exception: # noqa: BLE001 - empty/unreachable lake: serve health, 404 reads + return None + _manifest, _manifest_at = df, time.time() + return df + + +@app.get("/healthz") +def healthz(): + mode, _ = _source() + m = _manifest_frame() + return {"status": "ok", "source": mode, + "points": None if m is None else m.height} + + +@app.get("/history") +def history(lat: float, lon: float): + """One ERA5 point's full daily history, as parquet bytes (cached on disk — + the cache file IS the response body, so a warm hit is a sendfile).""" + i, j = era5lake.to_idx(lat, lon) + path = os.path.join(CACHE_DIR, f"p{i}_{j}.parquet") + if not os.path.exists(path): + mode, cfg = _source() + if mode == "none": + raise HTTPException(status_code=503, detail="lake not configured") + m = _manifest_frame() + if m is not None and m.filter( + (pl.col("lat_idx") == i) & (pl.col("lon_idx") == j)).is_empty(): + raise HTTPException(status_code=404, detail="point not in the lake") + try: + if mode == "local": + df = era5lake._read_local(i, j) + else: + df = era5lake._read_bucket(i, j, cfg) + except era5lake.LakeUnavailable as e: + raise HTTPException(status_code=404, detail=str(e)) from e + os.makedirs(CACHE_DIR, exist_ok=True) + tmp = f"{path}.{os.getpid()}.partial" + df.write_parquet(tmp) + os.replace(tmp, path) # atomic: concurrent requests see whole files only + return FileResponse(path, media_type="application/vnd.apache.parquet", + headers={"X-Lake-Point": f"{i},{j}"}) + + +class Query(BaseModel): + sql: str + + +_FORBIDDEN = re.compile( + r"\b(insert|update|delete|drop|create|alter|attach|copy|truncate|install|load" + r"|export|import|pragma|set|call)\b", re.I) + + +def _connect(): + """A fresh in-memory DuckDB with the lake views registered. Per-request: + connections are cheap, and a throwaway one means a query can't leave state + behind for the next. Returns None when no lake is configured.""" + import duckdb # noqa: PLC0415 - only the lake role pays the import + + mode, cfg = _source() + if mode == "local": + daily_glob = os.path.join(era5lake.local_dir(), era5lake.PREFIX, + "daily/*/*/*/*.parquet") + manifest_src = os.path.join(era5lake.local_dir(), era5lake.MANIFEST_KEY) + elif mode == "bucket": + daily_glob = f"s3://{cfg['bucket']}/{era5lake.PREFIX}/daily/*/*/*/*.parquet" + manifest_src = f"s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}" + else: + return None + con = duckdb.connect() + if mode == "bucket": + con.execute("LOAD httpfs") # baked into the image at build time + host = cfg["endpoint"].split("://", 1)[-1] + con.execute(f"SET s3_endpoint='{host}'") + con.execute("SET s3_url_style='path'") + con.execute(f"SET s3_region='{cfg['region']}'") + con.execute("SET s3_access_key_id=?", [cfg["access_key"]]) + con.execute("SET s3_secret_access_key=?", [cfg["secret_key"]]) + con.execute(f""" + CREATE VIEW era5_daily AS + SELECT * FROM read_parquet('{daily_glob}', hive_partitioning=1)""") + con.execute(f"CREATE VIEW manifest AS SELECT * FROM read_parquet('{manifest_src}')") + return con + + +@app.post("/query") +def query(q: Query): + sql = q.sql.strip().rstrip(";").strip() + if not sql.lower().startswith(("select", "with")) or ";" in sql \ + or _FORBIDDEN.search(sql): + raise HTTPException(status_code=400, + detail="only a single SELECT statement is allowed") + con = _connect() + if con is None: + raise HTTPException(status_code=503, detail="lake not configured") + try: + out = con.execute( + f"SELECT * FROM ({sql}) LIMIT {QUERY_ROW_CAP}").fetchall() + cols = [d[0] for d in con.description] + except Exception as e: # noqa: BLE001 - surface the engine's message to the caller + raise HTTPException(status_code=400, detail=f"query failed: {e}") from e + finally: + con.close() + return {"columns": cols, "rows": out, "row_cap": QUERY_ROW_CAP} diff --git a/backend/requirements-seed.txt b/backend/requirements-seed.txt index 5791fa4..5c33726 100644 --- a/backend/requirements-seed.txt +++ b/backend/requirements-seed.txt @@ -7,3 +7,10 @@ icechunk xarray zarr +# The live store compresses with PCodec; without this extra every read fails +# with "codec not available: 'pcodec'". +numcodecs[pcodec] +# gen_era5_lake.py's bucket writer. +boto3 +numcodecs[pcodec] +boto3 diff --git a/backend/requirements.txt b/backend/requirements.txt index 62fb0fc..127dc27 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -25,3 +25,6 @@ websockets==16.0 # Recurring worker-tier jobs — city warming, IndexNow pings (see # notifications/scheduler.py). In-process; no broker/queue needed at this scale. apscheduler==3.11.3 +# The lake role's SQL engine (lake_app.py /query): embedded, parallel +# partition-pruned scans over the ERA5 hive table. Only lake_app imports it. +duckdb==1.4.2 diff --git a/backend/seed_era5.py b/backend/seed_era5.py index 03be936..4d8e1b5 100644 --- a/backend/seed_era5.py +++ b/backend/seed_era5.py @@ -16,10 +16,10 @@ Idempotent by default: a cell that already has a cached history is skipped (pass --overwrite to reseed). Not run in CI or the app image — the icechunk/xarray/zarr stack lives only in requirements-seed.txt. -IMPORTANT — the S3/Icechunk *access* layer (_open_store / fetch_point) targets a -young, fast-moving API and the store's exact variable/coordinate names; it is NOT -exercised by the test suite. VERIFY it with `--dry-run` on the seed box before a -full run. The hourly→daily transform (hourly_to_daily) IS unit-tested and is the +The access layer was VERIFIED against the live store 2026-07-23 (prefix +`icechunkV2`, group `single/temporal`, coord `valid_time`, ECMWF short variable +names, and the pcodec codec — install `numcodecs[pcodec]`). Re-verify with +`--dry-run` if the store revs. The hourly→daily transform (hourly_to_daily) IS unit-tested and is the part that must be numerically correct. Source: https://registry.opendata.aws/earthmover-era5/ (group="temporal", anonymous) @@ -38,19 +38,19 @@ START_DATE = climate.START_DATE # "1980-01-01" # Anonymous AWS Open Data Icechunk store; overridable in case the registry moves. ERA5_BUCKET = os.environ.get("THERMOGRAPH_ERA5_BUCKET", "earthmover-icechunk-era5") -ERA5_PREFIX = os.environ.get("THERMOGRAPH_ERA5_PREFIX", "") # verify on the seed box +ERA5_PREFIX = os.environ.get("THERMOGRAPH_ERA5_PREFIX", "icechunkV2") ERA5_REGION = os.environ.get("THERMOGRAPH_ERA5_REGION", "us-east-1") -ERA5_GROUP = os.environ.get("THERMOGRAPH_ERA5_GROUP", "temporal") # time-contiguous layout +ERA5_GROUP = os.environ.get("THERMOGRAPH_ERA5_GROUP", "single/temporal") # time-contiguous layout # ERA5 store variable name -> our short name. ERA5 gusts (i10fg) are real here, so the # seed keeps genuine measured gusts rather than the Meteostat/estimate fallback. ERA5_VARS = { - "2m_temperature": "t2m", # K - "2m_dewpoint_temperature": "d2m", # K (→ RH with t2m) - "total_precipitation": "tp", # m, hourly accumulation - "10m_u_component_of_wind": "u10", # m/s - "10m_v_component_of_wind": "v10", # m/s - "instantaneous_10m_wind_gust": "i10fg", # m/s + "t2m": "t2m", # 2m temperature, K + "d2m": "d2m", # 2m dewpoint, K (→ RH with t2m) + "tp": "tp", # total precipitation, m, hourly accumulation + "u10": "u10", # 10m wind u, m/s + "v10": "v10", # 10m wind v, m/s + "fg10": "i10fg", # 10m wind gust, m/s (store short name; transform expects i10fg) } _M_TO_IN = 39.37007874 @@ -117,9 +117,9 @@ def fetch_point(ds, lat: float, lon: float, start: str, end: str) -> pl.DataFram may differ per store revision — verify with --dry-run.""" sub = (ds[list(ERA5_VARS)] .sel(latitude=lat, longitude=lon % 360.0, method="nearest") - .sel(time=slice(start, end)) + .sel(valid_time=slice(start, end)) .load()) - data = {"time": sub["time"].values} + data = {"time": sub["valid_time"].values} for src, dst in ERA5_VARS.items(): data[dst] = sub[src].values return pl.DataFrame(data) diff --git a/backend/tests/data/test_climate.py b/backend/tests/data/test_climate.py index 79c3d71..95e869f 100644 --- a/backend/tests/data/test_climate.py +++ b/backend/tests/data/test_climate.py @@ -311,6 +311,44 @@ def _full_history_frame(n=4000): })) +def test_lake_is_first_history_source(monkeypatch, tmp_path): + """A configured ERA5 lake short-circuits the whole third-party chain: neither + NASA POWER nor Open-Meteo is consulted.""" + monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) + monkeypatch.setattr(climate, "_archive_cooldown_until", 0.0) + cell = {"id": "lake_primary", "center_lat": 47.6, "center_lon": -122.3} + + full = _full_history_frame() + monkeypatch.setattr(climate.era5lake, "fetch_history_lake", + lambda c, start=None: full) + def _nasa_should_not_run(c): raise AssertionError("NASA should not be called") + def _om_should_not_run(c): raise AssertionError("Open-Meteo should not be called") + monkeypatch.setattr(climate, "_fetch_history_nasa", _nasa_should_not_run) + monkeypatch.setattr(climate, "_fetch_history", _om_should_not_run) + + df, meta = climate._load_history(cell) + assert meta["source"] == "era5-lake" + assert df.height == full.height + + +def test_lake_miss_falls_through_to_nasa(monkeypatch, tmp_path): + """An unconfigured lake (or a point outside it) costs nothing: NASA serves + exactly as before the lake existed.""" + monkeypatch.setattr(climate, "CACHE_DIR", str(tmp_path)) + monkeypatch.setattr(climate, "_archive_cooldown_until", 0.0) + cell = {"id": "lake_miss", "center_lat": 47.6, "center_lon": -122.3} + + def _no_lake(c, start=None): + raise climate.era5lake.LakeUnavailable("no lake configured") + monkeypatch.setattr(climate.era5lake, "fetch_history_lake", _no_lake) + full = _full_history_frame() + monkeypatch.setattr(climate, "_fetch_history_nasa", lambda c: full) + + df, meta = climate._load_history(cell) + assert meta["source"] == "nasa-power" + assert df.height == full.height + + def test_nasa_is_primary_history_source(monkeypatch, tmp_path): """NASA POWER is the primary history source: a full NASA frame is accepted and Open-Meteo is never consulted.""" diff --git a/backend/tests/data/test_era5lake.py b/backend/tests/data/test_era5lake.py new file mode 100644 index 0000000..198a5cf --- /dev/null +++ b/backend/tests/data/test_era5lake.py @@ -0,0 +1,90 @@ +"""The ERA5 lake: grid math, layout keys, and the read/slice client (no network +— the local-dir source stands in for the bucket).""" +import datetime + +import polars as pl +import pytest + +from data import era5lake + + +def _daily_frame(start="1940-01-01", days=32000): + d0 = datetime.date.fromisoformat(start) + dates = [d0 + datetime.timedelta(days=k) for k in range(days)] + n = len(dates) + return pl.DataFrame({ + "date": dates, + "tmax": [70.0] * n, "tmin": [50.0] * n, "precip": [0.1] * n, + "wind": [8.0] * n, "gust": [14.0] * n, "humid": [55.0] * n, + "fmax": [71.0] * n, "fmin": [48.0] * n, + }) + + +def test_grid_roundtrip_and_bounds(): + # Nearest-index snap, poles clamped, longitude wrapped to [0, 360). + assert era5lake.to_idx(90.0, 0.0) == (0, 0) + assert era5lake.to_idx(-90.0, 0.0) == (720, 0) + assert era5lake.to_idx(51.5074, -0.1278) == (154, 1439) + i, j = era5lake.to_idx(35.6762, 139.6503) + lat, lon = era5lake.to_coords(i, j) + assert abs(lat - 35.6762) <= era5lake.STEP / 2 + 1e-9 + assert abs(lon - 139.6503) <= era5lake.STEP / 2 + 1e-9 + + +def test_tile_and_keys_align_with_hive_layout(): + assert era5lake.tile_of(154, 1439) == (12, 119) + assert era5lake.point_key(154, 1439) == "era5/points/lat=154/lon=1439.parquet" + assert era5lake.daily_part_key(12, 119, 1994, 3) == \ + "era5/daily/tile=12_119/year=1994/month=03/part.parquet" + + +def test_s3_config_requires_both_keys(monkeypatch): + monkeypatch.delenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", raising=False) + monkeypatch.delenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", raising=False) + assert era5lake.s3_config() is None + monkeypatch.setenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", "k") + assert era5lake.s3_config() is None + monkeypatch.setenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", "s") + cfg = era5lake.s3_config() + assert cfg["bucket"] == "era5-thermograph" + opts = era5lake.storage_options(cfg) + assert opts["aws_virtual_hosted_style_request"] == "false" + + +def _write_point(tmp_path, cell, df): + i, j = era5lake.to_idx(cell["center_lat"], cell["center_lon"]) + path = tmp_path / era5lake.point_key(i, j) + path.parent.mkdir(parents=True, exist_ok=True) + df.write_parquet(path) + + +CELL = {"center_lat": 51.5, "center_lon": -0.125} + + +def test_fetch_slices_to_start_and_keeps_schema(monkeypatch, tmp_path): + monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path)) + _write_point(tmp_path, CELL, _daily_frame()) + df = era5lake.fetch_history_lake(CELL, start="1980-01-01") + assert df["date"].min() == datetime.date(1980, 1, 1) + assert set(df.columns) == {"date", "tmax", "tmin", "precip", "wind", + "gust", "humid", "fmax", "fmin"} + # Without a start, the whole 1940+ record comes back. + assert era5lake.fetch_history_lake(CELL).height == 32000 + + +def test_fetch_rejects_short_and_missing(monkeypatch, tmp_path): + monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path)) + with pytest.raises(era5lake.LakeUnavailable): + era5lake.fetch_history_lake(CELL) # nothing written + _write_point(tmp_path, CELL, _daily_frame(days=500)) + with pytest.raises(era5lake.LakeUnavailable): # implausibly short record + era5lake.fetch_history_lake(CELL) + + +def test_fetch_unconfigured_raises(monkeypatch): + monkeypatch.delenv("THERMOGRAPH_LAKE_LOCAL_DIR", raising=False) + monkeypatch.delenv("THERMOGRAPH_LAKE_URL", raising=False) + monkeypatch.delenv("THERMOGRAPH_LAKE_S3_ACCESS_KEY", raising=False) + monkeypatch.delenv("THERMOGRAPH_LAKE_S3_SECRET_KEY", raising=False) + with pytest.raises(era5lake.LakeUnavailable): + era5lake.fetch_history_lake(CELL) diff --git a/backend/tests/web/test_lake_app.py b/backend/tests/web/test_lake_app.py new file mode 100644 index 0000000..ea832c8 --- /dev/null +++ b/backend/tests/web/test_lake_app.py @@ -0,0 +1,100 @@ +"""The lake service over a local-dir lake fixture: health, the point hot path +(+ its disk cache), and the SELECT-only SQL surface.""" +import datetime +import io + +import polars as pl +import pytest +from fastapi.testclient import TestClient + + +@pytest.fixture +def lake(tmp_path, monkeypatch): + """A tiny two-point lake (points, hive partitions, manifest) on disk.""" + from data import era5lake + + points = [(154, 1439), (154, 0)] # around London + d0 = datetime.date(1940, 1, 1) + dates = [d0 + datetime.timedelta(days=k) for k in range(12000)] + n = len(dates) + rows = [] + for i, j in points: + df = pl.DataFrame({ + "date": dates, + "tmax": [70.0] * n, "tmin": [50.0] * n, "precip": [0.1] * n, + "wind": [8.0] * n, "gust": [14.0] * n, "humid": [55.0] * n, + "fmax": [71.0] * n, "fmin": [48.0] * n, + }) + p = tmp_path / era5lake.point_key(i, j) + p.parent.mkdir(parents=True, exist_ok=True) + df.write_parquet(p) + ti, tj = era5lake.tile_of(i, j) + part = df.with_columns(pl.lit(i, dtype=pl.Int32).alias("lat_idx"), + pl.lit(j, dtype=pl.Int32).alias("lon_idx")) + hp = tmp_path / era5lake.daily_part_key(ti, tj, 1940, 1) + hp.parent.mkdir(parents=True, exist_ok=True) + part.filter((pl.col("date").dt.year() == 1940) + & (pl.col("date").dt.month() == 1)).write_parquet(hp) + lat, lon = era5lake.to_coords(i, j) + rows.append({"lat_idx": i, "lon_idx": j, "lat": lat, "lon": lon, + "tile": f"{ti}_{tj}", "rows": n, + "date_min": "1940-01-01", "date_max": str(dates[-1])}) + (tmp_path / "era5").mkdir(exist_ok=True) + pl.DataFrame(rows).write_parquet(tmp_path / era5lake.MANIFEST_KEY) + + monkeypatch.setenv("THERMOGRAPH_LAKE_LOCAL_DIR", str(tmp_path)) + monkeypatch.setenv("THERMOGRAPH_LAKE_CACHE", str(tmp_path / "cache")) + import lake_app + monkeypatch.setattr(lake_app, "CACHE_DIR", str(tmp_path / "cache")) + monkeypatch.setattr(lake_app, "_manifest", None) + return TestClient(lake_app.app) + + +def test_healthz_reports_points(lake): + body = lake.get("/healthz").json() + assert body["status"] == "ok" + assert body["source"] == "local" + assert body["points"] == 2 + + +def test_history_serves_parquet_and_caches(lake, tmp_path): + r = lake.get("/history", params={"lat": 51.5074, "lon": -0.1278}) + assert r.status_code == 200 + assert r.headers["x-lake-point"] == "154,1439" + df = pl.read_parquet(io.BytesIO(r.content)) + assert df.height == 12000 + assert (tmp_path / "cache" / "p154_1439.parquet").exists() + # Second hit comes off the cache file and is byte-identical. + assert lake.get("/history", + params={"lat": 51.5074, "lon": -0.1278}).content == r.content + + +def test_history_404_for_point_outside_lake(lake): + r = lake.get("/history", params={"lat": -45.0, "lon": 170.0}) + assert r.status_code == 404 + + +def test_query_selects_with_partition_predicates(lake): + r = lake.post("/query", json={"sql": + "SELECT year, month, count(*) AS n FROM era5_daily " + "WHERE year = 1940 AND month = 1 GROUP BY year, month"}) + assert r.status_code == 200 + body = r.json() + assert body["columns"] == ["year", "month", "n"] + assert body["rows"][0][2] == 2 * 31 # two points x January 1940 + + +def test_query_manifest_view(lake): + r = lake.post("/query", json={"sql": "SELECT count(*) FROM manifest"}) + assert r.status_code == 200 + assert r.json()["rows"][0][0] == 2 + + +@pytest.mark.parametrize("sql", [ + "DROP TABLE era5_daily", + "SELECT 1; SELECT 2", + "INSERT INTO era5_daily VALUES (1)", + "COPY era5_daily TO 'x'", +]) +def test_query_rejects_non_select(lake, sql): + assert lake.post("/query", json={"sql": sql}).status_code == 400 diff --git a/infra/deploy/secrets/README.md b/infra/deploy/secrets/README.md index 3fc627d..afc09a6 100644 --- a/infra/deploy/secrets/README.md +++ b/infra/deploy/secrets/README.md @@ -15,7 +15,7 @@ no per-host duplication. | File | Holds | Encrypted? | |------|-------|------------| | `common.yaml` | Secrets shared by every host — Discord tokens/secret, VAPID keypair, `THERMOGRAPH_AUTH_SECRET`, metrics token, indexnow key, SMTP creds | ✅ | -| `prod.yaml` | Prod-only overrides — `POSTGRES_PASSWORD`, `REGISTRY_TOKEN` | ✅ | +| `prod.yaml` | Prod-only overrides — `POSTGRES_PASSWORD`, `REGISTRY_TOKEN`, ERA5 lake bucket creds (`THERMOGRAPH_LAKE_S3_ACCESS_KEY` / `THERMOGRAPH_LAKE_S3_SECRET_KEY`, consumed by the `lake` Swarm service and `gen_era5_lake.py`) | ✅ | | `beta.yaml` | Beta-only overrides | ✅ | | `example.yaml` | Format reference / CI fixture (fake values) | ✅ | | `../../.sops.yaml` | Which age recipient files are encrypted to (plaintext config) | — | diff --git a/infra/deploy/stack/autoscale.sh b/infra/deploy/stack/autoscale.sh index 2a36ade..0431ddc 100755 --- a/infra/deploy/stack/autoscale.sh +++ b/infra/deploy/stack/autoscale.sh @@ -21,7 +21,9 @@ set -eu STACK_NAME="${STACK_NAME:-thermograph}" -SERVICE="${STACK_NAME}_web" +# Which stack service to scale (the stack runs one autoscaler instance per +# scaled service — web and lake today; same loop, different bounds). +SERVICE="${STACK_NAME}_${TARGET_SERVICE:-web}" MIN="${MIN_REPLICAS:-1}" MAX="${MAX_REPLICAS:-3}" UP_AT="${SCALE_UP_CPU:-220}" diff --git a/infra/deploy/stack/thermograph-stack.yml b/infra/deploy/stack/thermograph-stack.yml index b34f77b..afb3c3f 100644 --- a/infra/deploy/stack/thermograph-stack.yml +++ b/infra/deploy/stack/thermograph-stack.yml @@ -82,6 +82,10 @@ services: # not the compose bridge's 172.19.0.1 (an overlay has no host gateway). # provision-mail.sh's DOCKER_MAIL_GATEWAY/SUBNET cover this listener. THERMOGRAPH_SMTP_HOST: ${STACK_SMTP_HOST:-172.18.0.1} + # History reads ask the lake service (Swarm VIP over 1..2 replicas) + # before any third-party source; unset on beta/LAN, where compose has no + # lake service and climate.py falls straight through to NASA. + THERMOGRAPH_LAKE_URL: http://lake:8141 volumes: - appdata:/state - applogs:/app/logs @@ -121,6 +125,7 @@ services: THERMOGRAPH_SINGLETON_PG: "1" RUN_MIGRATIONS: "0" THERMOGRAPH_SMTP_HOST: ${STACK_SMTP_HOST:-172.18.0.1} + THERMOGRAPH_LAKE_URL: http://lake:8141 volumes: - appdata:/state - applogs:/app/logs @@ -138,6 +143,45 @@ services: restart_policy: condition: on-failure + # The ERA5 lake indexer: SQL-style, partition-pruned reads over the + # era5-thermograph bucket with a node-local parquet cache, so a cell's + # history is one LAN hop for web instead of an S3 fetch (or a NASA call). + # Same backend image; THERMOGRAPH_ROLE=lake makes the entrypoint start + # lake_app (no database, no migrations). Prod-only by construction: it only + # exists in this stack file, and its S3 credentials + # (THERMOGRAPH_LAKE_S3_ACCESS_KEY/_SECRET_KEY) arrive via the same rendered + # /host/thermograph.env as every other secret. Without them it stays + # healthy and answers 503/404, and web falls through to NASA — the lake is + # an accelerator, never a point of failure. + lake: + image: ${REGISTRY_HOST:-git.thermograph.org}/${BACKEND_IMAGE_PATH:-emi/thermograph/backend}:${BACKEND_IMAGE_TAG:?required} + entrypoint: ["/host/env-entrypoint.sh"] + environment: + THERMOGRAPH_ROLE: lake + PORT: 8141 + THERMOGRAPH_SERVICE_ROLE: backend + WORKERS: "1" + THERMOGRAPH_LAKE_CACHE: /state/lake-cache + volumes: + # Replicas share the cache; lake_app's tmp+rename writes make that safe. + - lakecache:/state + - /opt/thermograph/infra/deploy/stack/env-entrypoint.sh:/host/env-entrypoint.sh:ro + - /etc/thermograph/stack.env:/host/thermograph.env:ro + networks: + - internal + deploy: + replicas: ${LAKE_MIN_REPLICAS:-1} + placement: + constraints: ["node.role == manager"] + resources: + limits: + cpus: "${LAKE_CPUS:-2}" + restart_policy: + condition: on-failure + update_config: + order: start-first + failure_action: rollback + frontend: image: ${REGISTRY_HOST:-git.thermograph.org}/${FRONTEND_IMAGE_PATH:-emi/thermograph/frontend}:${FRONTEND_IMAGE_TAG:?required} entrypoint: ["/host/env-entrypoint.sh"] @@ -200,6 +244,39 @@ services: restart_policy: condition: any + # Second autoscaler instance, watching `lake` (1..2 replicas). The lake is + # read-only and cache-heavy, so it saturates on decode CPU well below web's + # thresholds — scale up sooner, still down slowly. + autoscaler-lake: + image: docker:27-cli + entrypoint: ["/bin/sh", "/host/autoscale.sh"] + environment: + STACK_NAME: ${STACK_NAME:-thermograph} + TARGET_SERVICE: lake + MIN_REPLICAS: ${LAKE_MIN_REPLICAS:-1} + MAX_REPLICAS: ${LAKE_MAX_REPLICAS:-2} + SCALE_UP_CPU: ${LAKE_SCALE_UP_CPU:-120} + SCALE_DOWN_CPU: ${LAKE_SCALE_DOWN_CPU:-30} + POLL_SECONDS: ${POLL_SECONDS:-15} + UP_SAMPLES: ${UP_SAMPLES:-3} + DOWN_SAMPLES: ${DOWN_SAMPLES:-20} + COOLDOWN_SECONDS: ${COOLDOWN_SECONDS:-180} + volumes: + - /var/run/docker.sock:/var/run/docker.sock + - /opt/thermograph/infra/deploy/stack/autoscale.sh:/host/autoscale.sh:ro + networks: + - internal + deploy: + replicas: 1 + placement: + constraints: ["node.role == manager"] + resources: + limits: + cpus: "0.2" + memory: 64m + restart_policy: + condition: any + networks: internal: driver: overlay @@ -221,3 +298,6 @@ volumes: applogs: external: true name: ${APPLOGS_VOLUME:-thermograph_applogs} + # Lake parquet cache: node-local and disposable (a cold cache just refills + # from the bucket), so NOT external — the stack owns its lifecycle. + lakecache: {}