tutorial

How to Monitor Benthos Stream Processor with Vigilmon

Benthos (now Redpanda Connect) processes millions of messages per second — but input connector failures, processor panics, and dead-letter queue growth are silent. Learn how to monitor Benthos component health, message throughput, and pipeline uptime with Vigilmon.

Benthos (recently rebranded as Redpanda Connect) is a high-performance stream processor that connects hundreds of input and output components. A failed Kafka consumer input, a processor that panics on malformed JSON, or an output buffer that fills up will silently stall your data pipeline — messages stop flowing and downstream services get nothing, often without a single error log visible from outside the process.

Vigilmon gives you external visibility into Benthos pipeline health through Benthos's built-in HTTP metrics server and heartbeat monitoring for end-to-end pipeline liveness. This tutorial shows you how to set both up.

Why Benthos Needs External Monitoring

Benthos exposes rich internal metrics via its built-in HTTP server (/metrics in Prometheus format, /stats as JSON) and a ready endpoint (/ready). But internal metrics require an active scraper — if nobody is watching Prometheus, the data disappears. External monitoring with Vigilmon adds:

  • Component health alerting — Benthos's /ready endpoint returns 503 when inputs or outputs are unhealthy; Vigilmon fires an alert the moment this happens
  • Dead-letter queue growth detection — a custom health endpoint that reads Benthos's error counters and returns 503 when the DLQ grows beyond a threshold
  • Throughput stall detection — heartbeat monitors that fire when message processing stops even if the process is still alive
  • Multi-region availability checking — validate Benthos REST endpoints from outside your network

Step 1: Use Benthos's Built-In Health Endpoints

Benthos exposes HTTP endpoints by default on port 4195. You can probe these directly from Vigilmon.

Enable the HTTP Server

In your benthos.yaml config:

http:
  address: 0.0.0.0:4195
  enabled: true
  read_timeout: 5s
  root_path: /benthos

metrics:
  prometheus:
    prefix: benthos

Benthos then exposes:

  • GET /ready — returns 200 OK when all inputs and outputs are connected, 503 when any component is not ready
  • GET /metrics — Prometheus metrics for all components
  • GET /ping — basic liveness (always 200 if the process is running)

Test the Endpoint

# Check readiness — this is what Vigilmon will probe
curl -i http://localhost:4195/ready

# View Prometheus metrics for throughput and error rates
curl -s http://localhost:4195/metrics | grep -E 'input_received|output_sent|processor_error'

# Input component stats
curl -s http://localhost:4195/metrics | grep benthos_input

Example /ready responses:

# All components healthy
HTTP/1.1 200 OK
{"ok": true}

# Kafka input disconnected
HTTP/1.1 503 Service Unavailable
{"error": "input/kafka: connection refused"}

Step 2: Build a Dead-Letter Queue Health Endpoint

Benthos's /ready endpoint only reports connection health — it doesn't tell you that messages are being dropped into the DLQ or that processor error rates have spiked. Add a custom health sidecar that reads Benthos's Prometheus metrics and returns a structured health response.

# benthos_health.py
import os
import requests
from flask import Flask, jsonify

app = Flask(__name__)
BENTHOS_METRICS_URL = os.environ.get('BENTHOS_METRICS_URL', 'http://localhost:4195/metrics')

def parse_prometheus_metric(text, metric_name):
    for line in text.splitlines():
        if line.startswith(metric_name) and not line.startswith('#'):
            parts = line.rsplit(' ', 1)
            if len(parts) == 2:
                return float(parts[1])
    return 0.0

@app.route('/health/benthos')
def health():
    try:
        resp = requests.get(BENTHOS_METRICS_URL, timeout=5)
        if resp.status_code != 200:
            return jsonify({'status': 'down', 'reason': 'metrics_unavailable'}), 503

        metrics = resp.text
        processor_errors = parse_prometheus_metric(metrics, 'benthos_processor_error_total')
        input_received  = parse_prometheus_metric(metrics, 'benthos_input_received_total')
        output_sent     = parse_prometheus_metric(metrics, 'benthos_output_sent_total')
        output_errors   = parse_prometheus_metric(metrics, 'benthos_output_error_total')

        # DLQ / processor error threshold
        max_errors = float(os.environ.get('MAX_PROCESSOR_ERRORS', '100'))
        if processor_errors > max_errors:
            return jsonify({
                'status': 'degraded',
                'reason': 'processor_error_spike',
                'processor_errors': processor_errors,
                'threshold': max_errors,
            }), 503

        # Output error rate > 5%
        if output_sent > 0 and (output_errors / (output_sent + output_errors)) > 0.05:
            return jsonify({
                'status': 'degraded',
                'reason': 'high_output_error_rate',
                'output_errors': output_errors,
                'output_sent': output_sent,
            }), 503

        return jsonify({
            'status': 'ok',
            'processor_errors': processor_errors,
            'input_received': input_received,
            'output_sent': output_sent,
        }), 200

    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503

if __name__ == '__main__':
    app.run(port=8091)

Run this sidecar alongside your Benthos process and expose it on a port Vigilmon can reach.

Alternatively: Use a Benthos HTTP Output as a Health Probe

You can also configure Benthos itself to POST a health signal to an internal endpoint after each processed batch:

# benthos.yaml — add a metrics-based branch to your pipeline
pipeline:
  processors:
    - bloblang: |
        root = this
        # Tag each message with processing timestamp
        root.processed_at = now()

output:
  broker:
    outputs:
      - kafka:
          addresses: ["${KAFKA_BROKERS}"]
          topic: processed-events
      - http_client:
          url: "${VIGILMON_HEARTBEAT_URL}"
          verb: GET
          batching:
            count: 500
            period: 60s

This sends a GET request to your Vigilmon heartbeat URL after every 500 messages or 60 seconds, whichever comes first.

Step 3: Configure Vigilmon Monitors

HTTP Monitor for Benthos Readiness

  1. Log in to vigilmon.online → Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. URL: https://your-benthos-host.example.com:4195/ready
  4. Check interval: 1 minute
  5. Expected:
    • Status code: 200
    • Response time threshold: 3000ms
  6. Alert channel: Slack + PagerDuty for production pipelines
  7. Save

HTTP Monitor for DLQ / Error Rate

Add a second monitor for the health sidecar:

  • URL: https://your-benthos-host.example.com:8091/health/benthos
  • Expected: 200, body contains "status":"ok"
  • Interval: 2 minutes
  • Alert channel: Slack (P2 — degraded before it becomes an outage)

Step 4: Heartbeat Monitoring for Pipeline Throughput

Benthos can be running and its /ready endpoint can return 200 while the pipeline is processing zero messages — if the upstream source is empty or the consumer group has no assigned partitions. Vigilmon heartbeat monitors prove that messages are actually flowing.

Set Up the Heartbeat Monitor

  1. In Vigilmon → Monitors → New Monitor → Heartbeat
  2. Name: benthos-event-ingestion-pipeline
  3. Expected interval: 5 minutes
  4. Grace period: 3 minutes
  5. Save → copy heartbeat URL

Send Heartbeats via Benthos HTTP Output

Configure Benthos to ping Vigilmon as part of its output pipeline. Add an HTTP branch that fires every N messages:

# benthos.yaml
output:
  broker:
    outputs:
      - label: primary_output
        kafka:
          addresses: ["${KAFKA_BROKERS}"]
          topic: "${OUTPUT_TOPIC}"

      - label: vigilmon_heartbeat
        http_client:
          url: "${VIGILMON_HEARTBEAT_URL}"
          verb: GET
          max_in_flight: 1
          batching:
            count: 1000
            period: "${HEARTBEAT_INTERVAL:5m}"

The batching config sends a single GET to the heartbeat URL after every 1000 messages or 5 minutes — Vigilmon treats the absence of this ping as a stall alert.

Step 5: Alert Routing

| Monitor | Alert Channel | Priority | |---|---|---| | Benthos readiness /ready | Slack + PagerDuty | P1 | | DLQ / error rate /health/benthos | Slack | P2 | | Heartbeat: message throughput | Slack + email | P2 | | Benthos liveness /ping | Email | P3 |

For pipelines with strict SLAs, add a response time threshold on the /ready endpoint — Benthos component reconnect latency above 10 seconds signals that an upstream broker is under pressure, even before the endpoint returns 503.

Summary

Benthos pipeline failures are invisible from the outside. You need monitoring at three layers: component readiness, error rates, and message throughput.

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /ready | Input/output component connection health | | HTTP monitor on DLQ sidecar | Processor error spikes, high output error rate | | Heartbeat monitor | End-to-end message flow, pipeline liveness |

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