# 使用Elasticsearch、自建Logstash和Kibana构建日志管理平台 使用CSS服务的Elasticsearch集群搭建的统一日志管理平台可以实时地、统一地、方便地管理日志，让日志驱动运维、运营等，提升服务管理效率。 #### 应用场景 本文以Elasticsearch、Filebeat、Logstash和Kibana为例，搭建一个统一日志管理平台。使用Filebeat采集ECS中的日志，发送到Logstash进行数据处理，再存储到Elasticsearch中，最后通过Kibana进行日志的可视化查询与分析。该方案可以用于以下场景： - 日志管理：集中管理应用程序和系统日志，快速定位问题。 - 安全监控：检测和响应安全威胁，进行入侵检测和异常行为分析。 - 业务分析：分析用户行为，优化产品和服务。 - 性能监控：监控系统和应用程序性能，实时发现瓶颈。 #### 方案架构 ELKB（Elasticsearch、Logstash、Kibana、Beats）提供了一整套日志场景解决方案，是目前主流的一种日志系统。 - Elasticsearch是一个开源分布式的搜索和分析引擎，用于存储、搜索和分析大量数据。 - Logstash是一个服务器端的数据管道，负责收集、解析和丰富数据后，将其发送到Elasticsearch。 - Kibana为Elasticsearch提供一个开源的数据分析和可视化平台，用于对Elasticsearch中的数据进行搜索、查看和交互。 - Beats：轻量级的数据收集器（如Filebeat、Metricbeat等），安装在服务器上，负责收集和转发数据到Logstash。 使用Elasticsearch和Logstash构建日志管理平台的方案架构如[图1]所示。 图1ELKB架构 ![](https://support.huaweicloud.com/bestpractice-css/figure/zh-cn_image_0000001961178937.png "点击放大") 1. 数据收集 - Beats作为数据收集器，负责从各种源收集数据并发送到Logstash。 - Logstash可以独立收集数据，或从Beats接收数据，对数据进行过滤、转换和增强。 2. 数据处理 Logstash在将数据发送到Elasticsearch之前，对数据进行必要的处理，如解析结构化日志、过滤无用信息等。 3. 数据存储 Elasticsearch作为核心存储组件，Elasticsearch索引和存储来自Logstash的数据，提供快速搜索和数据检索功能。 4. 数据分析与可视化 使用Kibana对Elasticsearch中的数据进行分析和可视化，创建仪表板和报告，以直观展示数据。 ELKB系统中各组件的版本兼容性请参见。 #### 方案优势 - 实时性：提供实时数据收集和分析能力。 - 灵活性：支持各种数据源和灵活的数据处理流程。 - 易用性：用户界面友好，简化了数据操作和可视化过程。 - 扩展性：水平扩展能力强，可以处理PB级别的数据。 #### 前提条件 - 已创建非安全模式的Elasticsearch集群。 - 已申请弹性云服务器ECS，并安装了Java环境，购买ECS请参见[快速购买和使用Linux ECS](https://support.huaweicloud.com/qs-ecs/ecs_01_0103.html)。 #### 操作步骤 1. 登录ECS，部署并配置Filebeat。 1. 下载Filebeat，版本建议选择7.6.2。下载地址： 2. 配置Filebeat的配置文件"filebeat.yml"。 例如，采集"/root/"目录下以"log"结尾的所有文件，配置文件"filebeat.yml"中的内容如下： ``` filebeat.inputs: - type: log enabled: true # 采集的日志文件路径。 paths: - /root/*.log filebeat.config.modules: path: ${path.config}/modules.d/*.yml reload.enabled: false # Logstash的hosts信息 output.logstash: hosts: ["192.168.0.126:5044"] processors: ``` 2. 部署并配置Logstash。 ![](https://support.huaweicloud.com/bestpractice-css/public_sys-resources/note_3.0-zh-cn.png) 为了获得更优的性能，自建Logstash中的JVM参数建议配置为"ECS/docker"内存的一半。 1. 下载Logstash，版本建议选择7.6.2。下载地址： 2. 确保Logstash与CSS集群的网络互通，在虚拟机上执行命令curl http://{ip}:{port}测试虚拟机和ELasticsearch集群的连通性，返回200，表示已经连通。 3. 配置Logstash的配置文件"logstash-sample.conf"。 配置文件"logstash-sample.conf"中的内容如下： ``` input { beats { port => 5044 } } # 对数据的切割与截取信息 filter { grok { match => { "message" => '\[%{GREEDYDATA:timemaybe}\] \[%{WORD:level}\] %{GREEDYDATA:content}' } } mutate { remove_field => ["@version","tags","source","input","prospector","beat"] } } # CSS集群的信息 output { elasticsearch { # 目的端集群的节点访问地址，无需添加协议 hosts => ["xxx.xxx.xxx.xxx:9200", "xxx.xxx.xxx.xxx:9200"] # 事件写入的索引配置 index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}" # 安全集群必配 (非安全集群请删除以下配置) # user => "xxx" # 访问集群的用户名 # password => "xxx" # 用户名对应的密码 # 目的端集群启用SSL时，则需额外配置以下信息 # ssl => true # cacert => "/opt/logstash/extend/certs" # 用于认证目的端集群的CA证书文件路径 # ssl_certificate_verification => false # 目的端集群是否启用安全证书认证 } } ``` 表1配置项说明 | **配置**项名称 || **说明** | |:---|---|:---| | output | hosts | 目的端集群的节点访问地址。支持配置多个IP地址。 取值格式：\["\<节点IP地址1\>:\<端口号\>", "\<节点IP地址2\>:\<端口号\>"\] | | output | index | 事件写入的索引名称，即指定数据写入到哪个索引。 - 单个索引：直接输入单个索引名称（如"my_index"）。 - 多个索引：基于字段的动态命名或多个条件输出块来匹配多个索引。 | | output | user | 访问目的端集群的用户名。 安全模式的集群必填。 | | output | password | 访问目的端集群的用户名对应的密码。 安全模式的集群必填。 | | output | ssl | 目的端集群是否启用SSL。 取值范围： - true：使用HTTPS协议传输数据。 - false：使用HTTP协议传输数据。 | | output | cacert | 用于认证目的端集群的CA证书文件路径。 取值格式：\<证书路径\>\<证书名称\>（如/opt/logstash/extend/certs） - 当目的端集群是CSS服务的Elasticsearch或OpenSearch时，获取默认CA证书的"证书名称"和"证书路径"。操作指导请参见[查看默认证书](https://support.huaweicloud.com/usermanual-css/css_01_0247.html)。 - 当目的端集群是自建或第三方Elasticsearch或OpenSearch时，则将目的端集群的安全证书上传至Logstash并获取"证书名称"和"证书路径"。操作指导请参见[上传自定义证书](https://support.huaweicloud.com/usermanual-css/css_01_0247.html)。 | | output | ssl_certificate_verification | 目的端集群是否启用安全证书认证。 取值范围： - true（默认值）：使用安全证书认证目的端集群。 - false：忽略安全证书认证。 | ![](https://support.huaweicloud.com/bestpractice-css/public_sys-resources/note_3.0-zh-cn.png) "filter"的grok模式可以借助Grok Debugger（）进行调试。 3. 配置Elasticsearch索引匹配模板。 1. 登录[云搜索服务管理控制台](https://console.huaweicloud.com/elasticsearch/)。 2. 在左侧导航栏，选择"集群管理 \> Elasticsearch"。 3. 在集群列表，选择目标集群，单击操作列的"Kibana"，登录Kibana。 4. 在Kibana左侧导航栏选择"Dev Tools"，进入操作页面。 5. 创建一个索引匹配模板。 例如，创建一个索引模板，配置索引默认采用3分片、0副本，索引中定义了"@timestamp"、"content"、"host.name"、"level"、"log.file.path"、"message"、"timemaybe"等字段。 ``` PUT _template/filebeat { "index_patterns": ["filebeat*"], "settings": { # 定义分片数。 "number_of_shards": 3, # 定义副本数。 "number_of_replicas": 0, "refresh_interval": "5s" }, # 定义字段。 "mappings": { "properties": { "@timestamp": { "type": "date" }, "content": { "type": "text" }, "host": { "properties": { "name": { "type": "text" } } }, "level": { "type": "keyword" }, "log": { "properties": { "file": { "properties": { "path": { "type": "text" } } } } }, "message": { "type": "text" }, "timemaybe": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||strict_date_optional_time||epoch_millis||EEE MMM dd HH:mm:ss zzz yyyy" } } } } ``` 4. 准备测试数据。 在ECS上执行以下命令，生成测试数据，并将数据写到"/root/tmp.log"中： ``` bash -c 'while true; do echo [$(date)] [info] this is the test message; sleep 1; done;' >> /root/tmp.log & ``` 生成的测试数据样例如下： ``` [Thu Feb 13 14:01:16 CST 2020] [info] this is the test message ``` 5. 在ECS上执行以下命令，启动Logstash。 ``` nohup ./bin/logstash -f /opt/pht/logstash-7.6.2/logstash-sample.conf & ``` 6. 在ECS上执行以下命令，启动Filebeat。 ``` nohup ./filebeat & ``` 7. 在Kibana中创建索引模式。 1. 在Kibana页面，单击左上角菜单图标，选择"Management \> Stack Management"。 2. 在Stack Management页面，选择"Index Patterns"，单击"Create index pattern"创建索引模式。 3. 在"Index pattern name"输入框中输入索引名称前缀（如"filebeat\*"）。 4. 单击"Next step"。 5. 在"Time field"下拉框中选择时间字段（如"@timestamp"）。 6. 单击"Create index pattern"创建索引模式。 8. 通过Kibana查询并制作报表。 1. 在Kibana页面，单击左上角菜单图标，选择"Discover"。 2. 在左侧下拉菜单中选择"filebeat\*"索引模式。 3. 在搜索栏输入过滤条件，在时间筛选器中选择时间范围，查看数据。类似的效果如[图2]所示。 图2Discover界面示例 ![](https://support.huaweicloud.com/bestpractice-css/figure/zh-cn_image_0000001961259117.png "点击放大")