SC ReconciliationReconciliation Automation
  1. Home
  2. Architecture
  3. Monitoring and Alerting for Reconciliation Pipelines
  4. Instrumenting Custom Pipeline Metrics with Prometheus

Instrumenting Custom Pipeline Metrics with Prometheus

↑ Part of Monitoring and Alerting for Reconciliation Pipelines.

A reconciliation batch job that runs for four minutes on a cron schedule and then exits leaves nothing behind for Prometheus’s pull model to find — by the time a scrape would land, the process is gone. That gap is why teams end up flying blind on exactly the pipeline stages that fail most often: ingest throughput, match-tier breakdown, and exception volume by reason code all live in log lines that nobody greps until an incident is already underway. This page instruments a batch reconciliation job with prometheus_client, defines a label schema that will not blow up your time-series cardinality, and pushes the results through a Pushgateway so a short-lived process still produces a durable, scrapeable metric snapshot.

Operational Trigger Signals

Add custom instrumentation when the pipeline’s own operational history shows these measurable gaps:

  1. No per-stage throughput visibility: total job runtime is logged, but ingest, matching, and write-back are not broken out, so a slowdown in one stage is invisible until the whole job blows its SLA window.
  2. MTTR consistently above 30 minutes on batch failures: on-call engineers reconstruct what happened from unstructured logs instead of querying a metric, because no reconciliation_exceptions_total counter exists to tell them which tier or reason spiked.
  3. Dashboards show gaps during every run: Prometheus’s scrape targets list the job as down for the 3–8 minutes it actually executes, because a batch process that exits before the next scrape interval is structurally unscrapeable without a push-based bridge.
  4. No visibility into match-score distribution near the fuzzy threshold: recalibrating a similarity floor after an incident is guesswork without a histogram showing how many records cleared the gate by a wide margin versus a razor-thin one.

Once two or more of these hold across consecutive runs, instrument the job directly rather than parsing logs after the fact. The metrics feed the same alerting layer that watches exception queue depth and age and the dashboards described in Building Grafana Dashboards for Reconciliation Health, so the label schema chosen here should match what those consumers expect.

Step-by-Step Implementation

  1. Pick metric types per signal — Counters for monotonic events (records ingested, matched, exceptions), Gauges for point-in-time state (queue depth, run duration), Histograms for distributions (stage latency, match score).
  2. Build a per-run registry — instantiate a fresh CollectorRegistry inside the job rather than using the global default, so pushed samples reflect only this run and never accumulate stale series across invocations.
  3. Define a bounded label schema — labels are closed sets (stage, match_type, tier, reason), never free-text identifiers like PO numbers or vendor names.
  4. Wrap each stage in a timer — a context manager records wall-clock duration into the stage-latency histogram without duplicating timing code at every call site.
  5. Push at process exit, not mid-run — call push_to_gateway once in a finally block so a Pushgateway crash never masks whether the batch itself succeeded.
  6. Scrape the Pushgateway from Prometheus — add a static target with honor_labels: true so the job’s own labels win over the scrape-added instance label.

The implementation below wires up the Matching & Reconciliation Algorithms engine’s stage boundaries to metrics and pushes them once, at the very end of the run:

PYTHON
import logging
import time
from collections.abc import Iterator
from contextlib import contextmanager

from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram, push_to_gateway

logger = logging.getLogger(__name__)

PUSHGATEWAY_ADDR = "pushgateway.internal:9091"

# One fresh registry per run — never the global default — so a crashed
# or retried job cannot leak stale series into the next push.
registry = CollectorRegistry()

RECORDS_INGESTED = Counter(
    "reconciliation_records_ingested_total",
    "Raw records read from the source feed before normalization.",
    ["source_feed"],
    registry=registry,
)
RECORDS_MATCHED = Counter(
    "reconciliation_records_matched_total",
    "Records committed to the ledger, tagged by how they were matched.",
    ["match_type"],  # closed set: exact, fuzzy_validated, tolerance_adjusted
    registry=registry,
)
EXCEPTIONS_TOTAL = Counter(
    "reconciliation_exceptions_total",
    "Records routed to the exception queue, tagged by tier and reason.",
    ["tier", "reason"],  # closed sets — never a raw vendor or SKU value
    registry=registry,
)
QUEUE_DEPTH = Gauge(
    "reconciliation_exception_queue_depth",
    "Exception queue depth at job completion, per severity tier.",
    ["tier"],
    registry=registry,
)
RUN_DURATION = Gauge(
    "reconciliation_run_duration_seconds",
    "Wall-clock duration of the full batch run.",
    registry=registry,
)
STAGE_LATENCY = Histogram(
    "reconciliation_stage_latency_seconds",
    "Per-stage processing time within a single batch run.",
    ["stage"],
    buckets=(0.05, 0.1, 0.5, 1, 5, 15, 30, 60, 120),
    registry=registry,
)
MATCH_SCORE = Histogram(
    "reconciliation_match_score",
    "Similarity score distribution for fuzzy-validated matches.",
    ["match_type"],
    buckets=(0.5, 0.6, 0.7, 0.8, 0.85, 0.9, 0.95, 1.0),
    registry=registry,
)


@contextmanager
def timed_stage(stage_name: str) -> Iterator[None]:
    """Record wall-clock time for one pipeline stage into STAGE_LATENCY."""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        STAGE_LATENCY.labels(stage=stage_name).observe(elapsed)
        logger.info("stage=%s elapsed_s=%.3f", stage_name, elapsed)


def run_batch(source_feed: str, run_id: str) -> None:
    """Execute one reconciliation batch and push metrics before exit."""
    run_start = time.perf_counter()
    try:
        with timed_stage("ingest"):
            records = ingest(source_feed)
            RECORDS_INGESTED.labels(source_feed=source_feed).inc(len(records))

        with timed_stage("match"):
            matched, exceptions = match_records(records)
            for match_type, count in matched.items():
                RECORDS_MATCHED.labels(match_type=match_type).inc(count)
            for score, match_type in matched.get("scores", []):
                MATCH_SCORE.labels(match_type=match_type).observe(score)

        with timed_stage("exception_routing"):
            for tier, reason, count in exceptions:
                EXCEPTIONS_TOTAL.labels(tier=tier, reason=reason).inc(count)
                QUEUE_DEPTH.labels(tier=tier).inc(count)

        logger.info("run_id=%s completed successfully", run_id)
    finally:
        # Push happens in finally so a downstream failure still reports
        # partial progress instead of leaving the run invisible.
        RUN_DURATION.set(time.perf_counter() - run_start)
        push_to_gateway(
            PUSHGATEWAY_ADDR,
            job="reconciliation_batch",
            grouping_key={"instance": run_id},
            registry=registry,
        )
        logger.info("pushed metrics to %s for run_id=%s", PUSHGATEWAY_ADDR, run_id)

grouping_key is what makes repeated runs safe: each push under a distinct instance value writes a separate metric group on the Pushgateway rather than overwriting the previous run’s samples, and Prometheus scrapes the union of all current groups on its normal interval.

Label Cardinality Discipline

Every unique combination of label values on a metric creates a new time series, and Prometheus’s memory footprint scales with that count directly:

∣series∣=∏i=1n∣Li∣|\text{series}| = \prod_{i=1}^{n} |L_i|

where LiL_i is the set of distinct values label ii can take. A tier label with 3 values and a reason label with 8 values produces at most 24 series for reconciliation_exceptions_total — bounded and cheap. Swap reason for a raw vendor name or PO number and that same metric produces one series per distinct value ever seen, unbounded and permanent, because Prometheus never forgets a series once it exists. Keep every label a closed enum defined in code, never a value read directly off the record being processed.

Pushgateway pattern for short-lived batch metrics A batch reconciliation job runs for a few minutes and populates an in-process registry of Counters, Gauges, and Histograms. At exit, in a finally block, it pushes the registry once to a Pushgateway using a grouping key derived from the run id, so repeated runs do not overwrite each other. Prometheus scrapes the Pushgateway as a normal static target on its regular interval, and the resulting series feed both Alertmanager for alerting and Grafana for dashboards. Batch reconciliation job In-process registry Counters · Gauges · Histograms closed label sets only finally: push_to_gateway grouping_key = instance Pushgateway holds last-pushed group per key durable across job exit Prometheus server scrapes Pushgateway target honor_labels: true Alertmanager exception & latency rules Grafana reconciliation health boards

Configuration Reference

These are the metric names, types, and labels this instrumentation exposes. Keep names stable once dashboards and alert rules depend on them — renaming a metric breaks every downstream query silently.

Metric name Type Labels Notes
reconciliation_records_ingested_total Counter source_feed Raw records read, pre-normalization.
reconciliation_records_matched_total Counter match_type exact, fuzzy_validated, tolerance_adjusted.
reconciliation_exceptions_total Counter tier, reason Closed enums only; reason maps to the failure taxonomy.
reconciliation_exception_queue_depth Gauge tier Snapshot at job end, feeds queue-age alerting.
reconciliation_run_duration_seconds Gauge — Total wall-clock time for the run.
reconciliation_stage_latency_seconds Histogram stage ingest, match, exception_routing.
reconciliation_match_score Histogram match_type Bucketed near the similarity floor for threshold review.

Debugging & Recovery

  • reason values drifting from the taxonomy — the closed set behind the reason label should mirror the failure-reason vocabulary used in Exception Handling & Discrepancy Resolution; if the two drift apart, exception counts and the exception queue’s own triage view disagree about what actually broke.
  • Stale groups on the Pushgateway — if a job’s instance value is reused across runs instead of being unique per invocation, each push overwrites the prior group and gauges like reconciliation_run_duration_seconds silently freeze at the last successful value. Derive the grouping key from a run id, not a fixed job name.
  • Cardinality spikes after a schema change — a new label value that is not a closed enum (a vendor id slipping into reason, for example) shows up as a sudden jump in prometheus_tsdb_symbol_table_size_bytes on the Prometheus server itself; treat that as a regression in the instrumentation code, not a Prometheus tuning problem.
  • Pushgateway as a single point of failure — because it is not clustered by default, a Pushgateway outage silently drops every batch job’s metrics for the duration. Alert on Prometheus’s own up{job="pushgateway"} target health separately from the batch-job metrics it forwards.
  • Metrics never clear between runs — the Pushgateway keeps the last pushed value for a grouping key indefinitely, even after the job stops running entirely. Delete stale groups explicitly (DELETE to the Pushgateway’s /metrics/job/.../instance/... endpoint) as part of decommissioning a job, or a dashboard will keep showing a healthy last run that is actually months old.
  • Histogram buckets misaligned with SLAs — if the reconciliation_stage_latency_seconds buckets do not straddle the actual SLA boundary (for example, all buckets under 5s when the match stage sometimes runs 20s), histogram_quantile queries return misleading tails. Re-derive bucket boundaries from a week of real stage timings rather than guessing.

FAQ

Why use a Pushgateway instead of just scraping the batch job directly?

Prometheus’s normal model requires a target to be reachable and running at scrape time. A reconciliation batch job that runs for a few minutes on a schedule is usually not running when the next scrape interval lands, so a direct scrape target shows as down for most of its lifecycle and captures nothing. The Pushgateway sits in between: the job pushes its final metric values once at exit, and Prometheus scrapes the always-available Pushgateway on its normal interval instead of the ephemeral job.

How do I keep label cardinality under control for the match-score histogram?

Bucket the histogram itself rather than adding a label per score bucket, and keep the only label (match_type) a closed enum with a handful of values. The failure mode to avoid is adding a label for something high-cardinality like vendor_id or po_number directly on a Counter or Histogram — that turns a handful of time series into one per distinct value ever seen, and Prometheus never automatically expires old series from a still-active metric.

Should I delete pushed metrics after the job completes successfully?

No, not on every run — the whole point of the Pushgateway pattern is that the last-pushed values persist between runs so a dashboard can show “most recent run” state between executions. Only delete a group explicitly when a job is permanently retired; deleting after every successful run would recreate the exact gap-in-visibility problem the Pushgateway pattern exists to solve.