Monitoring & Observability for Marine Pipelines #

Operational intent: this topic covers what to measure in a running marine spatial pipeline so that a failure is noticed by a machine rather than by a reader of the published product. It sits inside Deployment & Cloud Infrastructure for Marine Pipelines, alongside the orchestration and Kubernetes scheduling topics, and it exists because those two together still leave the most important class of failure invisible.

Marine pipelines fail green. Every failure mode named across this site — an implicit datum assumption, a flipped separation sign, a missing PROJ grid, an axis-order swap, nodata read as depth, a dedup key that merges two vessels — produces a run in which every process exits zero, every pod completes, every dashboard is calm, and the output is wrong. Infrastructure monitoring is built to answer “is it running”, and the answer is yes. Nothing in a standard observability stack asks “is it right”.

Reference Metric Set #

Two of these deserve emphasis because they are unusual. pipeline_gate_residual publishes the measured value at each validation gate rather than the pass/fail — the actual control-point drift in metres, the actual mean residual against benchmarks. A gate that passes at 0.048 m against a 0.05 m threshold, three runs in a row after a year at 0.006 m, is a system about to fail, and the boolean cannot say so. And pipeline_proj_grid_count is a one-line check at container start that catches the missing-grid failure which otherwise changes coordinates by a metre with no other symptom.

Freshness Is the Cheapest High-Value Metric #

Most silent marine pipeline failures are, in the end, a stopped input: a feed whose connector half-opened, a survey upload that stopped arriving, a tide-gauge API that started returning an empty array with a 200 status. In all of these the pipeline runs on schedule, processes what it has, and publishes something. The product is stale rather than wrong, and staleness is invisible in every graph except one.

A freshness gauge is the age of the newest record in each dataset, published continuously rather than computed on request:

"""Dataset freshness as a Prometheus gauge, one series per product."""
from __future__ import annotations

import logging
import time

from prometheus_client import Gauge

logger = logging.getLogger(__name__)

DATASET_AGE = Gauge(
    "pipeline_dataset_age_seconds",
    "Age of the newest record in a published dataset",
    ["dataset", "environment"],
)


def publish_freshness(dataset: str, newest_epoch_s: float, *, environment: str) -> float:
    """Record how stale a dataset is. Returns the age for logging or assertion."""
    if newest_epoch_s <= 0:
        raise ValueError(f"{dataset}: no records found; cannot compute freshness")
    age = time.time() - newest_epoch_s
    if age < 0:
        # A record from the future means a clock or a timestamp-parsing problem,
        # which is itself worth alerting on rather than clamping away.
        raise ValueError(f"{dataset}: newest record is {-age:.0f} s in the future")
    DATASET_AGE.labels(dataset=dataset, environment=environment).set(age)
    logger.info("freshness %s: %.0f s", dataset, age)
    return age

The rule that makes this work is that the gauge is published by the consumer of the dataset, not by its producer. A producer that has stopped running publishes nothing at all, and a metric that stops being reported is easy to miss; a consumer that is still running publishes a steadily rising age, which is unmissable. Alert on the age crossing a multiple of the dataset’s expected update interval — three times is a reasonable default — rather than on a fixed number of hours, so the same rule works for a two-second AIS feed and a monthly bathymetric mosaic.

Alert on Change, Not on Absolutes #

Marine data is seasonal, weather-dependent and fleet-dependent. Traffic falls in winter storms. Survey volumes fall when the vessel is in port. Message rates fall when a satellite pass ends. A static threshold on any of these produces alerts that correlate with the weather, and a team that has learned to ignore them.

The alternative is to compare each metric against its own recent history. A simple and robust form is the ratio of the current value to the median of the same window on the previous n days, which handles both the daily cycle and the seasonal drift without any modelling:

# Prometheus rule: the feed is delivering far less than it usually does at
# this hour, regardless of what "usual" means this month.
- alert: AISIngestRateCollapsed
  expr: |
    rate(pipeline_rows_in_total{stage="ais_ingest"}[15m])
      /
    quantile_over_time(0.5,
      rate(pipeline_rows_in_total{stage="ais_ingest"}[15m])[7d:1h] offset 1d
    ) < 0.4
  for: 20m
  labels: {severity: page, team: data}
  annotations:
    summary: "AIS ingest at {{ $value | humanizePercentage }} of its 7-day norm"

The for: 20m clause matters as much as the expression. Satellite feeds are bursty by nature and a fifteen-minute window will dip below any threshold occasionally; requiring the condition to hold for twenty minutes removes essentially all of that without delaying a real outage meaningfully.

Structured Logs Are the Second Half #

Metrics say that something changed; logs say what. The two are complementary and the join between them is the run identifier. Every log line emitted anywhere in a marine pipeline should carry the run id, the stage name and the dataset it concerns, as structured fields rather than as English embedded in a message:

import json
import logging


class RunContextFilter(logging.Filter):
    """Attach the run identity to every record, so logs join to metrics."""

    def __init__(self, run_id: str, stage: str) -> None:
        super().__init__()
        self.run_id, self.stage = run_id, stage

    def filter(self, record: logging.LogRecord) -> bool:
        record.run_id = self.run_id
        record.stage = self.stage
        return True


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "run_id": getattr(record, "run_id", None),
            "stage": getattr(record, "stage", None),
            "msg": record.getMessage(),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)

The specific thing this buys on a marine pipeline is the ability to answer “which survey lines went through the version of the code that had the wrong grid” — a question that comes up after every incident, and that is unanswerable if the logs record only what happened and not which run it happened in. The run manifest carries the same identifier, so a metric anomaly leads to a run, the run leads to its logs and its manifest, and the manifest names the inputs and the image digest.

Tracing a Batch Pipeline Without a Request #

Distributed tracing was designed for request-response systems: a user request enters, fans out across services, and the trace is the tree of spans under it. A marine pipeline has no request. It has a nightly run that touches four hundred survey tiles across three stages over ninety minutes, and the useful unit of causality is the tile, not a call.

The adaptation that works is to treat the run as the trace and the tile as the span. Each tile’s processing emits one span carrying the tile identifier, the stage, the duration, and the attributes that matter for diagnosis — input row count, output row count, the CRS in and out, the number of rejections. A trace then shows four hundred parallel spans and, crucially, shows which ones were slow and which ones were anomalous in a data sense rather than a timing sense.

That last part is what makes it worth the effort on a batch system. Timing outliers on a tile pipeline are usually uninteresting — a tile is slow because it is dense. Data outliers are the whole game: the one tile out of four hundred whose output row count is a tenth of its input, or whose CRS attribute reads EPSG:4326 when every other tile reads EPSG:32618. Sorting spans by a data attribute rather than by duration surfaces those immediately, and nothing in a metrics dashboard can, because per-tile detail is exactly the cardinality that metrics cannot carry.

def tile_span_attributes(tile_id: str, result) -> dict[str, object]:
    """Attributes worth attaching to a per-tile span. Diagnosis, not timing."""
    return {
        "tile.id": tile_id,
        "tile.rows_in": result.rows_in,
        "tile.rows_out": result.rows_out,
        "tile.rows_rejected": result.rows_rejected,
        "tile.retention": (result.rows_out / result.rows_in) if result.rows_in else 0.0,
        "tile.crs_in": str(result.crs_in),
        "tile.crs_out": str(result.crs_out),
        "tile.drift_p95_m": result.drift_p95_m,
        "tile.coverage_fraction": result.coverage_fraction,
    }

Where a full tracing backend is more weight than the deployment justifies, the same value is available for far less: write one structured log line per tile with these fields, and query it. The point is not the tracing system; it is that per-tile attributes exist somewhere queryable, so that “which tile went wrong” is a filter rather than an investigation.

Validation Gates and Quality Control #

Gate 1 — every published dataset has a live freshness series. A dataset with no pipeline_dataset_age_seconds series is unmonitored, and the check is a query against the metrics backend rather than a code review. Run it in CI against the list of published products.

Gate 2 — rows in equals rows out plus rows rejected. For every stage, per run. A discrepancy is an unaccounted row, which is either loss or duplication; both matter and neither shows up anywhere else.

Gate 3 — the alert rules themselves are tested. An alert that has never fired is an alert that may not work. Replay a recorded outage against the rule set in CI — Prometheus’s promtool test rules does exactly this — and assert that each rule fires on the incident it was written for. This is the observability equivalent of the fleet-wide habit of feeding a checker input it must reject.

Retention: keep the summary far longer than the samples #

Metric storage is sized for operations — days to weeks of high-resolution samples, which is right for answering “why is the cluster slow now”. Marine data questions arrive on a completely different timescale: a hydrographic office comparing this season’s survey against one from three years ago wants to know whether the two pipelines were in the same state, and the metrics that would answer it expired long ago.

The resolution is to separate the two retention needs rather than to extend one of them. Keep raw samples at their default retention for operational debugging. Separately, at the end of every run, write a small, permanent summary — the gate residuals, row counts, coverage fraction, library versions and image digest — into the run manifest that ships with the product. That record is a few hundred bytes, it lives with the data rather than in the monitoring system, and it survives every migration the monitoring stack will go through in the years the product is in use.

The rule of thumb worth adopting: if a number would be needed to defend a published product rather than to operate the pipeline, it belongs in the manifest, not only in the time-series database. Operational metrics answer questions about now; a marine data product is asked questions about a run that finished years ago, and the two storage systems have almost nothing in common.

A related trap is aggregation. Long-retention tiers usually downsample by averaging, which is exactly wrong for the gate residuals: the useful statistic is the maximum drift observed, and averaging it away turns a run with one bad tile into a run that looks clean. Record the maximum and the p95 explicitly as their own series, so the downsampler has nothing to destroy.

Common Failure Modes #

Monitoring the orchestrator instead of the data. Prefect’s own dashboard shows flow-run success, which is the infrastructure question again. A flow that ran perfectly on empty input is a green flow run and an empty product.

Cardinality explosion from per-vessel labels. Labelling a metric by MMSI produces tens of thousands of series and will take down a Prometheus instance. Aggregate to fleet, class or region; keep per-vessel detail in logs, which are built for high cardinality.

Alerts routed to whoever is on call. A datum error is not actionable by the infrastructure team and a node failure is not actionable by the data team. Route by team label from the start; alerts sent to people who cannot act on them are how a team learns to ignore alerts.

Pipeline Integration and Downstream Handoff #

The metrics endpoint is scraped per pod; the freshness gauges are published by consumers; the structured logs ship to whatever aggregator the organisation already runs. The one artefact this topic adds to the pipeline’s outputs is the per-run metric snapshot, written into the run manifest alongside the parameters: the gate residuals, row counts and coverage fractions as they stood at publication.

That snapshot is what makes an old product diagnosable. Two years later, the metrics backend has long since expired its retention, but the manifest still records that this particular mosaic was published with a control-point drift of 0.041 m and 96.2% coverage — which is exactly the information needed to decide whether it can be compared with a new one.

Related #

• Instrumenting Dask Workers With Prometheus — worker-level metrics without cardinality explosion
• Alerting on Silent Pipeline Data Loss — the row-conservation rule as a working alert
• Orchestrating Marine Pipelines With Dask and Prefect — where stage boundaries, and therefore metric emission points, are defined
• Kubernetes Job Scheduling for Bathymetric Processing — the infrastructure signals this topic deliberately does not duplicate
• Deployment & Cloud Infrastructure for Marine Pipelines — the run manifest these metrics are snapshotted into

Up: Deployment & Cloud Infrastructure for Marine Pipelines