tutorial

How to Monitor ZeroMQ with Vigilmon

ZeroMQ is a high-performance async messaging library — not a broker — which means there's no central server to ping. Learn how to monitor ZeroMQ socket health, queue fill rates, message drops, context lifecycle, and pattern-specific metrics (pub/sub, push/pull, req/rep) with Vigilmon HTTP and heartbeat monitors.

ZeroMQ (ØMQ) is not a message broker — it's a high-performance asynchronous messaging library that turns sockets into distributed computing primitives. There's no central server to restart, no admin UI to log into, and no built-in health endpoint to probe. Your ZeroMQ topology is as healthy as its constituent processes are, and when one silently falls behind or stops processing, the only way to know is by instrumenting your own code.

This is exactly where Vigilmon fits: you add lightweight health and heartbeat instrumentation to your ZeroMQ workers, expose HTTP endpoints, and let Vigilmon probe them from outside your process. This tutorial shows you how to monitor socket health, queue fill rates, message drops, context lifecycle, and the three core ZeroMQ patterns — pub/sub, push/pull, and req/rep — with Vigilmon HTTP probes and heartbeat monitors.

Why ZeroMQ Needs External Monitoring

ZeroMQ's brokerless design means there's no single point of failure to monitor — but it also means failures are distributed and invisible:

  • A PUSH socket can silently drop messages when a downstream PULL socket's high-water mark is full
  • A PUB socket continues sending to zero subscribers without error
  • A REP socket that crashes mid-request leaves the paired REQ socket hanging forever
  • A context with open sockets that lost their peer connections sits idle consuming memory

External monitoring with Vigilmon surfaces these conditions by probing health endpoints you expose from your ZeroMQ worker processes:

  • Socket health — are all expected sockets bound and connected?
  • Queue fill rate — are high-water marks being approached?
  • Message drop counters — are messages being silently discarded?
  • Context lifecycle — is the ZeroMQ context alive?
  • Heartbeat pings — is each worker actively processing messages?

Step 1: Instrument Your ZeroMQ Workers

Add a lightweight Flask (or Express) health server alongside your ZeroMQ sockets. This server runs in a background thread and does not interfere with the ZeroMQ context.

Python Worker with Health Endpoint

# worker.py
import zmq
import threading
import time
import os
import requests
from flask import Flask, jsonify

# --- Shared metrics (updated by ZMQ threads) ---
metrics = {
    'context_alive': False,
    'sockets': {},
    'messages_processed': 0,
    'messages_dropped': 0,
    'last_message_at': None,
}
metrics_lock = threading.Lock()

# --- Health server ---
health_app = Flask(__name__)

@health_app.route('/health/zmq')
def health():
    with metrics_lock:
        m = dict(metrics)

    issues = []
    if not m['context_alive']:
        return jsonify(status='down', reason='context_dead'), 503

    for name, info in m['sockets'].items():
        if not info.get('bound') and not info.get('connected'):
            issues.append(f"socket_{name}_not_ready")

    stale_threshold = int(os.environ.get('ZMQ_STALE_SECONDS', '120'))
    if m['last_message_at'] is not None:
        age = time.time() - m['last_message_at']
        if age > stale_threshold:
            issues.append(f"no_messages_for_{int(age)}s")

    if m['messages_dropped'] > 0:
        issues.append(f"dropped_{m['messages_dropped']}_messages")

    if issues:
        return jsonify(status='degraded', issues=issues, **m), 503

    return jsonify(
        status='ok',
        messages_processed=m['messages_processed'],
        messages_dropped=m['messages_dropped'],
        context_alive=m['context_alive'],
    ), 200

def run_health_server():
    health_app.run(host='0.0.0.0', port=3008, use_reloader=False)

# --- ZeroMQ PULL worker example ---
def run_pull_worker(ctx):
    socket = ctx.socket(zmq.PULL)
    socket.set_hwm(1000)  # high-water mark: 1000 messages queued max
    socket.bind(f"tcp://*:{os.environ.get('ZMQ_PULL_PORT', '5555')}")

    with metrics_lock:
        metrics['sockets']['pull'] = {'bound': True, 'port': 5555}

    VIGILMON_URL = os.environ.get('VIGILMON_HEARTBEAT_URL', '')
    last_heartbeat = time.time()
    HEARTBEAT_INTERVAL = 60

    while True:
        try:
            # Non-blocking poll: check for messages every 100ms
            if socket.poll(timeout=100):
                msg = socket.recv()
                process_message(msg)
                with metrics_lock:
                    metrics['messages_processed'] += 1
                    metrics['last_message_at'] = time.time()

            now = time.time()
            if VIGILMON_URL and now - last_heartbeat > HEARTBEAT_INTERVAL:
                try:
                    requests.get(VIGILMON_URL, timeout=3)
                except Exception:
                    pass
                last_heartbeat = now

        except zmq.ZMQError as e:
            if e.errno == zmq.ETERM:
                break  # context was terminated

def process_message(msg):
    pass  # your processing logic

if __name__ == '__main__':
    ctx = zmq.Context.instance()
    with metrics_lock:
        metrics['context_alive'] = True

    # Start health server in background thread
    t_health = threading.Thread(target=run_health_server, daemon=True)
    t_health.start()

    # Run ZMQ worker
    try:
        run_pull_worker(ctx)
    finally:
        with metrics_lock:
            metrics['context_alive'] = False
        ctx.term()

Step 2: Pattern-Specific Health Endpoints

Different ZeroMQ patterns have different failure modes. Add pattern-specific endpoints for each:

PUB/SUB — Subscriber Health

# pub_sub_subscriber.py
import zmq, threading, time, os, requests
from flask import Flask, jsonify

sub_metrics = {
    'connected': False,
    'messages_received': 0,
    'last_received_at': None,
    'subscribed_topics': [],
}

health_app = Flask(__name__)

@health_app.route('/health/zmq/sub')
def sub_health():
    m = dict(sub_metrics)
    issues = []

    if not m['connected']:
        return jsonify(status='down', reason='not_connected_to_publisher'), 503

    stale = int(os.environ.get('ZMQ_SUB_STALE_SECONDS', '30'))
    if m['last_received_at'] and time.time() - m['last_received_at'] > stale:
        issues.append(f"no_messages_for_{int(time.time()-m['last_received_at'])}s")

    if not m['subscribed_topics']:
        issues.append('no_active_subscriptions')

    if issues:
        return jsonify(status='degraded', issues=issues, **m), 503
    return jsonify(status='ok', **m), 200

def run_subscriber():
    ctx = zmq.Context.instance()
    socket = ctx.socket(zmq.SUB)
    socket.connect(f"tcp://{os.environ['ZMQ_PUB_HOST']}:5556")
    socket.setsockopt(zmq.SUBSCRIBE, b'')  # subscribe to all topics
    sub_metrics['connected'] = True
    sub_metrics['subscribed_topics'] = ['all']

    while True:
        if socket.poll(timeout=1000):
            msg = socket.recv_multipart()
            process_event(msg)
            sub_metrics['messages_received'] += 1
            sub_metrics['last_received_at'] = time.time()

def process_event(msg):
    pass

REQ/REP — Request Timeout Monitor

@health_app.route('/health/zmq/req')
def req_health():
    ctx = zmq.Context.instance()
    socket = ctx.socket(zmq.REQ)
    socket.setsockopt(zmq.RCVTIMEO, 3000)   # 3s recv timeout
    socket.setsockopt(zmq.SNDTIMEO, 1000)   # 1s send timeout
    socket.setsockopt(zmq.LINGER, 0)        # don't block on close

    try:
        socket.connect(f"tcp://{os.environ['ZMQ_REP_HOST']}:5557")
        socket.send(b'ping')
        response = socket.recv()
        socket.close()
        if response == b'pong':
            return jsonify(status='ok', roundtrip_ms='<3000'), 200
        return jsonify(status='degraded', reason='unexpected_response'), 503
    except zmq.Again:
        socket.close()
        return jsonify(status='down', reason='req_rep_timeout'), 503
    except Exception as e:
        socket.close()
        return jsonify(status='down', error=str(e)), 503

PUSH/PULL — Queue Fill Approximation

ZeroMQ does not expose its internal send queue depth via the API, but you can track it by counting sent-vs-acknowledged messages or by monitoring high-water mark behavior:

push_metrics = {
    'messages_pushed': 0,
    'hwm': 1000,
    'hwm_hit_count': 0,
}

@health_app.route('/health/zmq/push')
def push_health():
    m = dict(push_metrics)
    if m['hwm_hit_count'] > 0:
        return jsonify(
            status='degraded',
            reason='hwm_pressure',
            hwm_hits=m['hwm_hit_count'],
            messages_pushed=m['messages_pushed'],
        ), 503
    return jsonify(status='ok', messages_pushed=m['messages_pushed']), 200

# In your PUSH worker:
def push_message(socket, msg):
    try:
        socket.send(msg, zmq.NOBLOCK)
        push_metrics['messages_pushed'] += 1
    except zmq.Again:
        # HWM reached — downstream PULL is not consuming fast enough
        push_metrics['hwm_hit_count'] += 1
        push_metrics['messages_dropped'] += 1

Step 3: Message Drop Monitoring

ZeroMQ drops messages silently when high-water marks are reached. Track this at the application level:

# drop_tracker.py
import zmq, os

class DroppingPushSocket:
    def __init__(self, ctx, endpoint, hwm=1000):
        self.socket = ctx.socket(zmq.PUSH)
        self.socket.set_hwm(hwm)
        self.socket.bind(endpoint)
        self.dropped = 0
        self.sent = 0

    def send(self, msg, flags=0):
        try:
            self.socket.send(msg, zmq.NOBLOCK | flags)
            self.sent += 1
        except zmq.Again:
            self.dropped += 1

    @property
    def drop_rate(self):
        total = self.sent + self.dropped
        return self.dropped / total if total > 0 else 0.0

Expose this through your health endpoint:

@health_app.route('/health/zmq/drops')
def drop_health():
    rate = push_socket.drop_rate if push_socket else 0.0
    if rate > 0.01:  # alert above 1% drop rate
        return jsonify(
            status='degraded',
            drop_rate_pct=round(rate * 100, 2),
            dropped=push_socket.dropped,
            sent=push_socket.sent,
        ), 503
    return jsonify(
        status='ok',
        drop_rate_pct=round(rate * 100, 2),
    ), 200

Step 4: Configure Vigilmon HTTP Monitors

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

Add pattern-specific monitors:

| Monitor URL | Pattern | Purpose | Interval | |---|---|---|---| | /health/zmq | All | Context health, drop count | 1 min | | /health/zmq/sub | PUB/SUB | Subscriber liveness, last message | 1 min | | /health/zmq/req | REQ/REP | Round-trip timeout probe | 2 min | | /health/zmq/push | PUSH/PULL | HWM pressure detection | 1 min | | /health/zmq/drops | All | Message drop rate | 1 min |

Step 5: Heartbeat Monitoring for ZeroMQ Workers

For ZeroMQ workers running in long-lived processes — especially PULL consumers that might silently stop consuming after a transient error — Vigilmon heartbeat monitors provide a critical liveness check beyond what a socket health endpoint can detect.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: zmq-event-processor
  3. Set the expected interval: 5 minutes
  4. Set the grace period: 10 minutes
  5. Save — copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123xyz

The heartbeat logic is already integrated in the PULL worker example from Step 1. For PUSH workers (senders), ping Vigilmon after each successful batch:

# In your PUSH worker
VIGILMON_URL = os.environ.get('VIGILMON_HEARTBEAT_URL', '')
last_heartbeat = time.time()

def send_batch(socket, messages):
    global last_heartbeat
    for msg in messages:
        push_message(socket, msg)

    now = time.time()
    if VIGILMON_URL and now - last_heartbeat > 60:
        try:
            requests.get(VIGILMON_URL, timeout=3)
        except Exception:
            pass
        last_heartbeat = now

Node.js Worker (zeromq.js)

// worker.js
const zmq = require('zeromq');
const axios = require('axios');

const sock = new zmq.Pull();
const VIGILMON_URL = process.env.VIGILMON_HEARTBEAT_URL;
let msgCount = 0;
let lastHeartbeat = Date.now();

const metrics = {
  contextAlive: true,
  messagesProcessed: 0,
  lastMessageAt: null,
};

// Express health server
const express = require('express');
const healthApp = express();
healthApp.get('/health/zmq', (req, res) => {
  const stale = Date.now() - (metrics.lastMessageAt || 0) > 120_000;
  if (!metrics.contextAlive) return res.status(503).json({ status: 'down' });
  if (stale && metrics.lastMessageAt) return res.status(503).json({
    status: 'degraded', reason: 'no_messages_120s',
  });
  res.json({ status: 'ok', messages: metrics.messagesProcessed });
});
healthApp.listen(3008);

async function run() {
  await sock.bind(`tcp://*:${process.env.ZMQ_PORT || '5555'}`);

  for await (const [msg] of sock) {
    processMessage(msg);
    metrics.messagesProcessed++;
    metrics.lastMessageAt = Date.now();
    msgCount++;

    const now = Date.now();
    if (now - lastHeartbeat > 60_000) {
      axios.get(VIGILMON_URL).catch(() => {});
      lastHeartbeat = now;
    }
  }
}

run().catch(err => {
  metrics.contextAlive = false;
  console.error(err);
});

Step 6: Alert Routing

| Monitor | Alert Channel | Priority | |---|---|---| | ZMQ context /health/zmq | Slack + PagerDuty | P1 | | REQ/REP timeout /health/zmq/req | Slack + PagerDuty | P1 | | SUB liveness /health/zmq/sub | Slack | P2 | | Drop rate /health/zmq/drops | Slack | P2 | | Heartbeat: PULL worker | Slack + email | P2 |

The REQ/REP monitor deserves special attention: a REP socket crash leaves the paired REQ forever blocked. Set the check interval to 2 minutes and fire PagerDuty immediately — REQ/REP deadlocks require manual recovery (restart both sides).

For PUSH/PULL topologies processing time-sensitive data, configure the drop rate monitor with a tight threshold (1%) and a short interval. Silent message drops at high throughput can mean significant data loss before anyone notices.

Summary

ZeroMQ's brokerless design makes it fast and flexible — but also invisible by default. Adding a lightweight health sidecar to each worker transforms opaque socket processes into monitorable services:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/zmq | Context health, drop count, message staleness | | HTTP monitor on /health/zmq/sub | PUB/SUB subscriber liveness | | HTTP monitor on /health/zmq/req | REQ/REP round-trip health and timeout detection | | HTTP monitor on /health/zmq/push | PUSH HWM pressure and queue fill rate | | Heartbeat monitor | Worker process liveness and processing continuity |

Get started free at vigilmon.online — your first ZeroMQ worker 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