tutorial

Monitoring LF Edge eKuiper with Vigilmon

eKuiper is a lightweight SQL-based IoT stream processing engine designed for resource-constrained edge devices — here's how to monitor its process health, SQL rules, MQTT/Kafka sources, sink delivery, and AI inference plugins with Vigilmon.

LF Edge eKuiper is a lightweight, SQL-based stream processing engine for IoT edge environments. Written in Go with a small binary footprint, it runs on IoT gateways, industrial PCs, and Kubernetes edge nodes — processing MQTT, Kafka, and EdgeX Foundry streams in real time using SQL rules, and writing results to MQTT, InfluxDB, HTTP endpoints, and more. Because eKuiper runs on constrained hardware at the network edge, failures are harder to detect than in cloud deployments: there's no automatic health dashboard, no cloud orchestration layer watching the process, and rule failures are often silent. Vigilmon gives you remote visibility into eKuiper's health from outside the edge device.

What You'll Set Up

  • eKuiper process health monitor via the REST management API (port 9081)
  • Rule execution health heartbeat (active rules vs. expected)
  • MQTT source connectivity heartbeat per rule
  • Kafka consumer health and lag monitor (if using Kafka sources)
  • Sink delivery success rate heartbeat per rule
  • Rule processing throughput heartbeat
  • Portable plugin / AI inference health heartbeat
  • EdgeX Foundry integration health monitor
  • Rule management API response time monitor
  • Edge device CPU and memory usage heartbeat

Prerequisites

  • LF Edge eKuiper running on an edge device or gateway
  • eKuiper REST management API accessible on port 9081 (default) from your monitoring host
  • A free Vigilmon account

Network note: eKuiper often runs on private edge networks. If the edge device is behind NAT, configure a reverse proxy or a VPN tunnel so Vigilmon can reach port 9081. Alternatively, use heartbeat-style monitoring (push-based), which only requires outbound HTTPS from the edge device.

Step 1: Monitor the eKuiper Process via REST API

eKuiper exposes a REST management API on port 9081. This API is used to create, start, and stop rules — and it's also the most direct way to verify the eKuiper process is alive.

If eKuiper's port 9081 is reachable from the public internet (or via a VPN/tunnel):

  1. Log in to vigilmon.online and click Add Monitor.
  2. Set Type to HTTP / HTTPS.
  3. URL: http://your-edge-device:9081/.
  4. Expected HTTP status: 200.
  5. Keyword check: enter Version — eKuiper's root response includes version information.
  6. Check interval: 1 minute.
  7. Set Alert after: 2 consecutive failures to avoid false alarms from brief network path issues.
  8. Click Save.

For a more targeted health check, use the /ping endpoint:

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-edge-device:9081/ping.
  3. Expected HTTP status: 200.
  4. Check interval: 1 minute.
  5. Click Save.

If the edge device is not publicly reachable, use the heartbeat approach below (Step 2) for all monitoring — the edge device pushes pings to Vigilmon rather than Vigilmon pulling from the device.

Step 2: Monitor Rule Execution Health

eKuiper rules are the core processing units — each rule defines a SQL query over a source stream and writes results to a sink. Rules can enter an error state if their source disconnects (e.g., MQTT broker unreachable), their sink fails (e.g., InfluxDB write error), or the SQL itself hits a runtime error. A rule in error state stops processing data silently.

Add a heartbeat monitor for your rule health checker:

  1. Click Add Monitor → Heartbeat / Cron.
  2. Expected interval: 2 minutes.
  3. Copy the heartbeat URL (e.g., https://vigilmon.online/api/push/YOUR_RULES_KEY).
  4. Click Save.

Create a rule health check script on the edge device:

#!/bin/bash
# Check eKuiper rule health — alert if any rule is in error state
RULES=$(curl -s http://localhost:9081/rules)
TOTAL=$(echo "$RULES" | python3 -c \
  "import sys,json; rules=json.load(sys.stdin); print(len(rules))")
RUNNING=$(echo "$RULES" | python3 -c \
  "import sys,json; rules=json.load(sys.stdin); \
   print(sum(1 for r in rules if r.get('status','') == 'Running'))")

# Alert if expected number of rules aren't all running
EXPECTED_RULES=5  # Set to your expected rule count

if [ "$RUNNING" -eq "$TOTAL" ] && [ "$TOTAL" -ge "$EXPECTED_RULES" ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_RULES_KEY" > /dev/null
fi

Schedule every 2 minutes in the edge device's crontab:

*/2 * * * * /opt/monitoring/ekuiper-rules-check.sh

If any rule stops running (enters error or stopped state), the heartbeat stops and Vigilmon alerts you.

Step 3: Monitor MQTT Source Connectivity

Most eKuiper deployments use MQTT as the primary data source — subscribing to topics from an MQTT broker (EMQX, Mosquitto, AWS IoT) and processing the stream with SQL rules. If the MQTT broker becomes unreachable, all rules sourced from MQTT stop receiving data silently.

Check MQTT broker connectivity from the edge device and send a heartbeat:

#!/bin/bash
# Check that eKuiper's MQTT broker is reachable
# Requires mosquitto_sub or mqtt CLI on the edge device
MQTT_HOST="your-mqtt-broker"
MQTT_PORT=1883
MQTT_TOPIC="test/ping"

# Test TCP connectivity to MQTT broker
if nc -z -w5 "$MQTT_HOST" "$MQTT_PORT" 2>/dev/null; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_MQTT_KEY" > /dev/null
fi

Set the heartbeat interval to 2 minutes.

Also check the MQTT source status via the eKuiper API — query the rule status for rules using MQTT sources:

#!/bin/bash
# Check specific rule using MQTT source
RULE_STATUS=$(curl -s http://localhost:9081/rules/my-mqtt-rule/status)
LAST_EXCEPTION=$(echo "$RULE_STATUS" | python3 -c \
  "import sys,json; s=json.load(sys.stdin); \
   print(s.get('lastException','none'))")

if [ "$LAST_EXCEPTION" = "none" ] || [ "$LAST_EXCEPTION" = "" ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_MQTT_RULE_KEY" > /dev/null
fi

If your MQTT broker port (1883 or 8883) is reachable from outside the edge network, add a Vigilmon TCP monitor directly:

  1. Click Add Monitor → TCP Port.
  2. Host: your-mqtt-broker-host, Port: 1883.
  3. Check interval: 1 minute.
  4. Click Save.

Step 4: Monitor Kafka Source Health and Consumer Lag

If eKuiper reads from Apache Kafka topics, consumer lag is the key metric — if eKuiper falls behind the Kafka producer (e.g., due to processing slowdowns or connectivity issues), the lag grows and data freshness degrades.

Monitor the Kafka broker TCP port from the edge device:

#!/bin/bash
# Check Kafka broker reachability and eKuiper consumer lag
KAFKA_HOST="your-kafka-broker"
KAFKA_PORT=9092
CONSUMER_GROUP="ekuiper"
TOPIC="iot-readings"

# TCP check
if ! nc -z -w5 "$KAFKA_HOST" "$KAFKA_PORT" 2>/dev/null; then
  exit 1  # Don't send heartbeat — Kafka unreachable
fi

# Consumer lag check (requires kafka-consumer-groups.sh on edge device)
LAG=$(kafka-consumer-groups.sh --bootstrap-server "${KAFKA_HOST}:${KAFKA_PORT}" \
  --describe --group "$CONSUMER_GROUP" 2>/dev/null | \
  awk '/'"$TOPIC"'/{sum+=$5} END{print sum+0}')

# Alert if lag > 1000 messages
if [ "${LAG:-0}" -lt "1000" ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_KAFKA_KEY" > /dev/null
fi

Set the heartbeat interval to 5 minutes. If the Kafka broker is unreachable or lag exceeds threshold, the heartbeat stops.

Step 5: Monitor Sink Delivery Success Rate

eKuiper writes processed results to sinks — MQTT topics, InfluxDB, HTTP REST endpoints, TimescaleDB, EdgeX Foundry, S3, email. Sink write failures cause processed results to be lost silently. Check the sink error count via the rule status API:

#!/bin/bash
# Check sink write errors across all rules
RULES=$(curl -s http://localhost:9081/rules | python3 -c \
  "import sys,json; print('\n'.join(r['id'] for r in json.load(sys.stdin)))")

ALL_HEALTHY=true
for RULE_ID in $RULES; do
  STATUS=$(curl -s "http://localhost:9081/rules/${RULE_ID}/status")
  SINK_ERRORS=$(echo "$STATUS" | python3 -c \
    "import sys,json; s=json.load(sys.stdin); \
     print(sum(v for k,v in s.items() if 'op_sink' in k and 'exceptions_total' in k))" 2>/dev/null || echo "0")
  
  if [ "${SINK_ERRORS:-0}" -gt "0" ]; then
    ALL_HEALTHY=false
    echo "Sink errors in rule ${RULE_ID}: ${SINK_ERRORS}"
  fi
done

if [ "$ALL_HEALTHY" = true ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_SINK_KEY" > /dev/null
fi

Add a Vigilmon heartbeat monitor (expected interval: 5 minutes). If any rule has accumulating sink errors, the heartbeat stops and you're alerted.

Step 6: Monitor Rule Processing Throughput

eKuiper should be processing a predictable volume of messages per second based on your connected sources. A drop to zero in message throughput means either the source stopped sending (MQTT disconnected, Kafka lag) or the rule has entered an error state.

#!/bin/bash
# Check processing throughput — alert if any rule processes zero messages
RULES=$(curl -s http://localhost:9081/rules | python3 -c \
  "import sys,json; print('\n'.join(r['id'] for r in json.load(sys.stdin)))")

HAS_TRAFFIC=false
for RULE_ID in $RULES; do
  STATUS=$(curl -s "http://localhost:9081/rules/${RULE_ID}/status")
  RECORDS_IN=$(echo "$STATUS" | python3 -c \
    "import sys,json; s=json.load(sys.stdin); \
     print(sum(v for k,v in s.items() if 'records_in_total' in k))" 2>/dev/null || echo "0")
  
  if [ "${RECORDS_IN:-0}" -gt "0" ]; then
    HAS_TRAFFIC=true
  fi
done

if [ "$HAS_TRAFFIC" = true ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_THROUGHPUT_KEY" > /dev/null
fi

Set the heartbeat interval to 5 minutes. No traffic across any rule for 5 minutes is a clear signal that upstream sources have gone silent.

Step 7: Monitor Portable Plugin and AI Inference Health

eKuiper's portable plugin system lets you run TensorFlow Lite or ONNX models for AI/ML inference at the edge. These plugins run as separate processes, and a plugin crash silently causes all rules that use that function to fail.

Check portable plugin process health:

#!/bin/bash
# Check eKuiper portable plugins are loaded and running
PLUGINS=$(curl -s http://localhost:9081/plugins/portable)
FAILED=$(echo "$PLUGINS" | python3 -c \
  "import sys,json; plugins=json.load(sys.stdin); \
   print(sum(1 for p in plugins if p.get('status','') not in ['loaded','running']))" 2>/dev/null || echo "0")

if [ "${FAILED:-0}" -eq "0" ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_PLUGIN_KEY" > /dev/null
fi

For inference plugins, also validate that inference rules are processing correctly by checking their rule status (Step 2 covers this, but you can add a separate heartbeat for inference-specific rules with a tighter interval if inference is latency-sensitive).

Step 8: Monitor EdgeX Foundry Integration

eKuiper is commonly deployed alongside EdgeX Foundry, reading sensor readings from the EdgeX message bus (EdgeX source) and writing processed results back to EdgeX (EdgeX sink). If the EdgeX stack degrades, eKuiper's EdgeX-sourced rules go silent.

Add an HTTP monitor for the EdgeX core data service (if accessible from the edge device):

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://localhost:59880/api/v3/ping (assuming EdgeX and eKuiper are co-located).
  3. Expected HTTP status: 200.
  4. Check interval: 1 minute.
  5. Click Save.

Add a heartbeat for the EdgeX integration health:

#!/bin/bash
# Check eKuiper → EdgeX integration
EDGEX_PING=$(curl -s -o /dev/null -w "%{http_code}" \
  http://localhost:59880/api/v3/ping 2>/dev/null)

# Check rules using EdgeX source are running
EDGEX_RULES=$(curl -s http://localhost:9081/rules | python3 -c \
  "import sys,json; rules=json.load(sys.stdin); \
   print(sum(1 for r in rules if r.get('status') == 'Running'))")

if [ "$EDGEX_PING" = "200" ] && [ "${EDGEX_RULES:-0}" -gt "0" ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_EDGEX_KEY" > /dev/null
fi

Step 9: Monitor Edge Device Resource Usage

eKuiper is designed for constrained hardware, but on a busy IoT gateway handling many rules and AI inference, CPU and memory usage can reach device limits. High CPU (>50%) degrades rule processing latency; high memory (>80%) can cause the OS to kill the eKuiper process.

#!/bin/bash
# Check eKuiper process resource usage
EKUIPER_PID=$(pgrep -x kuiperd || pgrep -x eKuiper)

if [ -z "$EKUIPER_PID" ]; then
  # Process not found — don't send heartbeat
  exit 1
fi

CPU=$(ps -p "$EKUIPER_PID" -o %cpu= | tr -d ' ')
MEM_MB=$(ps -p "$EKUIPER_PID" -o rss= | awk '{print $1/1024}')

# Alert thresholds: CPU > 50%, Memory > 200MB (adjust for your device)
CPU_OK=$(echo "$CPU < 50" | bc -l)
MEM_OK=$(echo "$MEM_MB < 200" | bc -l)

if [ "$CPU_OK" -eq "1" ] && [ "$MEM_OK" -eq "1" ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_RESOURCE_KEY" > /dev/null
fi

Set the heartbeat interval to 5 minutes. Adjust CPU and memory thresholds for your specific edge device hardware constraints.

Step 10: Monitor the Rule Management API

If you use a centralized management system to deploy and update eKuiper rules (common in fleet deployments), the eKuiper REST API's availability and response time determines whether rule deployments succeed. A slow or unavailable management API blocks rule updates across your edge fleet.

If your management system can reach eKuiper's port 9081 from outside the edge device, add a response time monitor:

#!/bin/bash
# Check eKuiper management API response time
START=$(date +%s%3N)
HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" \
  http://localhost:9081/rules --max-time 5)
END=$(date +%s%3N)
LATENCY=$((END - START))

# Alert if API is slow (>2000ms) or unavailable
if [ "$HTTP_CODE" = "200" ] && [ "$LATENCY" -lt "2000" ]; then
  curl -fsS -m 10 "https://vigilmon.online/api/push/YOUR_API_KEY" > /dev/null
fi

Step 11: Configure Alerting

Open Alert Channels in Vigilmon and configure notification routing:

Critical alerts (immediate, 24/7):

  • eKuiper REST API down (process crashed — all rules stopped)
  • Rule health heartbeat missed (one or more rules in error state)
  • Edge device resource heartbeat missed (process about to be killed by OOM)

Warning alerts (business hours or on-call):

  • MQTT source heartbeat missed (upstream broker issue)
  • Sink delivery heartbeat missed (output data loss starting)
  • Throughput heartbeat missed (rules processing no data)
  • EdgeX integration heartbeat missed (upstream EdgeX issue)

Recommended thresholds:

  • Process/API: alert after 2 consecutive failures (brief restarts are normal on embedded hardware)
  • Rule health heartbeat: alert after 1 missed ping
  • Resource heartbeat: alert after 1 missed ping
  • Source/sink heartbeats: alert after 2 missed pings (transient connectivity is common at the edge)

Summary: Your eKuiper Monitoring Stack

| Monitor | Type | What It Catches | |---|---|---| | eKuiper REST :9081/ping | HTTP | eKuiper process crash | | Rule health heartbeat | Heartbeat | Rules entering error state | | MQTT source heartbeat | Heartbeat | MQTT broker disconnection | | MQTT broker TCP :1883 | TCP Port | MQTT broker unreachable | | Kafka source heartbeat | Heartbeat | Kafka lag / broker connectivity | | Sink delivery heartbeat | Heartbeat | Output write errors | | Throughput heartbeat | Heartbeat | No messages processed | | Plugin health heartbeat | Heartbeat | AI inference plugin crash | | EdgeX integration heartbeat | Heartbeat | EdgeX upstream failure | | EdgeX core data :59880 | HTTP | EdgeX service health | | Resource usage heartbeat | Heartbeat | CPU/memory overload | | Management API heartbeat | Heartbeat | Rule deployment API slow |

eKuiper's edge deployment model means failures happen far from your monitoring infrastructure. The push-based heartbeat pattern is the key insight: instead of Vigilmon reaching into your edge devices, the edge devices push pings to Vigilmon — and silence is the alert. When any heartbeat stops arriving, you know immediately that something on that edge node needs attention.

Get started free at vigilmon.online.

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