Logs, Metrics and Alerts: Observability for GIS Pipelines
Problem statement
Two ways a scheduled GIS job goes wrong, and neither is visible:
$ tail -3 logs/nightly.log
starting
done$ ls -la data/out/parcels.gpkg
-rw-r--r-- 1 gis gis 812 Aug 11 02:31 parcels.gpkg # 812 bytes. It should be 400 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 published a header and no rows.
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
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 dropped 40% instead of 14%, starting on the 14th".
Three layers, three questions
Step-by-step solution
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%SZ"),
"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.get("features", 0) == 0:
alerts.append("output has zero features")
if previous and previous.get("features"):
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")
for step in metrics["steps"]:
if step["rows_in"] and step["retained_pct"] is not None and step["retained_pct"] < 50:
alerts.append(f"step '{step['step']}' dropped "
f"{100 - step['retained_pct']:.0f}% of rows")
return alertsThe 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 812-byte GeoPackage. 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.
Code examples
Example 1: an instrumented pipeline
#!/usr/bin/env python3
import json, logging, sys, time
from datetime import datetime, timezone
from pathlib import Path
import geopandas as gpd
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.with_suffix(".gpkg.part")
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 finalExample 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")
df["features_change_pct"] = df["features"].pct_change().mul(100).round(1)
df["duration_vs_median"] = (df["duration_s"] / df["duration_s"].median()).round(2)
print(df.tail(14).to_string())
anomalies = df[(df["features_change_pct"].abs() > 15) | (df["duration_vs_median"] > 2)]
print(f"\n{len(anomalies)} runs worth a look:")
print(anomalies.to_string())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(gdf, boundary):
import geopandas as gpd
return gpd.clip(gdf, boundary, keep_geom_type=True)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())The checker must be a different job from the pipeline. A pipeline cannot report that it did not run.
Explanation
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 4,812 features from parcels.gpkg, 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 clip step dropped 695 features" into "the clip step normally drops 690–700 features, and today it dropped 3,100". 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.
Edge cases or notes
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.