4、数据采集层搭建：日志采集原理、使用Filebeat/Logstash采集日志、自定义Agent开发入门

数据采集层，说白了就是合规监控系统的「耳朵」和「眼睛」。

系统能不能发现问题，全靠这一层把数据喂进来。我见过不少团队，一上来就搞规则引擎、搞告警，结果数据采不全，最后全白搭。嗯，咱们先把地基打牢。

4.1 日志采集的核心原理

日志采集听起来简单，不就是读文件嘛？但实际落地时，坑不少。

核心要解决三个问题：

• 怎么读？—— 文件在持续写，不能漏，也不能重复读。
• 怎么传？—— 采集到的日志要可靠地送到下游，网络断了怎么办？
• 怎么解析？—— 原始日志是纯文本，得转成结构化数据，才能做后续分析。

我个人习惯把日志采集分成两个阶段：采集端和传输端。

你想想看，如果采集端挂了，日志就丢了。所以生产环境一定要有「断点续传」机制。我曾经踩过这个坑——Filebeat 没配好，重启后从头读了一遍，结果 Kafka 里全是重复数据，下游的告警直接炸了。

4.2 使用 Filebeat 采集日志

Filebeat 是 Elastic 家的轻量级采集器。我为什么推荐它？因为它够轻，资源占用极低，适合部署在业务服务器上。

Filebeat 的工作流程：

1. 监听日志文件的变化（通过 harvester 实现）
2. 读取新增的行内容
3. 发送到配置的输出端（Elasticsearch、Logstash、Kafka 等）

下面是一个典型的 Filebeat 配置，我直接贴出来，你照着改就行：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  fields:
    app_name: my-app
    env: production

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "app-logs"
  partition.round_robin:
    reachable_only: true

# 关键配置：断点续传
filebeat.registry.path: /var/lib/filebeat/registry

我在项目中遇到过一个问题：业务日志量特别大，每秒几千行。Filebeat 默认的 harvester 数量不够，导致日志积压。后来我把 filebeat.harvester_limit 调大了，才解决。

4.3 使用 Logstash 做日志解析

Logstash 是 Elastic 栈里的「管道工」。它负责把原始日志解析成结构化数据。

为什么需要 Logstash？因为原始日志长这样：

2025-01-15 10:23:45 ERROR [UserService] 用户登录失败: 账号不存在, userId=12345

我们需要把它解析成 JSON：

{
  "timestamp": "2025-01-15T10:23:45",
  "level": "ERROR",
  "module": "UserService",
  "message": "用户登录失败: 账号不存在",
  "userId": 12345
}

Logstash 用 filter 插件来做这件事。最常用的是 grok 插件，它用正则表达式来匹配日志格式。

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \[%{DATA:module}\] %{GREEDYDATA:message}" }
  }
  mutate {
    add_field => { "app_name" => "my-app" }
  }
}

Logstash 的 pipeline 配置也很关键。我习惯把输入、过滤、输出分开写，方便维护：

input {
  beats {
    port => 5044
  }
}

filter {
  # 这里放解析逻辑
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"
  }
}

4.4 自定义 Agent 开发入门

有时候，现成的工具满足不了需求。比如要采集数据库的 binlog、要采集 API 的调用数据，或者要采集一些非标准格式的日志。这时候就得自己写 Agent。

自定义 Agent 的核心设计，我总结为三个模块：

模块	职责	技术选型建议
采集模块	从数据源读取原始数据	文件监听用 inotify，数据库用 JDBC，API 用 HTTP 轮询
处理模块	数据清洗、格式化、过滤	用管道模式，每个处理步骤是一个插件
发送模块	将数据发送到下游	支持 Kafka、HTTP、TCP 等多种输出

下面是一个极简的 Agent 示例，用 Python 写的，核心逻辑就几十行：

import time
import json
from kafka import KafkaProducer

class SimpleAgent:
    def __init__(self, log_path, kafka_topic):
        self.log_path = log_path
        self.producer = KafkaProducer(bootstrap_servers='localhost:9092',
                                      value_serializer=lambda v: json.dumps(v).encode())
        self.topic = kafka_topic
        self.offset = self._load_offset()

    def _load_offset(self):
        # 从本地文件加载上次读取的位置
        try:
            with open('offset.txt', 'r') as f:
                return int(f.read())
        except:
            return 0

    def _save_offset(self, offset):
        with open('offset.txt', 'w') as f:
            f.write(str(offset))

    def collect(self):
        with open(self.log_path, 'r') as f:
            f.seek(self.offset)
            for line in f:
                # 这里可以做简单的解析
                data = {
                    'raw': line.strip(),
                    'timestamp': time.time()
                }
                self.producer.send(self.topic, data)
            self._save_offset(f.tell())

    def run(self):
        while True:
            self.collect()
            time.sleep(1)

if __name__ == '__main__':
    agent = SimpleAgent('/var/log/app.log', 'app-logs')
    agent.run()

另外，Agent 的部署和管理也是个问题。我建议用 Supervisor 或 systemd 来管理 Agent 进程，确保它挂了能自动重启。如果 Agent 数量多，可以考虑用 Consul 或 Etcd 做服务发现和配置下发。

4.5 本章知识体系

下面这张图，是我自己梳理的数据采集层核心逻辑。你照着这个结构去搭建，基本不会跑偏。

数据采集层是整个合规监控系统的起点。Filebeat 适合做轻量采集，Logstash 擅长做解析，自定义 Agent 则用来处理特殊场景。我个人建议：能用现成工具就别自己造轮子，但一定要理解底层原理，这样出了问题才能快速定位。

好了，数据采集层就聊到这里。下一层是数据存储层，咱们到时候再细说。

---

公众号:蓝海资料掘金营，微信deep3321