4. 日志采集与集中管理：ELK/EFK架构搭建，Rsyslog与Fluentd配置实战，日志源接入

日志这东西，平时没人搭理它，一出事它就是唯一的救命稻草。我干了这么多年运维，见过太多因为日志没管好，排查故障时两眼一抹黑的场景。说白了，日志集中管理就是给整个网络装一个黑匣子，关键时刻能回放事故现场。

4.1 为什么必须做日志集中管理？

你想想看，一台交换机每天产生几千条日志，防火墙更多。如果每台设备都各自为政，出问题了你得一台一台登录去看，效率极低。更可怕的是——攻击者往往先删日志再跑路。

核心价值三点：

  • 统一检索：所有设备日志在一个地方查，不用到处翻
  • 实时告警：发现异常行为立刻通知，比如防火墙拒绝连接突增
  • 合规审计：等保2.0、ISO27001都要求日志留存6个月以上

我记得有一次帮客户做安全审计，发现他们的核心交换机日志只保留7天。等保检查时直接被扣分，后来连夜搭了ELK才补上。嗯，这种事真不能拖。

4.2 两大主流架构：ELK vs EFK

目前业界最成熟的就是ELK和EFK两套方案。区别其实就一个组件——日志采集器。

组件 ELK EFK
采集器 Logstash Fluentd
特点 功能强大，插件丰富，但吃内存 轻量级，资源占用低，适合容器环境
适用场景 传统IDC、日志量大的环境 Kubernetes、边缘节点、资源受限场景

我个人习惯：如果服务器资源充裕，首选ELK，Logstash的过滤能力确实强。如果跑在容器里或者机器配置不高，用Fluentd更稳。

4.3 ELK架构搭建实战

下面我直接给出一套能用的搭建步骤。注意，这是生产级别的精简版，去掉了花里胡哨的东西。

4.3.1 环境准备

三台服务器，最低配置2核4G：

  • Node1：Elasticsearch（数据节点）
  • Node2：Logstash + Kibana
  • Node3：Elasticsearch（备用节点，可选）

版本选择很重要。我踩过坑——ELK各组件版本必须严格一致，否则各种兼容性问题。建议用7.17.x，稳定且功能齐全。

4.3.2 安装Elasticsearch

# 导入GPG密钥
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -

# 添加仓库
echo "deb https://artifacts.elastic.co/packages/7.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-7.x.list

# 安装
sudo apt update && sudo apt install elasticsearch

# 修改配置 /etc/elasticsearch/elasticsearch.yml
# 关键参数：
cluster.name: my-elk-cluster
node.name: node-1
path.data: /var/lib/elasticsearch
path.logs: /var/log/elasticsearch
network.host: 0.0.0.0
http.port: 9200
discovery.seed_hosts: ["127.0.0.1"]
cluster.initial_master_nodes: ["node-1"]

# 启动
sudo systemctl enable elasticsearch
sudo systemctl start elasticsearch

注意：生产环境千万别把network.host设成0.0.0.0暴露公网。我曾经见过有人这么干，结果被挖矿程序扫到，ES直接被拿来当肉鸡。一定要加防火墙或绑定内网IP。

4.3.3 安装Logstash

sudo apt install logstash

# 创建配置文件 /etc/logstash/conf.d/syslog.conf
input {
  syslog {
    port => 514
    type => "syslog"
  }
}

filter {
  if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:timestamp} %{SYSLOGHOST:hostname} %{DATA:program}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:message}" }
    }
    date {
      match => [ "timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
  }
}

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

# 启动
sudo systemctl enable logstash
sudo systemctl start logstash

这里grok正则我用了好几年，基本覆盖了主流设备的syslog格式。如果你接的设备日志格式比较怪，可以用Grok Debugger在线调试，省得一遍遍重启Logstash。

4.3.4 安装Kibana

sudo apt install kibana

# 修改配置 /etc/kibana/kibana.yml
server.port: 5601
server.host: "0.0.0.0"
elasticsearch.hosts: ["http://localhost:9200"]

# 启动
sudo systemctl enable kibana
sudo systemctl start kibana

访问 http://你的IP:5601，看到Kibana界面就算成了。第一次进去需要创建索引模式，选syslog-*就行。

4.4 EFK架构搭建（轻量版）

如果你觉得Logstash太重，试试Fluentd。我最近在容器环境里全换成了EFK，内存占用直接降了60%。

# 安装Fluentd（使用td-agent）
curl -L https://toolbelt.treasuredata.com/sh/install-ubuntu-focal-td-agent4.sh | sh

# 配置 /etc/td-agent/td-agent.conf
<source>
  @type syslog
  port 514
  bind 0.0.0.0
  tag syslog
</source>

<filter syslog.*>
  @type parser
  key_name message
  <parse>
    @type syslog
  </parse>
</filter>

<match syslog.*>
  @type elasticsearch
  host localhost
  port 9200
  logstash_format true
  logstash_prefix fluentd
  flush_interval 5s
</match>

Fluentd的配置比Logstash简洁很多，而且它自带内存缓冲，网络断了不会丢数据。这一点我在生产环境验证过——ES重启期间，Fluentd缓存了大概20万条日志，恢复后全部写入，一条没丢。

4.5 日志源接入实战

架构搭好了，接下来就是接设备。这是最繁琐的一步，因为不同厂家的日志格式千奇百怪。

4.5.1 交换机日志接入（华为/Cisco）

以华为交换机为例：

# 华为交换机配置
info-center loghost 192.168.1.100  # 指向Logstash/Fluentd服务器
info-center source default channel loghost log level informational
info-center timestamp log

# Cisco交换机配置
logging host 192.168.1.100
logging trap informational
logging source-interface Vlan1
logging on

这里有个坑：华为默认日志时间戳是相对时间，必须加上info-center timestamp log才能显示绝对时间。我刚开始没加，查日志时发现时间全是错的，排查了半天才找到原因。

4.5.2 防火墙日志接入（深信服/Palo Alto）

防火墙日志通常包含安全事件，是审计的重点：

# 深信服防火墙
syslog-server 192.168.1.100 514
syslog-level notice

# Palo Alto防火墙
set shared log-settings syslog server ELK transport UDP
set shared log-settings syslog server ELK port 514
set shared log-settings syslog server ELK format BSD
set shared log-settings syslog server ELK facility LOG_LOCAL0

经验之谈：防火墙日志量巨大，建议只发送notice级别以上的日志。如果全量发送，一天可能几个GB，ES索引压力很大。我一般会做两级过滤——设备端只发重要日志，Logstash端再做一次去重和聚合。

4.5.3 服务器日志接入（Linux/Windows）

Linux服务器用Rsyslog最方便：

# 修改 /etc/rsyslog.conf
# 启用UDP接收
module(load="imudp")
input(type="imudp" port="514")

# 转发到Logstash
*.* @192.168.1.100:514

# 重启
systemctl restart rsyslog

Windows服务器需要安装Winlogbeat或NXLog。我个人推荐Winlogbeat，和ELK生态无缝集成：

# winlogbeat.yml 关键配置
winlogbeat.event_logs:
  - name: Application
  - name: Security
  - name: System

output.elasticsearch:
  hosts: ["192.168.1.100:9200"]
  index: "winlogbeat-%{+yyyy.MM.dd}"

4.6 架构总览图

下面这张图是我自己画的，把整个日志采集链路串起来了：

日志采集与集中管理架构图 日志源 交换机 (华为/Cisco) 防火墙 (深信服/Palo Alto) 服务器 (Linux/Windows) 日志采集层 Rsyslog (Linux服务器) Winlogbeat (Windows) Syslog (网络设备) 日志处理层 Logstash / Fluentd 解析、过滤、格式化 Grok正则匹配 存储层 Elasticsearch 索引、分片、副本 展示层 Kibana 仪表盘、告警、搜索 安全审计与合规 日志留存6个月、异常告警

4.7 避坑指南与最佳实践

最后分享几个我这些年总结的经验，都是真金白银换来的：

坑1：日志时间戳混乱

不同设备时区不同，有的用UTC，有的用CST。我建议所有设备统一使用UTC，在Kibana展示时再转换时区。否则日志排序全是乱的，根本没法查。

坑2：磁盘空间爆炸

日志增长速度快得吓人。我见过一个客户，防火墙日志一天产生50GB，三天就把磁盘撑爆了。解决方案：ES配置ILM（索引生命周期管理），设置30天后自动删除或归档到冷存储。

坑3：日志格式不兼容

华为和Cisco的syslog格式差异很大。华为是<14>2024-01-01T12:00:00，Cisco是<14>Jan 1 12:00:00。我建议在Logstash里写两套grok规则，用if语句区分设备类型。

嗯，日志集中管理这件事，说白了就是「平时养兵千日，用兵一时」。架构搭好了可能几个月都用不上，但一旦出安全事件，它就是你的救命稻草。别等到被攻击了才想起来搞日志，那时候已经晚了。

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

← 上一章 📖 返回目录 下一章 →