tutorial

How to Monitor Materialize with Vigilmon

"A complete guide to monitoring your Materialize real-time data warehouse—covering cluster health, source ingestion lag, view latency, replica sync, and Vigilmon integration."

How to Monitor Materialize with Vigilmon

Materialize is a real-time data warehouse that runs SQL queries directly on streams. Instead of running batch ETL jobs, you write SQL views that Materialize maintains incrementally as new data arrives. The result is always-fresh query results with millisecond latency—as long as the system is healthy.

When Materialize runs into trouble, the symptoms are subtle. A source might fall behind ingesting data. A replica might drift out of sync. A view might serve slightly stale results without any visible error. This tutorial shows you how to catch these problems early using Materialize's built-in observability tools and Vigilmon's monitoring infrastructure.

Prerequisites

  • A running Materialize environment (Materialize Cloud or self-hosted)
  • Access to the Materialize SQL console or psql
  • Vigilmon account at vigilmon.online
  • Familiarity with basic SQL

Materialize's Architecture

Materialize organizes compute resources into clusters—groups of workers that execute queries and maintain views. Each cluster can have multiple replicas for redundancy. Data enters Materialize via sources (Kafka, PostgreSQL CDC, S3) and is consumed via views and materialized views.

The key invariants to monitor are:

  1. Clusters and replicas are running
  2. Sources are ingesting data without lag
  3. Views are returning results within latency SLAs
  4. Replicas within a cluster are synchronized
  5. Storage utilization is within bounds

Step 1: Check Cluster and Replica Health

Query the mz_cluster_replicas system catalog table to see the status of each replica:

-- Connect to Materialize
psql "postgres://user:password@host:6875/materialize"

-- List all clusters and replicas
SELECT
  c.name AS cluster_name,
  r.name AS replica_name,
  r.size,
  rs.ready AS is_ready,
  rs.reason
FROM mz_clusters c
JOIN mz_cluster_replicas r ON c.id = r.cluster_id
LEFT JOIN mz_cluster_replica_statuses rs ON r.id = rs.replica_id
ORDER BY c.name, r.name;

A healthy cluster shows all replicas with is_ready = true. Any replica with is_ready = false is either starting, crashing, or out of memory.

For automated checks, expose this as a script:

#!/bin/bash
CONN="postgres://user:password@materialize-host:6875/materialize"

UNHEALTHY=$(psql "$CONN" -t -c "
  SELECT COUNT(*)
  FROM mz_cluster_replicas r
  LEFT JOIN mz_cluster_replica_statuses rs ON r.id = rs.replica_id
  WHERE rs.ready = false OR rs.ready IS NULL;
")

UNHEALTHY=$(echo "$UNHEALTHY" | tr -d '[:space:]')

if [ "$UNHEALTHY" -gt "0" ]; then
  echo "CRITICAL: $UNHEALTHY replica(s) not ready"
  exit 2
fi
echo "OK: All replicas ready"

Step 2: Monitor Source Ingestion Lag

Sources are your connection to external data. When a Kafka source falls behind, every downstream materialized view grows stale. Materialize exposes source lag through the mz_source_statistics system table:

-- Check source ingestion status
SELECT
  s.name AS source_name,
  s.type,
  ss.messages_received,
  ss.bytes_received,
  ss.updates_staged,
  ss.updates_committed,
  ss.envelope_errors
FROM mz_sources s
JOIN mz_source_statistics ss ON s.id = ss.id
WHERE s.type != 'subsource'
ORDER BY s.name;

Pay attention to envelope_errors—any non-zero value means Materialize is receiving malformed messages and dropping them.

For Kafka sources specifically, compare the updates_committed counter to the Kafka topic's end offset to derive lag:

#!/usr/bin/env python3
import psycopg2
from kafka import KafkaAdminClient, KafkaConsumer
from kafka.structs import TopicPartition

MATERIALIZE_DSN = "postgres://user:password@host:6875/materialize"
KAFKA_BOOTSTRAP = "kafka:9092"

def get_source_lag():
    conn = psycopg2.connect(MATERIALIZE_DSN)
    cur = conn.cursor()
    cur.execute("""
        SELECT s.name, ss.updates_committed
        FROM mz_sources s
        JOIN mz_source_statistics ss ON s.id = ss.id
        WHERE s.type != 'subsource'
    """)
    sources = cur.fetchall()
    conn.close()
    return sources

if __name__ == "__main__":
    sources = get_source_lag()
    for name, committed in sources:
        print(f"Source '{name}': {committed} messages committed")

Add a threshold check: if the committed count hasn't increased in 5 minutes but the source topic is still active, the source is stalled.

Step 3: Measure View Query Latency

The whole point of Materialize is low-latency query results. Monitor view latency with the mz_materialized_view_refreshes table and by timing actual queries:

-- Check if views have recent refresh activity
SELECT
  mv.name AS view_name,
  mv.cluster_id,
  mv.create_sql
FROM mz_materialized_views mv
ORDER BY mv.name;

For latency measurement, time a representative query against each critical materialized view:

#!/usr/bin/env python3
import psycopg2
import time

MATERIALIZE_DSN = "postgres://user:password@host:6875/materialize"
LATENCY_THRESHOLD_MS = 500  # 500ms SLA

VIEWS_TO_CHECK = [
    ("user_order_summary", "SELECT COUNT(*) FROM user_order_summary"),
    ("realtime_metrics", "SELECT COUNT(*) FROM realtime_metrics WHERE ts > NOW() - INTERVAL '1 minute'"),
]

conn = psycopg2.connect(MATERIALIZE_DSN)
exit_code = 0

for view_name, query in VIEWS_TO_CHECK:
    cur = conn.cursor()
    start = time.time()
    try:
        cur.execute(query)
        cur.fetchall()
        latency_ms = (time.time() - start) * 1000
        if latency_ms > LATENCY_THRESHOLD_MS:
            print(f"WARNING: {view_name} query took {latency_ms:.0f}ms (threshold {LATENCY_THRESHOLD_MS}ms)")
            exit_code = max(exit_code, 1)
        else:
            print(f"OK: {view_name} responded in {latency_ms:.0f}ms")
    except Exception as e:
        print(f"CRITICAL: {view_name} query failed: {e}")
        exit_code = 2
    finally:
        cur.close()

conn.close()
exit(exit_code)

Step 4: Monitor Replica Sync Status

When a cluster has multiple replicas, they should serve identical data. Replica divergence means some queries get stale results depending on which replica Materialize routes them to. Check synchronization status:

-- Check replica hydration status
SELECT
  c.name AS cluster_name,
  r.name AS replica_name,
  rh.hydrated,
  rh.reason
FROM mz_clusters c
JOIN mz_cluster_replicas r ON c.id = r.cluster_id
LEFT JOIN mz_cluster_replica_hydration_events rh ON r.id = rh.replica_id
ORDER BY c.name, r.name;

A replica that is not hydrated is still catching up to the current data frontier and should not serve production traffic. If a replica stays unhydrated for more than a few minutes after a restart, investigate memory and CPU.

You can also use Materialize's introspection indexes to check each replica's progress:

-- Check the watermark per replica
SELECT
  r.name AS replica_name,
  mva.object_id,
  mva.replica_id
FROM mz_cluster_replicas r
JOIN mz_arrangement_sizes mva ON r.id = mva.replica_id
WHERE mva.records > 0
ORDER BY r.name, mva.records DESC;

Step 5: Monitor Storage Utilization

Materialize persists state to object storage (S3-compatible). When storage grows unbounded, compaction cannot keep up and query performance degrades. Check storage via system tables:

-- Check storage usage per object
SELECT
  mo.id,
  mo.type,
  mo.name,
  ms.size_bytes
FROM mz_objects mo
LEFT JOIN mz_storage_usage ms ON mo.id = ms.object_id
WHERE ms.size_bytes IS NOT NULL
ORDER BY ms.size_bytes DESC
LIMIT 20;

Set an alert when any single object exceeds your budget. For most deployments, a single materialized view exceeding 100GB warrants investigation—either your retention window is too long or you need to optimize the view definition.

Step 6: Create a Health API Endpoint

Wrap your checks into a simple HTTP endpoint that Vigilmon can poll:

#!/usr/bin/env python3
from flask import Flask, jsonify
import psycopg2
import time

app = Flask(__name__)
MATERIALIZE_DSN = "postgres://user:password@host:6875/materialize"

@app.route('/health')
def health():
    checks = {}

    try:
        conn = psycopg2.connect(MATERIALIZE_DSN, connect_timeout=5)
        cur = conn.cursor()

        # Check replica readiness
        cur.execute("""
            SELECT COUNT(*) FROM mz_cluster_replicas r
            LEFT JOIN mz_cluster_replica_statuses rs ON r.id = rs.replica_id
            WHERE rs.ready = false OR rs.ready IS NULL
        """)
        unhealthy_replicas = cur.fetchone()[0]
        checks['replicas'] = 'ok' if unhealthy_replicas == 0 else f'{unhealthy_replicas} unhealthy'

        # Check source errors
        cur.execute("SELECT SUM(envelope_errors) FROM mz_source_statistics")
        errors = cur.fetchone()[0] or 0
        checks['source_errors'] = 'ok' if errors == 0 else f'{errors} errors'

        conn.close()
        all_ok = all(v == 'ok' for v in checks.values())
        return jsonify({'status': 'ok' if all_ok else 'degraded', 'checks': checks}), \
               200 if all_ok else 503

    except Exception as e:
        return jsonify({'status': 'error', 'error': str(e)}), 503

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

Step 7: Configure Vigilmon Monitors

Cluster Health Monitor

In Vigilmon, go to Monitors → New Monitor and create:

| Field | Value | |-------|-------| | Name | Materialize Cluster Health | | Type | HTTP | | URL | http://your-health-service:8080/health | | Expected Status | 200 | | Response Contains | "status": "ok" | | Interval | 60 seconds | | Timeout | 15 seconds |

Source Lag Monitor

Since source lag requires a database query, use Vigilmon's script monitor or expose it as an HTTP endpoint. Create a dedicated /lag route:

@app.route('/lag')
def lag_check():
    conn = psycopg2.connect(MATERIALIZE_DSN, connect_timeout=5)
    cur = conn.cursor()
    cur.execute("""
        SELECT SUM(envelope_errors) AS errors
        FROM mz_source_statistics
    """)
    errors = cur.fetchone()[0] or 0
    conn.close()
    ok = errors == 0
    return jsonify({'envelope_errors': errors}), 200 if ok else 503

Add a Vigilmon HTTP monitor for /lag with expected status 200.

SQL Query Latency Monitor

Use Vigilmon's TCP monitor to verify port 6875 (Materialize SQL interface) is accepting connections:

| Field | Value | |-------|-------| | Name | Materialize SQL Port | | Type | TCP | | Host | materialize-host | | Port | 6875 | | Interval | 30 seconds |

Pair this with the HTTP latency check to catch both connection failures and slow query performance.

Alert Routing

Configure notifications under Settings → Notifications:

  • Email: immediate for cluster down
  • Slack: #data-platform for source lag or replica sync issues
  • PagerDuty: for total cluster outage or persistent replica failures

Set Vigilmon to page after 2 consecutive failures (2 minutes of confirmed outage) to avoid alert fatigue from brief transient failures.

Monitoring Checklist

| Check | System | Threshold | |-------|--------|-----------| | Replica ready status | SQL query | 0 unready | | Source envelope errors | SQL query | 0 errors | | View query latency | Timed query | < 500ms | | Replica hydration | SQL query | All hydrated | | Storage per object | SQL query | < 100GB | | SQL port availability | TCP check | Port open | | Health endpoint | HTTP check | 200 OK |

Troubleshooting Common Issues

Replica not ready after restart:

  • Check memory limits—replica may have been OOM-killed
  • Review logs: mz_cluster_replica_metrics for resource spikes
  • Consider downgrading replica size to fit within node capacity

Source lag growing:

  • Check Kafka broker health and partition leadership
  • Verify network connectivity between Materialize and Kafka
  • Look for schema evolution issues causing envelope_errors

Query latency spike:

  • Check if the cluster replica is under CPU pressure
  • Look for large in-memory arrangements consuming memory
  • Consider adding a replica to spread query load

Conclusion

Materialize delivers real-time SQL on streams, but its operational health depends on keeping replicas synchronized, sources ingesting cleanly, and views staying within latency budgets. The system tables in Materialize give you rich insight into all of these dimensions—the challenge is surfacing that data consistently and alerting on it automatically.

By wrapping your key checks in a health API and pointing Vigilmon at it, you get continuous monitoring without managing separate monitoring infrastructure. Start with the replica health and source error checks, add latency monitoring for your most critical views, and use the storage utilization check to catch unbounded growth before it becomes an incident.

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