Logs, Metrics and Alerts: Observability for GIS Pipelines
On this page
Problem statement
Two ways a scheduled GIS job goes wrong, and neither is visible:
$ tail -3 logs/nightly.log
starting
doneThe second, reproduced on HM Land Registry's Rutland parcels (set up under Code examples): the ward boundary the job clips to comes back from its web service with no rows, the clip dutifully returns nothing, and the job writes its output and exits 0.
import os
boundary = gpd.read_file(ward_query("Oakham North Wesst")) # a one-letter typo upstream
empty_result = gpd.clip(parcels, boundary.to_crs(parcels.crs))
empty_result.to_file("data/out/parcels_empty.gpkg")
print(f"boundary rows: {len(boundary)}, output features: {len(empty_result)}, "
f"file size: {os.path.getsize('data/out/parcels_empty.gpkg'):,} bytes "
f"(a full run writes {os.path.getsize('data/raw/parcels.gpkg') / 1e6:.0f} MB)")boundary rows: 0, output features: 0, file size: 98,304 bytes (a full run writes 21 MB)
The first is a job with no observability at all: it started, it finished, and nothing in between is recorded. The second is worse — the job succeeded, the scheduler is green, and the output is empty because an upstream source returned no rows. It is not even small enough to look wrong: an empty GeoPackage is still 96 KB of tables and metadata, so a check that the file exists and is not tiny passes it.
Observability is the property of being able to answer questions about a running system from the outside. For a GIS pipeline that means three specific things: logs (what happened), metrics (how much, how long, how many), and alerts (what needs a human). Each answers a different question, and a pipeline missing any one of them has a blind spot.
Quick answer
Instrument three layers, each for a different question:
- Logs — a narrative with timestamps and levels; answers "what happened at 02:31?"
- Metrics — numbers per run: rows in and out per step, durations, bytes; answers "is this normal?"
- Alerts — the small set of conditions a human must act on; answers "do I need to get up?"
- Assert on outputs, not just on exceptions — an empty result is a failure
- Add a heartbeat, because the loudest failure is the job that never ran
import json, logging, sys, time
from datetime import datetime, timezone
from pathlib import Path
Path("logs").mkdir(exist_ok=True) # FileHandler will not create the folder
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-7s %(name)s %(message)s",
handlers=[logging.FileHandler("logs/pipeline.log", encoding="utf-8"),
logging.StreamHandler(sys.stdout)],
)
log = logging.getLogger("pipeline")
started = time.perf_counter()
metrics = {"run_id": datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"), "steps": []}
def step(name, fn, gdf, **params):
t0, before = time.perf_counter(), len(gdf)
log.info("→ %s %s", name, params or "")
out = fn(gdf, **params)
metrics["steps"].append({"step": name, "rows_in": before, "rows_out": len(out),
"seconds": round(time.perf_counter() - t0, 2)})
log.info("✓ %s: %d → %d rows in %.2fs", name, before, len(out),
metrics["steps"][-1]["seconds"])
return outThe row-count ledger is the single highest-value piece of instrumentation in a GIS pipeline: it turns "the output looks wrong" into "the clip step kept 876 parcels instead of 1,441, starting tonight".
Three layers, three questions
Logs, metrics and the alerts that deserve a human
Logs: a narrative with timestamps and levels
import logging
from pathlib import Path
def configure_logging(level="INFO", log_dir=Path("logs")) -> logging.Logger:
log_dir.mkdir(parents=True, exist_ok=True)
logging.basicConfig(
level=getattr(logging, level),
format="%(asctime)s %(levelname)-7s %(name)-12s %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S%z",
handlers=[
logging.FileHandler(log_dir / "pipeline.log", encoding="utf-8"),
logging.StreamHandler(),
],
)
logging.getLogger("fiona").setLevel(logging.WARNING) # quiet the libraries
logging.getLogger("pyogrio").setLevel(logging.WARNING)
return logging.getLogger("pipeline")Use the levels as they were intended, because alerting rules read them:
DEBUG— detail for a developer reproducing somethingINFO— the narrative: what step started, what it producedWARNING— recoverable oddities: 12 invalid geometries repaired, one file skippedERROR— this run did not do its jobCRITICAL— the system is broken, not just this run
log.info("read %s: %d features, crs=%s", path.name, len(gdf), gdf.crs)
log.warning("repaired %d invalid geometries", repaired)
log.error("clip produced 0 features — upstream boundary may be empty")
try:
step()
except Exception:
log.exception("step failed") # ERROR plus the traceback
raiselog.exception() inside an except block is the one-liner that keeps tracebacks out of stdout and inside the log, where they can be found later.
Structured logs, when something else will read them
import json, logging
class JsonFormatter(logging.Formatter):
def format(self, record):
payload = {
"ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
if hasattr(record, "extra_fields"):
payload.update(record.extra_fields)
if record.exc_info:
payload["exception"] = self.formatException(record.exc_info)
return json.dumps(payload)
handler = logging.FileHandler("logs/pipeline.jsonl", encoding="utf-8")
handler.setFormatter(JsonFormatter())
log.addHandler(handler)
log.info("step complete", extra={"extra_fields": {"step": "clip", "rows_out": 4106}})JSON Lines is greppable, loadable as a DataFrame, and ingestible by Loki, Elasticsearch or CloudWatch without a parser. Keep the human-readable handler too — you will read that one at 09:00.
Metrics: the numbers that make a run comparable
import time
from datetime import datetime, timezone
class RunMetrics:
def __init__(self, pipeline: str):
self.data = {
"pipeline": pipeline,
"run_id": datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ"), # not just the second
"started_utc": datetime.now(timezone.utc).isoformat(timespec="seconds"),
"steps": [], "counters": {}, "status": "running",
}
self._t0 = time.perf_counter()
def step(self, name, rows_in, rows_out, seconds):
self.data["steps"].append({
"step": name, "rows_in": rows_in, "rows_out": rows_out,
"dropped": rows_in - rows_out,
"retained_pct": round(rows_out / rows_in * 100, 2) if rows_in else None,
"seconds": round(seconds, 2),
})
def count(self, name, value=1):
self.data["counters"][name] = self.data["counters"].get(name, 0) + value
def finish(self, status="ok", **summary):
self.data.update(status=status, duration_s=round(time.perf_counter() - self._t0, 2),
finished_utc=datetime.now(timezone.utc).isoformat(timespec="seconds"),
**summary)
return self.dataThe metrics worth collecting in a GIS pipeline are specific: rows in and out per step, features written, total area, feature count against the previous run, bytes written, duration per step, and counts of repaired geometries and skipped files.
Alerts: the small set that deserves a human
def alert_conditions(metrics: dict, previous: dict | None) -> list[str]:
alerts = []
if metrics["status"] != "ok":
alerts.append(f"run failed: {metrics.get('error', 'unknown error')}")
if metrics["status"] == "ok" and metrics.get("features", 0) == 0:
alerts.append("output has zero features")
if metrics.get("features") and previous and previous.get("features"): # a failed run has no count
change = metrics["features"] / previous["features"] - 1
if abs(change) > 0.2:
alerts.append(f"feature count changed {change:+.1%} "
f"({previous['features']:,} → {metrics['features']:,})")
if previous and metrics["duration_s"] > previous.get("duration_s", 0) * 3:
alerts.append(f"run took {metrics['duration_s']:.0f}s, "
f"3× longer than the last run")
usual = {s["step"]: s.get("retained_pct") for s in (previous or {}).get("steps", [])}
for step in metrics["steps"]:
before = usual.get(step["step"])
if step["retained_pct"] is not None and before is not None and abs(step["retained_pct"] - before) > 20:
alerts.append(f"step '{step['step']}' kept {step['retained_pct']:.0f}% of rows, "
f"against {before:.0f}% last run")
return alertsTwo details in there were learned the hard way, in Example 1. A failed run has no feature count, so every read of metrics["features"] must allow for its absence — otherwise the alerting code raises a KeyError inside the failure handler, the real error is replaced by the KeyError, and no alert goes out. And a step's retention is compared with the same step last time, not with a fixed 50%: clipping Rutland's parcels to one ward keeps 7% of them every night, and a fixed threshold would page someone for it every night. The discipline that keeps alerting useful is deciding, for each condition, what a human would do about it. If the answer is "nothing", it belongs in the metrics, not in an alert. A channel full of alerts nobody acts on is the same as no alerting at all.
Assert on the output, not only on exceptions
def assert_output_sane(gdf, previous_count=None) -> None:
if gdf.empty:
raise ValueError("output has zero features")
if gdf.crs is None:
raise ValueError("output has no CRS")
if gdf.geometry.isna().any():
raise ValueError(f"{gdf.geometry.isna().sum()} null geometries in output")
if not gdf.geometry.is_valid.all():
raise ValueError(f"{(~gdf.geometry.is_valid).sum()} invalid geometries in output")
if previous_count and len(gdf) < previous_count * 0.5:
raise ValueError(f"feature count halved: {previous_count} → {len(gdf)}")This is the check that catches the empty GeoPackage — by counting features, not bytes. Nothing raised, nothing crashed — the data was simply wrong, and only an expectation about the result can detect that.
The heartbeat: detecting the run that never happened
import json, time
from pathlib import Path
STATE = Path("logs/last_run.json")
def record_success(metrics: dict) -> None:
STATE.parent.mkdir(parents=True, exist_ok=True)
STATE.write_text(json.dumps(metrics, indent=2), encoding="utf-8")
def check_freshness(max_age_hours=26) -> str | None:
"""Run this from a separate, simpler schedule."""
if not STATE.exists():
return "no run has ever been recorded"
age_h = (time.time() - STATE.stat().st_mtime) / 3600
if age_h > max_age_hours:
return f"last successful run was {age_h:.1f} hours ago"
return NoneA failing job is loud. A job that never started — because the machine was off, the scheduler was disabled, or the container failed to pull — is completely silent, and only a freshness check or an external heartbeat service will notice.
Send alerts where they will be seen
import json, os, urllib.request
def send_alert(text: str, details: dict | None = None) -> bool:
url = os.environ.get("ALERT_WEBHOOK_URL")
if not url:
print(f"ALERT (no webhook configured): {text}")
return False
payload = {"text": f"🔴 {text}", "attachments": [{"text": json.dumps(details, indent=2)}]
if details else []}
request = urllib.request.Request(
url, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(request, timeout=10) as response:
return response.status < 300
except Exception as exc:
print(f"alert delivery failed: {exc}") # never let alerting break the job
return FalseNote the swallowed exception: a failure to deliver an alert must not fail the pipeline, and it must not mask the original error either — so it is logged and ignored.
Instrumenting a Rutland clipping pipeline
The data used on this page
HM Land Registry's INSPIRE parcels for Rutland, the ONS ward boundary the pipeline clips them to (fetched from the ONS web service each run, the way a real job would), and a local stand-in for the alert webhook that records what it receives.
import io
import json
import os
import threading
import zipfile
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
import geopandas as gpd
import pyogrio
import requests
pyogrio.set_gdal_config_options({"GML_DOWNLOAD_SCHEMA": "NO"})
for folder in ["data/raw", "data/out", "logs"]:
Path(folder).mkdir(parents=True, exist_ok=True)
if not Path("data/raw/parcels.gpkg").exists():
reply = requests.Session().get("https://use-land-property-data.service.gov.uk/datasets/inspire/download/"
"Rutland.zip", timeout=600)
reply.raise_for_status()
zipfile.ZipFile(io.BytesIO(reply.content)).extract("Land_Registry_Cadastral_Parcels.gml", "downloads")
gpd.read_file("downloads/Land_Registry_Cadastral_Parcels.gml").to_file("data/raw/parcels.gpkg")
parcels = gpd.read_file("data/raw/parcels.gpkg")
def ward_query(name):
"""The ONS ward boundary as GeoJSON — zero features if the name matches nothing."""
reply = requests.get("https://services1.arcgis.com/ESMARspQHYMw9BZ9/arcgis/rest/services/"
"Wards_December_2024_Boundaries_UK_BGC/FeatureServer/0/query",
params={"where": f"WD24NM='{name}'", "outFields": "WD24CD,WD24NM",
"outSR": 27700, "f": "geojson"}, timeout=120)
reply.raise_for_status()
return io.BytesIO(reply.content)
received = []
class Hook(BaseHTTPRequestHandler):
def do_POST(self):
received.append(json.loads(self.rfile.read(int(self.headers["Content-Length"]))))
self.send_response(204)
self.end_headers()
def log_message(self, *args):
pass
hook = HTTPServer(("127.0.0.1", 0), Hook)
threading.Thread(target=hook.serve_forever, daemon=True).start()
os.environ["ALERT_WEBHOOK_URL"] = f"http://127.0.0.1:{hook.server_port}/alerts"
print(f"{len(parcels):,} parcels; webhook stand-in on port {hook.server_port}")21,253 parcels; webhook stand-in on port 35345
Example 1: an instrumented pipeline
The steps are real: drop parcels without geometry, clip to a ward fetched from the ONS, and add each parcel's area.
import time
def clean(gdf, config):
return gdf[gdf.geometry.notna() & ~gdf.geometry.is_empty]
def clip_to(gdf, config):
boundary = gpd.read_file(ward_query(config["ward"])).to_crs(gdf.crs)
log.info("boundary %r: %d feature(s)", config["ward"], len(boundary))
return gpd.clip(gdf, boundary, keep_geom_type=True)
def enrich(gdf, config):
return gdf.assign(area_m2=gdf.area.round(1))
log = logging.getLogger("parcels")
def run(config) -> dict:
metrics = RunMetrics("parcels")
previous = json.loads(Path("logs/last_run.json").read_text()) \
if Path("logs/last_run.json").exists() else None
try:
t0 = time.perf_counter()
gdf = gpd.read_file(config["input"])
metrics.step("read", 0, len(gdf), time.perf_counter() - t0)
log.info("read %s: %d features, crs=%s", config["input"], len(gdf), gdf.crs)
for name, fn in [("clean", clean), ("clip", clip_to), ("enrich", enrich)]:
t0, before = time.perf_counter(), len(gdf)
gdf = fn(gdf, config)
metrics.step(name, before, len(gdf), time.perf_counter() - t0)
log.info("%s: %d → %d features", name, before, len(gdf))
if before and len(gdf) < before * 0.5:
log.warning("%s dropped more than half the rows", name)
assert_output_sane(gdf, previous.get("features") if previous else None)
dest = Path(config["output"])
tmp = dest.parent / ".partial" / dest.name
tmp.parent.mkdir(parents=True, exist_ok=True)
gdf.to_file(tmp, driver="GPKG")
tmp.replace(dest)
final = metrics.finish("ok", features=len(gdf), output=str(dest),
bytes=dest.stat().st_size)
except Exception as exc:
log.exception("pipeline failed")
final = metrics.finish("failed", error=f"{type(exc).__name__}: {exc}")
Path("logs/runs").mkdir(parents=True, exist_ok=True)
Path(f"logs/runs/{final['run_id']}.json").write_text(json.dumps(final, indent=2))
for alert in alert_conditions(final, previous):
send_alert(alert, final)
raise
Path("logs/runs").mkdir(parents=True, exist_ok=True)
Path(f"logs/runs/{final['run_id']}.json").write_text(json.dumps(final, indent=2))
Path("logs/last_run.json").write_text(json.dumps(final, indent=2))
for alert in alert_conditions(final, previous):
log.warning("ALERT: %s", alert)
send_alert(alert, final)
log.info("done: %d features in %.1fs", final["features"], final["duration_s"])
return finalThree runs of it: a normal night; a night when the ward service returns nothing; and a night when the file arrives with 40% of its parcels missing — a random 60% of them, standing in for a damaged download.
parcels.sample(frac=0.6, random_state=0).to_file("data/raw/parcels_partial.gpkg")
nights = {
"normal": {"input": "data/raw/parcels.gpkg", "ward": "Oakham North West"},
"empty boundary": {"input": "data/raw/parcels.gpkg", "ward": "Oakham North Wesst"},
"partial file": {"input": "data/raw/parcels_partial.gpkg", "ward": "Oakham North West"},
}
for night, config in nights.items():
received.clear()
try:
result = run({**config, "output": "data/out/parcels.gpkg"})
outcome = f"ok, {result['features']:,} features"
except Exception as exc:
outcome = f"raised {type(exc).__name__}: {exc}"
print(f"{night:15} {outcome}")
for message in received:
print(f"{'':15} webhook: {message['text']}")
print("run records:", sorted(p.name for p in Path("logs/runs").glob("*.json")))normal ok, 1,441 features
empty boundary raised ValueError: output has zero features
webhook: 🔴 run failed: ValueError: output has zero features
partial file ok, 876 features
webhook: 🔴 feature count changed -39.2% (1,441 → 876)
run records: ['20261010T233527.243221Z.json', '20261010T233527.875324Z.json', '20261010T233528.046576Z.json']
The empty boundary is stopped by the output assertion, and alerted as a failed run. The partial file is not an error at all as far as Python is concerned — every parcel in it is a valid parcel, the output is not empty and not even halved — and only the comparison with the previous run notices. The first version of this example named each run by the second it started in, and the three runs above finished inside one second: they overwrote one another and left a single record. The run id now carries microseconds.
Example 2: turn run records into a trend
import json
from pathlib import Path
import pandas as pd
records = [json.loads(p.read_text()) for p in sorted(Path("logs/runs").glob("*.json"))]
df = pd.DataFrame([{
"run_id": r["run_id"],
"started": r["started_utc"],
"status": r["status"],
"duration_s": r.get("duration_s"),
"features": r.get("features"),
} for r in records]).set_index("run_id")
ok = df[df["status"] == "ok"].copy() # a failed run has no count to compare
ok["features_change_pct"] = ok["features"].pct_change().mul(100).round(1)
ok["duration_vs_median"] = (ok["duration_s"] / ok["duration_s"].median()).round(2)
print(df[["status", "features"]].to_string())
anomalies = ok[(ok["features_change_pct"].abs() > 15) | (ok["duration_vs_median"] > 2)]
print(f"\n{len(anomalies)} runs worth a look:")
print(anomalies.to_string()) status features
run_id
20261010T233527.243221Z ok 1441.0
20261010T233527.875324Z failed NaN
20261010T233528.046576Z ok 876.0
1 runs worth a look:
started status duration_s features features_change_pct duration_vs_median
run_id
20261010T233528.046576Z 2026-10-10T23:35:28+00:00 ok 0.17 876.0 -39.2 0.42
Compare successful runs with successful runs. Computed over every record, pct_change() meets the failed run's missing count, returns NaN for the night after it, and the partial delivery slips through the trend report — which is exactly what the first version of this example did. Anomaly detection for a GIS pipeline rarely needs anything cleverer than "compare with the median of the last thirty runs".
Example 3: per-step timing without cluttering the code
import functools, logging, time
log = logging.getLogger("pipeline")
def timed(name=None):
def decorator(fn):
label = name or fn.__name__
@functools.wraps(fn)
def wrapper(*args, **kwargs):
t0 = time.perf_counter()
log.info("→ %s", label)
try:
result = fn(*args, **kwargs)
except Exception:
log.exception("✗ %s failed after %.2fs", label, time.perf_counter() - t0)
raise
elapsed = time.perf_counter() - t0
size = f" ({len(result):,} rows)" if hasattr(result, "__len__") else ""
log.info("✓ %s in %.2fs%s", label, elapsed, size)
return result
return wrapper
return decorator
@timed("clip to boundary")
def clip_to_boundary(gdf, boundary):
return gpd.clip(gdf, boundary, keep_geom_type=True)
oakham = gpd.read_file(ward_query("Oakham North West"))
print(len(clip_to_boundary(parcels, oakham)), "parcels; the timing line went to the log:")
print(Path("logs/pipeline.log").read_text().splitlines()[-1].split(" ", 2)[-1])1441 parcels; the timing line went to the log:
INFO pipeline ✓ clip to boundary in 0.04s (1,441 rows)
Example 4: a freshness check on its own schedule
#!/usr/bin/env python3
"""check_freshness.py — run this every hour; it is not the pipeline."""
import json, sys, time
from pathlib import Path
JOBS = {
"parcels": {"state": Path("logs/last_run.json"), "max_age_h": 26},
"roads": {"state": Path("logs/roads_last_run.json"), "max_age_h": 26 * 7},
}
def main() -> int:
problems = []
for name, spec in JOBS.items():
state = spec["state"]
if not state.exists():
problems.append(f"{name}: no successful run recorded")
continue
age_h = (time.time() - state.stat().st_mtime) / 3600
record = json.loads(state.read_text())
if age_h > spec["max_age_h"]:
problems.append(f"{name}: last success {age_h:.1f}h ago "
f"(run {record.get('run_id')})")
for problem in problems:
print(problem, file=sys.stderr)
send_alert(f"stale pipeline: {problem}")
return 1 if problems else 0
if __name__ == "__main__":
raise SystemExit(main())import contextlib, io
JOBS.pop("roads") # only the parcels job exists here
for label in ["just after a run", "30 hours later"]:
if label == "30 hours later":
old = time.time() - 30 * 3600
os.utime("logs/last_run.json", (old, old))
stderr = io.StringIO()
with contextlib.redirect_stderr(stderr):
code = main()
print(f"{label:17} exit {code}", stderr.getvalue().strip())just after a run exit 0
30 hours later exit 1 parcels: last success 30.0h ago (run 20261010T233528.046576Z)
The checker must be a different job from the pipeline. A pipeline cannot report that it did not run.
Three layers for three questions
The three layers exist because they answer questions at different resolutions, and because they are consumed by different people at different times.
Logs are the narrative, and their value is in reconstruction: at 09:00, after a failure, you want to know what happened in order. That is why timestamps, levels and the actual numbers matter — "read data/raw/parcels.gpkg: 21253 features, crs=EPSG:27700" is worth a hundred lines of "starting step 3". Levels are not decoration either; they are the interface that lets a log aggregator or a grep separate the noise from the signal.
Metrics are the numbers, and their value is in comparison. A single run's feature count means little; the same count across thirty runs means everything, because it makes normal visible. That is what turns "the output has 876 parcels" into "the output normally has 1,441 parcels, and tonight it has 876" — the only signal Example 1's damaged file produced. Per-step row counts are the GIS-specific version of this, and they localise a problem to a step in seconds.
Alerts are the interrupt, and their value depends entirely on restraint. Every alert that does not lead to action trains people to ignore the channel, and an ignored channel is indistinguishable from no alerting. The useful test for each condition is: what would a human do when this fires? If there is no answer, log it and move on.
The fourth thing — the heartbeat — exists because of an asymmetry that catches everyone eventually. A failing job produces a signal; a job that never ran produces nothing at all. Machines get rebuilt, schedules get disabled, containers fail to pull, credentials expire. The only way to detect absence is to have something else check for the presence of a recent success, which is why a freshness check must live outside the pipeline it watches.
Log hygiene, alert fatigue and external heartbeats
print()is not logging: No timestamps, no levels, no handlers. It also block-buffers when redirected — uselogging, or at leastflush=True.- Set library log levels:
fiona,pyogrioandmatplotlibare chatty at DEBUG. Quiet them explicitly. - Rotate log files:
logging.handlers.RotatingFileHandlerprevents a long-running job filling a disk. - Never log secrets: Connection strings in log lines are a real leak path. Redact in a filter.
- Alert fatigue is a failure mode: Fewer, better alerts beat comprehensive ones nobody reads.
- The heartbeat must be external: A pipeline cannot report its own absence.
- Sample, do not log per feature: One line per feature in a million-feature job is a gigabyte of noise; log every N, or aggregate.
Internal links
- How to Get Alerted When an Automated GIS Job Fails
- How to Log and Summarise Errors in a Batch GIS Job in Python
- My GIS Pipeline Fails Silently: Fixing Swallowed Errors and Wrong Exit Codes
- How to Record Run Metadata and Data Lineage in a GIS Pipeline
- How to Validate Pipeline Inputs and Outputs Automatically in Python
- What Is a GIS Data Pipeline? The Anatomy Explained
FAQ
What is the difference between logs, metrics and alerts?
Logs are the narrative of what happened, metrics are numbers that make runs comparable, and alerts are the small set of conditions that require a person. Each answers a question the others cannot.
What should a GIS pipeline actually measure?
Rows in and out per step, features written, duration per step, bytes written, and counts of repaired geometries or skipped files — plus a comparison against the previous run.
How do I detect a run that produced wrong data without failing?
Assert on the output: non-empty, CRS present, geometries valid, and the feature count within a sensible band of the last run. An exception handler cannot catch what did not raise.
How do I know the job did not run at all?
Write a state file on every success and check its age from a separate schedule, or ping an external heartbeat service. The pipeline itself cannot report its own absence.
What should trigger an alert?
Only conditions a human would act on: the run failed, the output is empty, the feature count moved sharply, or the job is overdue. Everything else belongs in the metrics.
Should logs be JSON?
Emit both: a human-readable handler for the file you will actually read, and JSON Lines if something else — Loki, Elasticsearch, CloudWatch — ingests them.
How long should I keep run records?
Years. They are kilobytes, and their value is in the trend: comparing this run against the last thirty is what makes an anomaly visible at all.