tutorial

How to Monitor Citus (Distributed PostgreSQL) with Vigilmon

Citus extends PostgreSQL with sharding and distributed query execution — but a coordinator crash or worker node failure silently breaks queries. Learn how to monitor Citus coordinator health, worker node availability, and distributed query routing with Vigilmon.

Citus turns PostgreSQL into a distributed database by adding a coordinator node and a fleet of worker nodes. This gives you horizontal scale for analytical workloads and large tenants — but it also means that a coordinator crash takes down your entire cluster, a worker node failure makes its shards unavailable, and a distributed query planner error can silently return partial results. Standard PostgreSQL monitoring catches none of this.

Vigilmon gives you external visibility into Citus cluster health through HTTP endpoint monitoring of your application's database health routes, plus heartbeat monitors for distributed ETL jobs and query workers. This tutorial walks through setting it up from scratch.

Why Citus Needs External Monitoring

A standard PostgreSQL uptime check tells you whether the coordinator is accepting connections. But in a Citus cluster, that's only the beginning:

  • Coordinator availability — if the coordinator node is down, all queries fail. But the coordinator can be up while worker nodes are degraded.
  • Worker node health — each worker node holds shards. If a worker goes down, any query touching its shards returns an error. Citus does not automatically reroute to replicas unless you've configured shard replication.
  • Distributed query health — a query can hang or return incomplete results if a worker is slow or if metadata tables (pg_dist_node, pg_dist_shard_placement) become inconsistent.
  • Rebalancer status — after adding or removing workers, Citus needs to rebalance shards. A stalled rebalancer leaves the cluster unevenly distributed.

External monitoring with Vigilmon adds:

  • Proactive alerting when your database health endpoint returns non-200 (indicating coordinator or worker issues)
  • Worker node count monitoring via a health endpoint that queries pg_dist_node
  • Heartbeat monitoring so you know when distributed jobs stop completing successfully
  • Multi-region probe consensus to prevent false positives from transient network partitions

Step 1: Build a Citus Health Endpoint

Unlike Vitess, Citus doesn't ship with a standalone HTTP health server. You need to expose one through your application or a lightweight sidecar.

Node.js / Express Health Sidecar

// health/citus.js
const express = require('express');
const { Pool } = require('pg');

const app = express();

const pool = new Pool({
  host: process.env.CITUS_COORDINATOR_HOST || 'localhost',
  port: parseInt(process.env.CITUS_PORT || '5432'),
  database: process.env.CITUS_DATABASE,
  user: process.env.CITUS_USER,
  password: process.env.CITUS_PASSWORD,
  max: 5,
  connectionTimeoutMillis: 3000,
});

app.get('/health/citus', async (req, res) => {
  const client = await pool.connect().catch(err => null);
  if (!client) {
    return res.status(503).json({ status: 'down', reason: 'coordinator_unreachable' });
  }

  try {
    // Check coordinator is alive
    await client.query('SELECT 1');

    // Check worker node count and status
    const workerResult = await client.query(`
      SELECT
        count(*) FILTER (WHERE isactive = true) AS active_workers,
        count(*) FILTER (WHERE isactive = false) AS inactive_workers,
        count(*) AS total_workers
      FROM pg_dist_node
      WHERE noderole = 'primary'
    `);

    const { active_workers, inactive_workers, total_workers } = workerResult.rows[0];

    if (parseInt(inactive_workers) > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'worker_nodes_inactive',
        active_workers: parseInt(active_workers),
        inactive_workers: parseInt(inactive_workers),
        total_workers: parseInt(total_workers),
      });
    }

    return res.status(200).json({
      status: 'ok',
      active_workers: parseInt(active_workers),
      total_workers: parseInt(total_workers),
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  } finally {
    client.release();
  }
});

app.get('/health/citus/query', async (req, res) => {
  const client = await pool.connect().catch(() => null);
  if (!client) {
    return res.status(503).json({ status: 'down', reason: 'coordinator_unreachable' });
  }

  try {
    // Run a distributed query — fails if any worker holding relevant shards is down
    const result = await client.query(
      'SELECT count(*) FROM pg_dist_shard_placement WHERE shardstate = 1'
    );
    const activePlacements = parseInt(result.rows[0].count);

    const totalResult = await client.query(
      'SELECT count(*) FROM pg_dist_shard_placement'
    );
    const totalPlacements = parseInt(totalResult.rows[0].count);

    if (activePlacements < totalPlacements) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'shard_placements_inactive',
        active: activePlacements,
        total: totalPlacements,
      });
    }

    return res.status(200).json({ status: 'ok', active_shard_placements: activePlacements });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  } finally {
    client.release();
  }
});

app.listen(3005);

Python / FastAPI Health Sidecar

# health/citus.py
from fastapi import FastAPI
from fastapi.responses import JSONResponse
import asyncpg
import os

app = FastAPI()

async def get_conn():
    return await asyncpg.connect(
        host=os.environ["CITUS_COORDINATOR_HOST"],
        port=int(os.environ.get("CITUS_PORT", 5432)),
        database=os.environ["CITUS_DATABASE"],
        user=os.environ["CITUS_USER"],
        password=os.environ["CITUS_PASSWORD"],
        timeout=3,
    )

@app.get("/health/citus")
async def citus_health():
    try:
        conn = await get_conn()
    except Exception as e:
        return JSONResponse({"status": "down", "reason": "coordinator_unreachable", "error": str(e)}, 503)

    try:
        row = await conn.fetchrow("""
            SELECT
                count(*) FILTER (WHERE isactive = true) AS active_workers,
                count(*) FILTER (WHERE isactive = false) AS inactive_workers
            FROM pg_dist_node
            WHERE noderole = 'primary'
        """)
        if row["inactive_workers"] > 0:
            return JSONResponse({
                "status": "degraded",
                "reason": "worker_nodes_inactive",
                "active_workers": row["active_workers"],
                "inactive_workers": row["inactive_workers"],
            }, 503)
        return {"status": "ok", "active_workers": row["active_workers"]}
    except Exception as e:
        return JSONResponse({"status": "down", "error": str(e)}, 503)
    finally:
        await conn.close()

Step 2: Configure Vigilmon HTTP Monitors for Citus

Primary Coordinator Monitor

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your Citus health endpoint: https://your-app.example.com/health/citus
  4. Set the check interval to 1 minute
  5. Under Expected response:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 5000ms
  6. Under Alert channels, assign your primary on-call Slack channel + PagerDuty
  7. Save the monitor

Shard Placement Monitor

Add a second monitor for distributed shard health:

  • URL: https://your-app.example.com/health/citus/query
  • Expected: 200, body contains "status":"ok"
  • Interval: 2 minutes
  • Alert channel: Slack (P2 — shard degradation before full outage)

Direct Coordinator TCP Monitor (Optional)

For low-level coordinator availability without an HTTP sidecar:

  • Monitor type: TCP
  • Host: citus-coordinator.internal
  • Port: 5432
  • Interval: 1 minute
  • This catches the coordinator process being completely down, even if your sidecar isn't deployed

Vigilmon's multi-region probes prevent false positives from network blips between Vigilmon's probe locations and your coordinator.

Step 3: Heartbeat Monitoring for Distributed ETL Jobs

Citus is commonly used for analytics pipelines, multi-tenant SaaS databases, and time-series data. These workloads often include background ETL jobs, partition maintenance tasks, and data aggregation workers. A worker stall doesn't raise a database error — it just stops producing results.

Vigilmon heartbeat monitors detect stalled jobs: each job sends a ping to Vigilmon upon successful completion. If pings stop, Vigilmon fires an alert.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Name it: citus-etl-worker
  3. Set expected interval: 30 minutes (adjust to your ETL schedule)
  4. Set grace period: 45 minutes
  5. Save — copy the heartbeat URL

Wire Into Your ETL Pipeline (Python)

# etl/aggregate_tenant_data.py
import asyncpg
import httpx
import os
import asyncio

VIGILMON_HEARTBEAT = os.environ["VIGILMON_HEARTBEAT_URL"]

async def aggregate_tenants():
    conn = await asyncpg.connect(
        host=os.environ["CITUS_COORDINATOR_HOST"],
        database=os.environ["CITUS_DATABASE"],
        user=os.environ["CITUS_USER"],
        password=os.environ["CITUS_PASSWORD"],
    )

    try:
        # Run distributed aggregation across all tenant shards
        await conn.execute("""
            INSERT INTO tenant_daily_summary (tenant_id, event_date, event_count)
            SELECT
                tenant_id,
                DATE_TRUNC('day', created_at) AS event_date,
                COUNT(*) AS event_count
            FROM events
            WHERE created_at >= NOW() - INTERVAL '1 day'
            GROUP BY tenant_id, DATE_TRUNC('day', created_at)
            ON CONFLICT (tenant_id, event_date) DO UPDATE
                SET event_count = EXCLUDED.event_count
        """)

        # Ping Vigilmon to confirm successful ETL run
        async with httpx.AsyncClient() as client:
            await client.get(VIGILMON_HEARTBEAT, timeout=5)

        print("ETL aggregation complete, heartbeat sent")
    except Exception as e:
        print(f"ETL failed: {e}")
        raise
    finally:
        await conn.close()

if __name__ == "__main__":
    asyncio.run(aggregate_tenants())

Wire Into Your Node.js Background Worker

// workers/citus_maintenance.js
const { Pool } = require('pg');
const axios = require('axios');

const pool = new Pool({
  host: process.env.CITUS_COORDINATOR_HOST,
  database: process.env.CITUS_DATABASE,
  user: process.env.CITUS_USER,
  password: process.env.CITUS_PASSWORD,
});

async function runMaintenance() {
  const client = await pool.connect();
  try {
    // Vacuum and reindex distributed tables on each worker
    await client.query(`
      SELECT run_command_on_workers($$VACUUM ANALYZE events$$)
    `);

    // Ping Vigilmon on success
    await axios.get(process.env.VIGILMON_HEARTBEAT_URL).catch(() => {});
    console.log('Maintenance complete');
  } catch (err) {
    console.error('Maintenance failed:', err.message);
  } finally {
    client.release();
  }
}

// Run every 6 hours
setInterval(runMaintenance, 6 * 60 * 60 * 1000);
runMaintenance();

Step 4: Alert Routing for Citus Failures

Citus failures have a specific blast radius depending on whether the coordinator or a worker fails:

| Failure | Impact | Alert Priority | |---|---|---| | Coordinator down | All queries fail | P1 | | Worker node down | Queries touching that worker's shards fail | P1 | | Shard placement inactive | Subset of data unavailable | P2 | | Rebalancer stalled | No new data on new workers | P3 |

Configure Vigilmon alert routing to match:

| Monitor | Alert Channel | Priority | |---|---|---| | App /health/citus (coordinator + workers) | Slack + PagerDuty | P1 | | App /health/citus/query (shard placements) | Slack | P2 | | TCP on port 5432 (coordinator) | PagerDuty | P1 | | Heartbeat: ETL worker | Slack + email | P2 |

Set response time thresholds for early warning:

  • Alert at 3000ms for the coordinator health endpoint (slow health query signals coordinator load)
  • Alert at 8000ms for the shard placement endpoint (distributed query latency climbing)

For multi-region Citus deployments, set up separate monitors per region and route to region-specific on-call channels so alerts go directly to the team nearest the affected datacenter.

Summary

Citus extends PostgreSQL with horizontal sharding — but that extension means standard PostgreSQL monitoring is no longer enough. External monitoring with Vigilmon covers the gap:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/citus | Coordinator health, active worker node count | | HTTP monitor on /health/citus/query | Shard placement availability | | TCP monitor on port 5432 | Raw coordinator process availability | | Heartbeat monitor | ETL job and maintenance worker liveness |

Get started free at vigilmon.online — your first Citus monitor is running in under two minutes.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles