tutorial

Monitoring Apache Fluss with Vigilmon

Apache Fluss is the real-time streaming layer of the modern lakehouse — but tablet server failures, replication lag, and coordinator outages are invisible without external monitoring. Here's how to monitor Fluss end-to-end with Vigilmon.

Apache Fluss (Apache Software Foundation incubation, 2024) is a streaming storage system built by the same Alibaba Flink team behind Apache Flink and Apache Paimon. Where Kafka gives you high-throughput event streaming and Paimon gives you efficient analytical reads, Fluss bridges the gap: it stores real-time streaming data in columnar Arrow format with native Flink integration, supports primary key upserts at streaming latency (milliseconds, not minutes), and serves as the real-time layer in the Streaming Lakehouse architecture. Fluss tables appear as Flink dynamic tables, enabling stream-table joins and Flink SQL on data that is seconds old. But with that power comes operational responsibility: Fluss TabletServer failures cause tablet replication degradation, coordinator outages halt tablet assignment, and write latency spikes silently stall your real-time pipelines. Vigilmon provides the external monitoring that catches these failures before they reach your consumers.

What You'll Set Up

  • HTTP probe for Fluss coordinator service health
  • Heartbeat monitors for TabletServer process liveness on each node
  • Write latency and throughput monitoring via a health sidecar
  • Replication lag monitoring for under-replicated tablets
  • Snapshot health monitoring for Fluss periodic snapshots
  • ZooKeeper/coordinator connectivity monitoring
  • Flink connector health via Flink JobManager API

Prerequisites

  • Apache Fluss 0.5+ cluster with at least one Coordinator and two TabletServers
  • Apache Flink 1.18+ with Fluss Flink connector deployed for Fluss table access
  • ZooKeeper ensemble used by Fluss for cluster coordination (or embedded coordinator mode)
  • A free Vigilmon account

Why Monitoring Fluss Matters

Fluss is the real-time layer of your data stack — the component closest to your production event streams. Its failure modes are fast and consequential:

  • Coordinator crash — The Fluss Coordinator assigns tablets to TabletServers. If the Coordinator goes down, new tablet assignments stop and partition rebalancing halts. Existing writes continue until a TabletServer fails; then the cluster cannot recover without the Coordinator.
  • TabletServer loss — Fluss distributes data in shards called tablets across TabletServers. Losing a TabletServer causes any tablets it led to become under-replicated. If replication factor is 2 and two TabletServers host the same tablet leader and follower, a single server failure makes that tablet unavailable.
  • Replication lag — Follower TabletServers replicate write-ahead logs from leaders. If a follower falls behind (due to disk I/O contention, GC pause, or network congestion), any subsequent leader failure will result in data loss for unacknowledged writes.
  • Write latency P99 spikes — Fluss is designed for sub-50ms write latency. Spikes in P99 write latency indicate disk bottlenecks or JVM GC pressure on a TabletServer, and directly degrade the real-time quality of Flink streaming pipelines reading from Fluss.
  • Snapshot failures — Fluss creates periodic snapshots for crash recovery. If snapshot creation fails or the snapshot becomes stale, recovery after a cluster crash may require replaying from a very early offset, increasing recovery time significantly.

Step 1: Monitor the Fluss Coordinator Service

The Fluss Coordinator exposes an HTTP management endpoint. Add a Vigilmon HTTP monitor for it:

  1. In Vigilmon, click Add Monitor → HTTP / HTTPS.
  2. Enter the Coordinator REST URL: http://fluss-coordinator:9333/api/v1/status.
  3. Set Check interval to 1 minute.
  4. Set Expected HTTP status to 200.
  5. Set Response body contains to "state":"ACTIVE".
  6. Click Save.

If your Fluss cluster uses a high-availability Coordinator setup (active/standby), also add a TCP port monitor for the Coordinator leader port to catch leader elections:

  1. Click Add Monitor → TCP Port.
  2. Enter fluss-coordinator:9333.
  3. Set Check interval to 1 minute.
  4. Click Save.

A coordinator failure causes all tablet assignment operations to fail within seconds. Configure a PagerDuty or Slack alert on this monitor with zero consecutive failure grace — alert immediately.

Step 2: Monitor TabletServer Process Liveness

Fluss TabletServers are the data nodes. Create a heartbeat monitor for each TabletServer host:

  1. In Vigilmon, click Add Monitor → Cron Heartbeat.
  2. Name it: Fluss TabletServer - node1.
  3. Set the expected interval to 2 minutes.
  4. Copy the heartbeat URL.

On each TabletServer host, create a cron job that pings Vigilmon only when the TabletServer process is healthy:

#!/bin/bash
# /etc/cron.d/fluss-tabletserver-heartbeat
# Runs every 1 minute on each TabletServer node

*/1 * * * * fluss curl -sf "http://localhost:9334/api/v1/status" \
  | grep -q '"state":"RUNNING"' && \
  curl -fsS "https://vigilmon.online/heartbeat/YOUR_TS_NODE1_KEY" > /dev/null 2>&1

Repeat for each TabletServer node with a unique heartbeat URL. Set the Vigilmon heartbeat interval to 3 minutes — a TabletServer that has been down for 3 minutes without a heartbeat triggers the alert.

For clusters with many nodes, use a consolidated check script that counts healthy TabletServers and only pings the heartbeat if the count meets the minimum required:

#!/bin/bash
# /opt/monitoring/check-fluss-tabletservers.sh
MIN_HEALTHY=3
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_CLUSTER_TS_KEY"

COORDINATOR_URL="http://fluss-coordinator:9333"

HEALTHY=$(curl -sf "$COORDINATOR_URL/api/v1/tabletservers" | \
  python3 -c "
import sys, json
data = json.load(sys.stdin)
servers = data.get('tabletServers', [])
print(sum(1 for s in servers if s.get('state') == 'ALIVE'))
" 2>/dev/null)

if [ "${HEALTHY:-0}" -ge "$MIN_HEALTHY" ]; then
  curl -fsS "$HEARTBEAT_URL"
fi

Step 3: Monitor Write Latency and Throughput

Fluss is designed for low-latency streaming writes. A P99 write latency above 50ms indicates a TabletServer bottleneck. Build a health sidecar that queries Fluss metrics and exposes a health endpoint:

#!/usr/bin/env python3
# /opt/monitoring/fluss_health.py
# Lightweight Fluss health sidecar — exposes /health/write on port 9400

from http.server import HTTPServer, BaseHTTPRequestHandler
import json, urllib.request, os

FLUSS_METRICS_URL = os.getenv("FLUSS_METRICS_URL", "http://fluss-coordinator:9333/api/v1/metrics")
WRITE_LATENCY_P99_THRESHOLD_MS = int(os.getenv("WRITE_LATENCY_P99_THRESHOLD_MS", "50"))

class FlussHealthHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path != "/health/write":
            self.send_response(404)
            self.end_headers()
            return

        try:
            with urllib.request.urlopen(FLUSS_METRICS_URL, timeout=5) as r:
                metrics = json.loads(r.read())

            # Fluss exposes write latency P99 in milliseconds
            write_p99 = metrics.get("writeLatencyP99Ms", 0)
            write_throughput = metrics.get("writeRecordsPerSecond", 0)

            if write_p99 > WRITE_LATENCY_P99_THRESHOLD_MS:
                self.send_response(503)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(json.dumps({
                    "status": "degraded",
                    "reason": "write_latency_p99_exceeded",
                    "write_latency_p99_ms": write_p99,
                    "threshold_ms": WRITE_LATENCY_P99_THRESHOLD_MS,
                }).encode())
            else:
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(json.dumps({
                    "status": "ok",
                    "write_latency_p99_ms": write_p99,
                    "write_throughput_rps": write_throughput,
                }).encode())
        except Exception as e:
            self.send_response(503)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps({"status": "down", "error": str(e)}).encode())

    def log_message(self, format, *args):
        pass  # suppress default request logging

if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", 9400), FlussHealthHandler)
    server.serve_forever()

Run this sidecar as a systemd service on your Coordinator node. Then add a Vigilmon HTTP monitor:

  1. Click Add Monitor → HTTP / HTTPS.
  2. Enter: http://fluss-coordinator:9400/health/write.
  3. Set Check interval to 1 minute.
  4. Set Expected HTTP status to 200.
  5. Set Response body contains to "status":"ok".
  6. Click Save.

Step 4: Monitor Replication Health

Under-replicated tablets indicate that follower TabletServers are falling behind the leader — a precursor to data loss if the leader fails. Add a replication health heartbeat:

#!/bin/bash
# /opt/monitoring/check-fluss-replication.sh
MAX_UNDER_REPLICATED=0  # Zero tolerance for under-replicated tablets in production
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_REPLICATION_KEY"

COORDINATOR_URL="http://fluss-coordinator:9333"

UNDER_REPLICATED=$(curl -sf "$COORDINATOR_URL/api/v1/tablets/underreplicated" | \
  python3 -c "
import sys, json
data = json.load(sys.stdin)
print(data.get('count', 0))
" 2>/dev/null)

if [ "${UNDER_REPLICATED:-99}" -le "$MAX_UNDER_REPLICATED" ]; then
  curl -fsS "$HEARTBEAT_URL"
fi

Schedule every 2 minutes. Set the Vigilmon heartbeat interval to 5 minutes. Any under-replicated tablet will prevent the heartbeat from firing within 5 minutes of onset, triggering an alert.

For clusters where brief under-replication during leader elections is acceptable, set MAX_UNDER_REPLICATED=2 and alert only when more than 2 tablets are simultaneously under-replicated.

Step 5: Monitor Fluss Snapshot Health

Fluss creates periodic snapshots of tablet data for crash recovery. If snapshots fall behind schedule or fail to complete, recovery time after a cluster crash grows significantly.

#!/bin/bash
# /opt/monitoring/check-fluss-snapshots.sh
# Alert if the latest snapshot is older than 2x the configured snapshot interval
SNAPSHOT_INTERVAL_SECONDS=3600   # match your fluss.snapshot.interval configuration
MAX_SNAPSHOT_AGE_SECONDS=$((SNAPSHOT_INTERVAL_SECONDS * 2))
HEARTBEAT_URL="https://vigilmon.online/heartbeat/YOUR_SNAPSHOT_KEY"

COORDINATOR_URL="http://fluss-coordinator:9333"

LATEST_SNAPSHOT_TIME=$(curl -sf "$COORDINATOR_URL/api/v1/snapshots/latest" | \
  python3 -c "
import sys, json
data = json.load(sys.stdin)
print(data.get('createdAtMs', 0))
" 2>/dev/null)

NOW_MS=$(date +%s%3N)
AGE_SECONDS=$(( (NOW_MS - ${LATEST_SNAPSHOT_TIME:-0}) / 1000 ))

if [ "$AGE_SECONDS" -lt "$MAX_SNAPSHOT_AGE_SECONDS" ]; then
  curl -fsS "$HEARTBEAT_URL"
fi

Run every 10 minutes. Set the Vigilmon heartbeat interval to 30 minutes. A stale snapshot won't cause immediate data loss, but it silently increases the blast radius of a cluster crash.

Step 6: Monitor Flink Connector Health

Fluss integrates natively with Apache Flink — Fluss tables appear as Flink dynamic tables. Monitor the Flink jobs that read from Fluss to catch connector-level failures:

  1. In Vigilmon, click Add Monitor → HTTP / HTTPS.
  2. Enter the Flink JobManager API: http://flink-jobmanager:8081/overview.
  3. Set Check interval to 1 minute.
  4. Set Expected HTTP status to 200.
  5. Click Save.

For each Flink job that reads from Fluss, add a heartbeat monitor and wire it into your Flink streaming job:

// In your Flink streaming job that reads from Fluss tables
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.URI;

public class FlussReadHealthMapper extends RichMapFunction<Row, Row> {
    private transient HttpClient httpClient;
    private long lastHeartbeat = 0;
    private final long HEARTBEAT_INTERVAL_MS = 60_000; // ping every minute

    @Override
    public void open(Configuration parameters) {
        httpClient = HttpClient.newHttpClient();
    }

    @Override
    public Row map(Row row) throws Exception {
        long now = System.currentTimeMillis();
        if (now - lastHeartbeat > HEARTBEAT_INTERVAL_MS) {
            try {
                httpClient.send(
                    HttpRequest.newBuilder()
                        .uri(URI.create(System.getenv("VIGILMON_FLINK_HEARTBEAT_URL")))
                        .GET().build(),
                    java.net.http.HttpResponse.BodyHandlers.discarding()
                );
                lastHeartbeat = now;
            } catch (Exception ignored) {}
        }
        return row;
    }
}

Add this as a map operator after your Fluss source read. If the Flink job fails or the Fluss source connector stops delivering records, the heartbeat expires and Vigilmon alerts within your configured grace period.

Step 7: Configure Alert Routing

Create a Fluss-specific alert channel in Vigilmon and route by severity:

| Monitor | Alert Channel | Severity | Response | |---|---|---|---| | Coordinator HTTP | PagerDuty + Slack | Critical | Coordinator crash — tablet assignment halted | | TabletServer heartbeat (any node) | PagerDuty + Slack | Critical | Data node loss — tablets under-replicated | | Replication health heartbeat | Slack | High | Under-replicated tablets — leader failure risk | | Write latency P99 HTTP | Slack | High | Bottleneck degrading real-time pipeline quality | | Snapshot health heartbeat | Email | Medium | Stale snapshots — recovery time risk | | Flink JobManager HTTP | Slack | High | Flink-Fluss integration failure | | Flink connector heartbeat | Slack | High | Consumer pipeline stalled |

For the coordinator and TabletServer monitors, set Consecutive failures before alert to 1 — these are single-point-of-failure events that warrant immediate paging.

For write latency, set it to 3 to avoid alerting on transient GC pauses that self-resolve within 3 minutes.

Conclusion

Apache Fluss is the real-time foundation of the Streaming Lakehouse — and its failure modes cascade quickly. A TabletServer failure causes tablet under-replication; a coordinator failure halts cluster recovery; a write latency spike degrades every Flink streaming pipeline reading live data from Fluss. With Vigilmon monitoring the Coordinator, TabletServers, replication health, write latency, and snapshot freshness, you have the external visibility to catch these failures before they surface as stale dashboards or missed SLAs.

Start with the Coordinator HTTP monitor and TabletServer heartbeats — these cover the critical path for all Fluss operations. Then add replication and write latency monitoring as your cluster scales. Sign up for a free Vigilmon account to get started.

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