tutorial

How to Monitor Apache Flume Agent Health and Data Ingestion Pipelines with Vigilmon

Silent Flume agent crashes and channel overflow failures silently drop log events and break your data ingestion pipelines. Learn how to monitor Flume agent availability, channel fill rates, and ingestion heartbeats with Vigilmon.

Apache Flume is the distributed, reliable service for collecting, aggregating, and moving large amounts of log and event data from many sources to a centralized store like HDFS, HBase, or Kafka — but when a Flume agent crashes due to a JVM heap overflow, a channel fills to capacity and starts dropping events, or an HDFS sink loses its connection and buffers exhaust, your log ingestion pipeline silently loses data with no user-visible alert. Flume's JMX metrics and the optional monitoring HTTP server give you instrumentation, but they require active polling and don't alert you when agents stop collecting.

Vigilmon gives you external visibility into Flume agent health through the built-in monitoring HTTP server and a custom health sidecar, plus heartbeat monitors that confirm your ingestion pipelines are actively delivering data. This tutorial covers both.

Why Flume Needs External Monitoring

Flume's built-in tooling (the JMX metrics endpoint, the optional Ganglia/Graphite/HTTP monitoring plugins, and YARN application tracking for embedded agents) requires active watching. External monitoring with Vigilmon adds:

  • Proactive alerting when a Flume agent process crashes or becomes unresponsive
  • Channel fill rate detection — a nearly full channel predicts imminent event loss before it happens
  • Sink failure tracking — HDFS sink errors or Kafka producer failures that cause channel backup
  • Data freshness heartbeats confirming events are actively flowing end-to-end, not just that the agent process is alive
  • Multi-agent topology visibility across all your collector, aggregator, and consolidator agents

These layers matter because Flume failures are often gradual: an agent can be running (JVM alive, HTTP monitoring returns 200) while its HDFS sink is failing retries, causing the file channel to fill toward capacity. The agent self-reports as healthy right up until channel overflow causes event drops.

Step 1: Enable and Probe Flume's Built-in HTTP Monitoring

Flume ships with an HTTP monitoring server that exposes agent metrics as JSON. Enable it in your flume.conf:

# flume.conf — enable HTTP monitoring server
agent.sources  = src1
agent.channels = ch1
agent.sinks    = sink1

# HTTP monitoring server on port 41414 (default)
agent.sources.src1.type = ...
agent.channels.ch1.type = memory
agent.sinks.sink1.type  = hdfs

# Add to the JVM launch arguments:
# -Dflume.monitoring.type=http
# -Dflume.monitoring.port=34545

Start the Flume agent with monitoring enabled:

flume-ng agent \
  --conf /opt/flume/conf \
  --conf-file /opt/flume/conf/flume.conf \
  --name agent \
  -Dflume.monitoring.type=http \
  -Dflume.monitoring.port=34545

Probe the monitoring endpoint directly:

# Returns JSON metrics for all components
curl -s http://flume-agent.example.com:34545/metrics | python3 -m json.tool

# Example response includes channel fill rates, event counts, error rates

Point a Vigilmon monitor at http://flume-agent.example.com:34545/metrics for a fast reachability check.

Step 2: Build a Flume Health Sidecar

For deeper health analysis — channel capacity thresholds, sink error rate checks, and freshness validation — build a thin sidecar that interprets the Flume metrics JSON.

Python Health Sidecar

# flume_health.py
from flask import Flask, jsonify
import requests
import os

app = Flask(__name__)

FLUME_METRICS_URL   = os.environ.get('FLUME_METRICS_URL', 'http://localhost:34545/metrics')
CHANNEL_FILL_WARN   = float(os.environ.get('CHANNEL_FILL_WARN',   '0.80'))  # 80% full → warning
CHANNEL_FILL_CRIT   = float(os.environ.get('CHANNEL_FILL_CRIT',   '0.95'))  # 95% full → critical
SINK_ERROR_RATE_MAX = float(os.environ.get('SINK_ERROR_RATE_MAX', '0.05'))  # 5% error rate → critical

@app.route('/health/flume')
def flume_health():
    try:
        r = requests.get(FLUME_METRICS_URL, timeout=5)
        if r.status_code != 200:
            return jsonify({'status': 'down', 'error': f'metrics_http_{r.status_code}'}), 503
        metrics = r.json()
    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503

    warnings = []
    errors   = []

    for component, data in metrics.items():
        comp_type = data.get('Type', '')

        # Check channel fill rate
        if comp_type == 'CHANNEL':
            capacity   = int(data.get('ChannelCapacity',   0))
            fill_count = int(data.get('ChannelFillPercentage', 0))
            # ChannelFillPercentage is 0–100; some versions use ChannelSize + ChannelCapacity
            channel_size = int(data.get('ChannelSize', 0))
            fill_rate = (channel_size / capacity) if capacity > 0 else 0

            if fill_rate >= CHANNEL_FILL_CRIT:
                errors.append(f'{component}_fill_{round(fill_rate*100)}pct')
            elif fill_rate >= CHANNEL_FILL_WARN:
                warnings.append(f'{component}_fill_{round(fill_rate*100)}pct')

        # Check sink error rate
        if comp_type == 'SINK':
            event_drain_attempt = int(data.get('EventDrainAttemptCount', 1))
            event_drain_success = int(data.get('EventDrainSuccessCount', 0))
            if event_drain_attempt > 0:
                error_rate = 1 - (event_drain_success / event_drain_attempt)
                if error_rate > SINK_ERROR_RATE_MAX:
                    errors.append(f'{component}_sink_error_rate_{round(error_rate*100)}pct')

    if errors:
        return jsonify({'status': 'down', 'errors': errors, 'warnings': warnings}), 503
    if warnings:
        return jsonify({'status': 'degraded', 'warnings': warnings}), 200
    return jsonify({'status': 'ok'}), 200

@app.route('/health/flume/throughput')
def flume_throughput():
    """Check that events are actively being drained (non-zero throughput)."""
    try:
        r = requests.get(FLUME_METRICS_URL, timeout=5)
        r.raise_for_status()
        metrics = r.json()
    except Exception as e:
        return jsonify({'status': 'down', 'error': str(e)}), 503

    total_events = 0
    for component, data in metrics.items():
        if data.get('Type') == 'SINK':
            total_events += int(data.get('EventDrainSuccessCount', 0))

    if total_events == 0:
        return jsonify({'status': 'stalled', 'total_events_drained': 0}), 503
    return jsonify({'status': 'ok', 'total_events_drained': total_events}), 200

if __name__ == '__main__':
    app.run(port=8097)

Node.js Health Sidecar

// flume-health.js
const express = require('express');
const axios   = require('axios');

const app = express();

const FLUME_METRICS_URL   = process.env.FLUME_METRICS_URL   || 'http://localhost:34545/metrics';
const CHANNEL_FILL_CRIT   = parseFloat(process.env.CHANNEL_FILL_CRIT   || '0.95');
const SINK_ERROR_RATE_MAX = parseFloat(process.env.SINK_ERROR_RATE_MAX || '0.05');

app.get('/health/flume', async (req, res) => {
  let metrics;
  try {
    const r = await axios.get(FLUME_METRICS_URL, { timeout: 5000 });
    metrics = r.data;
  } catch (e) {
    return res.status(503).json({ status: 'down', error: e.message });
  }

  const errors = [];

  for (const [component, data] of Object.entries(metrics)) {
    if (data.Type === 'CHANNEL') {
      const capacity = parseInt(data.ChannelCapacity || 1);
      const size     = parseInt(data.ChannelSize     || 0);
      const fill     = size / capacity;
      if (fill >= CHANNEL_FILL_CRIT) errors.push(`${component}_fill_${Math.round(fill * 100)}pct`);
    }
    if (data.Type === 'SINK') {
      const attempts = parseInt(data.EventDrainAttemptCount || 1);
      const success  = parseInt(data.EventDrainSuccessCount || 0);
      const errRate  = attempts > 0 ? 1 - (success / attempts) : 0;
      if (errRate > SINK_ERROR_RATE_MAX) errors.push(`${component}_error_rate_${Math.round(errRate * 100)}pct`);
    }
  }

  if (errors.length) return res.status(503).json({ status: 'down', errors });
  return res.status(200).json({ status: 'ok' });
});

app.listen(3013, () => console.log('flume-health listening on 3013'));

Step 3: Configure Vigilmon HTTP Monitor for Flume

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your Flume health endpoint: https://your-app.example.com/health/flume
  4. Set the check interval to 1 minute (Flume channel overflows can happen quickly)
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 5000ms
  6. Under Alert channels, assign your Slack or PagerDuty channel
  7. Save the monitor

Add a second monitor for the raw Flume metrics endpoint:

  • URL: http://flume-agent.example.com:34545/metrics
  • Expected: 200
  • Interval: 1 minute
  • Alert channel: data-ingestion on-call channel

Add a throughput monitor to catch stalled agents:

  • URL: https://your-app.example.com/health/flume/throughput
  • Expected: 200, body contains "status":"ok"
  • Interval: 5 minutes
  • Alert channel: data platform team Slack channel

If you run multiple Flume agents (collector tier, aggregator tier, consolidator tier), create a separate monitor for each agent's health endpoint and label them clearly in Vigilmon.

Step 4: Heartbeat Monitoring for Flume Ingestion Pipelines

Agent availability and channel health tell you the Flume infrastructure is alive — but they don't confirm that data is actually flowing end-to-end from sources through sinks to its destination. An HDFS sink could be silently writing empty files, a Kafka channel could be re-enqueuing events that fail deserialization, or your log sources could have stopped sending events due to a log rotation misconfiguration.

Vigilmon heartbeat monitors detect these end-to-end failures: a downstream consumer (the process that reads from HDFS, HBase, or Kafka after Flume delivers data) pings Vigilmon after each successful processing cycle.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: flume-hdfs-ingestion
  3. Set the expected interval: 15 minutes (or your expected data arrival cadence)
  4. Set the grace period: 30 minutes
  5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire It Into Your Downstream Consumer

#!/bin/bash
# flume_consumer.sh — downstream HDFS consumer with Vigilmon heartbeat
# This runs after HDFS sink files are closed (e.g., every 10 minutes via Flume rollInterval)

set -e

HDFS_FLUME_DIR="/flume/events/$(date +%Y/%m/%d/%H)"
VIGILMON_HB="${VIGILMON_HEARTBEAT_URL}"
MIN_FILES=1  # Expect at least 1 file per run period

# Count files written by Flume in the current hour's directory
FILE_COUNT=$(hdfs dfs -count "$HDFS_FLUME_DIR" 2>/dev/null | awk '{print $2}' || echo "0")

if [ "$FILE_COUNT" -ge "$MIN_FILES" ]; then
  echo "[$(date)] $FILE_COUNT file(s) found in $HDFS_FLUME_DIR. Pinging Vigilmon heartbeat..."
  curl -s --max-time 10 "$VIGILMON_HB" > /dev/null
  echo "[$(date)] Heartbeat sent."
else
  echo "[$(date)] WARNING: Only $FILE_COUNT file(s) found in $HDFS_FLUME_DIR — expected $MIN_FILES+. Heartbeat NOT sent."
  exit 1
fi

Python Consumer with Heartbeat

# flume_hdfs_consumer.py — process Flume-delivered HDFS files with heartbeat
import subprocess
import requests
import os
from datetime import datetime

HDFS_FLUME_BASE = os.environ.get('HDFS_FLUME_BASE', '/flume/events')
VIGILMON_HB     = os.environ['VIGILMON_HEARTBEAT_URL']
MIN_EVENTS      = int(os.environ.get('MIN_EXPECTED_EVENTS', '100'))

def count_hdfs_events(hour_path):
    result = subprocess.run(
        ['hdfs', 'dfs', '-count', hour_path],
        capture_output=True, text=True, timeout=15
    )
    if result.returncode != 0:
        raise RuntimeError(f'HDFS count failed: {result.stderr}')
    # Output: <dir_count> <file_count> <content_size> <path>
    parts = result.stdout.strip().split()
    return int(parts[2])  # content_size as a proxy for event volume

now = datetime.utcnow()
hour_path = f"{HDFS_FLUME_BASE}/{now.strftime('%Y/%m/%d/%H')}"

try:
    size = count_hdfs_events(hour_path)
    if size < MIN_EVENTS:
        print(f"WARNING: Only {size} bytes in {hour_path}. Expected {MIN_EVENTS}+.")
        raise ValueError("Insufficient data volume")
    print(f"OK: {size} bytes in {hour_path}.")
    requests.get(VIGILMON_HB, timeout=10)
    print("Vigilmon heartbeat sent.")
except Exception as e:
    print(f"ERROR: {e}")
    raise

Kafka Consumer Heartbeat (Flume → Kafka sink)

# flume_kafka_consumer.py — check Flume-delivered Kafka topic has recent messages
from kafka import KafkaConsumer, TopicPartition
from kafka.structs import OffsetAndMetadata
import requests
import os
import time

KAFKA_BROKERS     = os.environ.get('KAFKA_BROKERS',     'kafka:9092').split(',')
FLUME_TOPIC       = os.environ.get('FLUME_TOPIC',       'flume-events')
VIGILMON_HB       = os.environ['VIGILMON_HEARTBEAT_URL']
MAX_LAG_THRESHOLD = int(os.environ.get('MAX_LAG_THRESHOLD', '10000'))

consumer = KafkaConsumer(bootstrap_servers=KAFKA_BROKERS)
partitions = consumer.partitions_for_topic(FLUME_TOPIC)

total_lag = 0
for p in (partitions or []):
    tp = TopicPartition(FLUME_TOPIC, p)
    end_offsets = consumer.end_offsets([tp])
    consumer.assign([tp])
    consumer.seek_to_committed(tp)
    committed = consumer.position(tp)
    total_lag += end_offsets[tp] - committed

consumer.close()

if total_lag > MAX_LAG_THRESHOLD:
    print(f"ERROR: Kafka lag {total_lag} exceeds threshold {MAX_LAG_THRESHOLD}. Heartbeat NOT sent.")
    raise SystemExit(1)

print(f"OK: Kafka lag {total_lag} within threshold.")
requests.get(VIGILMON_HB, timeout=10)
print("Vigilmon heartbeat sent.")

Step 5: Alert Routing for Flume Failures

Flume failures have distinct urgency levels: agent crashes stop data collection immediately, channel overflows cause event loss within minutes, and sink errors/stalls cause gradual data lag.

Configure alert routing in Vigilmon:

| Monitor | Alert Channel | Priority | |---|---|---| | Flume metrics endpoint (port 34545) | Slack + PagerDuty | P1 | | Flume channel/sink health /health/flume | Slack + PagerDuty | P1 | | Flume throughput check /health/flume/throughput | Slack | P2 | | Heartbeat: HDFS ingestion pipeline | Slack + email | P1 | | Heartbeat: Kafka delivery pipeline | Slack + PagerDuty | P1 | | Heartbeat: batch processing consumer | Email | P2 |

Set response time thresholds for early warning:

  • Alert at 3000ms for the Flume metrics endpoint (slow responses signal JVM GC stalls — often a precursor to OOM crashes)
  • Alert at 8000ms for the health sidecar (slow sidecar responses signal connectivity issues to Flume)

For compliance or audit log pipelines where event loss is unacceptable, configure a zero-tolerance alert policy: any Flume channel filling beyond 80% triggers an immediate P1 alert, not just when it reaches overflow. Early warning gives the on-call team time to either scale the sink backend (more HDFS connections, Kafka partitions) or throttle upstream sources before events are dropped.

Summary

Flume failures span agent availability (immediate), channel health (pre-failure warning), and end-to-end data flow (gradual degradation). External monitoring catches all three:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on Flume metrics endpoint | Agent process liveness, JMX reachability | | HTTP monitor on channel/sink health sidecar | Channel fill rate, sink error rates | | HTTP monitor on throughput endpoint | Active event drain confirmation | | Heartbeat monitor on downstream consumer | End-to-end data delivery, HDFS/Kafka freshness |

Get started free at vigilmon.online — your first Flume ingestion pipeline monitor is running in under two minutes.

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