目录
- 前言
- 阅读对象
- 阅读导航
- 前置知识
- 笔记正文
- 一、ELK架构
- 1.1 经典的ELK
- 1.2 整合消息队列+Nginx架构
- 二、LogStash介绍
- 2.1 Logstash核心概念
- 2.1.1 Pipeline
- 2.1.2 Event
- 2.1.3 Codec (Code / Decode)
- 2.1.4 Queue
- 2.2 Logstash数据传输原理
- 2.3 Logstash的安装(以windows为例)
- 2.4 Logstash配置文件结构
- 2.5 使用示例
- 三、Beats介绍
- 3.1 FileBeat简介
- 3.1 FileBeat的工作原理
- 3.3 Logstash vs FileBeat
- 3.4 Filebeat安装
- 四、经典ELK整合实战
- 案例1:采集tomcat服务器日志,使用FileBeats将日志发送到Logstash
- 案例2:改进上面的,Logstash输出数据到Elasticsearch
- 案例3:利用Logstash过滤器解析日志
- 案例4:输出到Elasticsearch指定索引
- 学习总结
前言
这一篇笔记给大家分享一些日志管理相关的组件。为什么需要这个玩意呢?这主要还是因为分布式微服务的兴起啊!试想一下,现在的电商场景,动不动就百万流量进来,万一线上出了什么问题了,异常啊什么的,总该需要追溯日志吧?而且这些日志,在微服务场景下,通常是分布在多个机器里面的。那在这种海量业务日志里面,如何快速找到分布在各服务器里面的日志呢?注意关键词:
- 海量日志
- 多台服务器
这时候,所谓的ELK架构就出现了。
阅读对象
阅读导航
系列上一篇文章:《【ES专题】ElasticSearch功能详解与原理剖析》
前置知识
笔记正文
一、ELK架构
什么是ELK架构?其实ELK架构是ElasticSearch + Logstash + Kibana
组成的。ELK架构可以将系统日志、网站日志、应用系统日志等各种日志进行收集、过滤、清洗,然后进行集中存放并可用于实时检索、分析。
集中化日志管理思路:
日志收集 -> 格式化分析 -> 检索和可视化 -> 风险告警
ELK
架构分为两种,一种是经典的ELK,另外一种是加上消息队列(Redis或Kafka或RabbitMQ)和Nginx
结构。
1.1 经典的ELK
经典的ELK主要是由Filebeat + Logstash + Elasticsearch + Kibana
组成,如下图:(早期的ELK只有Logstash + Elasticsearch + Kibana
)
beats
:一种数据搜集组件logstash
:一种数据聚合和处理的组件elasticsearch
:用于索引数据和存储的组件kibana
:一种用于数据分析和可视化的组件
此架构主要适用于数据量小的开发环境,存在数据丢失的危险。
1.2 整合消息队列+Nginx架构
这个架构,算是对上面一个架构的升级。主要是在beats
跟logstash
之间加上了Redis或Kafka或RabbitMQ
做消息队列,一方面是为了解决速度之间不匹配的问题,一方面也是为了保证消息不丢失。
beats
:一种数据搜集组件redis
、kafka
、rabbitmq
:一种缓存中间件logstash
:一种数据聚合和处理的组件elasticsearch
:用于索引数据和存储的组件kibana
:一种用于数据分析和可视化的组件
此种架构,主要用在生产环境,可以处理大数据量。
二、LogStash介绍
Logstash 是免费且开放的服务器端数据处理管道,能够从多个来源采集数据,转换数据,然后将数据发送到您最喜欢的存储库中(多种输出源选择,如:mysql,es,mongo,甚至是文件)。
Logstash的使用场景包括但不限于以下几种情况:
- 日志收集:Logstash可以接收、处理并输出日志,支持各种各样的日志,包括系统日志、WEB容器日志(如Apache日志和Nginx日志和Tomcat日志等)、各种应用日志等
- 数据过滤:Logstash可以过滤和处理日志数据,包括删除不需要的数据、转换数据格式等
- 数据处理:Logstash可以处理日志数据,包括数据分析和统计、事件聚合等
- 数据存储:Logstash可以将处理后的日志数据存储起来,以供后续检索和分析
- 数据监控:Logstash可以监控系统运行状态,及时发现异常情况,帮助管理员快速定位问题
- 日志分析:Logstash可以分析日志数据,生成报告和图表,帮助管理员更好地了解系统运行情况
- 日志搜索:Logstash可以实时搜索和分析日志数据,快速定位问题,提高系统维护效率
2.1 Logstash核心概念
Logstash中有几个核心概念,需要大家掌握了解。它们分别是:Pipeline
、Event
、Codec (Code / Decode)
、Queue
2.1.1 Pipeline
Pipeline,管道,这玩意大家都不陌生。管道在我们软件设计中通常扮演着盛放某些【处理器】的【容器】
角色,所谓处理器如Logstash的filter
过滤器(过滤器
+ 管道
几乎是一种范式)。在ES的Pipeline,主要有如下三个特征:
- 包含了
input—filter—output
三个阶段的处理流程(核心!三个阶段分别是做什么的,2.1.3中有一个原理图可以看看)
它是数据在Logstash内部流转时的具体表现形式。
在input阶段,数据被转换为Event;
在filter阶段,数据会被加工处理;(这个阶段可选)
在output阶段,数据被转换为目标格式
注意:在pipeline里面,同时也包含input、filter、output阶段对应的input、filter、output插件(又叫:组件)
- 插件生命周期管理:Logstash为各个阶段核心组件提供了多种类型的插件,这些插件通常是在
Pipeline
中被应用 - 队列管理:【2.1.2Event事件】的管理
2.1.2 Event
数据在内部流转时的具体表现形式,数据在input 阶段被转换为Event。Event本质上就是一个Java Object
,在配置文件中,可以对Event
的属性进行增删改查
2.1.3 Codec (Code / Decode)
Codec
其实是编解码的意思。编解码就是把数据从一种格式转换为另一种格式。在Logstash里面,即:在input
阶段将原始数据decode
成Event;在output
阶段将Event
给encode
成目标数据的相关编解码器。它的原理图如下所示:
2.1.4 Queue
这个Queue,即:队列,存在于【2.1.1】提到的管道中,是一种用于存储event
的机制。它通常有两种类型:
Memory Queue
内存管道:显然,进程Crash,机器宕机,都会引起数据的丢失Persistent Queue
持久化管道:不同于内存管道,因为持久化机制的存在,就算宕机也不会丢失数据。数据保证会被消费,甚至可以替代 Kafka等消息队列缓冲区的作用
它的原理图,以及所处的位置,如下:
注意,它是介于input组件跟filter组件之间的
2.2 Logstash数据传输原理
我们在前面了解到,Logstash会在Pipeline
里面分三个阶段处理数据,这些数据处理的原理如下:
- 数据采集与输入:Logstash支持各种输入选择,能够以连续的流式传输方式,轻松地从日志、指标、Web应用以及数据存储中采集数据
- 实时解析和数据转换:通过Logstash过滤器解析各个事件,识别已命名的字段来构建结构,并将它们转换成通用格式,最终将数据从源端传输到存储库中
- 存储与数据导出:Logstash提供多种输出选择,可以将数据发送到指定的地方
一个很典型的数据采集示例,如下:(从数据库读取一些数据,经过Logstash处理之后丢到ES中)
2.3 Logstash的安装(以windows为例)
logstash官方文档
1)下载并解压logstash
下载地址: Logstash下载地址
选择版本:7.17.3
,跟es
保持一致
2)测试:运行最基本的logstash管道
# 进入目标目录
cd logstash-7.17.3# 启动logstash
.\bin\logstash.bat -e "input { stdin { } } output { stdout {} }"
测试结果:
2.4 Logstash配置文件结构
Logstash的管道配置文件对每一个处理阶段的核心组件,都提供了一个单独的支持不同类型插件的配置,用于处理管道事件。
比如:(包括但不限于,具体大伙可以去看看配置目录,种类非常之多)
input
组件的类型插件:stdin
:从标准输入读取数据file
:从文件读取数据,如常见的日志文件kafka
:从Kafka读取数据beats(filebeat)
:从filebeat读取数据Socket
、HTTP
、Websocket
:从远端通信管道中获取数据
filter
组件的类型插件:grok
:通过正则表达式匹配模式来提取和转换数据(本质上是正则表达式,但是多封装了一层,使用起来更方便)date
:将日期和时间戳转换为可识别的日期格式mutate
:对数据执行各种转换操作,例如更改字段的值、删除字段或添加字段等ruby
:在Ruby代码中执行自定义过滤器逻辑
output
组件类型的插件:stdout
:将数据输出到标准输出file
:将数据写入文件elasticsearch
:将数据发送到Elasticsearch进行存储或索引kafka
:将数据写入Kafka进行存储或传输beats(filebeat)
:将数据发送到filebeat进行日志收集Socket
、HTTP
、Websocket
:将数据写入远端节点
一个简单的使用配置logstash-demo.conf
如下:
input {stdin { }
}filter {grok {match => { "message" => "%{COMBINEDAPACHELOG}" }}date {match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]}
}output {elasticsearch { hosts => ["localhost:9200"]} stdout { codec => rubydebug }
}
如上所示,每个配置部分可以包含一个或多个插件。例如,上面的配置中就指定了多个filter
插件,Logstash会按照它们在配置文件中出现的顺序进行处理。
接着,再次启动Logstash,指定我们写的配置文件:
#运行
.\bin\logstash -f logstash-demo.conf
2.5 使用示例
在这里,我们使用来两个简单的使用示例来演示一下Logstash是如何做数据采集的。
示例一:Logstash导入csv数据到ES
1)测试数据集下载:Logstash为我们提供了一些使用示例数据
上面是一个csv
表格,存储的是【电影】相关信息。电影id + 电影名 + 电影类型
。注意,【电影类型geners】有多个值,用|
分隔开
2)准备logstash-movie.conf
配置文件:
- 指定了
input
组件使用的插件类型file
,标明文件所在目录 - 指定了多个
filter
处理插件。比如,使用,
来分隔不同的字段;使用split
操作对电影类型gener
做分割 - 指定了
output
组件,将转换后的数据写入到ES中
input {file {path => "/home/es/logstash-7.17.3/dataset/movies.csv"start_position => "beginning"sincedb_path => "/dev/null"}
}
filter {csv {separator => ","columns => ["id","content","genre"]}mutate {split => { "genre" => "|" }remove_field => ["path", "host","@timestamp","message"]}mutate {split => ["content", "("]add_field => { "title" => "%{[content][0]}"}add_field => { "year" => "%{[content][1]}"}}mutate {convert => {"year" => "integer"}strip => ["title"]remove_field => ["path", "host","@timestamp","message","content"]}}
output {elasticsearch {hosts => "http://localhost:9200"index => "movies"document_id => "%{id}"user => "elastic"password => "123456"}stdout {}
}
3)运行Logstash
.\bin\logstash -f logstash-movie.conf
示例二:同步数据库数据到Elasticsearch
需求:将数据库中的数据同步到ES,借助ES的全文搜索,提高搜索速度
- 需要把新增用户信息同步到Elasticsearch中
- 用户信息Update后,需要能被更新到Elasticsearch
- 支持增量更新
- 用户注销后,不能被ES所搜索到
实现思路:借助JDBC Input Plugin将数据从数据库读到Logstash
实现步骤如下:
1)拷贝jdbc依赖到logstash-7.17.3/drivers目录下
2)准备mysql-demo.conf
配置文件
input {jdbc {jdbc_driver_library => "/home/es/logstash-7.17.3/drivers/mysql-connector-java-5.1.49.jar"jdbc_driver_class => "com.mysql.jdbc.Driver"jdbc_connection_string => "jdbc:mysql://localhost:3306/test?useSSL=false"jdbc_user => "root"jdbc_password => "123456"#启用追踪,如果为true,则需要指定tracking_columnuse_column_value => true#指定追踪的字段,tracking_column => "last_updated"#追踪字段的类型,目前只有数字(numeric)和时间类型(timestamp),默认是数字类型tracking_column_type => "numeric"#记录最后一次运行的结果record_last_run => true#上面运行结果的保存位置last_run_metadata_path => "jdbc-position.txt"statement => "SELECT * FROM user where last_updated >:sql_last_value;"schedule => " * * * * * *"}
}
output {elasticsearch {document_id => "%{id}"document_type => "_doc"index => "users"hosts => ["http://localhost:9200"]user => "elastic"password => "123456"}stdout{codec => rubydebug}
}
3)运行logstash
.\bin\logstash -f mysql-demo.conf
另外,我们还需要额外在localhost:3306
下准备一个数据库test
,库中有表user
,关键字段:last_updated
#user表
CREATE TABLE `user` (`id` int NOT NULL AUTO_INCREMENT,`name` varchar(50) DEFAULT NULL,`address` varchar(50) DEFAULT NULL,`last_updated` bigint DEFAULT NULL,`is_deleted` int DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 ;#插入数据
INSERT INTO user(name,address,last_updated,is_deleted) VALUES("张三","广州天河",unix_timestamp(NOW()),0);
当我们插入数据后,就会在Logstash的控制台中看到如下数据:
甚至,当我们更新的时候update user set address="广州白云山",last_updated=unix_timestamp(NOW()) where name="张三";
也会看到Logstash的变化
三、Beats介绍
轻量型数据采集器文档地址
什么是Beats?Beats是数据采集器,是一组轻量级采集程序的统称。包括filebeat、metricbeat、packetbeat
和winlogbeat
等。它们是轻量级、快速且可靠的日志和指标采集解决方案,用于替代Logstash Forwarder
。Beats可以轻松地与Logstash和Elasticsearch进行对接,以便进行集中存储和索引。其中,filebeat是最常用的一个,用于日志采集。
Beats 是一个免费且开放的平台,集合了多种单一用途的数据采集器。它们可以从成百上千或成千上万台机器和系统向 Logstash 或 Elasticsearch 发送数据。
3.1 FileBeat简介
FileBeat专门用于转发和收集日志数据的轻量级采集工具。它可以作为代理安装在服务器上,FileBeat监视指定路径的日志文件,收集日志数据,并将收集到的日志转发到Elasticsearch或者Logstash。
3.1 FileBeat的工作原理
启动FileBeat时,会启动一个或者多个输入(Input)Processor
,这些Input监控指定的日志数据位置。FileBeat会针对每一个文件启动一个Harvester(收割机)
。Harvester
读取每一个文件的日志,将新的日志发送到libbeat
,libbeat
将数据收集到一起,并将数据发送给输出(Output)。
3.3 Logstash vs FileBeat
Logstash跟FileBeat有很多相似的地方,都是做数据采集并处理的,也都有输出组件。它们的联系与区别如下:
- Logstash是在jvm上运行的,资源消耗比较大;而FileBeat是基于golang编写的,功能较少但资源消耗也比较小,更轻量级
运行位置不同
- Logstash 和Filebeat都具有日志收集功能,Filebeat更轻量,占用资源更少
- Logstash 具有Filter功能,能过滤分析日志
Logstash功能更强大
- 一般结构都是Filebeat采集日志,然后发送到消息队列、Redis、MQ中,然后Logstash去获取,利用Filter功能过滤分析,然后存储到Elasticsearch中
这里是不是有点怪怪的?命名Logstash已经有数据采集攻能了,为什么还要个FileBeat呢?答案只有一个:前者在采集方面更高效!
- FileBeat和Logstash配合,实现背压机制。当将数据发送到Logstash或 Elasticsearch时,Filebeat使用背压敏感协议,以应对更多的数据量。如果Logstash正在忙于处理数据,则会告诉Filebeat 减慢读取速度。一旦拥堵得到解决,Filebeat就会恢复到原来的步伐并继续传输数据
3.4 Filebeat安装
1)下载并解压Filebeat
Filebeat下载地址
选择版本:7.17.3
2)编辑配置
修改filebeat.yml
以设置连接信息:结构上跟Logstash也很相似
output.elasticsearch:hosts: ["192.168.65.174:9200","192.168.65.192:9200","192.168.65.204:9200"]username: "elastic"password: "123456"
setup.kibana:host: "192.168.65.174:5601"
3)启用和配置数据收集模块
从安装目录中,运行:
# 查看可以模块列表
./filebeat modules list#启用nginx模块
./filebeat modules enable nginx
#如果需要更改nginx日志路径,修改modules.d/nginx.yml
- module: nginxaccess:var.paths: ["/var/log/nginx/access.log*"]#启用 Logstash 模块
./filebeat modules enable logstash
#在 modules.d/logstash.yml 文件中修改设置
- module: logstashlog:enabled: truevar.paths: ["/home/es/logstash-7.17.3/logs/*.log"]
4)启动 Filebeat
# setup命令加载Kibana仪表板。 如果仪表板已经设置,则忽略此命令。
./filebeat setup
# 启动Filebeat
./filebeat -e
四、经典ELK整合实战
案例1:采集tomcat服务器日志,使用FileBeats将日志发送到Logstash
一个简单的,使用FileBeats将日志发送到Logstash的使用案例。网上其实也大把了,我这里简单记录一下,方便我后续实战使用可以随时翻来看。
1)创建配置文件filebeat-logstash.yml
,配置FileBeats将数据发送到Logstash
chmod 644 filebeat-logstash.yml
vim filebeat-logstash.yml
#因为Tomcat的web log日志都是以IP地址开头的,所以我们需要修改下匹配字段。
# 不以ip地址开头的行追加到上一行
filebeat.inputs:
- type: logenabled: truepaths:- /home/es/apache-tomcat-8.5.33/logs/*access*.*multiline.pattern: '^\\d+\\.\\d+\\.\\d+\\.\\d+ 'multiline.negate: truemultiline.match: afteroutput.logstash:enabled: truehosts: ["192.168.65.204:5044"]
2)启动FileBeat,并指定使用指定的配置文件
./filebeat -e -c filebeat-logstash.yml
3)配置Logstash接收FileBeat收集的数据并打印到控制台
vim config/filebeat-console.conf# 配置从FileBeat接收数据input {beats {port => 5044}}output {stdout {codec => rubydebug}}
接着启动Logstash
# reload.automatic:修改配置文件时自动重新加载
bin/logstash -f config/filebeat-console.conf --config.reload.automatic
最后测试访问tomcat,logstash是否接收到了Filebeat传过来的tomcat日志
案例2:改进上面的,Logstash输出数据到Elasticsearch
如果我们需要将数据输出值ES而不是控制台的话,我们修改Logstash的output配置。
vim config/filebeat-elasticSearch.conf
input {beats {port => 5044}
}output {elasticsearch {hosts => ["http://localhost:9200"]user => "elastic"password => "123456"}stdout{codec => rubydebug}
}
再次启动Logstash
bin/logstash -f config/filebeat-elasticSearch.conf --config.reload.automatic
ES中会生成一个以logstash开头的索引,测试日志是否保存到了ES。
案例3:利用Logstash过滤器解析日志
从日志文件中收集到的数据包含了很多有效信息,比如IP、时间等,在Logstash中可以配置过滤器Filter对采集到的数据进行过滤处理,Logstash中有大量的插件可以供我们使用。
查看Logstash已经安装的插件
bin/logstash-plugin list
Grok插件
Grok是一种将非结构化日志解析为结构化的插件。这个工具非常适合用来解析系统日志、Web服务器日志、MySQL或者是任意其他的日志格式。
Grok是通过模式匹配的方式来识别日志中的数据,可以把Grok插件简单理解为升级版本的正则表达式。它拥有更多的模式,默认Logstash拥有120个模式。如果这些模式不满足我们解析日志的需求,我们可以直接使用正则表达式来进行匹配。grok模式的语法是:
%{SYNTAX:SEMANTIC}
SYNTAX(语法)
指的是Grok模式名称,SEMANTIC(语义)
是给模式匹配到的文本字段名。例如:
%{NUMBER:duration} %{IP:client}
duration表示:匹配一个数字,client表示匹配一个IP地址
默认在Grok中,所有匹配到的的数据类型都是字符串,如果要转换成int
类型(目前只支持int和float),可以这样:%{NUMBER:duration:int} %{IP:client}
在这里,我们使用Grok配置Logstash,修改如下:
vim config/filebeat-elasticSearch.conf
input {beats {port => 5044}
}filter {grok {match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }}
}output {elasticsearch {hosts => ["http://localhost:9200"]user => "elastic"password => "123456"}stdout{codec => rubydebug}
}
再次启动。经过Grok语法解析后,比如如下Tomcat日志:
192.168.65.103 - - [23/Jun/2022:22:37:23 +0800] "GET /docs/images/docs-stylesheet.css HTTP/1.1" 200 5780
解析后的字段:
除此以外,我们还可以过滤掉一些不需要的字段,还有将日期格式进行转换,我们可以使用Date插件来实现。经过一些列操作后,上述配置最终定格为:
vim config/filebeat-elasticSearch.conf
input {beats {port => 5044}
}filter {grok {match => { "message" => "%{IP:ip} - - \[%{HTTPDATE:date}\] \"%{WORD:method} %{PATH:uri} %{DATA:protocol}\" %{INT:status:int} %{INT:length:int}" }}mutate {enable_metric => "false"remove_field => ["message", "log", "tags", "input", "agent", "host", "ecs", "@version"]}date {match => ["date","dd/MMM/yyyy:HH:mm:ss Z","yyyy-MM-dd HH:mm:ss"]target => "date"}
}output {elasticsearch {hosts => ["http://localhost:9200"]user => "elastic"password => "123456"}stdout{codec => rubydebug}
}
案例4:输出到Elasticsearch指定索引
index来指定索引名称,默认输出的index名称为:logstash-%{+yyyy.MM.dd}。但注意,要在index中使用时间格式化,filter的输出必须包含 @timestamp字段,否则将无法解析日期。继续修改上述案例的Logstash配置:
vim config/filebeat-filter-es.confinput {beats {port => 5044}
}filter {grok {match => { "message" => "%{IP:ip} - - \[%{HTTPDATE:date}\] \"%{WORD:method} %{PATH:uri} %{DATA:protocol}\" %{INT:status:int} %{INT:length:int}" }}mutate {enable_metric => "false"remove_field => ["message", "log", "tags", "input", "agent", "host", "ecs", "@version"]}date {match => ["date","dd/MMM/yyyy:HH:mm:ss Z","yyyy-MM-dd HH:mm:ss"]target => "date"}
}output {stdout {codec => rubydebug
}
elasticsearch {index => "tomcat_web_log_%{+YYYY-MM}"hosts => ["http://localhost:9200"]user => "elastic"password => "123456"}
}
学习总结
- 学习了Logstash、Beats等日志搜集组件
- 学习了ELK架构及其简单应用