tutorial

Monitoring Pathway with Vigilmon

Pathway is a Python framework for real-time AI pipelines and RAG with automatic incremental computation. Here's how to monitor its workers, LLM API calls, vector DB sync, data connectors, and pipeline lag with Vigilmon.

Pathway is an open source Python framework for building real-time AI and data pipelines — RAG document ingestion, ML feature computation, CDC-driven analytics — using familiar pandas-like APIs and automatic incremental computation. A Pathway pipeline is a long-running Python process that continuously reads from sources (Kafka, S3, PostgreSQL, REST APIs), transforms data, optionally calls LLMs or writes to vector databases, and emits results to output sinks. When any link in that chain breaks — an LLM API goes down, a Kafka connector drops, memory grows unbounded — your real-time AI application silently degrades. Vigilmon monitors every layer of the Pathway stack so you catch failures before your RAG index goes stale or your ML features fall behind.

What You'll Set Up

  • Process health monitor for the Pathway worker via REST API
  • Data source connector health checks per input source
  • LLM API availability and error rate monitoring
  • Vector database write health (for RAG pipelines)
  • Incremental computation lag monitoring
  • Output sink write health
  • Memory usage alert
  • RAG freshness heartbeat
  • Pipeline error rate alerting

Prerequisites

  • Pathway pipeline deployed (Docker container, systemd service, or Kubernetes pod)
  • Pathway REST API server enabled (pw.io.http.rest_connector or pw.run_server())
  • Access to your LLM provider API endpoint (if applicable)
  • A free Vigilmon account

Step 1: Monitor the Pathway Worker Process

The Pathway worker is the Python process running your pipeline. A crash stops all incremental computation — source connectors disconnect, LLM calls stop, and your real-time index freezes.

Expose a health endpoint from your Pathway application:

import pathway as pw
from pathway.io.http import rest_connector

# Add a health check route alongside your pipeline
@pw.udf
def health_check(query: str) -> str:
    return '{"status": "ok", "pipeline": "running"}'

# Or use Pathway's built-in REST server
pw.run_server(host="0.0.0.0", port=8000, with_cache=False)

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-pathway-host:8000/health
  4. Check interval: 1 minute
  5. Expected HTTP status: 200
  6. Keyword check: running
  7. Click Save.

If your pipeline runs as a systemd service, also monitor the TCP port the Pathway REST server binds to as a secondary liveness signal.

Step 2: Monitor Data Source Connectors

Pathway reads from Kafka, S3, PostgreSQL, and REST APIs. A lost connector means your pipeline processes stale data silently.

Kafka Input Connector

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

PostgreSQL Input Connector (CDC)

  1. Click Add Monitor → TCP Port.
  2. Host: your PostgreSQL host.
  3. Port: 5432
  4. Check interval: 1 minute
  5. Click Save.

REST API Input Connector

If Pathway polls an external REST API as a data source:

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: the external REST API endpoint Pathway polls.
  3. Expected HTTP status: 200
  4. Check interval: 2 minutes
  5. Click Save.

For each connector, add the health metric to your Pathway pipeline's instrumentation:

import pathway as pw
import requests

class ConnectorHealthLogger:
    def log_connector_status(self, connector_name: str, ok: bool):
        # Emit to your metrics endpoint
        print(f"connector_health{{connector=\"{connector_name}\"}} {1 if ok else 0}")

Step 3: Monitor LLM API Health

If your Pathway pipeline calls an LLM for real-time AI processing (e.g., embedding generation for RAG, classification, summarization), LLM API downtime or elevated error rates directly break your pipeline's AI output.

LLM API Endpoint Monitor

  1. Click Add Monitor → HTTP / HTTPS.
  2. For OpenAI: URL https://api.openai.com/v1/models For a self-hosted model (Ollama, vLLM): URL http://your-llm-host:11434/api/tags
  3. Expected HTTP status: 200
  4. Check interval: 2 minutes
  5. Click Save.

LLM Error Rate via Pipeline Metrics

Add error rate instrumentation to your LLM calls in the Pathway pipeline:

import pathway as pw
import openai
import time

@pw.udf
def call_llm_with_monitoring(text: str) -> str:
    start = time.time()
    try:
        response = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": text}]
        )
        # Record success metric
        record_metric("llm_api_success", 1)
        record_metric("llm_api_latency_ms", (time.time() - start) * 1000)
        return response.choices[0].message.content
    except Exception as e:
        # Record failure metric
        record_metric("llm_api_error", 1)
        raise

Expose the error rate via an HTTP metrics endpoint and add a Vigilmon keyword check:

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-pathway-host:8000/metrics/llm
  3. Keyword check: ensure error_rate is below your threshold (e.g., check for error_rate":0. matching only values under 0.05).
  4. Check interval: 2 minutes
  5. Click Save.

Step 4: Monitor Vector Database Write Health (RAG Pipelines)

Pathway-powered RAG pipelines write embeddings to a vector database (Pinecone, Weaviate, Qdrant). Failed writes cause your RAG index to go stale — queries return outdated results without any obvious error.

Add write health tracking to your Pathway vector sink:

import pathway as pw
import requests

@pw.udf
def write_to_vector_db_with_health(embedding: list, doc_id: str) -> str:
    try:
        # Your vector DB write logic here
        result = vector_db_client.upsert([(doc_id, embedding)])
        record_metric("vector_write_success", 1)
        return "ok"
    except Exception as e:
        record_metric("vector_write_failure", 1)
        raise

Then monitor the sink health endpoint:

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-pathway-host:8000/metrics/vector-sink
  3. Expected HTTP status: 200
  4. Keyword check: write_failure":0
  5. Check interval: 2 minutes
  6. Click Save.

Step 5: Monitor Incremental Computation Lag

Pathway's key promise is low-latency incremental updates — your output should reflect new input within your SLA window. If lag grows, your real-time AI application is serving stale results.

Instrument end-to-end lag in your pipeline:

import pathway as pw
import time

@pw.udf
def tag_with_ingestion_time(data: str) -> dict:
    return {"data": data, "ingested_at": time.time()}

@pw.udf
def compute_lag_and_emit(tagged_data: dict) -> str:
    lag_ms = (time.time() - tagged_data["ingested_at"]) * 1000
    record_metric("pipeline_lag_ms", lag_ms)
    if lag_ms > 5000:  # Alert threshold: 5 second SLA
        record_metric("pipeline_sla_breach", 1)
    return tagged_data["data"]

Monitor the lag metric:

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-pathway-host:8000/metrics/lag
  3. Keyword check: sla_breach":0
  4. Check interval: 1 minute
  5. Click Save.

Step 6: Monitor the Pathway REST API Output

If Pathway exposes query results via HTTP (e.g., for real-time RAG retrieval), monitor the API endpoint directly:

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-pathway-host:8000/v1/query (or your configured endpoint path).
  3. Method: GET or POST depending on your Pathway REST connector configuration.
  4. Expected HTTP status: 200
  5. Check interval: 1 minute
  6. Click Save.

Step 7: Monitor Output Sink Health

Pathway writes results to Kafka, S3, or REST endpoints. A sink failure means processed data accumulates in memory without being delivered downstream.

Kafka Output Topic

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

For sink-level write success, use the same metrics endpoint pattern from Step 4:

@pw.udf
def write_to_kafka_with_health(record: dict) -> str:
    try:
        producer.produce(output_topic, key=record["id"], value=json.dumps(record))
        producer.flush()
        record_metric("kafka_sink_write_success", 1)
        return "ok"
    except Exception as e:
        record_metric("kafka_sink_write_failure", 1)
        raise

Step 8: Monitor Memory Usage

Pathway maintains incremental computation state in memory. Unbounded state growth causes OOM crashes — particularly for pipelines that join large historical datasets or accumulate windowed aggregations.

Expose memory usage from your Pathway process:

import psutil
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import threading

class MetricsHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/metrics/memory':
            process = psutil.Process(os.getpid())
            mem_info = process.memory_info()
            total_ram = psutil.virtual_memory().total
            usage_pct = (mem_info.rss / total_ram) * 100
            payload = {
                "rss_bytes": mem_info.rss,
                "memory_pct": round(usage_pct, 2),
                "oom_risk": usage_pct > 80
            }
            self.send_response(200)
            self.end_headers()
            self.wfile.write(json.dumps(payload).encode())

# Start metrics server in background thread
threading.Thread(
    target=lambda: HTTPServer(('0.0.0.0', 9090), MetricsHandler).serve_forever(),
    daemon=True
).start()

Then add the Vigilmon monitor:

  1. Click Add Monitor → HTTP / HTTPS.
  2. URL: http://your-pathway-host:9090/metrics/memory
  3. Keyword check: oom_risk":false
  4. Check interval: 2 minutes
  5. Click Save.

Step 9: RAG Freshness Heartbeat

For RAG pipelines, staleness is a silent failure: the pipeline is running but hasn't ingested a new document in hours. Add a heartbeat that fires each time a document is successfully ingested and indexed:

  1. Click Add Monitor → Cron Heartbeat.
  2. Set the expected interval to your target ingestion cadence (e.g., 30 minutes if you expect new documents at least every 30 minutes).
  3. Copy the heartbeat URL.
  4. Fire the heartbeat from your document ingestion sink:
import pathway as pw
import requests

@pw.udf
def index_document_with_freshness_ping(doc: dict) -> str:
    # Index the document
    index_result = vector_db_client.upsert_document(doc)

    # Ping Vigilmon to record a successful ingest
    try:
        requests.get(
            "https://vigilmon.online/heartbeat/rag-freshness-abc123",
            timeout=3
        )
    except Exception:
        pass  # Don't let heartbeat failures affect pipeline processing

    return index_result

If document ingestion stalls, the heartbeat goes silent and Vigilmon alerts you.

Step 10: Configure Alert Channels

Route alerts based on severity:

Worker crash (P0):

  • Monitor: Worker process HTTP health check
  • Alert condition: down for 1 check
  • Channel: PagerDuty or SMS
  • Reason: pipeline stops entirely; no incremental updates until process restarts

LLM API down (P1):

  • Monitor: LLM API endpoint HTTP check
  • Alert condition: down for 2 consecutive checks
  • Channel: Slack #ai-ops
  • Reason: real-time AI processing fails; RAG queries degrade to stale results

Incremental lag SLA breach (P1):

  • Monitor: Pipeline lag metrics HTTP check
  • Alert condition: sla_breach keyword present
  • Channel: Slack #ai-ops
  • Reason: real-time AI inference is serving stale data

Vector DB write failures (P1):

  • Monitor: Vector sink health HTTP check
  • Alert condition: write_failure > 0
  • Channel: Slack #ai-ops
  • Reason: RAG index going stale

Memory OOM risk (P2):

  • Monitor: Memory metrics HTTP check
  • Alert condition: oom_risk true
  • Channel: Slack #data-engineering + PagerDuty
  • Reason: OOM imminent; need to tune state retention or scale memory

RAG freshness heartbeat missed (P2):

  • Monitor: Cron Heartbeat
  • Alert condition: missed by 1 interval
  • Channel: Slack #ai-ops
  • Reason: document ingestion stalled; RAG index becoming stale

Summary

| What to monitor | Monitor type | Check interval | Alert condition | |---|---|---|---| | Pathway worker process | HTTP | 1 min | Down for 1 check | | Kafka/PG source connectors | TCP Port | 1 min | Unreachable | | LLM API availability | HTTP | 2 min | Down for 2 checks | | LLM API error rate | HTTP / keyword | 2 min | Error rate > 0 | | Vector DB write health | HTTP / keyword | 2 min | Write failures > 0 | | Incremental computation lag | HTTP / keyword | 1 min | SLA breach flag | | Pathway REST API output | HTTP | 1 min | Down for 1 check | | Output sink write health | HTTP / keyword | 2 min | Write failures > 0 | | Process memory usage | HTTP / keyword | 2 min | OOM risk flag | | RAG document freshness | Cron Heartbeat | Per ingest cadence | Heartbeat missed |

Pathway's incremental computation model makes it uniquely powerful for real-time AI pipelines — but that same continuous-processing model means a silent failure can let stale data accumulate undetected. Vigilmon closes that gap with end-to-end monitoring from source connector to vector index freshness, so your RAG pipeline stays fresh and your users never see yesterday's answers.

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