How to Monitor StreamSets Data Pipelines with Vigilmon

StreamSets builds and runs data pipelines that move data in real time and batch modes — connecting origins, applying transformations, and writing to destinations across streaming and batch systems. When a pipeline stops, stalls, or writes zero records, the consequences cascade quickly: downstream tables go stale, ML models train on outdated features, and business dashboards silently mislead.

In this tutorial you'll set up end-to-end monitoring for StreamSets pipelines using Vigilmon — free tier, no credit card.

Why StreamSets pipelines need dedicated monitoring

StreamSets DataCollector and StreamSets Platform have built-in pipeline metrics, but external monitoring fills critical gaps:

• Pipeline transitions to ERROR state — a pipeline moves from RUNNING to ERROR (e.g., a deserialization failure) and stops processing; StreamSets alerts are not configured by default
• Zero-throughput stalls — the pipeline stays in RUNNING state but input records per second drops to zero; the pipeline appears healthy but data isn't flowing
• Origin connection failures — Kafka, Kafka Connect, or a JDBC origin loses its connection and retries indefinitely while no data passes through
• Executor process crashes — the StreamSets Data Collector JVM crashes or is OOM-killed; the pipeline is gone but your orchestration doesn't notice
• Platform node failures — in a distributed StreamSets deployment, a worker node goes down and the scheduler doesn't reassign the pipeline

Vigilmon's heartbeat and HTTP monitors let you detect all of these from outside the pipeline, independent of StreamSets' own state.

What you'll need

• A StreamSets Data Collector instance or StreamSets Platform account
• Shell access to the machine running StreamSets Data Collector (for on-prem) or HTTP access to the StreamSets REST API
• A free Vigilmon account (sign up takes 30 seconds)

Step 1: Add a Vigilmon HTTP Client destination to batch pipelines

For batch pipelines, the cleanest integration is a StreamSets HTTP Client destination (or processor) at the end of the pipeline that pings a Vigilmon heartbeat URL after processing each batch.

Using the StreamSets HTTP Client destination:

1. Open your batch pipeline in the StreamSets Data Collector UI
2. Add an HTTP Client destination at the end of the pipeline
3. Set the Resource URL to your Vigilmon heartbeat URL: https://vigilmon.online/api/heartbeats/YOUR_HEARTBEAT_ID/ping
4. Set HTTP Method to GET
5. Set Batch Wait Time to your pipeline's expected batch interval
6. Connect the final processor in your pipeline to this destination

Now every time the pipeline processes a batch successfully, Vigilmon receives a heartbeat ping.

Step 2: Use the StreamSets REST API to check pipeline status

For long-running streaming pipelines, poll the StreamSets REST API from a monitoring script and ping Vigilmon only when the pipeline is RUNNING and producing records:

#!/usr/bin/env python3
import os
import sys
import requests

SDC_URL = "http://localhost:18630"
PIPELINE_ID = "your_pipeline_id"
SDC_USER = os.environ.get("SDC_USER", "admin")
SDC_PASS = os.environ.get("SDC_PASS")
HEARTBEAT_URL = "https://vigilmon.online/api/heartbeats/YOUR_HEARTBEAT_ID/ping"

# Fetch pipeline status
status_resp = requests.get(
    f"{SDC_URL}/rest/v1/pipeline/{PIPELINE_ID}/status",
    auth=(SDC_USER, SDC_PASS),
    timeout=10,
)
status_resp.raise_for_status()
status = status_resp.json().get("status")

# Fetch pipeline metrics
metrics_resp = requests.get(
    f"{SDC_URL}/rest/v1/pipeline/{PIPELINE_ID}/metrics",
    auth=(SDC_USER, SDC_PASS),
    timeout=10,
)
metrics_resp.raise_for_status()
metrics = metrics_resp.json()

# Extract records-in count from meter stats
records_in = (
    metrics.get("meters", {})
    .get("pipeline.batchInputRecords.meter", {})
    .get("count", 0)
)

if status == "RUNNING" and records_in > 0:
    requests.get(HEARTBEAT_URL, timeout=10)
    print(f"Pipeline RUNNING, {records_in} records in — heartbeat sent")
else:
    print(f"Pipeline status={status}, records_in={records_in} — heartbeat NOT sent")
    sys.exit(1)

Schedule this script to run every few minutes via cron or your orchestration tool.

Step 3: Create heartbeat monitors in Vigilmon

For each critical StreamSets pipeline, create a Vigilmon heartbeat monitor:

1. Log in to vigilmon.online and go to Monitors → New Monitor
2. Choose Heartbeat as the type
3. Name it after the pipeline, e.g. StreamSets: Kafka → Snowflake
4. Set expected interval to match your check frequency — e.g., 5 minutes for a streaming pipeline check running every 5 minutes
5. Set grace period to 2 minutes for streaming pipelines (shorter = faster alert) or 15 minutes for batch pipelines
6. Copy the heartbeat URL and add it to your monitoring script or HTTP Client destination
7. Save the monitor

Step 4: Monitor the StreamSets Data Collector HTTP endpoint

StreamSets Data Collector exposes a web UI and REST API on a configurable port (default 18630). Add an HTTP monitor to Vigilmon to detect when the SDC process goes down:

1. Go to Monitors → New Monitor
2. Choose HTTP / HTTPS
3. Set the URL to your SDC instance, e.g. http://streamsets.your-company.com:18630/
4. Set check interval to 1 minute
5. Set expected status code to 200
6. Save the monitor

Also add a TCP Port monitor on the same host and port for an additional layer:

1. Go to Monitors → New Monitor
2. Choose TCP Port
3. Enter hostname and port 18630
4. Save the monitor

Step 5: Configure alerts and create a status page

Alert channels:

1. Go to Alert Channels → Add Channel
2. Add a Slack channel for your data engineering team
3. For production streaming pipelines, add a PagerDuty integration for immediate escalation
4. Assign channels to all your StreamSets monitors

Example Vigilmon alert when a streaming pipeline heartbeat is missed:

{
  "monitor_name": "StreamSets: Kafka → Snowflake",
  "status": "down",
  "type": "heartbeat",
  "last_ping_at": "2024-01-15T14:30:00Z",
  "expected_by": "2024-01-15T14:37:00Z",
  "message": "Heartbeat overdue by 8 minutes"
}

Status page:

1. Go to Status Pages → New Status Page
2. Name it "Data Streaming Health"
3. Add all StreamSets pipeline heartbeat monitors and the SDC HTTP monitor
4. Publish the page

Share the status page URL with your data consumers so they can self-serve pipeline health status instead of pinging your team.

Putting it all together

Here's the complete cron-based polling setup:

# /etc/cron.d/streamsets-monitoring
# Run every 3 minutes
*/3 * * * * dataeng /opt/monitoring/check_streamsets.py >> /var/log/streamsets-monitor.log 2>&1

And the full monitoring matrix:

| Monitor | Type | What it catches | |---------|------|-----------------| | StreamSets: Kafka → Snowflake | Heartbeat | Pipeline ERROR state, zero throughput, JVM crash | | StreamSets: JDBC → S3 | Heartbeat | Batch pipeline failures and missed runs | | SDC web UI (port 18630) | HTTP | Data Collector process failure | | SDC port 18630 | TCP | Process crash before HTTP layer responds |

What's next

• Per-pipeline alert channels — assign different alert channels to different pipelines based on severity; a financial pipeline might page on-call, while a low-priority analytics pipeline sends an email
• SSL expiry monitoring — if your SDC instance runs behind HTTPS, Vigilmon monitors your TLS certificate expiry automatically
• Downstream destination monitoring — add HTTP monitors for Snowflake, Kafka cluster endpoints, or S3 bucket access so you catch write-target failures independently of the pipeline

Get started free at vigilmon.online — no credit card, monitors start running in under a minute.