feat: add observability metrics infrastructure

- Add scripts/metrics.py with 7 Prometheus metrics definitions
- Add CLI timing logs to register_stac.py
- Expose metrics endpoint in workflow pods (port 8000)
- Add prometheus-client dependency
- Background metrics server with trap cleanup

Author: wietzesuijker

notebooks/02_pyramid_performance.ipynb

@@ -399,7 +399,7 @@
     "plt.show()\n",
     "\n",
     "print(\n",
-    "    f\"\\n📊 Key Metric: {np.mean([s for z, s in zip(zooms, [measured[i]/expected[i] for i in range(len(zooms))], strict=False) if z <= 10]):.1f}× average speedup at production-relevant zooms\"\n",
+    "    f\"\\n📊 Key Metric: {np.mean([s for z, s in zip(zooms, [measured[i] / expected[i] for i in range(len(zooms))], strict=False) if z <= 10]):.1f}× average speedup at production-relevant zooms\"\n",
     ")"
    ]
   },
@@ -426,15 +426,15 @@
     "print(\"Return on Investment:\")\n",
     "print(\"=\" * 60)\n",
     "print(\"Storage Cost:\")\n",
-    "print(f\"  Native only: {native_storage:,} pixels ({native_storage/1e6:.0f} MB uncompressed)\")\n",
-    "print(f\"  With pyramids: {total_storage:,} pixels ({total_storage/1e6:.0f} MB uncompressed)\")\n",
+    "print(f\"  Native only: {native_storage:,} pixels ({native_storage / 1e6:.0f} MB uncompressed)\")\n",
+    "print(f\"  With pyramids: {total_storage:,} pixels ({total_storage / 1e6:.0f} MB uncompressed)\")\n",
     "print(f\"  Overhead: +{overhead_pct:.0f}%\")\n",
     "print(\"\\nPerformance Gain:\")\n",
     "print(\n",
-    "    f\"  z6-10 (low zoom): {np.mean([measured[i]/expected[i] for i, z in enumerate(zooms) if z <= 10]):.1f}× faster\"\n",
+    "    f\"  z6-10 (low zoom): {np.mean([measured[i] / expected[i] for i, z in enumerate(zooms) if z <= 10]):.1f}× faster\"\n",
     ")\n",
     "print(\n",
-    "    f\"  z12-14 (high zoom): {np.mean([measured[i]/expected[i] for i, z in enumerate(zooms) if z >= 12]):.1f}× faster\"\n",
+    "    f\"  z12-14 (high zoom): {np.mean([measured[i] / expected[i] for i, z in enumerate(zooms) if z >= 12]):.1f}× faster\"\n",
     ")\n",
     "print(\"\\nProduction Impact:\")\n",
     "print(\"  • Consistent 100-200ms tile generation across all zooms\")\n",

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "pika>=1.3.0",
     "tenacity>=8.0.0",
     "requests>=2.31.0",
+    "prometheus-client>=0.19.0",
 ]
 
 [project.optional-dependencies]

scripts/benchmark_geozarr.py

@@ -110,7 +110,7 @@ def main(argv: list[str] | None = None) -> int:
         if speedup > 1:
             logger.info(f"✅ GeoZarr is {speedup}x faster than EOPF")
         else:
-            logger.warning(f"⚠️  EOPF is {1/speedup:.2f}x faster than GeoZarr")
+            logger.warning(f"⚠️  EOPF is {1 / speedup:.2f}x faster than GeoZarr")
 
         return 0
 

scripts/benchmark_tile_performance.py

@@ -154,7 +154,7 @@ def benchmark_zoom_level(
         status = "✓" if result["success"] else "✗"
         logger.debug(
             f"  {status} z{z}/{x}/{y}: {result['latency_ms']:.1f}ms "
-            f"({result['size_bytes']/1024:.1f}KB)"
+            f"({result['size_bytes'] / 1024:.1f}KB)"
         )
 
     # Calculate statistics

scripts/metrics.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""Prometheus metrics instrumentation for data-pipeline scripts.
+
+This module provides shared metric definitions and a metrics server
+for exposing metrics to the Prometheus scraper in Kubernetes.
+
+Usage:
+    from scripts.metrics import start_metrics_server, CONVERSION_DURATION
+
+    start_metrics_server(port=8000)  # In main()
+
+    with CONVERSION_DURATION.labels(collection="sentinel-2-l2a").time():
+        convert_data()
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+from prometheus_client import Counter, Histogram, start_http_server
+
+logger = logging.getLogger(__name__)
+
+# Metrics port for Kubernetes ServiceMonitor to scrape
+DEFAULT_METRICS_PORT = 8000
+
+# Conversion workflow metrics
+CONVERSION_DURATION = Histogram(
+    "geozarr_conversion_seconds",
+    "Time to convert source to GeoZarr format",
+    labelnames=["collection", "resolution"],
+)
+
+CONVERSION_DATA_SIZE = Histogram(
+    "geozarr_conversion_bytes",
+    "Size of data converted in bytes",
+    labelnames=["collection"],
+    buckets=[1e6, 10e6, 100e6, 1e9, 10e9, 100e9],  # 1MB to 100GB
+)
+
+# STAC API interaction metrics
+STAC_REGISTRATION_TOTAL = Counter(
+    "stac_registration_total",
+    "Total STAC item registration attempts",
+    labelnames=["collection", "status"],  # status: success|failure|retry
+)
+
+STAC_HTTP_REQUEST_DURATION = Histogram(
+    "stac_http_request_seconds",
+    "STAC API HTTP request duration",
+    labelnames=["method", "endpoint", "status_code"],
+)
+
+# Preview generation metrics
+PREVIEW_GENERATION_DURATION = Histogram(
+    "preview_generation_seconds",
+    "Time to generate preview images",
+    labelnames=["collection", "preview_type"],  # preview_type: true_color|quicklook|s1_grd
+)
+
+PREVIEW_HTTP_REQUEST_DURATION = Histogram(
+    "preview_http_request_seconds",
+    "HTTP request duration for preview-related operations",
+    labelnames=["operation", "status_code"],
+)
+
+# AMQP workflow metrics
+AMQP_PUBLISH_TOTAL = Counter(
+    "amqp_publish_total",
+    "Total AMQP messages published",
+    labelnames=["exchange", "status"],  # status: success|failure
+)
+
+
+def start_metrics_server(port: int | None = None) -> None:
+    """Start Prometheus metrics HTTP server.
+
+    Args:
+        port: Port to listen on. Defaults to METRICS_PORT env var or 8000.
+
+    Note:
+        Should only be called once per process. Safe to call in Kubernetes
+        pod startup. Metrics exposed at http://localhost:<port>/metrics
+    """
+    if port is None:
+        port = int(os.getenv("METRICS_PORT", str(DEFAULT_METRICS_PORT)))
+
+    try:
+        start_http_server(port)
+        logger.info("Metrics server started on port %d", port)
+    except OSError as e:
+        # Port already in use (e.g., from previous run)
+        logger.warning("Failed to start metrics server on port %d: %s", port, e)
+
+
+def is_metrics_enabled() -> bool:
+    """Check if metrics collection is enabled.
+
+    Returns:
+        True if ENABLE_METRICS env var is set to "true" (case-insensitive).
+        Defaults to True if not set (opt-out model).
+    """
+    return os.getenv("ENABLE_METRICS", "true").lower() == "true"

scripts/register_stac.py

@@ -15,6 +15,7 @@
 import logging
 import os
 import sys
+import time
 from typing import Any, cast
 from urllib.parse import urlparse
 
@@ -436,6 +437,8 @@ def register_item(
 
 def main() -> int:
     """CLI entrypoint."""
+    start_time = time.perf_counter()
+
     parser = argparse.ArgumentParser(description="Register GeoZarr output to STAC API")
     parser.add_argument(
         "--stac",
@@ -510,11 +513,13 @@ def main() -> int:
             headers=headers,
         )
 
-        logger.info("Registration complete")
+        duration = time.perf_counter() - start_time
+        logger.info(f"Registration complete in {duration:.2f}s")
         return 0
 
     except Exception as exc:
-        logger.error(f" {exc}")
+        duration = time.perf_counter() - start_time
+        logger.error(f"Registration failed after {duration:.2f}s: {exc}")
         import traceback
 
         traceback.print_exc()

uv.lock

@@ -242,6 +242,11 @@ spec:
       source: |
         set -euo pipefail
 
+        # Start metrics server in background (for Prometheus scraping)
+        python -c "from scripts.metrics import start_metrics_server; start_metrics_server()" &
+        METRICS_PID=$!
+        trap "kill $METRICS_PID 2>/dev/null || true" EXIT
+
         echo "════════════════════════════════════════════════════════════════════════════"
         echo "  STEP 3/4: STAC REGISTRATION"
         echo "════════════════════════════════════════════════════════════════════════════"
@@ -280,10 +285,14 @@ spec:
         limits:
           memory: "2Gi"
           cpu: "1"
-      source: |
       source: |
         set -euo pipefail
 
+        # Start metrics server in background (for Prometheus scraping)
+        python -c "from scripts.metrics import start_metrics_server; start_metrics_server()" &
+        METRICS_PID=$!
+        trap "kill $METRICS_PID 2>/dev/null || true" EXIT
+
         echo "════════════════════════════════════════════════════════════════════════════"
         echo "  STEP 4/4: STAC AUGMENTATION"
         echo "════════════════════════════════════════════════════════════════════════════"