Alerting on Replication Lag with _scheduler/jobs #

A replication job reads running, its checkpoint is advancing, and yet the target is an hour behind the source because writes arrive faster than the worker can drain them — a lag that no crash alarm will ever catch. This page shows how to turn that backlog into a single, trustworthy alert: read the source update_seq, read the job’s through_seq and changes_pending from _scheduler/jobs, compute lag correctly without abusing opaque sequence strings, and fire only after threshold, warm-up, and flap-damping gates agree. It is the lag-detection companion to Async Monitoring & Webhooks; where Streaming Replication Events to Webhooks fires on state transitions, this page fires on a job that never changes state but quietly falls behind.

Immediate Triage / Prerequisites #

Before defining a single threshold, confirm what “lag” actually means for your job, because the obvious approach — subtracting sequence strings — is wrong. Read one job’s scheduler entry and look at the fields you will build on:

# the fields that matter for lag: changes_pending, through_seq, and the job state
curl -s http://localhost:5984/_scheduler/jobs | \
  python3 -c "import sys,json;[print(j['id'],j['info'].get('changes_pending'),j['info'].get('through_seq')) for j in json.load(sys.stdin)['jobs']]"

# the source head, to distinguish an idle job (nothing to do) from a stalled one
curl -s http://localhost:5984/sensor_data | \
  python3 -c "import sys,json;print('update_seq',json.load(sys.stdin)['update_seq'])"

The single most important fact: changes_pending is CouchDB’s own count of documents the job still has to process, and it is the correct primary lag signal. The sequence values (through_seq, update_seq) are opaque packed strings whose only portable part is a leading integer that is not a document count — you cannot meaningfully subtract two nodes’ sequences to get “how far behind” in docs. Use changes_pending for the magnitude, and use the source update_seq only to tell an idle job (changes_pending is 0 or absent, nothing to do) apart from a stalled one (changes_pending high and not falling). Prerequisites for the code below: Python 3.8+ with requests (pip install requests), admin credentials for _scheduler/jobs, and a target alert webhook URL. One more rule: the sampler’s own wall clock is the only clock you can trust for rates — never compare timestamps across the source and target nodes.

Step-by-Step Implementation #

Follow these steps to sample a job, compute lag, gate it, and emit an alert. Each step has a check so you can verify state before moving on.

1. Read the source update_seq. A GET on the source database returns update_seq, the head of its _changes feed. Extract only the integer prefix, and treat it as a coarse liveness indicator, not a doc count:

   import requests
   head = requests.get("http://localhost:5984/sensor_data").json()["update_seq"]
   source_prefix = int(str(head).split("-", 1)[0])   # leading int only; the tail is opaque
   assert source_prefix >= 0

2. Read the job’s changes_pending and through_seq. Pull the matching _scheduler/jobs entry. changes_pending is the backlog magnitude; through_seq tells you the job is still advancing between samples:

   jobs = requests.get("http://localhost:5984/_scheduler/jobs", auth=("admin", "password")).json()["jobs"]
   job = next(j for j in jobs if j["id"].startswith("rep_edge_to_cloud"))
   pending = job["info"].get("changes_pending") or 0

   Verify changes_pending is present. If the key is absent the job is caught up — that is 0 lag, not missing data.

3. Compute lag and classify idle vs stalled. Lag magnitude is changes_pending. Distinguish states: pending == 0 is caught up; pending > 0 and through_seq advancing between samples is draining (healthy under load); pending > 0 and through_seq frozen is stalled. Only sustained high lag or a stall warrants an alert.

4. Evaluate against a threshold with hysteresis and warm-up. Fire only when lag crosses a high water mark, and clear only when it falls below a lower one, so a value hovering at the boundary does not oscillate. Suppress everything until the job has been observed for a warm-up period — a job that just started legitimately shows a large changes_pending while it does its initial scan:

   firing = lag > HI if not currently_firing else lag > LO   # hysteresis band [LO, HI]
   if job_age_seconds < WARMUP_SECONDS:
       firing = False                                        # ignore initial catch-up

5. Flap-damp, then emit. Require the breach to persist across several consecutive samples before paging, and require several clean samples before resolving. Only after all gates agree do you POST an alert payload to your Alertmanager- or PagerDuty-style receiver.

Complete Working Example #

The script below is self-contained and runnable. It samples every replication job from _scheduler/jobs, uses changes_pending as the lag magnitude, and gates each job through warm-up, a hysteresis band, and consecutive-breach flap damping before emitting a de-duplicated alert payload. It carries per-job state between samples in memory, uses only the sampler’s own clock for timing, and never subtracts opaque sequence strings.

import sys
import time
from dataclasses import dataclass, field

import requests


@dataclass
class JobState:
    """Per-job alert state carried across samples."""
    firing: bool = False
    breach_streak: int = 0
    clear_streak: int = 0
    first_seen: float = field(default_factory=time.time)
    last_through_seq: str | None = None


def seq_prefix(seq) -> int:
    """Leading integer of a CouchDB sequence; the opaque tail is never compared."""
    return int(str(seq).split("-", 1)[0]) if seq is not None else 0


def evaluate(states, jobs, *, hi=5000, lo=1000, warmup=120,
             breach_needed=3, clear_needed=3, now=None):
    """Yield an alert action per job whose gated lag state changed.

    Lag magnitude is changes_pending (a real document count). update_seq /
    through_seq are used only to tell idle from stalled, never subtracted.
    """
    now = now or time.time()
    alerts = []
    for job in jobs:
        jid = job["id"]
        info = job.get("info") or {}
        pending = info.get("changes_pending") or 0
        through = info.get("through_seq")
        st = states.setdefault(jid, JobState())

        advancing = through != st.last_through_seq
        st.last_through_seq = through
        stalled = pending > 0 and not advancing          # backlog and not moving
        age = now - st.first_seen

        # Warm-up: a freshly scheduled job legitimately shows a big initial backlog.
        if age < warmup:
            continue

        # Hysteresis band: fire above hi, clear only below lo.
        over = pending > (lo if st.firing else hi) or stalled

        if over:
            st.breach_streak += 1
            st.clear_streak = 0
        else:
            st.clear_streak += 1
            st.breach_streak = 0

        # Flap damping: require N consecutive samples before flipping state.
        if not st.firing and st.breach_streak >= breach_needed:
            st.firing = True
            alerts.append(_payload(jid, pending, stalled, "firing"))
        elif st.firing and st.clear_streak >= clear_needed:
            st.firing = False
            alerts.append(_payload(jid, pending, stalled, "resolved"))
    return alerts


def _payload(job_id, pending, stalled, status):
    return {
        "labels": {"alertname": "CouchReplicationLag", "job": job_id,
                   "severity": "critical" if stalled else "warning"},
        "annotations": {"summary": f"replication lag on {job_id}",
                        "description": f"changes_pending={pending} stalled={stalled}"},
        "status": status,   # firing | resolved -> Alertmanager-compatible
        "generatorURL": f"http://localhost:5984/_scheduler/jobs",
    }


def sample_jobs(base_url, session):
    return session.get(f"{base_url}/_scheduler/jobs", timeout=30).json()["jobs"]


def send_alerts(webhook_url, alerts, session):
    if alerts:  # Alertmanager accepts a JSON array of alert objects
        session.post(webhook_url, json=alerts, timeout=10).raise_for_status()


if __name__ == "__main__":
    base = "http://localhost:5984"
    webhook = "https://alertmanager.example/api/v2/alerts"
    s = requests.Session()
    s.auth = ("admin", "password")
    states: dict[str, JobState] = {}
    while True:
        try:
            jobs = sample_jobs(base, s)
            fired = evaluate(states, jobs)
            send_alerts(webhook, fired, s)
            for a in fired:
                print(a["status"], a["labels"]["job"], a["annotations"]["description"])
        except requests.RequestException as exc:
            print("sample failed:", exc, file=sys.stderr)
        time.sleep(30)   # sampler cadence; also the flap-damping time unit

Run it against a healthy pipeline and it stays quiet; drive a job into a sustained backlog and after the warm-up and consecutive-breach gates pass it emits one firing payload, then exactly one resolved payload when the backlog drains — no per-sample noise.

Gotchas & Edge Cases #

• Do not do arithmetic on sequence strings. through_seq and update_seq are opaque packed tokens; their leading integer is not a document count and differs in meaning across nodes. Use changes_pending for magnitude. If you truly need a seq-based check, only compare a job’s own through_seq to itself across samples to test whether it is advancing.
• Idle is not stalled. A job with changes_pending of 0 (or the key absent) has nothing to do and is healthy, even though its checkpoint stops moving. Gate every alert on changes_pending > 0, or you will page on quiet databases at night.
• Per-job and aggregate lag answer different questions. Alert per job to find which link is behind; also emit an aggregate (sum of changes_pending) to catch a fleet-wide slowdown that no single job trips. Do not replace one with the other — a single hot job can hide inside a low aggregate.
• Warm-up prevents false alarms on restart. A freshly scheduled or rescheduled job shows a large changes_pending during its initial scan. Without a warm-up window every deploy or crashing → running recovery pages you. Tie the warm-up to when the sampler first saw the job, not to CouchDB’s timestamps.
• The sampler’s clock is the only trustworthy one. Rates, warm-up ages, and streak timers must be measured on the machine running the evaluator. Source and target node clocks can skew; mixing them into lag math produces phantom alerts — the same clock-skew failure mode that manufactures conflicts in offline-first pipelines.

Verification & Observability #

Confirm the alarm behaves before you trust it on call. Drive a backlog by writing faster than replication drains, then watch changes_pending climb and the alert fire exactly once:

# watch the backlog magnitude climb and fall (the value the alarm keys on)
watch -n 5 "curl -s http://localhost:5984/_scheduler/jobs | \
  python3 -c \"import sys,json;[print(j['id'],j['info'].get('changes_pending')) for j in json.load(sys.stdin)['jobs']]\""

# confirm the job is still advancing (draining) rather than truly stalled
curl -s http://localhost:5984/_scheduler/jobs | \
  python3 -c "import sys,json;[print(j['id'],j['info'].get('through_seq')) for j in json.load(sys.stdin)['jobs']]"

A correct configuration produces one firing alert per sustained backlog and one resolved when it clears, with no flapping while lag hovers near the threshold. Export changes_pending per job and its aggregate as gauges so you can chart lag over time and tune HI/LO from real percentiles rather than guesses; the exporter for exactly these gauges is built in Prometheus Metrics & Exporter Integration for CouchDB Replication. To deliver these alerts reliably with signing and retries rather than a single best-effort POST, route them through the machinery in Streaming Replication Events to Webhooks; to catch the different failure of a running job whose checkpoint has frozen, add Monitoring CouchDB Replication Checkpoints via the API.

FAQ #

Related #

• Async Monitoring & Webhooks
• Streaming Replication Events to Webhooks
• Monitoring CouchDB Replication Checkpoints via the API

Part of: Async Monitoring & Webhooks