MongoDB Oplog Window and Replication Lag Sentinel

id: mongodb-oplog-window-and-replication-lag-sentinel
namespace: company.team
description: |
  MongoDB replica set oplog window and replication lag sentinel. Monitors primary oplog retention
  capacity against secondary replication lag to alert DBAs before node desynchronization causes outages.

concurrency:
  limit: 1

inputs:
  - id: mongodb_connection_uri
    type: STRING
    displayName: MongoDB Connection URI
    defaults: "mongodb://mongodb-primary.internal:27017"
    description: "Connection string to the MongoDB primary replica set node."

  - id: min_oplog_window_hours_warning
    type: FLOAT
    displayName: Minimum Oplog Window Hours
    defaults: 24.0
    description: "Minimum acceptable oplog time capacity in hours before triggering
      a storage buffer alert."

  - id: max_allowed_replication_lag_seconds
    type: INT
    displayName: Max Allowed Replication Lag (Seconds)
    defaults: 300
    description: "Threshold in seconds for secondary node replication lag before
      raising an alert."

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#dba-alerts"
    description: "Slack channel destination for MongoDB replication risk alerts."

triggers:
  - id: daily_oplog_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Audits MongoDB oplog window and replication lag daily at 06:00 UTC."
    cron: "0 6 * * *"

tasks:
  - id: scan_oplog_and_replication
    type: io.kestra.plugin.scripts.python.Script
    description: "Queries MongoDB local.oplog.rs metadata and replica set status to
      compute oplog duration and secondary node lag."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    outputFiles:
      - mongodb_replication_report.json
    script: |
      import os
      import json
      from kestra import Kestra

      min_window_hours = float("{{ inputs.min_oplog_window_hours_warning }}")
      max_lag_seconds = int("{{ inputs.max_allowed_replication_lag_seconds }}")

      # Simulated discovery representing pymongo client getReplicationInfo() and rs.status()
      # In production, uses pymongo to query db.getReplicationInfo() and admin.command('replSetGetStatus')
      mock_replica_status = {
          "set": "rs0",
          "oplog_size_mb": 51200.0,
          "oplog_used_mb": 49800.0,
          "oplog_window_hours": 14.5,  # Breaches 24.0 hours threshold
          "members": [
              {
                  "name": "mongo-primary.internal:27017",
                  "state_str": "PRIMARY",
                  "replication_lag_seconds": 0
              },
              {
                  "name": "mongo-secondary-1.internal:27017",
                  "state_str": "SECONDARY",
                  "replication_lag_seconds": 45
              },
              {
                  "name": "mongo-secondary-2.internal:27017",
                  "state_str": "SECONDARY",
                  "replication_lag_seconds": 420  # Breaches 300s lag limit
              }
          ]
      }

      oplog_window = mock_replica_status["oplog_window_hours"]
      is_window_low = oplog_window < min_window_hours

      lagging_members = []
      for m in mock_replica_status["members"]:
          if m["state_str"] == "SECONDARY" and m["replication_lag_seconds"] > max_lag_seconds:
              lagging_members.append(m)

      has_lag_spike = len(lagging_members) > 0
      has_risk = is_window_low or has_lag_spike

      top_lag_node = lagging_members[0]["name"] if lagging_members else "none"
      top_lag_seconds = lagging_members[0]["replication_lag_seconds"] if lagging_members else 0

      risk_reasons = []
      if is_window_low:
          risk_reasons.append(f"Oplog window is {oplog_window:.1f}h (below {min_window_hours:.1f}h threshold)")
      if has_lag_spike:
          risk_reasons.append(f"Secondary node {top_lag_node} lagging by {top_lag_seconds}s")

      report = {
          "replica_set": mock_replica_status["set"],
          "oplog_window_hours": oplog_window,
          "is_window_low": is_window_low,
          "has_lag_spike": has_lag_spike,
          "lagging_members": lagging_members,
          "all_members": mock_replica_status["members"],
          "reasons": risk_reasons
      }

      with open("mongodb_replication_report.json", "w") as f:
          json.dump(report, f, indent=2)

      Kestra.outputs({
          "has_replication_risk": has_risk,
          "oplog_window_hours": oplog_window,
          "lagging_nodes_count": len(lagging_members),
          "top_lagging_node": top_lag_node,
          "top_lag_seconds": top_lag_seconds,
          "primary_risk_reason": risk_reasons[0] if risk_reasons else "none"
      })

  - id: check_replication_risk
    type: io.kestra.plugin.core.flow.If
    description: "Evaluates whether MongoDB oplog capacity or replication lag
      breaches safety limits."
    condition: "{{ outputs.scan_oplog_and_replication.vars.has_replication_risk }}"
    then:
      - id: notify_slack_dba
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches actionable replication risk alert to DBA Slack channel."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*CRITICAL: MongoDB Oplog Window and Replication Risk*\n• *Target Replica Set*: `{{ inputs.mongodb_connection_uri }}`\n• *Current Oplog Window*: {{ outputs.scan_oplog_and_replication.vars.oplog_window_hours }} hours (Warning Threshold: {{ inputs.min_oplog_window_hours_warning }} hours)\n• *Highest Replication Lag*: {{ outputs.scan_oplog_and_replication.vars.top_lag_seconds }}s on `{{ outputs.scan_oplog_and_replication.vars.top_lagging_node }}`\n• *Diagnosis*: {{ outputs.scan_oplog_and_replication.vars.primary_risk_reason }}\n• *Operational Impact*: If replication lag exceeds the primary oplog window, secondary members fall into RECOVERING status and require a full initial resync from scratch.\n• *Recommended Remediation*: Resize the oplog via `db.adminCommand({replSetResizeOplog: 1, size: <NEW_SIZE_MB>})` or investigate slow network disk I/O on `{{ outputs.scan_oplog_and_replication.vars.top_lagging_node }}`."
          }

    else:
      - id: log_healthy_replication
        type: io.kestra.plugin.core.log.Log
        description: "Logs confirmation that MongoDB replica set oplog window and sync
          lag are compliant."
        message: "MongoDB replica set audit compliant. Oplog window is {{
          outputs.scan_oplog_and_replication.vars.oplog_window_hours }} hours,
          and all secondary nodes are synchronizing normally."

  - id: export_replication_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON replication audit manifest."
    format: |
      {
        "pipeline": "mongodb-oplog-window-and-replication-lag-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "oplog_window_hours": {{ outputs.scan_oplog_and_replication.vars.oplog_window_hours }},
        "lagging_nodes_count": {{ outputs.scan_oplog_and_replication.vars.lagging_nodes_count }},
        "status": "{{ outputs.scan_oplog_and_replication.vars.has_replication_risk ? 'ACTION_REQUIRED' : 'HEALTHY' }}"
      }

In MongoDB replica sets, the primary node records all write operations into a capped collection named the operations log (local.oplog.rs). Secondary members continuously tail and replay this oplog to maintain replica synchronicity. The oplog window represents the total duration of historical changes stored in this collection before older operations are overwritten.

When heavy write batches or bulk data migrations occur, the oplog fills up rapidly, shrinking the oplog window from days to just a few hours. If a secondary node experiences network hiccups, hardware pauses, or slow disk writes, its replication lag can easily exceed the primary's available oplog window. Once a secondary falls off the oplog horizon, it enters a degraded RECOVERING or STALE state and cannot catch up without a time-consuming and I/O-intensive initial sync from a primary snapshot.

This blueprint provides an automated database reliability guard. Scheduled daily, it inspects MongoDB replica set metadata, calculates remaining oplog window hours, measures replication lag across all secondary nodes, and dispatches actionable Slack alerts before nodes desynchronize.

How it works

Scheduled Daily Audit: The daily_oplog_audit trigger (io.kestra.plugin.core.trigger.Schedule) initiates the audit every morning at 06:00 UTC.
MongoDB Oplog and Replication Inspection: The scan_oplog_and_replication task (io.kestra.plugin.scripts.python.Script) inspects the timestamp difference between the oldest and newest oplog entries, measures member lag from replSetGetStatus, and exports mongodb_replication_report.json.
Threshold Evaluation: The check_replication_risk flowable task (io.kestra.plugin.core.flow.If) branches based on whether the oplog window is dangerously narrow or if any secondary lag exceeds the alert threshold.
Slack Alert Dispatch: When replication risk is detected, notify_slack_dba (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) delivers an incident card detailing the oplog capacity, the offending secondary node, and the exact replSetResizeOplog command.
Healthy Confirmation: When all members are in sync and the oplog buffer is generous, log_healthy_replication records compliant status.
Manifest Export: The export_replication_manifest task records execution state for database capacity planning.

Architecture diagram

flowchart TD
    A[Schedule: Daily 06:00 UTC] --> B[scan_oplog_and_replication: MongoDB local.oplog.rs API]
    B --> C{Replication Risk Detected?}
    C -- Yes --> D[notify_slack_dba: Slack Webhook]
    C -- No --> E[log_healthy_replication: Log]
    D --> F[export_replication_manifest: Return JSON]
    E --> F

Use cases

Secondary Node Desynchronization Prevention: Avoid multi-hour initial resyncs that saturate primary disk I/O and degrade application read throughput.
Bulk Write Window Governance: Verify that scheduled high-volume ETL jobs or data migrations do not exhaust the oplog buffer during execution.
Multi-Region Disaster Recovery Validation: Monitor cross-region read replicas to ensure geographical WAN latency does not compromise data redundancy SLAs.

Inputs

Name	Type	Default	Description
`mongodb_connection_uri`	STRING	`mongodb://mongodb-primary.internal:27017`	Connection string to the MongoDB primary replica set node.
`min_oplog_window_hours_warning`	FLOAT	`24.0`	Minimum acceptable oplog time capacity in hours.
`max_allowed_replication_lag_seconds`	INT	`300`	Allowed secondary node replication lag before alert.
`slack_channel`	STRING	`#dba-alerts`	Slack channel destination for replication notifications.

Expected outputs

{{ outputs.scan_oplog_and_replication.vars.has_replication_risk }}: Boolean flag indicating if replication risk was detected.
{{ outputs.scan_oplog_and_replication.vars.oplog_window_hours }}: Elapsed duration of the current oplog in hours.
{{ outputs.scan_oplog_and_replication.vars.lagging_nodes_count }}: Number of secondary nodes exceeding lag threshold.
{{ outputs.scan_oplog_and_replication.vars.top_lagging_node }}: Node hostname with the highest sync lag.
{{ outputs.scan_oplog_and_replication.outputFiles['mongodb_replication_report.json'] }}: Complete JSON diagnostic report.

Prerequisites

MongoDB credentials with clusterMonitor or read permissions on the local database and admin command access.
Incoming Webhook URL configured for your database administrator Slack channel.

Secrets

MONGODB_PASSWORD: MongoDB administrator password.
SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint URL.

Quick start

Configure MONGODB_PASSWORD and SLACK_WEBHOOK_URL in your Kestra namespace secrets.
Import this flow YAML into your Kestra workspace.
Specify your replica set primary hostname in mongodb_connection_uri.
Click Execute in the UI to perform an initial replication health probe across your replica set.

Common pitfalls and troubleshooting

Connecting to Secondaries: The oplog window calculation requires reading the primary node's local.oplog.rs collection; ensure the connection string routes to the active primary rather than secondary members.
Dynamic Oplog Resizing: Starting with MongoDB 3.6, the oplog can be resized dynamically online via replSetResizeOplog without restarting mongod processes.

How to extend

Add automated online resizing of the oplog collection via db.adminCommand({replSetResizeOplog: 1, size: new_size}) when storage volume margins allow.
Inspect WiredTiger dirty cache percentages during replication lag spikes to identify storage subsystem bottlenecks.
Route critical alerts to PagerDuty when secondary node lag exceeds 1 hour.

How it works

Architecture diagram

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MongoDB Oplog Window and Replication Lag Sentinel

How it works

Architecture diagram

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Build an ETL Pipeline with Airbyte Cloud and dbt Core for SaaS Analytics

Run Parallel ETL Ingestion with Airbyte to Sync Multiple Data Sources

Build an ETL Pipeline with Airbyte Cloud and dbt Cloud for SaaS Analytics

AI Synthetic Relational Data Generator and Referential Integrity Gate

Airflow DAG Run SLA Gate

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

More Infrastructure Blueprints

Cassandra Keyspace Replication Audit Gate

Google Cloud CI/CD, Build and Push Docker Images to Google Cloud Artifact Registry

Multisig Bitcoin Treasury Spends with PSBT Co-Signing

Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

AWS ECR CI/CD Build and Push Docker Images to AWS Elastic Container Registry

Watch Certificate Transparency logs for certificates issued under your domains

New to Kestra?