Monitoring Quix Streams with Vigilmon

Quix Streams is an open source Python library for building stateful Kafka stream processing applications — tumbling windows, hopping windows, sliding aggregations, stream joins — using a pandas-inspired DataFrame API backed by RocksDB for persistent state and Kafka transactions for exactly-once semantics. Where Kafka Streams gives Java developers native Kafka integration, Quix Streams brings the same capabilities to Python data engineers and ML teams. When you deploy Quix Streams as a containerized Python process, you need visibility into Kafka consumer lag, RocksDB state health, window computation integrity, and transaction commit health before issues silently corrupt your stream processing outputs. Vigilmon provides that visibility with minimal instrumentation overhead.

What You'll Set Up

• Process health monitor for the Quix Streams Python application
• Kafka consumer lag monitoring per topic-partition
• RocksDB state size and compaction health
• Window computation heartbeat
• Message processing throughput alert
• Kafka transaction commit health
• Dead letter queue accumulation monitor
• Offset commit success monitoring
• Output topic write health

Prerequisites

• Quix Streams application running (Docker container, systemd service, or Kubernetes pod)
• Access to your Kafka cluster (Confluent Cloud, self-hosted, or Quix Cloud managed Kafka)
• Python metrics endpoint exposed from your application (HTTP)
• A free Vigilmon account

Step 1: Monitor the Quix Streams Process

The Python process running your Quix Streams application is the heart of your stream processor. A crash stops consumption from all assigned Kafka partitions until the consumer group rebalances — and during that window, lag accumulates.

Add an HTTP health endpoint to your Quix Streams application:

from quixstreams import Application
from http.server import BaseHTTPRequestHandler, HTTPServer
import threading
import json
import time

app = Application(
    broker_address="localhost:9092",
    consumer_group="my-stream-processor",
    auto_offset_reset="earliest",
)

# Global health state
health_state = {"status": "starting", "started_at": None}

class HealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/health':
            payload = {**health_state, "uptime_s": time.time() - (health_state["started_at"] or time.time())}
            self.send_response(200)
            self.end_headers()
            self.wfile.write(json.dumps(payload).encode())

    def log_message(self, *args):
        pass  # Suppress default access logs

def start_health_server():
    HTTPServer(('0.0.0.0', 8080), HealthHandler).serve_forever()

threading.Thread(target=start_health_server, daemon=True).start()

# Mark healthy once the app loop is running
health_state["status"] = "running"
health_state["started_at"] = time.time()

Then add the monitor in Vigilmon:

1. Log in to vigilmon.online and click Add Monitor.
2. Set Type to HTTP / HTTPS.
3. URL: http://your-quix-host:8080/health
4. Check interval: 1 minute
5. Expected HTTP status: 200
6. Keyword check: running
7. Click Save.

Step 2: Monitor Kafka Consumer Lag

Consumer group lag is the primary health signal for any Kafka stream processor — it tells you how far behind your application is from the latest messages on each topic-partition. Growing lag means your processor is slower than message production, and at some point your real-time use case becomes batch processing.

Expose consumer lag metrics from your application:

from confluent_kafka import Consumer, TopicPartition
from confluent_kafka.admin import AdminClient
import json

def get_consumer_lag(consumer: Consumer, topic: str, group_id: str) -> dict:
    """Get per-partition lag for a consumer group."""
    metadata = consumer.list_topics(topic)
    partitions = [
        TopicPartition(topic, p)
        for p in metadata.topics[topic].partitions.keys()
    ]

    # Get committed offsets
    committed = consumer.committed(partitions)

    # Get end offsets (high watermarks)
    end_offsets = consumer.get_watermark_offsets(partitions[0])

    lag_by_partition = {}
    for tp in committed:
        _, high = consumer.get_watermark_offsets(tp)
        lag = max(0, high - (tp.offset if tp.offset >= 0 else 0))
        lag_by_partition[tp.partition] = lag

    total_lag = sum(lag_by_partition.values())
    return {
        "total_lag": total_lag,
        "partitions": lag_by_partition,
        "lag_ok": total_lag < 10000  # Your SLA threshold
    }

Expose this via HTTP and monitor it:

1. Click Add Monitor → HTTP / HTTPS.
2. URL: http://your-quix-host:8080/metrics/consumer-lag
3. Keyword check: lag_ok":true
4. Check interval: 2 minutes
5. Click Save.

Step 3: Monitor RocksDB State Health

Quix Streams uses RocksDB as its state backend for windowed aggregations and joins. RocksDB disk usage grows continuously unless compaction keeps up — and if the disk fills, your Quix Streams process crashes with an IO error.

Instrument RocksDB health in your application:

import os
import shutil

def get_rocksdb_health(state_dir: str = "/tmp/quixstreams-state") -> dict:
    """Check RocksDB disk usage and report health."""
    if not os.path.exists(state_dir):
        return {"size_bytes": 0, "disk_free_pct": 100, "rocksdb_ok": True}

    # Get total size of RocksDB state directory
    state_size = sum(
        os.path.getsize(os.path.join(root, f))
        for root, dirs, files in os.walk(state_dir)
        for f in files
    )

    # Get disk free space
    disk_usage = shutil.disk_usage(state_dir)
    disk_free_pct = (disk_usage.free / disk_usage.total) * 100

    return {
        "size_bytes": state_size,
        "size_mb": round(state_size / 1_048_576, 2),
        "disk_free_pct": round(disk_free_pct, 1),
        "rocksdb_ok": disk_free_pct > 20  # Alert when disk < 20% free
    }

Add the Vigilmon monitor:

1. Click Add Monitor → HTTP / HTTPS.
2. URL: http://your-quix-host:8080/metrics/rocksdb
3. Keyword check: rocksdb_ok":true
4. Check interval: 5 minutes
5. Click Save.

Step 4: Heartbeat for Window Computation Health

Quix Streams window aggregations (tumbling, hopping, sliding) close on a schedule based on event time or processing time. A window that stops closing silently means your aggregated output stops updating — downstream consumers receive no new records.

Add a heartbeat that fires each time a window closes successfully:

from quixstreams import Application
import requests

app = Application(broker_address="localhost:9092", consumer_group="my-processor")
topic = app.topic("sensor-readings")

sdf = app.dataframe(topic)

# Tumbling window — emit heartbeat on each window close
sdf = (
    sdf
    .tumbling_window(duration_ms=60_000)  # 1-minute windows
    .sum()
    .final()
    .apply(lambda value: ping_heartbeat_on_window_close(value))
)

def ping_heartbeat_on_window_close(windowed_value):
    """Ping Vigilmon each time a window finalizes."""
    try:
        requests.get(
            "https://vigilmon.online/heartbeat/quix-window-abc123",
            timeout=3
        )
    except Exception:
        pass
    return windowed_value

In Vigilmon:

1. Click Add Monitor → Cron Heartbeat.
2. Set the expected interval to 2 minutes (2× your window duration for safety margin).
3. Copy and paste the heartbeat URL into the ping_heartbeat_on_window_close function.
4. Click Save.

Step 5: Monitor Message Processing Throughput

A drop in messages-per-second is often the first symptom of a Kafka consumer rebalance, RocksDB compaction stall, or upstream producer slowdown. Track throughput in your application:

import time
from collections import deque
import threading

class ThroughputTracker:
    def __init__(self, window_seconds: int = 60):
        self.window_seconds = window_seconds
        self.timestamps = deque()
        self._lock = threading.Lock()

    def record(self):
        now = time.time()
        with self._lock:
            self.timestamps.append(now)
            cutoff = now - self.window_seconds
            while self.timestamps and self.timestamps[0] < cutoff:
                self.timestamps.popleft()

    @property
    def messages_per_second(self) -> float:
        with self._lock:
            return len(self.timestamps) / self.window_seconds

throughput = ThroughputTracker()

# Call throughput.record() in your Quix Streams message handler

Expose via HTTP and monitor:

1. Click Add Monitor → HTTP / HTTPS.
2. URL: http://your-quix-host:8080/metrics/throughput
3. Keyword check: use a keyword like degraded":false (set degraded: true when throughput drops >30% below baseline).
4. Check interval: 2 minutes
5. Click Save.

Step 6: Monitor Kafka Transaction Health

Quix Streams uses Kafka transactions for exactly-once semantics. A failed transaction means messages are reprocessed — at-least-once delivery until the transaction is resolved, which can cause duplicate outputs in downstream systems.

Track transaction commit success in your application:

from quixstreams import Application

# Quix Streams wraps Kafka transactions internally — instrument at the
# commit callback level or via producer interceptor
transaction_metrics = {"commits": 0, "failures": 0}

def on_commit_success(offsets, error):
    if error:
        transaction_metrics["failures"] += 1
    else:
        transaction_metrics["commits"] += 1

# Expose via metrics endpoint
def get_transaction_health() -> dict:
    total = transaction_metrics["commits"] + transaction_metrics["failures"]
    failure_rate = transaction_metrics["failures"] / max(total, 1)
    return {
        "commit_total": total,
        "failure_total": transaction_metrics["failures"],
        "failure_rate": round(failure_rate, 4),
        "transactions_ok": failure_rate < 0.01  # Alert if > 1% failure rate
    }

1. Click Add Monitor → HTTP / HTTPS.
2. URL: http://your-quix-host:8080/metrics/transactions
3. Keyword check: transactions_ok":true
4. Check interval: 5 minutes
5. Click Save.

Step 7: Monitor Dead Letter Queue Accumulation

Messages that fail processing repeatedly are routed to a dead letter queue (DLQ) topic. DLQ accumulation indicates systematic message processing failures — malformed data, schema mismatches, or downstream errors.

1. Click Add Monitor → HTTP / HTTPS.
2. URL: http://your-quix-host:8080/metrics/dlq
3. Keyword check: dlq_size":0 (or dlq_ok":true if you allow a small DLQ backlog).
4. Check interval: 5 minutes
5. Click Save.

Instrument DLQ writes in your error handler:

from quixstreams import Application
from quixstreams.sinks import KafkaSink

dlq_metrics = {"dlq_total": 0}
dlq_topic = app.topic("my-processor-dlq")

def handle_message_error(message, error):
    """Route failed messages to DLQ and track count."""
    dlq_metrics["dlq_total"] += 1
    # Write to DLQ topic
    dlq_producer.produce(dlq_topic.name, value=message.value(), key=message.key())

Step 8: Monitor Offset Commit Health

Kafka offset commits are the checkpointing mechanism for at-least-once delivery. Failed commits cause message reprocessing after restart. Track commit health:

offset_metrics = {"commits": 0, "failures": 0}

# Hook into Quix Streams commit callbacks
def on_offset_committed(offsets, error):
    if error:
        offset_metrics["failures"] += 1
    else:
        offset_metrics["commits"] += 1

def get_offset_health() -> dict:
    total = offset_metrics["commits"] + offset_metrics["failures"]
    failure_rate = offset_metrics["failures"] / max(total, 1)
    return {
        "commit_total": total,
        "failure_rate": round(failure_rate, 4),
        "offsets_ok": failure_rate < 0.005
    }

1. Click Add Monitor → HTTP / HTTPS.
2. URL: http://your-quix-host:8080/metrics/offsets
3. Keyword check: offsets_ok":true
4. Check interval: 2 minutes
5. Click Save.

Step 9: Monitor Output Topic Write Health

Quix Streams writes processed records to output Kafka topics. Failed producer writes mean downstream consumers stop receiving data — even though your processor appears healthy.

1. Click Add Monitor → TCP Port.
2. Host: your Kafka broker.
3. Port: 9092
4. Check interval: 1 minute
5. Click Save.

Also track producer-level write success in your application:

producer_metrics = {"writes": 0, "failures": 0}

def on_delivery(err, msg):
    if err:
        producer_metrics["failures"] += 1
    else:
        producer_metrics["writes"] += 1

# Expose as metrics endpoint

Step 10: Configure Alert Channels

Set alert routing to reach the right responders:

Process crash (P0):

• Monitor: Application HTTP health check
• Alert condition: down for 1 check
• Channel: PagerDuty or SMS
• Reason: consumer group goes offline; lag accumulates until rebalance

Consumer lag SLA breach (P1):

• Monitor: Consumer lag HTTP check
• Alert condition: lag_ok false
• Channel: Slack #streaming-ops
• Reason: real-time processing falling behind; downstream systems receiving stale data

Window computation stalled (P1):

• Monitor: Cron Heartbeat
• Alert condition: missed by 1 interval
• Channel: Slack #streaming-ops
• Reason: aggregated output has stopped updating

RocksDB disk critical (P1):

• Monitor: RocksDB health HTTP check
• Alert condition: rocksdb_ok false
• Channel: Slack #streaming-ops + PagerDuty
• Reason: OOM or disk-full crash imminent

DLQ accumulation (P2):

• Monitor: DLQ metrics HTTP check
• Alert condition: dlq_size > 0
• Channel: Slack #streaming-ops
• Reason: systematic message processing failures requiring investigation

Transaction failure rate (P2):

• Monitor: Transaction health HTTP check
• Alert condition: transactions_ok false
• Channel: Slack #streaming-ops
• Reason: exactly-once guarantees degraded; duplicate outputs possible

Summary

| What to monitor | Monitor type | Check interval | Alert condition | |---|---|---|---| | Quix Streams process | HTTP | 1 min | Down for 1 check | | Kafka consumer lag | HTTP / keyword | 2 min | lag_ok false | | RocksDB disk health | HTTP / keyword | 5 min | rocksdb_ok false | | Window computation | Cron Heartbeat | 2× window duration | Heartbeat missed | | Processing throughput | HTTP / keyword | 2 min | degraded true | | Transaction commits | HTTP / keyword | 5 min | transactions_ok false | | Dead letter queue | HTTP / keyword | 5 min | DLQ size > 0 | | Offset commits | HTTP / keyword | 2 min | offsets_ok false | | Output Kafka broker | TCP Port | 1 min | Unreachable |

Quix Streams brings the power of stateful stream processing to Python without requiring Java expertise — but exactly-once semantics, RocksDB state, and windowed aggregations all have failure modes that are invisible without active monitoring. Vigilmon surfaces these failures quickly so your streaming pipeline stays healthy and your downstream consumers never silently receive stale or duplicate data.

Get started with Vigilmon free →