文档首页/ 云搜索服务 CSS/ 最佳实践/ 日志管理与分析/ 使用Elasticsearch、自建Logstash和Kibana构建日志管理平台

更新时间：2026-04-23 GMT+08:00

使用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所示。

图1 ELKB架构

数据收集
- Beats作为数据收集器，负责从各种源收集数据并发送到Logstash。
- Logstash可以独立收集数据，或从Beats接收数据，对数据进行过滤、转换和增强。
数据处理
 Logstash在将数据发送到Elasticsearch之前，对数据进行必要的处理，如解析结构化日志、过滤无用信息等。
数据存储
 Elasticsearch作为核心存储组件，Elasticsearch索引和存储来自Logstash的数据，提供快速搜索和数据检索功能。
数据分析与可视化
 使用Kibana对Elasticsearch中的数据进行分析和可视化，创建仪表板和报告，以直观展示数据。

ELKB系统中各组件的版本兼容性请参见https://www.elastic.co/support/matrix#matrix_compatibility。

方案优势

实时性：提供实时数据收集和分析能力。
灵活性：支持各种数据源和灵活的数据处理流程。
易用性：用户界面友好，简化了数据操作和可视化过程。
扩展性：水平扩展能力强，可以处理PB级别的数据。

前提条件

已创建非安全模式的Elasticsearch集群。
已申请弹性云服务器ECS，并安装了Java环境，购买ECS请参见快速购买和使用Linux ECS。

操作步骤

登录ECS，部署并配置Filebeat。
1. 下载Filebeat，版本建议选择7.6.2。下载地址：https://www.elastic.co/downloads/past-releases#filebeat-oss
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:
```

部署并配置自建Logstash。

为了获得更优的性能，自建Logstash中的JVM参数建议配置为“ECS/docker”内存的一半。

下载Logstash，版本建议选择7.6.2。下载地址：https://www.elastic.co/downloads/past-releases#logstash-oss
确保Logstash与CSS集群的网络互通，在虚拟机上执行命令curl http:// {ip}:{port}测试虚拟机和ELasticsearch集群的连通性，返回200，表示已经连通。

配置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>:<端口号>"]
	index	事件写入的索引名称，即指定数据写入到哪个索引。单个索引：直接输入单个索引名称（如“my_index”）。多个索引：基于字段的动态命名或多个条件输出块来匹配多个索引。
	user	访问目的端集群的用户名。安全模式的集群必填。
	password	访问目的端集群的用户名对应的密码。安全模式的集群必填。
	ssl	目的端集群是否启用SSL。取值范围： true：使用HTTPS协议传输数据。 false：使用HTTP协议传输数据。
	cacert	用于认证目的端集群的CA证书文件路径。取值格式：<证书路径><证书名称>（如/opt/logstash/extend/certs）当目的端集群是CSS服务的Elasticsearch或OpenSearch时，获取默认CA证书的“证书名称”和“证书路径”。操作指导请参见查看默认证书。当目的端集群是自建或第三方Elasticsearch或OpenSearch时，则将目的端集群的安全证书上传至Logstash并获取“证书名称”和“证书路径”。操作指导请参见上传自定义证书。
	ssl_certificate_verification	目的端集群是否启用安全证书认证。取值范围： true（默认值）：使用安全证书认证目的端集群。 false：忽略安全证书认证。

Logstash的“filter”进行模式配置时，可以借助Grok Debugger（https://grokdebugger.com/）。

配置Elasticsearch集群的索引匹配模板。

登录云搜索服务管理控制台。
在左侧导航栏，选择“集群管理 > Elasticsearch”。
在集群列表，选择目标集群，单击操作列的“Kibana”，登录Kibana。
在Kibana左侧导航栏选择“Dev Tools”，进入操作页面。

创建一个索引匹配模板。

例如，创建一个索引模板，配置索引默认采用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"
          }
        }
    }
}

在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

执行如下命令，启动Logstash。

nohup ./bin/logstash -f /opt/pht/logstash-6.8.6/logstash-sample.conf &

执行如下命令，启动Filebeat。
```
./filebeat
```
通过Kibana进行查询并制作报表。
1. 进入Elasticsearch集群的Kibana操作界面。
2. 单击左侧导航栏的“Discover”进行查询与分析，类似的效果如图2所示。
  图2 Discover界面示例

父主题：日志管理与分析

上一篇：日志管理与分析

下一篇：集群性能调优

意见反馈

文档内容是否对您有帮助？

有帮助没帮助

提供反馈

提交成功！非常感谢您的反馈，我们会继续努力做到更好！您可在我的云声建议查看反馈及问题处理状态。

系统繁忙，请稍后重试

如您有其它疑问，您也可以通过华为云社区问答频道来与我们联系探讨

云宝助手提问云社区提问