未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

Flume 监控与运维

没有监控的 Flume 就像在黑夜里开车。本文系统介绍 Flume 的监控体系搭建，帮助你在问题发生的第一时间感知并定位。

---

两种监控端点

Flume 内置两种监控方式，二选一启用：

方式	启动参数	访问地址	适用场景
HTTP 监控	-Dflume.monitoring.type=http -Dflume.monitoring.port=34545	http://host:34545/metrics	快速调试、Prometheus 集成
Ganglia 监控	-Dflume.monitoring.type=ganglia -Dflume.monitoring.hostname=ganglia-host -Dflume.monitoring.port=8649	Ganglia Web	已有 Ganglia 环境时使用

推荐方式：HTTP 监控（配合 Prometheus）

---

JMX 监控配置

开启 JMX

在 conf/flume-env.sh 中添加 JMX 参数，同时开启 HTTP 监控。下例关闭了 JMX 认证与 SSL，仅限隔离测试网络；生产环境须配置认证、加密及网络访问控制，并替换示例主机地址：

# conf/flume-env.sh

export JAVA_OPTS="
  -Dcom.sun.management.jmxremote=true
  -Dcom.sun.management.jmxremote.port=54321
  -Dcom.sun.management.jmxremote.rmi.port=54321
  -Dcom.sun.management.jmxremote.ssl=false
  -Dcom.sun.management.jmxremote.authenticate=false
  -Djava.rmi.server.hostname=192.168.1.100
"

# HTTP 监控端点（端口 34545）
export JAVA_OPTS="$JAVA_OPTS
  -Dflume.monitoring.type=http
  -Dflume.monitoring.port=34545
"

启动命令

flume-ng agent \
  --conf conf \
  --conf-file conf/agent.conf \
  --name agent \
  -Dflume.root.logger=INFO,LOGFILE \
  -Dflume.monitoring.type=http \
  -Dflume.monitoring.port=34545

---

HTTP 监控端点：指标解读

启动后，访问监控端点获取所有组件指标：

curl http://localhost:34545/metrics | python3 -m json.tool

典型输出结构：

{
  "SOURCE.tailsource": {
    "EventReceivedCount": 1500000,
    "EventAcceptedCount": 1499800,
    "AppendBatchReceivedCount": 1500,
    "AppendBatchAcceptedCount": 1499,
    "Type": "SOURCE",
    "StartTime": 1711008000000,
    "StopTime": 0
  },
  "CHANNEL.memchannel": {
    "EventPutSuccessCount": 1499800,
    "EventPutAttemptCount": 1499800,
    "EventTakeSuccessCount": 1498900,
    "EventTakeAttemptCount": 1498900,
    "ChannelSize": 900,
    "ChannelCapacity": 300000,
    "ChannelFillPercentage": 0.3,
    "Type": "CHANNEL",
    "StartTime": 1711008000000,
    "StopTime": 0
  },
  "SINK.hdfssink": {
    "EventDrainSuccessCount": 1498900,
    "EventDrainAttemptCount": 1498900,
    "BatchCompleteCount": 749,
    "BatchEmptyCount": 0,
    "BatchUnderflowCount": 0,
    "Type": "SINK",
    "StartTime": 1711008000000,
    "StopTime": 0
  }
}

关键指标含义

指标名	所在组件	含义	健康状态判断
EventPutSuccessCount	CHANNEL	Source 成功放入 Channel 的事件总数	持续增长 = Source 正常工作
EventTakeSuccessCount	CHANNEL	Sink 从 Channel 取出的事件总数	应接近 EventPutSuccessCount
ChannelSize	CHANNEL	当前 Channel 中堆积的事件数	不超过 ChannelCapacity 的 80%
ChannelFillPercentage	CHANNEL	Channel 使用率（%）	> 90% 需要告警
ChannelCapacity	CHANNEL	Channel 最大容量	配置值，用于对比 ChannelSize
EventDrainSuccessCount	SINK	Sink 成功写出的事件总数	接近 EventTakeSuccessCount
EventReceivedCount	SOURCE	Source 接收到的原始事件数	持续增长 = 数据源有数据
EventAcceptedCount	SOURCE	Source 成功接受（放入 Channel）的事件数	等于 EventReceivedCount = Source 无丢弃

TPS 计算（采样间隔法）：

# 采样两次，计算差值即得到 TPS
t1=$(curl -s http://localhost:34545/metrics | python3 -c \
  "import sys,json; d=json.load(sys.stdin); \
   print(d['CHANNEL.memchannel']['EventPutSuccessCount'])")
sleep 60
t2=$(curl -s http://localhost:34545/metrics | python3 -c \
  "import sys,json; d=json.load(sys.stdin); \
   print(d['CHANNEL.memchannel']['EventPutSuccessCount'])")
echo "TPS = $(( (t2 - t1) / 60 )) events/sec"

---

Prometheus 集成

使用 flume-ng-prometheus-reporter 插件

Step 1：下载插件

# 下载 Prometheus 报告插件（社区维护）
wget https://github.com/jrkinley/flume-ng-prometheus/releases/download/v0.1/\
flume-ng-prometheus-reporter-0.1-SNAPSHOT.jar

# 放置到 Flume plugins 目录
mkdir -p /opt/flume/plugins.d/prometheus-reporter/lib
cp flume-ng-prometheus-reporter-*.jar \
   /opt/flume/plugins.d/prometheus-reporter/lib/

Step 2：启动时指定 Prometheus 监控类型

flume-ng agent \
  --conf conf \
  --conf-file conf/agent.conf \
  --name agent \
  -Dflume.monitoring.type=com.github.jrkinley.flume.PrometheusSink \
  -Dflume.monitoring.port=9091 \
  -Dflume.monitoring.namespace=flume

Step 3：也可以用 Prometheus JMX Exporter（更通用，与前述插件方案择一使用；两者不能同时占用 9091 端口）

# 下载 JMX Exporter Agent
wget https://repo1.maven.org/maven2/io/prometheus/jmx/\
jmx_prometheus_javaagent/0.20.0/jmx_prometheus_javaagent-0.20.0.jar

# 将下载的 JAR 放到后续 -javaagent 指定的位置（需具备目录写权限）
install -m 0644 jmx_prometheus_javaagent-0.20.0.jar /opt/flume/lib/jmx_prometheus_javaagent-0.20.0.jar

# 准备 JMX Exporter 配置文件
cat > /opt/flume/conf/flume-jmx-exporter.yml << 'EOF'
lowercaseOutputName: true
lowercaseOutputLabelNames: true
rules:
  - pattern: "org.apache.flume.source<type=(.+)><>(.+)"
    name: flume_source_$2
    labels:
      source: "$1"
  - pattern: "org.apache.flume.channel<type=(.+)><>(.+)"
    name: flume_channel_$2
    labels:
      channel: "$1"
  - pattern: "org.apache.flume.sink<type=(.+)><>(.+)"
    name: flume_sink_$2
    labels:
      sink: "$1"
EOF

# 在 JAVA_OPTS 中加载 JMX Exporter Agent（监听 9091 端口）
export JAVA_OPTS="$JAVA_OPTS \
  -javaagent:/opt/flume/lib/jmx_prometheus_javaagent-0.20.0.jar=\
9091:/opt/flume/conf/flume-jmx-exporter.yml"

Step 4：Prometheus 抓取配置

# prometheus.yml
scrape_configs:
  - job_name: 'flume'
    static_configs:
      - targets: ['flume-host-01:9091', 'flume-host-02:9091']
    scrape_interval: 30s
    metrics_path: /metrics

Grafana 监控面板核心指标

导入 Grafana 后建议监控以下 Panel：

Panel 名称	指标	告警阈值
Channel 使用率	flume_channel_channelfillpercentage	> 85% 告警
事件吞吐量（TPS）	rate(flume_channel_eventputsuccesscount[1m])	突降 50% 告警
Source 丢弃率	(EventReceivedCount - EventAcceptedCount) / EventReceivedCount	> 1% 告警
Sink 失败率	(EventDrainAttemptCount - EventDrainSuccessCount) / EventDrainAttemptCount	> 0.1% 告警
GC 停顿时间	JVM GC pause 指标	> 500ms 告警

---

日志管理

log4j.properties 配置

# conf/log4j.properties

# 根日志级别：INFO 适合生产，DEBUG 仅调试时使用
flume.root.logger=INFO,LOGFILE
log4j.rootLogger=${flume.root.logger}

# 控制台输出（生产建议关闭，避免性能损耗）
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{dd MMM yyyy HH:mm:ss,SSS} %-5p [%t] (%C.%M:%L) %x - %m%n

# 滚动文件输出（生产推荐）
log4j.appender.LOGFILE=org.apache.log4j.RollingFileAppender
log4j.appender.LOGFILE.File=/var/log/flume/flume.log
log4j.appender.LOGFILE.MaxFileSize=100MB
log4j.appender.LOGFILE.MaxBackupIndex=10
log4j.appender.LOGFILE.layout=org.apache.log4j.PatternLayout
log4j.appender.LOGFILE.layout.ConversionPattern=%d{dd MMM yyyy HH:mm:ss,SSS} %-5p [%t] (%C.%M:%L) %x - %m%n

# 按日期滚动（与 RollingFileAppender 二选一）
# log4j.appender.LOGFILE=org.apache.log4j.DailyRollingFileAppender
# log4j.appender.LOGFILE.DatePattern='.'yyyy-MM-dd

# 特定组件 DEBUG（排查某个 Source/Channel/Sink 时）
# log4j.logger.org.apache.flume.sink.hdfs=DEBUG
# log4j.logger.org.apache.flume.source.taildir=DEBUG

常用日志关键词过滤：

# 查看 Channel Full 错误（Source 被阻塞的标志）
grep "ChannelException" /var/log/flume/flume.log | tail -20

# 查看 Sink 连接失败
grep "Failed to" /var/log/flume/flume.log | tail -20

# 查看 GC 情况（需要开启 GC 日志）
grep "GC pause" /var/log/flume/gc.log | tail -20

# 实时跟踪 ERROR 级别日志
tail -f /var/log/flume/flume.log | grep -E "ERROR|WARN"

---

进程守护

Systemd 配置（推荐）

# /etc/systemd/system/flume-agent.service

[Unit]
Description=Apache Flume Data Collection Agent
After=network.target

[Service]
Type=simple
User=flume
Group=flume

# flume-env.sh 是 shell 脚本，不应作为 systemd EnvironmentFile 读取
# 由下方 flume-ng 按 --conf /opt/flume/conf 加载

# 启动命令
ExecStart=/opt/flume/bin/flume-ng agent \
  --conf /opt/flume/conf \
  --conf-file /opt/flume/conf/agent.conf \
  --name agent \
  -Dflume.root.logger=INFO,LOGFILE \
  -Dflume.monitoring.type=http \
  -Dflume.monitoring.port=34545

# 异常退出时自动重启（5 秒延迟）
Restart=on-failure
RestartSec=5s

# 进程停止超时（优雅关闭时间）
TimeoutStopSec=60

# 工作目录
WorkingDirectory=/opt/flume

# 标准输出重定向
StandardOutput=append:/var/log/flume/flume-stdout.log
StandardError=append:/var/log/flume/flume-stderr.log

[Install]
WantedBy=multi-user.target

# 启用并启动服务
systemctl daemon-reload
systemctl enable flume-agent
systemctl start flume-agent

# 查看服务状态
systemctl status flume-agent

# 查看实时日志
journalctl -u flume-agent -f

Supervisor 配置（备选）

# /etc/supervisor/conf.d/flume-agent.conf

[program:flume-agent]
command=/opt/flume/bin/flume-ng agent
  --conf /opt/flume/conf
  --conf-file /opt/flume/conf/agent.conf
  --name agent
  -Dflume.root.logger=INFO,LOGFILE
  -Dflume.monitoring.type=http
  -Dflume.monitoring.port=34545
directory=/opt/flume
user=flume
autostart=true
autorestart=true
startretries=3
startsecs=10
stopwaitsecs=60
stdout_logfile=/var/log/flume/supervisor-stdout.log
stderr_logfile=/var/log/flume/supervisor-stderr.log

---

优雅升级与重启

有条件的优雅停止步骤：

SIGTERM 不保证 Channel 自动排空。停止前应通过上游缓冲、暂停入口或切换采集流量控制新增事件，再核对积压与下游投递；MemoryChannel 中未处理的事件可能在停机时丢失。FileChannel 的持久化也不等于端到端不丢失、不重复，须验证恢复与业务数据一致性。以下进程操作要求已确认只匹配一个目标 Agent。

# Step 1：查找 Flume 进程 ID
FLUME_PID=$(pgrep -f "org.apache.flume.node.Application")
echo "Flume PID: $FLUME_PID"

# Step 2：发送 SIGTERM 信号（请求优雅关闭，不保证 Channel 排空）
kill -TERM $FLUME_PID

# Step 3：等待进程退出（最多等 60 秒）
if timeout 60 bash -c "while kill -0 $FLUME_PID 2>/dev/null; do sleep 1; done"; then
  echo "Flume Agent 进程已退出；仍需核对关闭日志与数据状态"
else
  echo "等待退出超时或检查失败，请排查；不要直接强制终止" >&2
  exit 1
fi

# Step 4（仅 FileChannel 场景）：核对关闭日志及 checkpoint/data 目录状态
# 不以进程退出作为最终 Checkpoint 写入完成或数据安全的证明；重启后验证恢复和下游数据

# 错误做法：kill -9 强制杀死
# kill -9 $FLUME_PID  # 会导致 MemoryChannel 数据丢失，FileChannel 需要恢复

滚动重启（多 Agent 场景）：

# 逐一重启 Agent，不同时停止所有节点
for host in flume-01 flume-02 flume-03; do
  echo "重启 $host 上的 Flume Agent..."
  ssh $host "systemctl restart flume-agent"

  # 检查指标端点是否响应（30 秒检查一次，最多 3 次）
  # 非空 JSON 不代表业务健康；生产还须验证吞吐、积压、错误和下游投递
  for i in {1..3}; do
    sleep 30
    status=$(curl -s "http://$host:34545/metrics" 2>/dev/null | python3 -c \
      "import sys,json; d=json.load(sys.stdin); \
       print('OK' if d else 'FAIL')" 2>/dev/null)
    if [ "$status" = "OK" ]; then
      echo "$host 指标端点已响应；继续前须另行确认业务健康"
      break
    fi
    echo "等待 $host 恢复... ($i/3)"
  done
  if [ "$status" != "OK" ]; then
    echo "$host 指标端点检查失败，停止滚动重启" >&2
    exit 1
  fi
done

---

常见故障排查表

故障现象	可能原因	排查步骤	解决方案
ChannelFillPercentage > 90%	Sink 写入速度慢	检查 Sink 目标端（HDFS/Kafka）是否正常	提升 Sink batchSize，增加 threadsPoolSize
ChannelException: Space for event exhausted	Channel 容量不足	确认 capacity 配置，计算实际 TPS	增大 capacity 或换 SpillableMemoryChannel
Flume 进程 OOM 崩溃	堆内存不足	查看 GC 日志，检查 MemoryChannel capacity	增大 -Xmx，减小 capacity，开启 G1GC
Too many open files	文件描述符耗尽	lsof -p $FLUME_PID | wc -l	ulimit -n 65536，修改 /etc/security/limits.conf
Source 不采集新数据	position 文件异常或权限问题	检查 positionFile 内容和权限	先备份并排查 positionFile；确需删除时，应核对 Taildir 的 skipToEnd 等起读配置，不能保证从头读取，并须验证重复采集与漏采风险
HDFS Sink 无法写入	Kerberos Ticket 过期	检查 klist	配置 kinit 定时刷新或使用 Keytab
消费延迟突然升高	上游日志量激增	对比 Source EventReceived 与平时基线	临时增加 Flume Agent 节点，水平扩展
Agent 频繁重启	配置错误或依赖服务不可达	查看 supervisor/systemd 日志	检查 agent.conf 语法，确认目标端连接

---

掌握检查

• [ ] 能在 flume-env.sh 中同时开启 JMX（端口 54321）和 HTTP 监控（端口 34545）
• [ ] 能解读 HTTP 监控端点返回的 JSON，说出 ChannelFillPercentage 的告警意义
• [ ] 理解 EventPutSuccessCount 和 EventTakeSuccessCount 的差值代表什么
• [ ] 能配置 Prometheus JMX Exporter 采集 Flume 指标
• [ ] 能在 log4j.properties 中配置滚动日志文件，避免磁盘被打满
• [ ] 能用 systemd 配置 Flume 服务自动重启，并实现优雅停止

练习题

1. 你负责一台 Flume Agent，发现 ChannelFillPercentage 从 30% 在 10 分钟内上升到 92%，但 Source 的 EventAcceptedCount 增长正常。请完整描述排查思路，并给出 3 种可能的根因和对应解决方案。
2. 公司要求 Flume 的所有关键指标接入 Prometheus，并在 Grafana 创建看板。请写出完整的配置步骤，包括 Flume 端、Prometheus 端的关键配置片段。
3. 生产环境需要对 Flume Agent 进行版本升级，但要求数据不能丢失，服务中断不超过 30 秒。请给出完整的升级操作流程，并列明上游缓冲、Channel 持久化或排空、冗余切换、回滚与数据核对的成立条件；若现有架构无法满足目标，请说明限制并提出改造方案，不得仅凭 SIGTERM 或滚动重启宣称达标。

学习连接

前置知识: 性能调优实战（原引用“10-性能调优实战”未能确定有效目标，暂不可跳转） - JVM 与组件调优 下一步: Flume 企业级实战（原引用“12-Flume企业级实战”未能确定有效目标，暂不可跳转） - 生产案例 相关概念: Prometheus 监控体系（原引用“Prometheus监控”未能确定有效目标，暂不可跳转）、Grafana 面板设计（原引用“Grafana可视化”未能确定有效目标，暂不可跳转）

---