Coastal & Marine Spatial Analysis Pipelines

Alerting on Silent Pipeline Data Loss

The most expensive marine pipeline failures are the ones where everything succeeds and less data comes out than went in. A filter threshold retuned slightly too aggressively, a join that silently drops unmatched rows, a schema change that makes a column null and a downstream dropna remove it — all of these exit zero. The identity that catches them is arithmetic, and turning it into an alert that fires on real loss without paging on normal rejection is the work. This page does that, within the monitoring workflow.

Root cause: every stage has three outputs and most report one

For any stage, every input row ends in one of three places: emitted downstream, deliberately rejected, or unaccounted. The third category should be empty and is the one nobody counts.

rowsin=rowsout+rowsrejected+unaccounted\text{rows}_\text{in} = \text{rows}_\text{out} + \text{rows}_\text{rejected} + \text{unaccounted}

Most pipelines instrument rows_out because it is the interesting number, sometimes rows_in, and rarely rows_rejected — which makes the identity uncheckable and the unaccounted term invisible.

from prometheus_client import Counter

ROWS_IN = Counter("pipeline_rows_in_total", "Rows entering a stage", ["stage", "dataset"])
ROWS_OUT = Counter("pipeline_rows_out_total", "Rows leaving a stage", ["stage", "dataset"])
ROWS_REJ = Counter("pipeline_rows_rejected_total", "Rows deliberately rejected",
                   ["stage", "dataset", "reason"])

Three counters and one extra label. The reason label is bounded — a stage has a handful of rejection reasons, enumerated in code — so it does not create the cardinality problem that a per-row identifier would.

The Stage Where the Arithmetic Stops Working A four-stage pipeline with row counts annotated at every boundary. At the decode stage, 41.2 million rows enter, 40.9 million leave and 0.3 million are rejected with recorded reasons, so the identity balances. At the dedupe stage it balances again. At the enrich stage, 38.1 million enter, 34.6 million leave, 0.1 million are rejected, and 3.4 million are unaccounted for; the stage joins against a vessel registry and silently drops rows with no match. At the publish stage the identity balances. A note records that only the enrich stage's discrepancy indicates a fault, that every stage exited zero, and that the unaccounted term is the only signal distinguishing the two situations. decode 41.2 → 40.9 + 0.3 ✓ dedupe 40.9 → 38.1 + 2.8 ✓ enrich (join) 38.1 → 34.6 + 0.1 3.4 M unaccounted ✗ publish 34.6 → 34.6 + 0 ✓ Every stage exited zero. Three balance and one does not, and the one that does not is dropping 9% of the feed because a registry join found no match for vessels the registry has not heard of. Note that dedupe rejects 2.8 M rows and is fine: those are counted, with a reason. The distinction is not how many rows disappear but whether the pipeline can say where they went. Without the rejected counter, the enrich stage and the dedupe stage look identical from outside.

Step-by-step

1. Instrument the stage so the identity is checkable 

import logging
from contextlib import contextmanager

logger = logging.getLogger(__name__)


@contextmanager
def accounted_stage(stage: str, dataset: str, n_in: int):
    """Force a stage to declare where every input row went."""
    ROWS_IN.labels(stage=stage, dataset=dataset).inc(n_in)
    ledger = {"out": 0, "rejected": {}}
    try:
        yield ledger
    finally:
        ROWS_OUT.labels(stage=stage, dataset=dataset).inc(ledger["out"])
        for reason, n in ledger["rejected"].items():
            ROWS_REJ.labels(stage=stage, dataset=dataset, reason=reason).inc(n)
        unaccounted = n_in - ledger["out"] - sum(ledger["rejected"].values())
        if unaccounted:
            logger.error("%s/%s: %d row(s) unaccounted for (in=%d out=%d rejected=%d)",
                         stage, dataset, unaccounted, n_in, ledger["out"],
                         sum(ledger["rejected"].values()))

Making the ledger an explicit object the stage must fill in is the design choice that matters. A stage that returns a filtered DataFrame and nothing else cannot report where the missing rows went, and no amount of instrumentation outside it can recover that.

2. The alert, expressed on the unaccounted fraction 

- alert: PipelineRowsUnaccounted
  expr: |
    (
      increase(pipeline_rows_in_total[1h])
      - increase(pipeline_rows_out_total[1h])
      - sum without (reason) (increase(pipeline_rows_rejected_total[1h]))
    )
    / clamp_min(increase(pipeline_rows_in_total[1h]), 1)
    > 0.001
  for: 15m
  labels: {severity: page, team: data}
  annotations:
    summary: "{{ $labels.stage }}/{{ $labels.dataset }}: {{ $value | humanizePercentage }} of rows vanished"
    runbook: "Rows entered the stage and were neither emitted nor rejected with a reason."

A tenth of a per cent rather than zero, because floating-point counter increments and scrape boundaries produce small discrepancies. clamp_min guards the division when a stage is idle, which is otherwise the most common cause of a spurious page at 3 a.m.

3. A second alert on the rejection rate, relative to its own history

Unaccounted rows are unambiguous. A rise in rejections is ambiguous — it can be a bad batch of input or a retuned filter — and it needs a comparison against the stage’s own normal:

- alert: PipelineRejectionRateJumped
  expr: |
    sum without (reason) (rate(pipeline_rows_rejected_total[30m]))
    /
    clamp_min(quantile_over_time(0.5,
      sum without (reason) (rate(pipeline_rows_rejected_total[30m]))[14d:1h] offset 1d
    ), 1e-9)
    > 3
  for: 30m
  labels: {severity: warn, team: data}
  annotations:
    summary: "{{ $labels.stage }} rejecting {{ $value | printf \"%.1f\" }}× its 14-day norm"

Warning rather than page, because a rejection spike is frequently legitimate — a storm, a jammed area, a new source with dirty data — and the response is to look rather than to wake up.

4. Prove the rules work 

# promtool test rules — replay a recorded incident against the rule set.
rule_files: [pipeline-alerts.yml]
evaluation_interval: 1m
tests:
  - interval: 1m
    input_series:
      - series: 'pipeline_rows_in_total{stage="enrich",dataset="ais"}'
        values: '0+1000x120'
      - series: 'pipeline_rows_out_total{stage="enrich",dataset="ais"}'
        values: '0+1000x60 60000+910x60'      # loses 9% from minute 60
      - series: 'pipeline_rows_rejected_total{stage="enrich",dataset="ais",reason="bounds"}'
        values: '0+0x120'
    alert_rule_test:
      - eval_time: 95m
        alertname: PipelineRowsUnaccounted
        exp_alerts:
          - exp_labels: {severity: page, team: data, stage: enrich, dataset: ais}

This is the fleet habit applied to alerting: feed the rule set the incident it exists to catch, and assert that it fires. An alert that has never fired in anger and has never been tested is a configuration file, not a control.

A 9% Loss Hides Inside Normal Throughput Variation Two panels over the same twenty-four hours. The upper panel plots rows out per hour, showing a strong daily cycle with a peak-to-trough ratio of about three. A nine per cent step down beginning at hour twelve is drawn against it and is visually indistinguishable from the normal variation; a threshold alert placed anywhere either fires constantly during the nightly trough or never fires at all. The lower panel plots the unaccounted fraction, which is flat at zero for the first twelve hours and steps to nine per cent, crossing the 0.1 per cent alert line immediately. A note records that the second signal has no daily cycle because it is a ratio of two quantities that vary together. rows out per hour — the loss is invisible 9% loss starts here a threshold either fires nightly or never unaccounted fraction — flat, then not 0.1% alert line fires within 15 minutes no daily cycle: it is a ratio of two quantities that vary together

Verification and acceptance test 

import pytest


def test_every_stage_declares_its_ledger(flow):
    for stage in flow.stages:
        assert hasattr(stage, "ledger"), (
            f"{stage.name} filters rows without declaring where they went")


def test_conservation_holds_on_a_clean_run(run_metrics):
    for (stage, dataset), m in run_metrics.items():
        unaccounted = m["in"] - m["out"] - m["rejected"]
        assert abs(unaccounted) <= max(1, m["in"] * 1e-6), (
            f"{stage}/{dataset}: {unaccounted} rows unaccounted for")


def test_the_alert_rules_pass_promtool():
    import subprocess
    out = subprocess.run(["promtool", "test", "rules", "alert-tests.yml"],
                         capture_output=True, text=True)
    assert out.returncode == 0, out.stdout + out.stderr

Routing, and the alert that nobody owns

A conservation alert names a stage, and the stage has an owner. Wiring that into the routing is what turns a firing alert into a fix rather than into a forwarded message.

route:
  receiver: platform-oncall
  group_by: [alertname, stage, dataset]
  routes:
    - matchers: [team="data"]
      receiver: data-oncall
      group_wait: 2m
      group_interval: 10m
    - matchers: [team="platform"]
      receiver: platform-oncall

Two properties of that configuration matter more than the receiver names. group_by including stage and dataset means two stages losing rows simultaneously produce two alerts rather than one merged notification, which matters because they are usually unrelated faults. And group_wait: 2m gives a burst of related alerts time to arrive together, so a single upstream problem affecting four stages arrives as one grouped page rather than four.

The harder half is organisational. A conservation alert on the enrich stage is actionable by whoever owns the enrichment logic, and on a small team that is obvious. On a larger one, stages acquire owners informally and the label goes stale: a team="data" label on a stage that was handed to the platform team six months ago routes every alert to people who cannot act on it, and the response degrades to forwarding.

The cheap defence is to derive the label from the same place the code ownership lives:

def team_for_stage(stage: str, codeowners: dict[str, str]) -> str:
    """Resolve the owning team from CODEOWNERS rather than from a hand-kept map."""
    path = f"pipeline/stages/{stage}.py"
    for pattern, team in codeowners.items():
        if path.startswith(pattern.rstrip("*")):
            return team
    raise KeyError(
        f"no CODEOWNERS entry covers {path}; an alert on this stage would have no owner")

Raising rather than defaulting is deliberate. An unowned stage should fail the build that tries to deploy it, because an alert with no owner is an alert that fires into an empty room — and the discovery that it did so always happens during the incident it was meant to prevent.

An Alert With No Owner Fires Into an Empty Room A routing chain. A firing conservation alert carries stage and dataset labels. The stage label is resolved against the repository's CODEOWNERS file to a team, and the team label routes the alert to that team's on-call rotation. Two stages are shown resolving successfully to the data and platform teams. A third stage has no CODEOWNERS entry; rather than defaulting to a catch-all receiver, the resolution raises and fails the deploy that tried to ship the stage. A note records that the alternative — a default receiver — produces an alert that is delivered, acknowledged by nobody who can act on it, and forwarded, and that this is discovered during the incident the alert existed to prevent. alert fires stage=enrich dataset=ais CODEOWNERS lookup pipeline/stages/enrich.py → team=data data on-call platform on-call no CODEOWNERS entry the deploy fails here not: route to a catch-all A default receiver produces an alert that is delivered, acknowledged by nobody who can act on it, and forwarded. That is discovered during the incident the alert existed to prevent, which is the worst possible moment. Failing the deploy costs a pull request; the alternative costs the incident.

Edge cases and gotchas

  • A stage that legitimately emits more rows than it consumes. An explode, a resample or a gap-fill increases the row count, and the identity as written flags it every run. Add an expected_ratio to the stage’s declaration and assert against rows_out / rows_in rather than equality — the point is that the ratio is declared, not that it is one.
  • Counter resets during a scrape window. increase() handles a reset, but a worker restarting twice inside the window can still produce a small negative. The clamp_min on the denominator handles the division; a > 0.001 threshold rather than > 0 absorbs the numerator.
  • Alerting on the run rather than on the window. A batch pipeline that runs nightly has no meaningful one-hour rate. Emit a per-run summary metric at the end of the run and alert on that, or the rule evaluates a window in which nothing happened and reports zero of zero.

Up: Monitoring & Observability for Marine Pipelines