tutorial

How to Monitor Apache StreamPipes Uptime and Health with Vigilmon

Apache StreamPipes is a self-service data analytics platform for IoT and streaming pipelines — but pipeline executor failures, adapter disconnections, and broker connectivity loss can silently stall your entire event processing graph. Here's how to monitor StreamPipes with Vigilmon.

Apache StreamPipes is a self-service Industrial IoT data analytics platform that allows non-technical users to build real-time streaming pipelines using a graphical drag-and-drop interface. It sits on top of Apache Kafka and Apache Flink, connects to dozens of industrial protocols (OPC-UA, MQTT, Modbus), and runs pipeline processors as distributed microservices. When a pipeline executor crashes, an adapter loses its MQTT broker connection, or Kafka becomes unreachable, StreamPipes continues to display pipelines as "running" while no data flows through them — making external health monitoring essential.

Vigilmon gives you external visibility into StreamPipes availability through HTTP probe monitoring and heartbeat monitoring for pipeline throughput. This tutorial walks through both.

Why Apache StreamPipes Monitoring Matters

StreamPipes' microservice architecture introduces failure modes that the built-in dashboard does not surface:

  • Pipeline processor crashes leave pipelines listed as "running" in the UI while the underlying Flink job has exited — no data is processed, no alert is raised
  • Adapter disconnections (OPC-UA server unreachable, MQTT broker restart) stop ingest at the source; downstream processors receive nothing but appear healthy
  • Kafka connectivity loss disconnects the message bus between pipeline stages — processors complete successfully but their output never reaches the next stage
  • Backend API unavailability means users cannot create, modify, or monitor pipelines — the REST API is the control plane for all automation and CI/CD pipelines
  • CouchDB or InfluxDB unavailability causes metadata loss or measurement gaps — pipelines may run but their state and measurements are not persisted

External monitoring through Vigilmon verifies whether StreamPipes is actually reachable and processing — not just whether the Docker containers are running.

Step 1: Build a StreamPipes Health Endpoint

StreamPipes exposes a REST API that you can probe for health. The backend runs on port 8030 by default. Test the native readiness endpoint:

curl -i http://your-streampipes-host:8030/streampipes-backend/api/v2/
# HTTP/1.1 200 OK  (returns API version info)

For a deeper check that verifies pipeline execution health, build a sidecar that queries the pipeline status API:

Node.js Sidecar Health Endpoint

// streampipes-health.js — checks pipeline execution health
const express = require('express');
const axios = require('axios');

const app = express();
const SP_API = process.env.SP_API_URL || 'http://localhost:8030/streampipes-backend/api/v2';
const SP_USER = process.env.SP_USER || 'admin@streampipes.apache.org';
const SP_PASS = process.env.SP_PASSWORD || '';

async function getAuthToken() {
  const res = await axios.post(`${SP_API}/auth/login`, {
    username: SP_USER,
    password: SP_PASS,
  }, { timeout: 10000 });
  return res.data.accessToken;
}

app.get('/health/streampipes', async (req, res) => {
  try {
    const token = await getAuthToken();

    // Check all pipelines — any stopped pipeline that should be running is a problem
    const pipelineRes = await axios.get(`${SP_API}/pipelines`, {
      headers: { Authorization: `Bearer ${token}` },
      timeout: 10000,
    });

    const pipelines = pipelineRes.data || [];
    const stopped = pipelines.filter(p => p.running === false);

    if (stopped.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'stopped_pipelines',
        stopped: stopped.map(p => p.name),
        total: pipelines.length,
      });
    }

    return res.status(200).json({
      status: 'ok',
      running_pipelines: pipelines.length,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3001);

Python Sidecar Health Endpoint

# streampipes_health.py — FastAPI sidecar for StreamPipes health
import os
import httpx
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()
SP_API = os.environ.get("SP_API_URL", "http://localhost:8030/streampipes-backend/api/v2")
SP_USER = os.environ.get("SP_USER", "admin@streampipes.apache.org")
SP_PASS = os.environ.get("SP_PASSWORD", "")

async def get_token(client: httpx.AsyncClient) -> str:
    r = await client.post(f"{SP_API}/auth/login",
                          json={"username": SP_USER, "password": SP_PASS},
                          timeout=10.0)
    r.raise_for_status()
    return r.json()["accessToken"]

@app.get("/health/streampipes")
async def streampipes_health():
    async with httpx.AsyncClient() as client:
        try:
            token = await get_token(client)
        except Exception as e:
            return JSONResponse(status_code=503,
                                content={"status": "down", "check": "auth", "error": str(e)})

        try:
            r = await client.get(f"{SP_API}/pipelines",
                                 headers={"Authorization": f"Bearer {token}"},
                                 timeout=10.0)
            r.raise_for_status()
            pipelines = r.json()
        except Exception as e:
            return JSONResponse(status_code=503,
                                content={"status": "down", "check": "pipelines", "error": str(e)})

        stopped = [p["name"] for p in pipelines if not p.get("running", True)]
        if stopped:
            return JSONResponse(status_code=503, content={
                "status": "degraded",
                "reason": "stopped_pipelines",
                "stopped": stopped,
                "total": len(pipelines),
            })

    return {"status": "ok", "running_pipelines": len(pipelines)}

Deploy and verify:

curl -i https://your-app.example.com/health/streampipes
# HTTP/1.1 200 OK
# {"status":"ok","running_pipelines":5}

Step 2: Configure a Vigilmon HTTP Monitor for StreamPipes

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your health endpoint: https://your-app.example.com/health/streampipes
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 15000ms (auth + pipeline list fetch can be slow)
  6. Under Alert channels, assign your Slack or email channel
  7. Save the monitor

For a lighter-weight check against the native StreamPipes API:

  1. Create a monitor pointing to http://your-streampipes-host:8030/streampipes-backend/api/v2/
  2. Set the expected status code to 200
  3. Set the response time threshold to 5000ms

Failure Coverage

| Failure Mode | Container Monitor | Vigilmon | |---|---|---| | Backend JVM crash | ✓ | ✓ | | Pipeline executor stopped (Flink job exited) | ✗ | ✓ | | Adapter disconnected (no ingest) | ✗ | ✓ (via heartbeat) | | Kafka connectivity loss | ✗ | ✓ (via heartbeat) | | CouchDB metadata unavailable | ✗ | ✓ |

Step 3: Heartbeat Monitoring for StreamPipes Pipeline Throughput

Industrial IoT pipelines that process sensor data continuously are the best candidates for heartbeat monitoring. When an OPC-UA adapter disconnects or a Kafka topic falls behind, the pipeline continues to appear healthy in the UI while no data is flowing. A Vigilmon heartbeat detects this silent stall.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: streampipes-pipeline-throughput
  3. Set the expected interval: 5 minutes (match your pipeline's ingest frequency)
  4. Set the grace period: 10 minutes
  5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire It Into a StreamPipes Data Sink

Create a custom StreamPipes data sink microservice that pings Vigilmon whenever it receives and processes a measurement batch:

// VigilmonHeartbeatSink.java — StreamPipes data sink that pings Vigilmon
package org.apache.streampipes.extensions.sinks.vigilmon;

import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
import org.apache.streampipes.model.runtime.Event;
import java.net.HttpURLConnection;
import java.net.URL;

public class VigilmonHeartbeatSink implements IStreamPipesDataSink {
    private String heartbeatUrl;
    private int batchCount = 0;
    private final int pingEvery;

    public VigilmonHeartbeatSink(String heartbeatUrl, int pingEvery) {
        this.heartbeatUrl = heartbeatUrl;
        this.pingEvery = pingEvery;
    }

    @Override
    public void onEvent(Event event, EventSinkRuntimeContext ctx) throws SpRuntimeException {
        batchCount++;
        if (batchCount % pingEvery == 0) {
            try {
                HttpURLConnection conn = (HttpURLConnection) new URL(heartbeatUrl).openConnection();
                conn.setRequestMethod("GET");
                conn.setConnectTimeout(3000);
                conn.setReadTimeout(3000);
                conn.getResponseCode();
                conn.disconnect();
            } catch (Exception ignored) {}
        }
    }
}

Python Adapter Heartbeat

For Python-based StreamPipes adapters, ping after each successful message forward:

import requests, os

HEARTBEAT_URL = os.environ["VIGILMON_HEARTBEAT_URL"]
forward_count = 0

def on_message_forwarded(event: dict):
    global forward_count
    forward_count += 1
    if forward_count % 100 == 0:
        try:
            requests.get(HEARTBEAT_URL, timeout=3)
        except Exception:
            pass

Step 4: Monitor Multiple StreamPipes Components

A production StreamPipes deployment spans multiple services. Monitor each independently:

[streampipes-backend]     :8030/streampipes-backend/api/v2/
[streampipes-pipeline-health]  /health/streampipes (sidecar)
[streampipes-couchdb]     :5984/_up
[streampipes-kafka]       :9092 (TCP monitor)
[streampipes-pipeline-throughput]  (heartbeat)

Name monitors using the bracket convention so alert notifications immediately identify which component requires attention.

Group all StreamPipes monitors into a Vigilmon Status Page for your operations team or IoT platform stakeholders.

Step 5: Alert Routing for StreamPipes Outages

StreamPipes powers real-time IoT dashboards and industrial automation — a pipeline stall can mean missing sensor readings, delayed anomaly detection, or failed safety alerts in operational technology environments.

Configure alert channels in Vigilmon:

  1. Backend API monitor → Slack + PagerDuty (P1 — platform unavailable)
  2. Pipeline health sidecar → Slack + PagerDuty (P1 — active pipelines stopped)
  3. Heartbeat monitor → Slack + PagerDuty (P1 — data flow stalled)
  4. CouchDB / Kafka monitors → Slack (P2 — infrastructure degraded)

Set response time thresholds carefully:

  • Alert at 10000ms on the pipeline health sidecar — slow pipeline API responses indicate a Flink executor under load
  • Alert at 3000ms on the native backend API — the root /api/v2/ endpoint should be fast; a slow response means the JVM is GC-paused or overloaded

Summary

Apache StreamPipes pipeline failures are silent — the UI shows running status while no data flows through stalled executors or disconnected adapters. Vigilmon monitors the full path from API availability through active throughput.

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /api/v2/ | Backend API reachability | | HTTP monitor on /health/streampipes | Pipeline execution status, stopped pipelines | | Heartbeat monitor | Active data flow through pipelines | | TCP monitor on Kafka :9092 | Message bus connectivity |

Get started free at vigilmon.online — your first StreamPipes monitor takes under two minutes to configure.

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