tutorial

Monitoring Faust with Vigilmon

Faust is an async Python stream processing library for Kafka, originally built at Robinhood. Here's how to monitor its workers, Kafka tables, asyncio event loop, rebalance health, and agent throughput with Vigilmon.

Faust is an open source Python stream processing library originally developed at Robinhood for real-time trading systems and event-driven microservices. Built on asyncio, Faust brings Kafka Streams-style processing to Python — async agents that process messages from Kafka topics, distributed key-value tables backed by RocksDB and Kafka changelog topics, and cron-style periodic tasks alongside your streaming logic. A Faust deployment typically runs multiple Python worker processes that partition Kafka topic workloads across themselves via consumer group coordination. When a worker crashes, an asyncio event loop saturates, or a RocksDB table fills its disk, your streaming application degrades silently. Vigilmon gives you continuous visibility into Faust worker health, Kafka consumer lag, table state, and processing throughput so failures surface immediately instead of after your users notice.

What You'll Set Up

  • HTTP health probe for Faust's built-in web server
  • Kafka consumer group lag monitoring
  • Faust table (RocksDB) disk health monitor
  • Agent processing throughput alert
  • Kafka consumer group rebalance frequency monitor
  • asyncio event loop lag alert
  • Changelog topic write health
  • Worker discovery / fleet count monitor
  • Schema registry connectivity check
  • Alert channels with priority routing

Prerequisites

  • Faust application with one or more worker processes running
  • Faust web server enabled (default port 6066)
  • Access to your Kafka cluster and schema registry (if using Avro/Protobuf)
  • A free Vigilmon account

Step 1: Monitor the Faust Web Server

Faust ships an embedded web server (default port 6066) that exposes health endpoints, metrics, and worker discovery information. Monitoring this is the simplest first step — a Faust worker crash takes the web server down with it.

  1. Log in to vigilmon.online and click Add Monitor.
  2. Set Type to HTTP / HTTPS.
  3. URL: http://your-faust-host:6066/
  4. Check interval: 1 minute
  5. Expected HTTP status: 200
  6. Click Save.

Faust also exposes a dedicated health endpoint:

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-faust-host:6066/health
  3. Expected HTTP status: 200
  4. Keyword check: status":"ok" or "alive":true
  5. Check interval: 1 minute
  6. Click Save.

If you run multiple Faust workers on different hosts, add a monitor for each worker's port 6066. A monitor per worker lets you identify which specific worker crashed rather than knowing only that the fleet lost a member.

Step 2: Monitor Kafka Consumer Lag

Faust workers consume Kafka topics and distribute partitions across the worker fleet. Consumer group lag measures how far behind your workers are from the latest messages on each partition. Rising lag means your agents are processing slower than messages arrive.

Expose consumer lag from your Faust application using the built-in monitor:

import faust
from faust import web

app = faust.App(
    'myapp',
    broker='kafka://localhost:9092',
    store='rocksdb://',
    web_port=6066,
)

@app.page('/metrics/consumer-lag')
async def consumer_lag_page(self, request: web.Request) -> web.Response:
    """Expose consumer group lag per topic-partition."""
    assignment = app.consumer.assignment()
    lag_data = {}
    total_lag = 0

    for tp in assignment:
        committed = await app.consumer.position(tp)
        # Get high watermark
        _, high = await app.consumer.get_watermark_offsets(tp)
        lag = max(0, high - (committed or 0))
        lag_data[f"{tp.topic}:{tp.partition}"] = lag
        total_lag += lag

    return self.json({
        "total_lag": total_lag,
        "partitions": lag_data,
        "lag_ok": total_lag < 5000
    })

Add the monitor:

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

Step 3: Monitor Faust Table (RocksDB) Health

Faust tables are distributed key-value stores backed by RocksDB locally and replicated to Kafka changelog topics. If the local RocksDB disk fills up, Faust fails with an IO error and the worker crashes.

Add a table health endpoint to your Faust app:

import os
import shutil
import faust
from faust import web

@app.page('/metrics/rocksdb')
async def rocksdb_health(self, request: web.Request) -> web.Response:
    state_dir = app.conf.datadir  # Default: /tmp/faust-<app-id>
    
    total_size = 0
    if os.path.exists(state_dir):
        for root, dirs, files in os.walk(state_dir):
            total_size += sum(
                os.path.getsize(os.path.join(root, f))
                for f in files
            )
    
    disk = shutil.disk_usage(state_dir if os.path.exists(state_dir) else "/tmp")
    disk_free_pct = (disk.free / disk.total) * 100

    return self.json({
        "state_dir": state_dir,
        "state_size_mb": round(total_size / 1_048_576, 2),
        "disk_free_pct": round(disk_free_pct, 1),
        "rocksdb_ok": disk_free_pct > 20
    })
  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-faust-host:6066/metrics/rocksdb
  3. Keyword check: rocksdb_ok":true
  4. Check interval: 5 minutes
  5. Click Save.

Step 4: Monitor Agent Processing Throughput

Faust agents are async coroutines that process messages from Kafka topics. A drop in agent throughput is an early warning of event loop saturation, partition rebalancing, or an upstream Kafka producer slowdown.

Instrument message throughput in your agents:

import faust
import time
from collections import deque
import threading

app = faust.App('myapp', broker='kafka://localhost:9092')
topic = app.topic('orders', value_type=dict)

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

    def record(self):
        now = time.monotonic()
        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) / max(self.window_seconds, 1)

    @property
    def is_degraded(self) -> bool:
        if self.baseline_mps == 0:
            return False
        return self.messages_per_second < (self.baseline_mps * 0.7)

order_throughput = ThroughputTracker()

@app.agent(topic)
async def process_order(orders):
    async for order in orders:
        order_throughput.record()
        # Process the order...
        await handle_order(order)

@app.page('/metrics/throughput')
async def throughput_metrics(self, request: web.Request) -> web.Response:
    return self.json({
        "messages_per_second": round(order_throughput.messages_per_second, 2),
        "degraded": order_throughput.is_degraded,
        "baseline_mps": order_throughput.baseline_mps
    })
  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-faust-host:6066/metrics/throughput
  3. Keyword check: degraded":false
  4. Check interval: 2 minutes
  5. Click Save.

Step 5: Monitor Kafka Rebalance Health

Kafka consumer group rebalances happen whenever a Faust worker joins or leaves — during deployments, crashes, or scale-out events. A brief rebalance is expected; frequent or long rebalances indicate instability and cause processing gaps while partitions are reassigned.

Add rebalance tracking to your Faust app:

import time

rebalance_state = {
    "count_last_hour": 0,
    "last_rebalance_at": None,
    "last_duration_ms": None,
    "rebalance_ok": True
}

@app.on_rebalance_start.connect
async def on_rebalance_start(app, **kwargs):
    rebalance_state["_start"] = time.time()
    rebalance_state["count_last_hour"] += 1
    rebalance_state["last_rebalance_at"] = time.time()
    # Alert if > 3 rebalances in an hour
    rebalance_state["rebalance_ok"] = rebalance_state["count_last_hour"] <= 3

@app.on_rebalance_return.connect
async def on_rebalance_complete(app, **kwargs):
    start = rebalance_state.pop("_start", time.time())
    rebalance_state["last_duration_ms"] = round((time.time() - start) * 1000)

@app.page('/metrics/rebalance')
async def rebalance_metrics(self, request: web.Request) -> web.Response:
    return self.json(rebalance_state)
  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-faust-host:6066/metrics/rebalance
  3. Keyword check: rebalance_ok":true
  4. Check interval: 5 minutes
  5. Click Save.

Step 6: Monitor asyncio Event Loop Lag

Faust runs on a single asyncio event loop per worker. If a coroutine blocks (a slow database call, a blocking I/O operation, a CPU-intensive computation without await), the event loop backs up and every other coroutine waiting in the queue experiences increased latency. Event loop lag above 1 second is a strong signal of event loop saturation.

Add an event loop lag monitor:

import asyncio
import time
import faust
from faust import web

app = faust.App('myapp', broker='kafka://localhost:9092')
loop_lag_state = {"lag_ms": 0.0, "loop_ok": True}

@app.timer(interval=5.0)
async def measure_loop_lag():
    """Measure how long the event loop takes to schedule a callback."""
    start = time.monotonic()
    await asyncio.sleep(0)  # Yield to event loop and measure roundtrip
    lag = (time.monotonic() - start) * 1000
    loop_lag_state["lag_ms"] = round(lag, 2)
    loop_lag_state["loop_ok"] = lag < 1000  # Alert if > 1 second lag

@app.page('/metrics/loop-lag')
async def loop_lag_metrics(self, request: web.Request) -> web.Response:
    return self.json(loop_lag_state)
  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-faust-host:6066/metrics/loop-lag
  3. Keyword check: loop_ok":true
  4. Check interval: 1 minute
  5. Click Save.

Step 7: Monitor Changelog Topic Write Health

Faust tables replicate state changes to Kafka changelog topics. If changelog writes fail, table state diverges between workers on restart — recovered state will be missing or inconsistent.

Track changelog write health:

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

# Instrument Faust's store to track changelog writes
# Faust's RocksDB store sends changelog records via the internal Kafka producer

@app.page('/metrics/changelog')
async def changelog_health(self, request: web.Request) -> web.Response:
    total = changelog_metrics["writes"] + changelog_metrics["failures"]
    failure_rate = changelog_metrics["failures"] / max(total, 1)
    return self.json({
        "write_total": total,
        "failure_total": changelog_metrics["failures"],
        "failure_rate": round(failure_rate, 4),
        "changelog_ok": failure_rate < 0.01
    })

Also add a TCP port monitor on your Kafka broker to catch the underlying connectivity failure:

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

Step 8: Monitor Worker Fleet Size

Faust workers discover each other via Kafka for partition coordination. If a worker crashes unexpectedly, the fleet is smaller than intended — some partitions are processed by fewer workers, potentially increasing per-worker load and latency.

@app.page('/metrics/workers')
async def worker_fleet(self, request: web.Request) -> web.Response:
    """Report current worker count from Kafka consumer group."""
    # Faust exposes connected worker count via the cluster/topology
    worker_count = len(app.consumer._current_assignment or [])
    
    # You can also query the Kafka consumer group directly
    expected_workers = int(os.environ.get("EXPECTED_WORKER_COUNT", "3"))
    
    return self.json({
        "worker_count": worker_count,
        "expected_count": expected_workers,
        "fleet_ok": worker_count >= expected_workers
    })
  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-faust-host:6066/metrics/workers
  3. Keyword check: fleet_ok":true
  4. Check interval: 2 minutes
  5. Click Save.

Step 9: Monitor Schema Registry (Avro/Protobuf)

If your Faust application uses Avro or Protobuf schemas (via the faust-avro package or Confluent Schema Registry), schema registry downtime causes serialization failures on every message — agents crash attempting to deserialize incoming records.

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-schema-registry-host:8081/subjects (default Confluent Schema Registry endpoint).
  3. Expected HTTP status: 200
  4. Check interval: 2 minutes
  5. Click Save.

Also monitor schema compatibility:

import requests

@app.page('/metrics/schema-registry')
async def schema_registry_health(self, request: web.Request) -> web.Response:
    registry_url = os.environ.get("SCHEMA_REGISTRY_URL", "http://localhost:8081")
    try:
        resp = requests.get(f"{registry_url}/subjects", timeout=3)
        return self.json({
            "registry_reachable": resp.status_code == 200,
            "registry_ok": resp.status_code == 200
        })
    except Exception as e:
        return self.json({"registry_reachable": False, "registry_ok": False, "error": str(e)})

Step 10: Configure Alert Channels

Route alerts based on impact severity:

Worker process crash (P0 — page immediately):

  • Monitor: Faust web server /health HTTP check
  • Alert condition: down for 1 check
  • Channel: PagerDuty or SMS
  • Reason: Kafka partitions unprocessed until rebalance; consumer lag spike

Event loop saturation (P0):

  • Monitor: asyncio loop lag HTTP check
  • Alert condition: loop_ok false
  • Channel: PagerDuty
  • Reason: > 1 second event loop lag means agents stop making progress; equivalent to a frozen worker

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 SLA

RocksDB disk critical (P1):

  • Monitor: RocksDB health HTTP check
  • Alert condition: rocksdb_ok false
  • Channel: Slack #streaming-ops + PagerDuty
  • Reason: disk-full crash imminent; table state may be corrupted on recovery

Fleet below minimum (P1):

  • Monitor: Worker fleet HTTP check
  • Alert condition: fleet_ok false
  • Channel: Slack #streaming-ops
  • Reason: fewer workers than expected; remaining workers under increased load

Schema registry down (P1):

  • Monitor: Schema registry HTTP check
  • Alert condition: down for 2 consecutive checks
  • Channel: Slack #streaming-ops
  • Reason: all Avro/Protobuf deserialization fails; agents crash

Frequent rebalances (P2):

  • Monitor: Rebalance metrics HTTP check
  • Alert condition: rebalance_ok false
  • Channel: Slack #streaming-ops
  • Reason: > 3 rebalances/hour indicates worker instability

Agent throughput drop (P2):

  • Monitor: Throughput HTTP check
  • Alert condition: degraded true
  • Channel: Slack #streaming-ops
  • Reason: > 30% throughput drop indicating upstream slowdown or rebalance

Summary

| What to monitor | Monitor type | Check interval | Alert condition | |---|---|---|---| | Faust web server | HTTP | 1 min | Down for 1 check | | Worker /health endpoint | HTTP / keyword | 1 min | alive false | | Kafka consumer lag | HTTP / keyword | 2 min | lag_ok false | | RocksDB disk health | HTTP / keyword | 5 min | rocksdb_ok false | | Agent throughput | HTTP / keyword | 2 min | degraded true | | Rebalance frequency | HTTP / keyword | 5 min | rebalance_ok false | | asyncio event loop lag | HTTP / keyword | 1 min | loop_ok false | | Changelog topic health | HTTP / keyword | 5 min | changelog_ok false | | Worker fleet count | HTTP / keyword | 2 min | fleet_ok false | | Schema registry | HTTP | 2 min | Down for 2 checks | | Kafka broker (TCP) | TCP Port | 1 min | Unreachable |

Faust's asyncio foundation makes it exceptionally efficient for high-throughput Python streaming — but asyncio's cooperative multitasking model means a single blocking coroutine can freeze the entire event loop, and RocksDB table failures are often silent until a worker crashes mid-recovery. Vigilmon monitors the full Faust operational surface from the asyncio event loop to the Kafka changelog so your stream processing pipeline stays healthy and your on-call team gets paged before users do.

Get started with Vigilmon free →

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