ELK日志分析系统-EW帮帮网

Elasticsearch 是一个开源分布式搜索引擎，具有分布式、零配置、自动发现、索引自动分片、索引副本机制、restful 风格接口、多数据源、自动搜索负载等特点。
Logstash 是一个完全开源的工具，可以对日志进行收集、过滤，并将其存储供以后使用（如搜索）。
Kibana 也是一个开源和免费的工具，它可以为 Logstash 和 ElasticSearch 提供的日志分析提供友好的 Web 界面，帮助汇总、分析和搜索重要数据日志。

进行日志处理分析，一般需要经过以下几步：

将日志进行集中化管理。
将日志格式化（Logstash）并输出到 Elasticsearch。
对格式化后的数据进行索引和存储（Elasticsearch）。
前端数据的展示（Kibana）。

1.2 Elasticsearch

1.2.1 ElasticSearch 概述

Elasticsearch 是一个基于 Lucene 的搜索服务器，提供了分布式多用户能力的全文搜索引擎，基于 RESTful web 接口。它用 Java 开发，并作为 Apache 许可条款下的开放源码发布，是当前流行的企业级搜索引擎。其设计用于云计算中，能够达到实时搜索，具有稳定、可靠、快速、安装使用方便等特点。

1.2.2 Elasticsearch 核心概念

接近实时（NRT）：Elasticsearch 是一个接近实时的搜索平台，这意味着从索引一个文档到这个文档能够被搜索到有一个轻微的延迟（通常是 1 秒）。
集群（cluster）：一个集群由一个或多个节点组织在一起，共同持有整个数据，并一起提供索引和搜索功能。集群由一个唯一的名字标识，默认名字是 "elasticsearch"。节点只能通过指定集群名字来加入该集群。
节点（node）：节点是集群中的一个服务器，作为集群的一部分，存储数据并参与集群的索引和搜索功能。节点也由名字标识，默认是一个随机的漫威漫画角色名字。节点可通过配置集群名称加入指定集群，默认加入 "elasticsearch" 集群。
索引（index）：索引是具有相似特征的文档集合，由名字标识（必须全为小写字母），对索引中文档进行索引、搜索、更新和删除时需使用该名字，集群中可定义任意多索引。
类型（type）：在索引中可定义一种或多种类型，是索引的逻辑分类 / 分区，语义由用户决定，通常为具有共同字段的文档定义类型。
文档（document）：文档是可被索引的基础信息单元，以 JSON 格式表示，在 index/type 中可存储任意多文档，且文档必须被索引 / 赋予索引的 type。
分片和复制（shards&replicas）：索引可存储超出单个节点硬件限制的数据，通过将索引划分成多份（分片）来解决。分片允许水平分割 / 扩展内容容量，以及在分片上进行分布式、并行操作以提高性能 / 吞吐量。为应对分片 / 节点失败，Elasticsearch 允许创建分片的拷贝（复制分片），提供高可用性和扩展搜索量 / 吞吐量。每个索引可分成多个分片，可复制 0 次或多次，索引创建时指定分片和复制数量，创建后可动态改变复制数量但不能改变分片数量。默认情况下，Elasticsearch 7.0 以前的版本中每个索引被分片 5 个主分片和 1 个复制，7.0 以后的版本中默认索引的主分片数量为 1，副本分片数量为 1。

1.3 Logstash

1.3.1 Logstash 介绍

Logstash 由 JRuby 语言编写，运行在 Java 虚拟机（JVM）上，是一款强大的数据处理工具，可实现数据传输、格式处理、格式化输出，具有强大的插件功能，常用于日志处理。其设计理念是只做三件事：数据输入、数据加工、数据输出。

1.3.2 Logstash 工作的三个阶段

input 数据输入端：可以接收来自任何地方的源数据，包括 file（从文件中读取）、syslog（监听在 514 端口的系统日志信息，并解析成 RFC3164 格式）、redis（从 redis-server list 中获取）、beat（接收来自 Filebeat 的事件）。
Filter 数据中转层：主要进行格式处理、数据类型转换、数据过滤、字段添加修改等，常用过滤器有 grok（通过正则解析和结构化任何文本）、mutate（在事件字段执行一般转换，可重命名、删除、替换和修改事件字段）、drop（完全丢弃事件，如 debug 事件）、clone（复制事件，可能添加或者删除字段）、geoip（添加有关 IP 地址地理位置信息）。
output：是 logstash 工作的最后一个阶段，负责将数据输出到指定位置，兼容大多数应用，常用的有 elasticsearch（发送事件数据到 Elasticsearch，便于查询、分析、绘图）、file（将事件数据写入到磁盘文件上）、mongodb（将事件数据发送至高性能 NoSQL mongodb，便于永久存储、查询、分析，大数据分片）、redis（将数据发送至 redis-server，常用于中间层暂时缓存）、graphite（发送事件数据到 graphite）、statsd（发送事件数据到 statsd）。

1.4 Kibana

1.4.1 Kibana 介绍

Kibana 是一个设计用于和 Elasticsearch 配置工作的开源分析和可视化平台，可以用它进行搜索、查看、集成 Elasticsearch 中的数据索引，利用各种图表、报表、地图组件轻松对数据进行可视化分析。

1.4.2 Kibana 主要功能

包括与 Elasticsearch 无缝集成、整合数据、复杂数据分析、让更多团队成员收益、接口灵活、配置简单、可视化多数据源、简单数据导出等。

二、环境设置

2.1 实验拓扑图

ES/Kibana 服务器：192.168.10.103，主机名：elk1
生产服务器：192.168.10.101，主机名：web01
其他服务器：192.168.10.104

说明：新版的 logstash 自带缓存功能，不需要再使用 zookeeper+kafka 做消息中间件；elasticsearch 服务器的数量需要奇数个，因为有脑裂的问题。

2.2 主机清单

主机名	IP 地址	操作系统	软件包
elk1	192.168.10.101	Openeuler 24	Elasticsearch、logstash、kbana
elk2	192.168.10.102	Openeuler 24	Elasticsearch、filebeat、httpd
elk3	192.168.10.103	Openeuler 24	Elasticsearch

三、部署 ES 群集

3.1 基本配置（以 elk1 为例，其它两台操作类似）

3.1.1 设置主机名

按照主机清单为每一台主机设置主机名（elk1、elk2、elk3），命令如下：

hostnamectl set-hostname elk1  # 在elk1上执行
hostnamectl set-hostname elk2  # 在elk2上执行
hostnamectl set-hostname elk3  # 在elk3上执行

3.1.2 在两台 ES 主机上设置 hosts 文件

cat <<EOF >>/etc/hosts
192.168.10.101 elk1
192.168.10.102 elk2
192.168.10.103 elk3
EOF

3.1.3 关闭所有节点的防火墙

systemctl stop firewalld
systemctl disable firewalld
setenforce 0
sed -i "s/SELINUX=enforcing/SELINUX=disabled/" /etc/selinux/config

3.1.4 创建 es 运行用户

useradd es

3.1.5 安装 java 环境

dnf -y install java-11
java -version

3.1.6 为用户设置资源访问限制

vim /etc/security/limits.conf

添加以下内容：

es soft nofile 65535
es soft memlock unlimited
es soft nproc 65535
es hard nofile 65535
es hard nproc 65535
es hard memlock unlimited

说明：

es soft nofile 65535：一个进程最多能打开的文件数。
es soft nproc 65535：一个用户最多能创建的进程数。
es soft memlock unlimited：最大锁定内存地址空间（unlimited 不限制）。

vim /etc/sysctl.conf

添加以下配置：

vm.max_map_count=655360

sysctl -p

说明：vm.max_map_count的默认值是 65536，该参数限制一个进程可以拥有的 VMA（虚拟内存区域）的数量，会影响中间件可以开启的线程数量，值过小可能导致中间件无法开启足够线程而报错。

3.2 安装 Elasticsearch

tar zxvf elasticsearch-7.10.0-linux-x86_64.tar.gz
mv elasticsearch-7.10.0 /usr/local/elasticsearch

3.2.1 更改 Elasticsearch 主配置文件

vim /usr/local/elasticsearch/config/elasticsearch.yml

修改内容如下：

node.name: elk1  # 本节点主机名，以elk1节点为例，其它节点请改成相对应的节点名称
cluster.name: kgc-elk-cluster  # 群集名称
path.data: /elk/data  # 数据文件路径
path.logs: /elk/logs  # 日志文件路径
bootstrap.memory_lock: false  # 锁定物理内存
network.host: 0.0.0.0  # 监听地址
http.port: 9200  # 监听端口
discovery.seed_hosts: ["elk1", "elk2", "elk3"]  # 群集中的主机列表
cluster.initial_master_nodes: ["elk1"]  # master主机名称，群集初始化会将此节点选举为master

3.2.2 创建数据存放路径并授权

mkdir -p /elk/data
mkdir -p /elk/logs
chown -R es:es /elk/
chown -R es:es /usr/local/elasticsearch/

3.2.3 启动 es

su - es  # es限制使用root启动，需要切换至es用户
nohup /usr/local/elasticsearch/bin/elasticsearch &
sudo netstat anpt | grep 9200

3.2.4 查看节点信息

curl http://192.168.10.103:9200/_cat/nodes

输出结果中带有星号的是 master 节点。

四、安装 Logstash

4.1 在 logstash 服务器上安装 logstash

systemctl stop firewalld
setenforce 0
tar zxvf logstash-7.10.0-linux-x86_64.tar.gz
yum -y install java-11
mv logstash-7.10.0 /usr/local/logstash
chmod -R 777 /usr/local/logstash/data/

4.2 测试安装结果

4.2.1 直接将消息输出到屏幕

/usr/local/logstash/bin/logstash -e 'input { stdin {} } output { stdout { codec => rubydebug } }'

执行该命令后稍等，出现消息格式，可输入字符串作为消息内容，退出按 Ctrl+C。若命令失败，删除 logstash 目录的 data 子目录。

4.2.2 将消息输出到 es（如果没有安装 head 组件，此步骤省去）

/usr/local/logstash/bin/logstash -e 'input { stdin {} } output { stdout { codec => rubydebug } output { elasticsearch { hosts => ["192.168.10.103:9200"] } }'

输入字符串作为消息，使用以下方法查看 es 中索引信息：

查看索引列表：curl -XGET "http://localhost:9200/_cat/indices?v"
索引分片信息：curl -XGET "http://localhost:9200/_cat/shards?v"
查询索引前 10 条数据：curl -XGET "http://localhost:9200/索引名/_search?pretty" -H 'Content-Type: application/json' -d '{"query":{"match_all":{}},"size":10}'

4.3 Logstash 配置文件

Logstash 配置文件基本由 input、output 以及 filter（根据需要）三部分组成。标准配置文件格式如下：

input {...}
filter {...}
output {...}

4.3.1 收集系统日志并输出到 elasticsearch

chmod o+r /var/log/messages  # 让Logstash可以读取日志
touch /usr/local/logstash/system.conf
vim /usr/local/logstash/system.conf

输入以下内容：

input {
  file {
    path => "/var/log/messages"
    type => "system"
    start_position => "beginning"
  }
}
output {
  elasticsearch {
    hosts => ["192.168.10.101:9200"]
    index => "system-%{+YYYY.M.dd}"
  }
}

运行 logstash：

/usr/local/logstash/bin/logstash -f /usr/local/logstash/system.conf

查看索引列表，会发现多出一个索引。

五、Filebeat

5.1 在产生日志的客户端服务器上安装 filebeat（以 elk2 主机为例）

yum -y install httpd
systemctl start httpd
echo "www.kec.com" > /var/www/html/index.html
curl 192.168.10.102 www.kgc.com
cat /var/log/httpd/access_log
cat /var/log/httpd/error_log

5.2 安装 filebeat

tar zxvf filebeat-7.10.0-linux-x86_64.tar.gz
mv filebeat-7.10.0-linux-x86_64 /usr/local/filebeat

5.3 配置 web01 服务器 filebeat 的输出

5.3.1 修改 filebeat 文件，将 filebeat 输出的日志输入到远程的 logstash

mv filebeat.yml filebeat.yml.bak
cd /usr/local/filebeat/
vim filebeat.yml

修改内容如下：

filebeat.inputs:
- type: log
  paths:
    - /var/log/httpd/access_log
output.logstash:
  hosts: ["192.168.10.102:5044"]

5.3.2 启动 filebeat 服务

/usr/local/filebeat/filebeat -c /usr/local/filebeat/filebeat.yml &

5.4 修改 logstash 的配置文件，使日志输出到 elasticsearch

如果 logstash 处于开启状态，先关闭：

netstat -anpt | grep 5044  # 查看logstash进程
kill -9 进程号  # 杀死进程

修改配置文件：

vim /usr/local/logstash/config/beats.conf

输入以下内容：

input {
  beats {
    port => "5044"
    codec => "json"
  }
}
output {
  elasticsearch {
    hosts => ["192.168.10.103:9200"]
    index => "weblog-beat-%{+YY.M.dd}"
  }
}

说明：Codec 是 logstash 从 1.3.0 版开始新引入的概念，用于 decode、encode 事件，使得 logstash 可以更好地与其他有自定义数据格式的运维产品共存。

5.5 运行 logstash 并验证

/usr/local/logstash/bin/logstash -f /usr/local/logstash/config/beats.conf --path.data=/usr/local/logstash/config.d/web01 &

查看索引，会发现新增了一个 weblog-beat-2025.04.28 的索引。

六、安装 Kibana

6.1 在 elk1 上安装 Kibana

tar zxvf kibana-7.10.0-linux-x86_64.tar.gz
mv kibana-7.10.0-linux-x86_64 /usr/local/kibana

6.2 修改 Kibana 主配置文件

vim /usr/local/kibana/config/kibana.yml

修改内容如下：

server.port: 5601  # 监听端口
server.host: "0.0.0.0"  # 监听地址
elasticsearch.hosts: "http://192.168.10.103:9200"  # ES主机的IP地址
kibana.index: ".kibana"

chown -R es:es /usr/local/kibana/

6.3 启动 Kibana 服务

su - es
nohup /usr/local/kibana/bin/kibana &

6.4 验证 Kibana

6.4.1 浏览器访问

在浏览器中输入http://192.168.10.103:5601，进入 Kibana 界面。

6.4.2 添加索引模式

将 elk1 主机的系统日志添加到 kibana，若看不到日志，可访问网站生成日志信息后刷新 kibana。添加第二个日志时，在主菜单中找到 Management--》Stack Management，然后点击 Kibana 下的 Index Patterns，再点击右上角的 Create index pattern。

6.5 删除索引数据

6.5.1 删除单个索引

curl -XDELETE "http://localhost:9200/索引名"

6.5.2 批量删除

使用通配符（如 logs-*）或逗号分隔索引名，例如：

curl -XDELETE "http://localhost:9200/system-2025.04.28"

七、ELK 实战应用场景

7.1 微服务日志追踪

在微服务架构中，一个请求可能会经过多个服务节点，传统日志分散存储难以追踪完整链路。ELK 通过以下方实现日志追踪：

统一日志格式：在每个微服务中配置 Logstash 或 Filebeat，将日志统一格式化为 JSON，包含请求 ID（TraceID）、服务名称等字段。
分布式追踪关联：通过 Elasticsearch 的聚合查询，根据 TraceID 将分散在不同服务的日志关联，形成完整请求链路。
Kibana 可视化链路：使用 Kibana 的仪表盘功能，绘制请求耗时、错误分布等图表，快速定位性能瓶颈。

实战命令示例：

# 在微服务容器中启动Filebeat收集日志
docker run -d --name filebeat \
-v /var/log/microservice:/var/log/microservice \
-v /path/to/filebeat.yml:/usr/share/filebeat/filebeat.yml \
docker.elastic.co/beats/filebeat:7.10.0

7.2 安全合规审计

ELK 在安全日志分析中的应用：

日志采集与标准化：通过 Logstash 收集防火墙、IDS、认证系统等安全设备日志，使用 Grok 过滤器解析为结构化数据。
异常行为检测：利用 Elasticsearch 的聚合分析，识别异常登录、高频访问等安全事件。
合规报表生成：Kibana 支持自定义报表，满足等保、GDPR 等合规要求的日志留存与审计需求。

安全日志配置示例：

# Logstash配置解析防火墙日志
input {
  file {
    path => "/var/log/firewall.log"
    type => "firewall"
  }
}
filter {
  grok {
    match => { "message" => "%{IP:src_ip} %{WORD:action} %{NUMBER:port} -> %{IP:dest_ip}" }
  }
  geoip {
    source => "src_ip"
    target => "src_geo"
  }
}
output {
  elasticsearch {
    hosts => ["es-cluster:9200"]
    index => "security-%{+YYYY.MM.dd}"
  }
}

7.3 性能瓶颈分析

ELK 用于系统性能监控的关键能力：

多维度指标采集：结合 Metricbeat 收集服务器 CPU、内存、磁盘 IO 等指标，与应用日志关联分析。
实时告警触发：通过 Elasticsearch 的阈值查询，当响应时间超过阈值时触发 Kibana 告警。
历史趋势分析：利用时间序列数据，分析业务高峰期的资源占用模式，为扩容提供依据。

性能分析仪表盘配置：

在 Kibana 中创建新仪表盘，添加以下可视化组件：
- 折线图：展示 API 响应时间趋势
- 柱状图：服务请求量分布
- 热图：资源占用高峰期时段
设置告警规则：

{
  "query": {
    "range": {
      "response_time": {
        "gte": 500  # 响应时间超过500ms触发告警
      }
    }
  }
}

八、ELK 集群优化策略

8.1 Elasticsearch 分片策略优化

8.1.1 分片数量计算

公式：主分片数 = 节点数 × 3 ~ 5（每个节点建议承载 3-5 个主分片）。
示例：3 节点集群建议主分片数为 5（Elasticsearch 7.0 + 默认主分片数为 1，需手动调整）。

8.1.2 动态调整副本数

# 修改索引副本数为2
PUT /my-index/_settings
{
  "number_of_replicas": 2
}

注意：分片数在索引创建后无法修改，需通过 reindex 重建索引。

8.2 冷热数据分层存储

8.2.1 热数据节点配置

硬件要求：SSD 磁盘、高内存（建议 32GB+）
JVM 参数优化：

# /usr/local/elasticsearch/config/jvm.options
-Xms16g
-Xmx16g
-XX:MaxDirectMemorySize=16g

8.2.2 冷数据节点配置

硬件要求：HDD 磁盘、较低配置
索引生命周期管理（ILM）配置：

# kibana.yml中启用ILM
xpack.il.m.enabled: true

# 创建生命周期策略
PUT /_ilm/policy/cold_data_policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "set_priority": {
            "priority": 100
          }
        }
      },
      "cold": {
        "min_age": "7d",
        "actions": {
          "move_to": {
            "node_attr": "cold"
          },
          "set_priority": {
            "priority": 50
          }
        }
      }
    }
  }
}

8.3 Logstash 性能调优

8.3.1 批量处理参数

# logstash.yml配置
pipeline.batch.size: 1000  # 每次处理1000条日志
pipeline.batch.delay: 50ms  # 最大等待50ms凑满批量

8.3.2 多管道并行处理

# 启动多管道
/usr/local/logstash/bin/logstash -f /config/pipeline1.conf -f /config/pipeline2.conf

九、ELK 与 Beats/APM 集成

9.1 Beats 家族组件

组件	功能	应用场景
Filebeat	日志文件采集	服务器日志、应用日志
Metricbeat	系统指标采集	服务器 CPU / 内存 / 磁盘监控
Packetbeat	网络流量分析	网络性能监控、协议解析
Auditbeat	系统审计日志	安全合规审计

Metricbeat 安装示例：

yum install metricbeat
vim /etc/metricbeat/metricbeat.yml
# 配置发送到ELK
output.elasticsearch:
  hosts: ["192.168.10.103:9200"]
service.start

9.2 APM（应用性能监控）集成

9.2.1 部署 APM Server

docker run -d --name apm-server \
-e "APM_SECRET_TOKEN=my-token" \
-e "ELASTICSEARCH_HOSTS=http://192.168.10.103:9200" \
-p 8200:8200 \
docker.elastic.co/apm/apm-server:7.10.0

9.2.2 应用埋点配置（以 Java 为例）

<!-- pom.xml添加依赖 -->
<dependency>
  <groupId>co.elastic.apm</groupId>
  <artifactId>elastic-apm-agent</artifactId>
  <version>1.20.0</version>
</dependency>

# 启动参数
java -javaagent:/path/to/elastic-apm-agent-1.20.0.jar \
-Delastic.apm.service_name=my-service \
-Delastic.apm.server_urls=http://192.168.10.103:8200 \
-Delastic.apm.secret_token=my-token \
-jar my-app.jar

十、常见问题与解决方案

10.1 Elasticsearch 脑裂问题

10.1.1 原因

网络分区导致节点间通信中断
集群节点数为偶数（建议奇数节点）

10.1.2 解决方案

配置discovery.zen.minimum_master_nodes为(节点数/2)+1

# elasticsearch.yml
discovery.zen.minimum_master_nodes: 2  # 3节点集群设置为2

启用延迟加入（适用于网络不稳定场景）

cluster.join.timeout: 60s

10.2 Logstash 数据丢失问题

10.2.1 原因

未开启持久化队列
输出端故障导致缓冲溢出

10.2.2 解决方案

启用持久化队列

# logstash.yml
queue.type: persisted
queue.directory: /var/logstash/queue
queue.max_size: 1024gb

配置重试策略

# output配置
elasticsearch {
  hosts => ["es1:9200", "es2:9200"]
  retry_max: 30
  retry_interval: 5s
}

10.3 Kibana 连接异常

10.3.1 排查步骤

检查 Kibana 配置文件elasticsearch.hosts是否正确

grep "elasticsearch.hosts" /usr/local/kibana/config/kibana.yml

测试 ES 连接性

curl -XGET http://192.168.10.103:9200/_cluster/health?pretty

查看 Kibana 日志

tail -f /usr/local/kibana/logs/kibana.log

十一、企业级部署最佳实践

11.1 高可用架构设计

11.1.1 三层架构

采集层：多节点 Logstash+Filebeat 分布式采集
存储分析层：Elasticsearch 集群（3 节点以上）
展示层：Kibana 集群 + Nginx 负载均衡

11.1.2 资源规划

组件	最低配置（单节点）	建议配置（生产）
Elasticsearch	8 核 CPU/16GB 内存 / 200GB SSD	16 核 CPU/64GB 内存 / 1TB SSD
Logstash	4 核 CPU/8GB 内存	8 核 CPU/16GB 内存
Kibana	2 核 CPU/4GB 内存	4 核 CPU/8GB 内存

11.2 日志保留策略

11.2.1 按重要性分级

核心业务日志：保留 180 天（如交易日志）
普通业务日志：保留 90 天（如访问日志）
系统日志：保留 30 天（如服务器操作日志）

112.2 自动清理脚本

#!/bin/bash
# clean_logs.sh
DATE=$(date -d "30 days ago" +%Y.%m.%d)
INDEX_PATTERN="system-*"

# 删除30天前的索引
curl -XDELETE "http://localhost:9200/${INDEX_PATTERN}-${DATE}"

# 添加定时任务
crontab -e
0 0 * * * /path/to/clean_logs.sh >/dev/null 2>&1

11.3 安全加固

11.3.1 访问控制

启用 Elasticsearch X-Pack 安全模块

# 启用安全模块
bin/elasticsearch-setup-passwords auto

配置 Kibana 认证

# kibana.yml
xpack.security.enabled: true
xpack.security.encryptionKey: "your-encryption-key"

11.3.2 传输加密

生成 SSL 证书

bin/elasticsearch-certutil ca
bin/elasticsearch-certutil cert --ca ca.crt

配置 ES 启用 HTTPS

# elasticsearch.yml
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: certs/elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: certs/elastic-certificates.p12