Coastal & Marine Spatial Analysis Pipelines

Instrumenting Dask Workers With Prometheus

Dask exposes Prometheus endpoints on the scheduler and every worker, and what they report is entirely about the framework: task counts, thread time, memory, spill volume. All useful, and none of it says whether the data was right. Adding the data metrics is straightforward and has one failure mode that takes a monitoring system down, which is why it is worth doing deliberately. This page covers both, within the monitoring workflow.

Root cause: workers are ephemeral and labels are permanent

Two properties of the setting shape everything below.

Workers come and go. A worker that processed four hundred tiles and then exited takes its gauges with it. Prometheus scrapes what exists; a gauge that was 400 when the worker died is simply absent at the next scrape, and any query summing gauges across workers under-reports by exactly the work of every worker that has finished.

Label values are permanent. Every distinct combination of label values creates a time series that Prometheus stores until retention expires, whether or not it is ever queried again. Labelling a counter by tile identifier on a four-hundred-tile job creates four hundred series per metric per run, and a year of nightly runs is 146,000 series from one metric.

# The line that takes down a Prometheus instance:
TILES.labels(tile_id=tile_id, mmsi=mmsi, file=path).inc()

# The same information, bounded:
TILES.labels(stage="grid", dataset="bathy_ne", outcome="ok").inc()

The first form has unbounded cardinality in three dimensions simultaneously. The second has one series per combination of a handful of stages, a handful of datasets and three outcomes — a few dozen in total, forever.

One Label Choice, Two Very Different Futures A chart of stored time-series count against elapsed days, over one year of nightly runs of a four-hundred-tile pipeline. The tile-labelled scheme grows linearly, adding four hundred new series every night because each run's tile identifiers are new values, and passing one hundred and forty thousand series by the end of the year. The stage-labelled scheme is a flat line at roughly thirty series, because the label values repeat every night. A note records that the flat scheme answers every question the pipeline actually asks of its metrics, and that per-tile detail belongs in logs or spans, which are designed for high cardinality. days of nightly runs stored series labels: tile_id, mmsi, file +400 series every night → 146,000 labels: stage, dataset, outcome — flat at ~30 0120240365 The flat scheme answers every question the pipeline asks of metrics. Per-tile detail belongs in logs and spans.

Step-by-step

1. Turn on what already exists 

from dask.distributed import Client

client = Client("tcp://scheduler:8786")
# Scheduler metrics: http://scheduler:8787/metrics
# Worker metrics:    http://<worker>:8788/metrics

These give task counts by state, thread CPU and non-CPU time, memory by category, spill volume and transfer bytes — precisely the three-way split described in profiling Dask graph overhead. Scrape both before adding anything: half the questions people write custom metrics for are already answered here.

2. Register data metrics in a worker plugin

A module-level Counter() is created when the module is imported, which on a worker is when the first task needing it arrives — not at worker start. Workers that join mid-run, or that never happen to run that task, expose nothing, and the metric appears and disappears from the scrape. A WorkerPlugin registers on every worker at setup and on every worker that joins later:

import logging

from distributed.diagnostics.plugin import WorkerPlugin
from prometheus_client import Counter, Histogram

logger = logging.getLogger(__name__)

# Bounded label sets, declared once.
ROWS = Counter("pipeline_rows_total", "Rows by stage and outcome",
               ["stage", "dataset", "outcome"])
GATE = Histogram("pipeline_gate_residual_m", "Measured residual at a validation gate",
                 ["gate", "dataset"],
                 buckets=(0.005, 0.01, 0.02, 0.05, 0.1, 0.25, 0.5, 1.0))
TILE_SECONDS = Histogram("pipeline_tile_seconds", "Wall time per tile",
                         ["stage"], buckets=(1, 5, 15, 60, 180, 600))


class MarineMetrics(WorkerPlugin):
    """Ensure the metric objects exist on every worker, including late joiners."""

    name = "marine-metrics"

    def setup(self, worker):
        worker.marine_metrics = {"rows": ROWS, "gate": GATE, "tile": TILE_SECONDS}
        logger.info("marine metrics registered on %s", worker.address)


client.register_plugin(MarineMetrics())

Note the histogram buckets on pipeline_gate_residual_m. They are chosen around the gate’s threshold rather than as a generic spread, so the quantiles the alert rule needs — is the p95 approaching 0.05 m — fall inside a bucket boundary rather than being interpolated across a wide one.

3. Emit from the task, with the outcome as a label 

import time
from contextlib import contextmanager


@contextmanager
def observe_tile(stage: str, dataset: str, rows_in: int):
    """Record duration, row counts and outcome for one unit of work."""
    t0 = time.perf_counter()
    outcome = "ok"
    try:
        yield
    except Exception:
        outcome = "error"
        raise
    finally:
        TILE_SECONDS.labels(stage=stage).observe(time.perf_counter() - t0)
        ROWS.labels(stage=stage, dataset=dataset, outcome=outcome).inc(rows_in)
        logger.info("tile %s/%s: %s in %.1f s", stage, dataset, outcome,
                    time.perf_counter() - t0)

Recording the outcome as a label rather than as a separate metric is what makes the error rate queryable in one expression, which is the form every alert rule wants.

4. Aggregate run-level facts at the scheduler, not in the query

Summing a counter across workers works while the workers exist. For a run-level total that must survive the workers, compute it on the scheduler and write it into the manifest:

def run_totals(client: Client) -> dict[str, float]:
    """Ask every live worker for its counters and sum them once, at the end."""
    per_worker = client.run(lambda dask_worker: {
        "rows_ok": ROWS.labels(stage="grid", dataset="bathy_ne", outcome="ok")._value.get(),
    })
    total = {k: sum(w[k] for w in per_worker.values()) for k in next(iter(per_worker.values()))}
    logger.info("run totals across %d worker(s): %s", len(per_worker), total)
    return total

This has the same limitation as any live query — it misses workers that already exited — so the reliable version accumulates on the scheduler as tasks complete. Either way the total belongs in the run manifest, because metric retention is measured in weeks and a marine product is questioned years later.

Summing Across Workers Misses the Ones That Left A timeline of a run with six workers. Three of them are scaled down partway through, after having completed a substantial number of tiles each. A query that sums a counter across the live workers at the end of the run returns only the three survivors' values, under-reporting the total by the work of the three that exited. Beside it, a scheduler-side accumulator that increments as each task completes holds the full total regardless of which workers still exist. A note records that the difference is invisible unless the totals are compared against an independent count, and that the manifest should carry the scheduler's number. six workers, three scaled down mid-run these three exit here run ends query summing live workers 241 tiles — under by 40% scheduler accumulator 400 tiles — correct The gap is invisible without an independent count. Put the scheduler's number in the manifest.

Verification and acceptance test 

import pytest
from prometheus_client import REGISTRY


def test_no_metric_has_an_unbounded_label():
    forbidden = {"tile_id", "mmsi", "file", "path", "run_id", "timestamp"}
    for metric in REGISTRY.collect():
        for sample in metric.samples:
            bad = forbidden & set(sample.labels)
            assert not bad, f"{metric.name} carries unbounded label(s) {bad}"


def test_gate_histogram_buckets_bracket_the_threshold():
    buckets = [0.005, 0.01, 0.02, 0.05, 0.1, 0.25, 0.5, 1.0]
    threshold = 0.05
    assert threshold in buckets, (
        "the alert threshold must be a bucket boundary, or the quantile near it "
        "is interpolated across a wide bucket and the alert fires late")


def test_every_stage_emits_rows_in_and_out(flow_metrics):
    for stage in ("ingest", "grid", "validate", "publish"):
        assert flow_metrics.get((stage, "ok")) is not None, f"{stage} emits no row counter"

The first test is the important one, and it is cheap enough to run in CI on every commit. Cardinality problems are not caught by review — the offending line looks reasonable — and they are caught immediately by iterating the registry.

Scraping workers that do not stay still

Prometheus discovers targets and then scrapes them on a schedule. Dask workers appear and disappear on a schedule of their own, and the mismatch produces three operational problems that have standard answers.

Discovery lag. A worker that lives for four minutes with a thirty-second discovery interval and a fifteen-second scrape may be scraped only a handful of times, and a worker that lives for ninety seconds may never be discovered at all. On Kubernetes, pod-level service discovery with a short refresh closes most of the gap; outside it, a push-based path is the only reliable option for short-lived workers.

The last scrape is always missing. A worker exits between scrapes and its final increments are lost. For counters this is a small under-count that does not accumulate error over long windows; for anything where the exact total matters, the scheduler-side accumulation from the previous section is the answer rather than a shorter scrape interval.

Pushgateway is the wrong default and occasionally the right answer. It holds metrics after the source has gone, which sounds like the fix and introduces its own problem: values persist until overwritten or deleted, so a worker that pushed and never returns leaves a stale series that looks live forever. Where it is used, push a grouping key that includes the run, and delete the group at the end of the run.

def push_run_summary(gateway: str, run_id: str, totals: dict[str, float]) -> None:
    """Push end-of-run totals, then delete the group so nothing goes stale."""
    from prometheus_client import CollectorRegistry, Gauge, delete_from_gateway, push_to_gateway
    reg = CollectorRegistry()
    for name, value in totals.items():
        Gauge(f"pipeline_run_{name}", f"end-of-run {name}", registry=reg).set(value)
    push_to_gateway(gateway, job="marine-run", grouping_key={"run_id": run_id}, registry=reg)
    logger.info("pushed run summary for %s: %s", run_id, totals)
    # The caller deletes the group once the summary has been scraped and archived.

The deletion is the part that is always omitted and always needed. Without it the gateway accumulates one permanent series per run, which is the cardinality problem from the top of this page arriving by a different route.

Some Workers Are Never Scraped At All A timeline with scrape ticks every fifteen seconds along the bottom. Three worker lifetimes are drawn above it. A long-lived worker spanning most of the run is scraped many times and its counters are well represented. A four-minute worker is scraped roughly sixteen times, which is adequate. A ninety-second worker that started just after a service-discovery refresh is scraped zero times, because discovery runs every thirty seconds and the worker was gone before its target was registered. A separate line shows the scheduler-side accumulator, which records completed tasks from all three workers regardless of whether any of them was ever scraped. scrape every 15 s long-lived ✓ 4 minutes ✓ 16 scrapes 90 seconds ✗ never discovered scheduler ✓ Discovery runs every 30 s. A worker that lives 90 s can be gone before its target is registered — and its tasks still counted.

Edge cases and gotchas

  • Two workers on one host share a port. Running multiple workers per node needs distinct --dashboard-address values or only one exposes metrics. Kubernetes side-steps it by giving each pod its own address; a bare-metal deployment does not.
  • Counters reset when a worker restarts, and that is fine. Prometheus’s rate() and increase() handle counter resets correctly. Gauges do not have that property, which is another reason to prefer counters for anything cumulative.
  • client.run executes on live workers only, and blocks. Using it in a hot path serialises the run behind the slowest worker. Reserve it for end-of-run collection, and prefer accumulating on the scheduler for anything needed mid-run.

Up: Monitoring & Observability for Marine Pipelines