正文 

当前位置： 首页 > 知识 > java教程

elasticsearch+logstash+filebeat+kibana搭建elk日志分析平台（官方推荐的beats架构）

博学谷狂野架构师 博学谷狂野架构师     2022-11-29     220

关键词：

Elasticsearch

俗话话说的号，没有金刚钻，也不揽那瓷器活；日志分析可以说是所有大小系统的标配了，不知道有多少菜鸟程序员有多喜欢日志，如果没了日志，那自己写的bug想不被别人发现，可就难了； 有了它，就可将bug们统统消化在自己手里。

当然了，作为一个架构师搭建动手搭建一个日志平台也基本是必备技能了，虽然我们说架构师基本不咋写代码了，但是如果需要的时候，还是能扛枪的

​​​​

ELK部署应用与工作机制

3.1 ELK日志分析平台介绍

ELK是三个开源软件的缩写，分别表示：Elasticsearch , Logstash和Kibana。Elasticsearch和Kibana我们上面做过讲解。 Logstash 主要是用来日志的搜集、分析、过滤日志的工具，适用大数据量场景， 一般采用c/s模式，client端安装在需要收集日志的主机上，server端负责将收到的各节点日志进行过滤、修改等操作， 再一并发往Elasticsearch上做数据分析。

一个完整的集中式日志系统，需要包含以下几个主要特点：

  • 收集－能够采集多种来源的日志数据
  • 传输－能够稳定的把日志数据传输到中央系统
  • 存储－如何存储日志数据
  • 分析－可以支持 UI 分析
  • 警告－能够提供错误报告，监控机制

ELK提供了一整套解决方案，并且都是开源软件，之间互相配合使用，完美衔接，高效的满足了很多场合的应用，是目前主流的一种日志分析平台。

3.2 ELK部署架构模式

3.2.1 简单架构

Elasticsearch

这是最简单的一种ELK部署架构方式， 由Logstash分布于各个节点上搜集相关日志、数据，并经过分析、过滤后发送给远端服务器上的Elasticsearch进行存储。 优点是搭建简单， 易于上手， 缺点是Logstash耗资源较大， 依赖性强， 没有消息队列缓存， 存在数据丢失隐患

3.2.2 消息队列架构

Elasticsearch

该队列架构引入了KAFKA消息队列， 解决了各采集节点上Logstash资源耗费过大， 数据丢失的问题， 各终端节点上的Logstash Agent 先将数据/日志传递给Kafka， 消息队列再将数据传递给Logstash， Logstash过滤、分析后将数据传递给Elasticsearch存储， 由Kibana将日志和数据呈现给用户。

3.2.3 BEATS架构

Elasticsearch

该架构的终端节点采用Beats工具收集发送数据， 更灵活，消耗资源更少，扩展性更强。同时可配置Logstash 和Elasticsearch 集群用于支持大集群系统的运维日志数据监控和查询， 官方也推荐采用此工具， 本章我们采用此架构模式进行配置讲解（如果在生产环境中， 可以再增加kafka消息队列， 实现了beats+消息队列的部署架构 ）。

Beats工具包含四种：

1、Packetbeat（搜集网络流量数据）

2、Topbeat（搜集系统、进程和文件系统级别的 CPU 和内存使用情况等数据）

3、Filebeat（搜集文件数据）

4、Winlogbeat（搜集 Windows 事件日志数据）

3.3 ELK工作机制

3.3.1 Filebeat工作机制

Filebeat由两个主要组件组成：prospectors 和 harvesters。这两个组件协同工作将文件变动发送到指定的输出中。

Elasticsearch

Harvester（收割机）：负责读取单个文件内容。每个文件会启动一个Harvester，每个Harvester会逐行读取各个文件，并将文件内容发送到制定输出中。Harvester负责打开和关闭文件，意味在Harvester运行的时候，文件描述符处于打开状态，如果文件在收集中被重命名或者被删除，Filebeat会继续读取此文件。所以在Harvester关闭之前，磁盘不会被释放。默认情况filebeat会保持文件打开的状态，直到达到close_inactive

filebeat会在指定时间内将不再更新的文件句柄关闭，时间从harvester读取最后一行的时间开始计时。若文件句柄被关闭后，文件发生变化，则会启动一个新的harvester。关闭文件句柄的时间不取决于文件的修改时间，若此参数配置不当，则可能发生日志不实时的情况，由scan_frequency参数决定，默认10s。Harvester使用内部时间戳来记录文件最后被收集的时间。例如：设置5m，则在Harvester读取文件的最后一行之后，开始倒计时5分钟，若5分钟内文件无变化，则关闭文件句柄。默认5m】。

Prospector（勘测者）：负责管理Harvester并找到所有读取源。

filebeat.prospectors:
- input_type: log
  paths:
    - /apps/logs/*/info.log

Prospector会找到/apps/logs/*目录下的所有info.log文件，并为每个文件启动一个Harvester。Prospector会检查每个文件，看Harvester是否已经启动，是否需要启动，或者文件是否可以忽略。若Harvester关闭，只有在文件大小发生变化的时候Prospector才会执行检查。只能检测本地的文件。

Filebeat如何记录发送状态：

将文件状态记录在文件中（默认在/var/lib/filebeat/registry）。此状态可以记住Harvester收集文件的偏移量。若连接不上输出设备，如ES等，filebeat会记录发送前的最后一行，并再可以连接的时候继续发送。Filebeat在运行的时候，Prospector状态会被记录在内存中。Filebeat重启的时候，利用registry记录的状态来进行重建，用来还原到重启之前的状态。每个Prospector会为每个找到的文件记录一个状态，对于每个文件，Filebeat存储唯一标识符以检测文件是否先前被收集。

Filebeat如何保证数据发送成功：

Filebeat之所以能保证事件至少被传递到配置的输出一次，没有数据丢失，是因为filebeat将每个事件的传递状态保存在文件中。在未得到输出方确认时，filebeat会尝试一直发送，直到得到回应。若filebeat在传输过程中被关闭，则不会再关闭之前确认所有时事件。任何在filebeat关闭之前未确认的事件，都会在filebeat重启之后重新发送。这可确保至少发送一次，但有可能会重复。可通过设置shutdown_timeout 参数来设置关闭之前的等待事件回应的时间（默认禁用）。

3.3.2 Logstash工作机制

Logstash事件处理有三个阶段：inputs → filters → outputs。是一个接收，处理，转发日志的工具。支持系统日志，webserver日志，错误日志，应用日志等。

Elasticsearch

Input：输入数据到logstash。

支持的输入类型：

file：从文件系统的文件中读取，类似于tail -f命令

syslog：在514端口上监听系统日志消息，并根据RFC3164标准进行解析

redis：从redis service中读取

beats：从filebeat中读取

Filters：数据中间处理，对数据进行操作。

一些常用的过滤器为：

grok：解析任意文本数据，Grok 是 Logstash 最重要的插件。它的主要作用就是将文本格式的字符串，转换成为具体的结构化的数据，配合正则表达式使用。内置120多个解析语法。

​​官方提供的grok表达式​​

mutate：对字段进行转换。例如对字段进行删除、替换、修改、重命名等。

drop：丢弃一部分events不进行处理。

clone：拷贝 event，这个过程中也可以添加或移除字段。

geoip：添加地理信息(为前台kibana图形化展示使用)

Outputs：outputs是logstash处理管道的最末端组件。

一个event可以在处理过程中经过多重输出，但是一旦所有的outputs都执行结束，这个event也就完成生命周期。

常见的outputs为：

elasticsearch：可以高效的保存数据，并且能够方便和简单的进行查询。

file：将event数据保存到文件中。

graphite：将event数据发送到图形化组件中，一个很流行的开源存储图形化展示的组件。

Codecs：codecs 是基于数据流的过滤器，它可以作为input，output的一部分配置。 Codecs可以帮助你轻松的分割发送过来已经被序列化的数据。 常见的codecs：

json：使用json格式对数据进行编码/解码。

multiline：将多个事件中数据汇总为一个单一的行。比如：java异常信息和堆栈信息。

3.4 Logstash安装配置

在192.168.116.141机器节点上进行安装：

  1. 下载解压
    下载：

cd /usr/local
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.10.2-linux-x86_64.tar.gz

  1. 解压：

tar -xvf logstash-7.10.2-linux-x86_64.tar.gz

  1. 创建数据存储与日志记录目录

[root@localhost logstash-7.10.2]# mkdir -p /usr/local/logstash-7.10.2/data
[root@localhost logstash-7.10.2]# mkdir -p /usr/local/logstash-7.10.2/logs

  1. 修改配置文件：

vi /usr/local/logstash-7.10.2/config/logstash.yml

  1. 配置内容：

# 数据存储路径
path.data: /usr/local/logstash-7.10.2/data
# 监听主机地址
http.host: "192.168.116.141"
# 日志存储路径
path.logs: /usr/local/logstash-7.10.2/logs
#启动监控插件
xpack.monitoring.enabled: true 
#Elastic集群地址
xpack.monitoring.elasticsearch.hosts:["http://192.168.116.140:9200","http://192.168.116.140:9201","http://192.168.116.140:9202"]

  1. 创建监听配置文件：

vi /usr/local/logstash-7.10.2/config/logstash.conf

  1. 配置：

input 
  beats 
    # 监听端口
    port => 5044
  


output 

  stdout 
    # 输出编码插件
    codec => rubydebug
  

  elasticsearch 
    # 集群地址
    hosts => ["http://192.168.116.140:9200","http://192.168.116.140:9201","http://192.168.116.140:9202"]

  1. 启动服务：
    以root用户身份执行：

## 后台启动方式
nohup  /usr/local/logstash-7.10.2/bin/logstash -f /usr/local/logstash-7.10.2/config/logstash.conf &
##
./logstash -f ../config/logstash.conf

  1. 成功启动后会显示以下日志：

[2020-10-15T06:57:40,640][INFO ][logstash.agent           ] Successfully started Logstash API endpoint :port=>9600


  2. 访问地址：​​http://192.168.116.141:9600/​​, 可以看到返回信息：

  3. Elasticsearch

3.5 Filebeat安装配置

在192.168.116.141机器节点上操作：

  1. 下载解压
    与ElasticSearch版本一致， 下载7.10.2版本。

cd /usr/local
wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.10.2-linux-x86_64.tar.gz

  1. 解压：

tar -xvf filebeat-7.10.2-linux-x86_64.tar.gz

  1. 修改配置文件

vi /usr/local/filebeat-7.10.2/filebeat.yml

  1. 修改内容：

# 需要收集发送的日志文件
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/messages
# 如果需要添加多个日志，只需要添加
- type: log
  enabled: true
  paths:
    - /var/log/test.log    
# filebeat 配置模块， 可以加载多个配置
filebeat.config.modules:  
  path: $path.config/modules.d/*.yml  
  reload.enabled: false    
# 索引分片数量设置
setup.template.settings:
  index.number_of_shards: 2
# kibana 信息配置
setup.kibana:
  host: "192.168.116.140:5601"
# logstash 信息配置 （注意只能开启一项output设置， 如果采用logstash, 将output.elasticsearch关闭）
output.logstash:  
  hosts: ["192.168.116.141:5044"]
# 附加metadata元数据信息
processors:
  - add_host_metadata: ~
  - add_cloud_metadata: ~

  1. 启动服务

## 后台启动
nohup /usr/local/filebeat-7.10.2/filebeat -e -c /usr/local/filebeat-7.10.2/filebeat.yml &
## 
./filebeat -e -c filebeat.yml

  1. 启动成功后显示日志：

2020-12-15T07:09:33.922-0400    WARN    beater/filebeat.go:367  Filebeat is unable to load the Ingest Node pipelines for the configured modules because the Elasticsearch output is not configured/enabled. If you have already loaded the Ingest Node pipelines or are using Logstash pipelines, you can ignore this warning.
2020-12-15T07:09:33.922-0400    INFO    crawler/crawler.go:72   Loading Inputs: 1
2020-12-15T07:09:33.923-0400    INFO    log/input.go:148        Configured paths: [/var/log/messages]
2020-12-15T07:09:33.923-0400    INFO    input/input.go:114      Starting input of type: log; ID: 14056778875720462600 
2020-12-15T07:09:33.924-0400    INFO    crawler/crawler.go:106  Loading and starting Inputs completed. Enabled inputs: 1
2020-12-15T07:09:33.924-0400    INFO    cfgfile/reload.go:150   Config reloader started

  1. 我们监听的是/var/log/messages系统日志信息， 当日志发生变化后， filebeat会通过logstash上报到Elasticsearch中。 我们可以查看下集群的全部索引信息：
    ​​​http://192.168.116.140:9200/_cat/indices?v​​​

    可以看到， 已经生成了名为logstash-2021.07.20-000001索引。

3.6 Kibana配置与查看数据


  2. 进入Kibana后台， 进行配置：
    ​​​http://192.168.116.140:5601​​

  3. Elasticsearch

  4. 进入【Management】--> 在Index Pattern中输入"logstash-*" --> 点击【next step】, 选择"@timestamp",

  5. Elasticsearch

  6. 点击【 Create index pattern 】进行创建。

  7. Elasticsearch

  8. 查看数据
    进入【Discover】, 可以查看到收集的数据：

Elasticsearch

如果没有显示， 可以重新调整Time Range时间范围。

本文由育博学谷狂野架构师发布如果本文对您有帮助，欢迎关注和点赞；如果您有任何建议也可留言评论或私信，您的支持是我坚持创作的动力转载请注明出处！

logstash上报数据到elasticsearch(代码片段)

logstash上报数据到elasticsearch前提是已经安装、部署完logstash和elasticsearchWindows安装启动logstash_zhangphil的博客-CSDN博客_logstashwindows安装Windows安装启动logstash(1)下载logstash，下载链接：DownloadLogstashFree|GetStartedN 查看详情

用elasticsearch,logstash,kibana搭建实时日志收集系统

用ElasticSearch,LogStash,Kibana搭建实时日志收集系统介绍这套系统，logstash负责收集处理日志文件内容存储到elasticsearch搜索引擎数据库中。kibana负责查询elasticsearch并在web中展示。logstash收集进程收获日志文件内容后，先输出到redis中... 查看详情

elasticsearch+logstash+kibana(代码片段)

ElasticSearch+Logstash+kibana 1、介绍：　　Logstash：搬运工　　ElasticSearch：搜索引擎　　Kilbana：可视化系统 　　ElasticSearch是基于lucene的搜索框架，它提供了一个分布式多用户能力的全文搜索引擎。　　基于restfulweb接口　　上... 查看详情

logstash：如何配置metricbeat及logstash为elasticsearch8.x收集数据

如何配置Metricbeat及Logstash为Elasticsearch8.0收集数据如何配置Metricbeat及Logstash为Elasticsearch8.0收集数据_哔哩哔哩_bilibili 查看详情

使用logstash同步数据至elasticsearch，springboot中集成elasticsearch实现搜索

安装logstash、同步数据至ElasticSearch为什么使用logstash来同步，CSDN上有一篇文章简要的分析了以下几种同步工具的优缺点：https://blog.csdn.net/laoyang360/article/details/51694519。下面开始实践：　　1.下载Logstash安装包，需要注意版本与elast... 查看详情

读取 Logstash 日志但不会推送到 elasticsearch

】读取Logstash日志但不会推送到elasticsearch【英文标题】：Logstashlogsarereadbutarenotpushedtoelasticsearch【发布时间】：2019-12-0312:10:20【问题描述】：我有以下logstash配置：inputfilecodec=>"json_lines"path=>["/etc/logstash/input.log"]sincedb_path=>" 查看详情

filebeat+logstash+elasticsearch+granfa(代码片段)

filebeat+logstash+elasticsearch+granfahttps://www.cnblogs.com/wenchengxiaopenyou/p/9034213.html一。背景前端web服务器为nginx，采用filebeat+logstash+elasticsearch+granfa进行数据采集与展示，对客户端ip进行地域统计，监控服务器响应时间等。二。业务整体架... 查看详情

filebeat+logstash+elasticsearch收集haproxy日志

...，传输到kafka，在用logstash从消息队列中抓取日志存储到elasticsearch中。但是在5.5版本中，使用filebeat收集日志，减少对客户机的性能影响，filebeat收集日志传输到logstash的5044端口，logstash接收日志，然后传输 查看详情

如何在elasticsearch中查看logstash打到elasticsearch的数据

 #catsyslog02.conf#filename:syslog02.conf#注意这个是要用#号注释掉input{file{path=>["/var/log/*.log"]}}output{elasticsearch{hosts=>["12x.xx.15.1xx:9200"]}}查看配置文件是否有问题:#../bin/logstash-fsyslog02.conf-tS 查看详情

elasticsearch+logstash+kibana搭建教程

#ELK：Elasticsearch+Logstash+Kibana搭建教程Shipper：日志收集者。负责监控本地日志文件的变化，及时把日志文件的最新内容收集起来，输出到Redis暂存。Indexer：日志存储者。负责从Redis接收日志，写入到本地文件。Broker：日志Hub，用来... 查看详情

logstash导入数据到elasticsearch

一：在Windows环境　　1下载解压Logstash的压缩包     2将oracle的驱动放入Logstash的解压包中     3在Logstash的包中创建jdbc.conf     4整理sql语句在一个文件下  &nb 查看详情

elasticsearch数据库同步插件logstash

 　　1.下载和elasticsearch相同版本的logstash.　　2.进行解压后,进入bin下,新建一个文件mysql.conf,并输入inputstdinoutputstdout3.cmd进入bin下,输入logstash-fmysql.conf 启动后,输入　　http://127.0.0.1:9600/4.ok;说明已经启动了   查看详情

Logstash 将所有内容放入所有 Elasticsearch 索引中

】Logstash将所有内容放入所有Elasticsearch索引中【英文标题】：LogstashputseverythingintoallElasticsearchindexes【发布时间】：2017-05-1007:32:05【问题描述】：我在/etc/logstash/conf.d下有4个文件。这4个文件用于读取tomcat的日志和apache的日志，监... 查看详情

logstash同步mysql数据到elasticsearch(代码片段)

目录1MySql数据到Elasticsearch1.1下载logstash1.2解压logstash1.3在logstash目录创建mysql文件夹1.4将mysql驱动文件和数据库查询文件放进mysql中1.5在config目录下创建mysqltoes.conf文件1.6mysqltoes.conf配置1.7启动logstash2配置语法讲解3启动方式4filebeat基... 查看详情

从 Logstash 在 Elasticsearch 上使用 geopoint

】从Logstash在Elasticsearch上使用geopoint【英文标题】：UsinggeopointonElasticsearchfromLogstash【发布时间】：2019-04-2113:29:46【问题描述】：我想在docker的elk堆栈中使用geoip。根据这个文档，我添加了配置https://www.elastic.co/blog/geoip-in-the-elastic-... 查看详情

logstash+kibana+多elasticsearch集群部署(代码片段)

ELK原理与介绍?ELK是三个开源软件的缩写，分别表示：Elasticsearch,Logstash,Kibana,它们都是开源软件。新增了一个FileBeat，它是一个轻量级的日志收集处理工具(Agent)，Filebeat占用资源少，适合于在各个服务器上搜集日志后传输给Logstash... 查看详情

Logstash 拆分日志并单独插入到elasticsearch中

】Logstash拆分日志并单独插入到elasticsearch中【英文标题】：Logstashsplitlogandinsertitseparatelyintoelasticsearch【发布时间】：2021-09-1405:55:54【问题描述】：我正在编写一个logstash配置文件，而我收到的一个日志给我带来了问题，团队向我... 查看详情

腾讯云logstash实战4-使用logstash消费kafka数据并写入到elasticsearch

...h的一个典型应用场景，就是消费kafka中的数据并且写入到Elasticsearch,使用腾讯云Logstash产品，可以通过简单的配置快速地完成这一过程。在“管道管理”页面，点击“新建管道”按钮，创建一个管道：进入管道配置页面，点击“... 查看详情