ELFK 分布式日志收集系统

'# ELFK 分布式日志收集系统

一、背景与问题

在分布式系统中，日志收集一直是个棘手的难题。传统日志系统存在以下痛点：

1. 日志分散：每个服务节点独立存储日志，难以集中分析
2. 格式不统一：不同服务使用不同日志格式，难以统一处理
3. 实时性差：传统日志分析需要人工下载日志文件
4. 查询效率低：海量日志文件难以快速检索
5. 数据丢失风险：节点宕机导致日志丢失

ELFK（Elasticsearch + Logstash + Fluentd + Kibana）通过分布式架构和流式处理，解决了这些核心问题。其核心价值在于：

• 实时流式处理（Logstash）
• 弹性存储（Elasticsearch）
• 可视化分析（Kibana）
• 轻量级日志采集（Fluentd）

二、基本原理

ELFK的核心架构包含四个组件，形成完整的日志处理闭环：

1. Fluentd（日志采集）

• 负责收集各节点日志
• 支持多种日志源（文件、syslog、网络等）
• 使用插件化架构，可扩展性强

2. Logstash（日志处理）

• 作为数据管道，进行日志解析、过滤、转换
• 三阶段处理模型：Input → Filter → Output
• 支持正则表达式、Grok解析、字段转换等

3. Elasticsearch（日志存储）

• 基于倒排索引的搜索引擎
• 支持动态映射（自动字段类型识别）
• 分片机制实现水平扩展

4. Kibana（日志展示）

• 提供可视化界面
• 支持图表、仪表盘、日志搜索
• 与Elasticsearch深度集成

三、环境准备

在部署前需要准备以下环境：

# 安装依赖
sudo apt-get install -y openjdk-8-jdk
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.10.2-linux-x86_64.tar.gz
tar -xzf elasticsearch-7.10.2-linux-x86_64.tar.gz
sudo mv elasticsearch-7.10.2 /usr/local/elasticsearch

# 安装Logstash
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.10.2.tar.gz
tar -xzf logstash-7.10.2.tar.gz
sudo mv logstash-7.10.2 /usr/local/logstash

# 安装Fluentd
gem install fluentd

四、核心实现

1. Fluentd 日志采集配置

# /etc/fluentd/fluent.conf
<source>
  type tail
  path /var/log/*.log
  format json
  time_key log_time
  time_format %Y-%m-%d %H:%M:%S
</source>

<match **>
  @type elasticsearch
  host elasticsearch
  port 9200
  logstash_format true
</match>

关键点说明：

• 使用tail插件监控日志文件
• 设置log_time字段作为时间戳
• logstash_format启用Logstash兼容模式
• time_format指定时间格式

2. Logstash 日志处理配置

# /etc/logstash/conf.d/logstash.conf
input {
  beats {
    port => 5044
  }
}

filter {
  # 正则解析日志
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }

  # 字段转换
  mutate {
    convert => { "status" => "integer" }
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "%{+YYYY.MM.dd}"
  }
  stdout {
    codec => rubydebug
  }
}

关键点说明：

• 使用beats插件接收Fluentd发送的数据
• grok解析Apache日志格式
• convert将字段转换为整型
• index策略按日期分片

3. Elasticsearch 索引管理

# 创建索引模板
PUT _index_template/log_template
{
  "index_patterns": ["log-*"],
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1
  },
  "mappings": {
    "properties": {
      "timestamp": { "type": "date" },
      "level": { "type": "keyword" }
    }
  }
}

关键点说明：

• 使用模板管理索引生命周期
• 设置3个分片保证可用性
• 定义timestamp和level字段类型
• 通过log-*模式匹配所有日志索引

五、完整案例

案例：微服务日志收集系统

场景：一个包含3个微服务（auth、payment、order）的系统，需要集中收集日志并实时分析

架构图：

[微服务] -> [Fluentd] -> [Logstash] -> [Elasticsearch] -> [Kibana]

步骤：

1. 部署Fluentd：

   • 在每个服务节点部署Fluentd
   • 配置/etc/fluentd/fluent.conf收集日志

2. 配置Logstash：

   • 在中央节点部署Logstash
   • 配置logstash.conf处理日志
   • 添加filter处理异常日志

3. 部署Elasticsearch：

   • 配置elasticsearch.yml设置集群名称
   • 启动Elasticsearch服务

4. 部署Kibana：

   • 配置kibana.yml连接Elasticsearch
   • 创建日志仪表盘

完整配置示例：

# Logstash 额外配置
filter {
  if [level] == "ERROR" {
    mutate {
      add_field => { "severity" => "high" }
    }
  }
}

六、源码解析

1. Logstash 的 Grok 解析

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

• COMBINEDAPACHELOG 是预定义模式
• 匹配格式：[ip] [user] [date] [time] "[request]" [status] [size]

2. Elasticsearch 的分片策略

{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1
  }
}

• 分片数决定数据分布
• 副本数影响读写性能
• 最佳实践：分片数 = (节点数 × 分片数) / 副本数

3. Kibana 的可视化配置

{
  "title": "Error Log Analysis",
  "panels": [
    {
      "id": "1",
      "type": "bar",
      "gridPos": { "h": 10, "w": 12, "x": 0, "y": 0 },
      "targets": [{ "refId": "A", "table": "log-*" }],
      "series": [
        { "type": "count", "mode": "absolute", "field": "level" }
      ]
    }
  ]
}

七、进阶使用

1. 日志分级处理

filter {
  if [level] == "DEBUG" {
    mutate {
      remove_field => [ "level" ]
    }
  }
}

2. 实时告警

output {
  if [level] == "ERROR" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "alerts-%{+YYYY.MM.dd}"
    }
  }
}

3. 日志压缩策略

PUT _ilm/policy/log_policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "7d",
        "actions": {
          "rollover": {
            "max_size": "50gb"
          }
        }
      },
      "warm": {
        "min_age": "30d",
        "actions": {
          "tiered_storage": {
            "storage_type": "cold"
          }
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

八、性能与工程实践

1. 性能优化方案

2. 安全风险分析

• 数据泄露：未加密传输可能导致日志泄露
• 权限控制不足：未设置RBAC策略可能被非法访问
• SQL注入：未过滤输入可能导致Elasticsearch注入攻击

防护措施：

• 使用HTTPS加密传输
• 配置Elasticsearch安全模块（xpack.security）
• 对输入数据进行严格校验

3. 高可用方案

# Elasticsearch 集群配置
cluster.name: my-cluster
node.name: node-1
discovery.seed_hosts: ["node-2", "node-3"]
cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]

九、常见问题与踩坑

1. 日志丢失问题

现象：部分日志未出现在Elasticsearch中
原因：

• Logstash缓冲区满
• Elasticsearch写入失败
• Fluentd采集失败

解决办法：

# 增加Logstash缓冲
output {
  elasticsearch {
    buffer_type => "memory"
    buffer_size => 1000
  }
}

2. 查询性能差

现象：Elasticsearch查询响应时间过长
原因：

• 分片数设置不合理
• 缺乏合适的索引
• 查询语句不优化

解决办法：

# 创建索引时指定字段
PUT /logs
{
  "mappings": {
    "properties": {
      "timestamp": { "type": "date" }
    }
  }
}

3. 分片过多问题

现象：集群负载过高
原因：分片数设置过大的情况下，可能导致过多分片
解决办法：

# 调整分片数
PUT /log-2023.10.01/_settings
{
  "number_of_shards": 2
}

十、最佳实践

1. 标准化日志格式

• 使用JSON格式统一日志
• 包含标准字段（timestamp、level、message、logger）

2. 索引生命周期管理

• 设置合理的保留策略（7天热数据，30天温数据）
• 自动删除旧索引

3. 分布式监控

• 使用Prometheus监控ELFK集群
• 配置Alertmanager告警

4. 安全加固

• 开启Elasticsearch安全功能
• 使用RBAC策略控制访问
• 定期更新密码和证书

十一、总结

ELFK体系通过分布式架构和流式处理，解决了传统日志系统的关键痛点。其核心价值在于：

• 实时性：通过Logstash实现流式处理
• 可扩展性：Elasticsearch支持水平扩展
• 易用性：Kibana提供可视化界面
• 灵活性：Fluentd的插件化架构

实际使用中需要注意：

• 避免过度使用分片
• 合理配置缓冲机制
• 注重安全防护
• 定期维护索引

ELFK适用于需要实时日志分析、多源日志收集的场景，但在资源受限的边缘计算环境或日志量较小的场景中，可能需要选择轻量级方案。通过合理配置和优化，ELFK能够构建一个高效、可靠的日志管理系统。