How to Monitor Fluentd Health and Buffer Queue Depth with Vigilmon

Fluentd is the CNCF-graduated log aggregator that unifies your logging layer — collecting from hundreds of sources, transforming, and routing to dozens of outputs. When Fluentd falls behind, its behavior is insidious: buffers fill silently, output retries accumulate without visible errors, and eventually chunks start being dropped. By the time your dashboard shows missing logs, you've already lost data.

Vigilmon adds external health monitoring for Fluentd through its built-in monitoring agent HTTP endpoint and optional health sidecars. This tutorial covers input plugin health, buffer queue depth, output retry counts, worker thread status, memory buffer usage, and Vigilmon integration.

Why Fluentd Needs External Monitoring

Fluentd's internal monitoring_agent exposes a JSON API, but most teams never configure alerting on it. Vigilmon external monitoring adds:

• Process liveness checks that fire immediately if fluentd crashes or hangs
• Buffer depth monitoring to catch overflow before data loss occurs
• Output retry alerting that identifies which output plugin is failing and how severely
• Worker thread health for multi-worker configurations
• Memory usage tracking to detect buffer memory leaks before OOM kills the process

These complement — but don't replace — your log pipeline dashboards.

Step 1: Enable Fluentd's Monitoring Agent

Fluentd ships a built-in HTTP monitoring agent. Add it to your fluent.conf:

# fluent.conf

# Enable the monitoring agent HTTP API
<source>
  @type monitor_agent
  bind 0.0.0.0
  port 24220
</source>

Restart Fluentd and verify the monitoring endpoint:

# Get all plugin status
curl -s http://localhost:24220/api/plugins.json | python3 -m json.tool

# Get config dump
curl -s http://localhost:24220/api/config.json | python3 -m json.tool

The plugins.json response includes per-plugin state including buffer queue and retry counts for each output plugin.

Step 2: Build a Fluentd Health Aggregation Endpoint

Vigilmon probes HTTP endpoints and evaluates status codes and body content. Build a health sidecar that interprets the Fluentd monitoring agent API and returns a clear pass/fail:

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

app = Flask(__name__)

FLUENTD_MONITOR_URL = os.environ.get('FLUENTD_MONITOR_URL', 'http://localhost:24220')
BUFFER_QUEUE_WARN = int(os.environ.get('BUFFER_QUEUE_WARN', '50'))    # warn at 50 chunks queued
BUFFER_QUEUE_CRIT = int(os.environ.get('BUFFER_QUEUE_CRIT', '200'))   # critical at 200 chunks
RETRY_WARN = int(os.environ.get('RETRY_WARN', '5'))                    # warn at 5 retries
RETRY_CRIT = int(os.environ.get('RETRY_CRIT', '20'))                   # critical at 20 retries

@app.route('/health/fluentd')
def fluentd_health():
    try:
        resp = requests.get(
            f'{FLUENTD_MONITOR_URL}/api/plugins.json',
            timeout=5
        )
        resp.raise_for_status()
        data = resp.json()
    except Exception as e:
        return jsonify(status='down', error=str(e)), 503

    plugins = data.get('plugins', [])
    max_buffer_queue = 0
    max_retry_count = 0
    degraded_plugins = []

    for plugin in plugins:
        plugin_id = plugin.get('plugin_id', 'unknown')
        plugin_type = plugin.get('type', 'unknown')

        # Buffer queue depth (chunks waiting to be flushed)
        buffer_queue = plugin.get('buffer_queue_length', 0)
        if buffer_queue > max_buffer_queue:
            max_buffer_queue = buffer_queue

        # Output retry count
        retry_count = plugin.get('retry_count', 0)
        if retry_count > max_retry_count:
            max_retry_count = retry_count

        # Flag severely degraded plugins
        if buffer_queue >= BUFFER_QUEUE_CRIT or retry_count >= RETRY_CRIT:
            degraded_plugins.append({
                'id': plugin_id,
                'type': plugin_type,
                'buffer_queue': buffer_queue,
                'retry_count': retry_count,
            })

    if degraded_plugins:
        return jsonify(
            status='degraded',
            reason='plugins_in_critical_state',
            plugins=degraded_plugins,
            max_buffer_queue=max_buffer_queue,
            max_retry_count=max_retry_count,
        ), 503

    if max_buffer_queue >= BUFFER_QUEUE_WARN or max_retry_count >= RETRY_WARN:
        return jsonify(
            status='degraded',
            reason='plugins_buffer_or_retry_elevated',
            max_buffer_queue=max_buffer_queue,
            max_retry_count=max_retry_count,
        ), 503

    return jsonify(
        status='ok',
        plugin_count=len(plugins),
        max_buffer_queue=max_buffer_queue,
        max_retry_count=max_retry_count,
    ), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', '5005')))

Install dependencies and run:

pip install flask requests
python fluentd_health.py &

Step 3: Add Memory Usage Monitoring

High buffer memory usage is an early warning sign before Fluentd gets OOM-killed. Add a separate memory health endpoint:

import psutil
import subprocess

@app.route('/health/fluentd/memory')
def fluentd_memory():
    try:
        # Find Fluentd PIDs
        result = subprocess.run(
            ['pgrep', '-f', 'fluentd'],
            capture_output=True, text=True
        )
        pids = [int(p) for p in result.stdout.strip().split('\n') if p]

        total_rss_mb = 0
        for pid in pids:
            try:
                proc = psutil.Process(pid)
                total_rss_mb += proc.memory_info().rss / (1024 * 1024)
            except psutil.NoSuchProcess:
                pass

        memory_limit_mb = int(os.environ.get('MEMORY_LIMIT_MB', '512'))
        fill_ratio = total_rss_mb / memory_limit_mb

        if fill_ratio >= 0.90:
            return jsonify(
                status='degraded',
                reason='memory_near_limit',
                rss_mb=round(total_rss_mb, 1),
                limit_mb=memory_limit_mb,
                fill_ratio=round(fill_ratio, 3),
            ), 503

        return jsonify(
            status='ok',
            rss_mb=round(total_rss_mb, 1),
            limit_mb=memory_limit_mb,
            fill_ratio=round(fill_ratio, 3),
        ), 200
    except Exception as e:
        return jsonify(status='unknown', error=str(e)), 200  # non-fatal

Step 4: Configure Vigilmon HTTP Monitors for Fluentd

Monitor 1: Fluentd Plugin Health

1. Log in to vigilmon.online and go to Monitors → New Monitor
2. Choose HTTP / HTTPS
3. Set the URL to https://your-host.example.com/health/fluentd
4. Set the check interval to 1 minute
5. Under Expected response: Status code 200, body contains "status":"ok"
6. Response time threshold: 5000ms
7. Assign your on-call alert channel (Slack + PagerDuty for production)
8. Save

Monitor 2: Fluentd Direct Monitoring Agent (Defense in Depth)

Add a direct probe on the Fluentd monitoring API as a fallback:

• URL: http://your-host:24220/api/plugins.json
• Expected: 200 (presence of response confirms process is alive)
• Interval: 2 minutes
• Alert: Slack only (secondary signal)

This monitor fires even if your Python health sidecar crashes.

Monitor 3: Memory Usage

• URL: https://your-host.example.com/health/fluentd/memory
• Expected: 200, body contains "status":"ok"
• Interval: 5 minutes
• Alert: Slack (typically P2 — early warning, not immediate page)

Step 5: Heartbeat Monitoring for Fluentd Input Throughput

Buffer and retry health confirm Fluentd is processing — but they don't confirm your log sources are actually delivering events. Use Fluentd's exec or http output plugin to ping Vigilmon periodically when log events are flowing:

# In fluent.conf — add a throughput heartbeat to your pipeline

<match **>
  @type copy

  # Your existing output (e.g., Elasticsearch)
  <store>
    @type elasticsearch
    host elasticsearch
    port 9200
    # ... rest of your ES config
  </store>

  # Vigilmon heartbeat — fires when any log event is matched
  <store>
    @type exec_filter
    command /bin/sh -c 'curl -sf "${VIGILMON_HEARTBEAT_URL}" > /dev/null; cat'
    time_key time
    tag_key tag
    # Rate-limit to once per minute
    flush_interval 60s
    buffer_type memory
    buffer_queue_limit 1
  </store>
</match>

In Vigilmon, create a Heartbeat monitor:

1. Go to Monitors → New Monitor → Heartbeat
2. Name: fluentd-pipeline-throughput
3. Expected interval: 5 minutes
4. Grace period: 10 minutes
5. Save and copy the heartbeat URL

Set VIGILMON_HEARTBEAT_URL as an environment variable in your Fluentd process environment.

Step 6: Alert Routing for Fluentd

| Monitor | Alert Channel | Priority | |---|---|---| | Plugin health /health/fluentd | Slack + PagerDuty | P1 | | Fluentd monitoring agent (direct) | Slack | P1 | | Memory usage /health/fluentd/memory | Slack | P2 | | Pipeline throughput heartbeat | Slack + email | P2 |

Response time tuning:

• Alert at 3000ms for plugin health (slow response signals Fluentd thread pressure)
• Alert at 1000ms for direct monitoring agent (should be near-instant JSON)

For multi-worker Fluentd deployments (workers N in config), scale your health monitoring to cover all worker processes. Set BUFFER_QUEUE_WARN and BUFFER_QUEUE_CRIT per your typical log volume — a batch output writing every 5 minutes will naturally show higher queue depths than a streaming output.

Summary

Fluentd failures are silent by design: events buffer, retries accumulate, and data loss happens gradually before any crash or error log appears. External monitoring with Vigilmon covers the gaps:

| Monitor Type | What It Covers | |---|---| | HTTP on /health/fluentd | Plugin state, buffer queue depth, retry counts | | HTTP on monitoring agent (direct) | Process liveness, defense in depth | | HTTP on /health/fluentd/memory | Buffer memory before OOM kill | | Heartbeat | Active log event flow through the pipeline |

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