Ubuntu Filebeat与Logstash如何协同工作

Ubuntu Filebeat与Logstash协同工作机制
Filebeat作为轻量级日志采集器，负责从Ubuntu服务器收集日志文件（如系统日志、应用日志），并通过Beats协议将数据发送至Logstash；Logstash作为数据处理管道，接收Filebeat传输的日志，进行过滤、解析、丰富等操作，最终将处理后的结构化数据发送至Elasticsearch等存储系统。两者协同实现了日志的高效采集、灵活处理与集中存储。

1. 安装Filebeat与Logstash

在Ubuntu系统上，通过APT包管理器安装两者（确保网络可访问Elastic官方仓库）：

# 更新软件包列表
sudo apt-get update

# 安装Filebeat
sudo apt-get install filebeat -y

# 安装Logstash
sudo apt-get install logstash -y

安装完成后，Filebeat默认配置文件位于/etc/filebeat/filebeat.yml，Logstash配置文件位于/etc/logstash/conf.d/（建议新建.conf文件，如filebeat.conf）。

2. 配置Filebeat采集日志并发送至Logstash

2.1 启用Filebeat模块（可选但推荐）

Filebeat内置模块（如system、nginx、mysql）可简化常见日志格式的收集与解析。以系统日志为例，启用模块并查看配置：

# 启用system模块（收集/var/log/syslog、/var/log/auth.log等）
sudo filebeat modules enable system

# 查看已启用模块
sudo filebeat modules list

模块启用后，Filebeat会自动生成对应输入配置（如监控/var/log/syslog）。

2.2 自定义日志输入（非模块场景）

若需监控自定义日志文件（如/var/log/myapp/*.log），手动编辑/etc/filebeat/filebeat.yml，添加log类型输入：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/myapp/*.log  # 监控所有.log文件

2.3 配置Logstash输出

修改Filebeat配置，将日志发送至Logstash（默认监听5044端口）：

output.logstash:
  hosts: ["localhost:5044"]  # 若Logstash在远程服务器，替换为IP地址

3. 配置Logstash接收与处理日志

创建Logstash配置文件（如/etc/logstash/conf.d/filebeat.conf），定义输入、过滤、输出三部分：

input {

  beats {
    
    port =>
     5044  # 监听Filebeat发送的端口
    host =>
 "0.0.0.0"  # 允许所有IP连接（生产环境建议限制为服务器IP）
  }

}


filter {

  # 示例1：解析系统日志（需配合system模块）
  if [fileset][module] == "system" {

    if [fileset][name] == "auth" {
  # 处理auth日志（如ssh登录）
      grok {
    
        match =>
 {
     "message" =>
 "%{
SYSLOGTIMESTAMP:timestamp}
 %{
SYSLOGHOST:hostname}
 sshd(?:\[%{
POSINT:pid}
\])?: %{
DATA:event}
 %{
DATA:method}
 for (invalid user )?%{
DATA:user}
 from %{
IPORHOST:ip}
 port %{
NUMBER:port}
 ssh2" }

      }

      date {
    
        match =>
 ["timestamp", "ISO8601"]  # 将日志时间转换为@timestamp字段
      }

      mutate {
    
        remove_field =>
 ["message", "timestamp"]  # 删除原始message和临时字段
      }

    }

  }


  # 示例2：解析Nginx访问日志（需自定义路径）
  # if [fileset][name] == "access" {

  #   grok {
    
  #     match =>
 {
     "message" =>
 "%{
COMBINEDAPACHELOG}
" }

  #   }

  #   date {
    
  #     match =>
 ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
  #   }

  # }

}


output {

  elasticsearch {
    
    hosts =>
     ["localhost:9200"]  # 发送至Elasticsearch
    index =>
 "filebeat-%{
+YYYY.MM.dd}
"  # 按日期生成索引
  }

  stdout {
     codec =>
 rubydebug }
  # 测试时输出到终端（可选）
}
    

说明：

• input：配置Beats输入插件，监听5044端口；
• filter：使用grok插件解析日志（如系统日志、Nginx日志），date插件转换时间格式，mutate插件清理无用字段；
• output：将处理后的数据发送至Elasticsearch（可同时输出至stdout用于调试）。

4. 启动与验证服务

4.1 启动服务

# 启动Filebeat
sudo systemctl start filebeat
sudo systemctl enable filebeat  # 设置开机自启

# 启动Logstash
sudo systemctl start logstash
sudo systemctl enable logstash  # 设置开机自启

4.2 验证数据传输

• 查看Filebeat状态：

  sudo systemctl status filebeat
  

  若状态为active (running)，则表示Filebeat正常运行。

• 查看Logstash日志：

  sudo journalctl -u logstash -f  # 实时查看Logstash日志
  

  若看到Successfully published events to Elasticsearch，则表示数据已成功传输。

• 检查Elasticsearch索引：
  通过Kibana或Elasticsearch API查看索引是否存在（如filebeat-2025.11.03）：

  curl -X GET "localhost:9200/_cat/indices?v"
  

5. 关键注意事项

• 端口开放：确保Ubuntu防火墙允许5044端口（Filebeat→Logstash）和9200端口（Logstash→Elasticsearch）通信：

  sudo ufw allow 5044/tcp
  sudo ufw allow 9200/tcp
  

• 性能优化：若日志量较大，可在Logstash的output部分添加bulk批量发送配置，或调整Filebeat的queue大小（如queue.mem.events: 4096）。
• 安全配置：生产环境建议启用Logstash的SSL/TLS加密（ssl => true）和Elasticsearch的身份认证（user/password）。

声明：本文内容由网友自发贡献，本站不承担相应法律责任。对本内容有异议或投诉，请联系2913721942#qq.com核实处理，我们将尽快回复您，谢谢合作！