分布式运用之企业级日志ELFK+logstash的过滤模块

一、ELFK集群部署（Filebeat+ELK）

在搭建ELK的基础上安装Filebeat服务，Filebeat服务可以布置在以下任意一台主机，本次实验将布置在apache服务器的节点上

步骤一：安装 Filebeat（在apache节点操作）

#上传软件包 filebeat-6.7.2-linux-x86_64.tar.gz 到/opt目录
tar zxvf filebeat-6.7.2-linux-x86_64.tar.gz
mv filebeat-6.7.2-linux-x86_64/ /usr/local/filebeat

步骤二：设置 filebeat 的主配置文件

 cd /usr/local/filebeatvim filebeat.ymlfilebeat.prospectors:- type: log         #指定 log 类型，从日志文件中读取消息enabled: truepaths:#指定监控的日志文件- /var/log/httpd/access_logtags: ["filebeat"]         #设置索引标签fields:                  #可以使用 fields 配置选项设置一些参数字段添加到 output 中service_name: apachelog_type: accessfrom: 192.168.73.107--------------Elasticsearch output-------------------(全部注释掉)----------------Logstash output---------------------output.logstash:hosts: ["192.168.73.107:5044"]      #指定 logstash 的 IP 和端口

步骤三：新建 Logstash 配置文件

cd /etc/logstash/conf.d
cp system.conf filebeat.conf 
vim filebeat.conf
input {beats { port => "5044" }}output {elasticsearch {hosts => ["192.168.73.105:9200","192.168.73.106:9200"]index =>"%{[fields][service_name]}-%{+YYYY.MM.dd}"}stdout {codec => rubydebug}
}

filebeat 功能的引入测试

1）创建两个访问页面

echo "<h1>this is filebeat test </h1>" >/var/www/html/test1.html
echo "h1> this is a test </h1>" >/var/www/html/test2.html

2)启动filebeat和logstash

#启动 filebeat
nohup ./filebeat -e -c filebeat.yml > filebeat.out &
#-e：输出到标准输出，禁用syslog/文件输出
#-c：指定配置文件
#nohup：在系统后台不挂断地运行命令，退出终端不会影响程序的运行

#启动 logstash
cd /etc/logstash/conf.d
logstash -f filebeat.conf

3）进行访问测试

此时：屏幕输出的日志以及发生改变（Logstash的新建配置文件中output模块中stdout）

4）kibana中创建于apache相关的索引，并且通过时序，可以查看到访问日志的信息

解决日志与kibana时间不一致的方法

二、 Logstash的过滤模块

2.1 Logstash配置文件中的模块

（1）input {}

指定输入流，通过file、beats、kafka、redis中获取数据

（2）filter {}

常用插件：

grok：对若干个大文本字段进行再分割，分割成一些小字段（?<字段名>正则表达式）字段名：正则表示匹配到的内容
date：对数据中的时间进行统一格式化
mutate：对一些无用的字段进行剔除，或增加字段
mutiline：对多行数据进行统一编排，多行合并和拆分

（3）ourput {}

elasticsearch stdout

2.2 Filter（过滤模块）中的插件

而对于 Logstash 的 Filter，这个才是 Logstash 最强大的地方。Filter 插件也非常多，我们常用到的 grok、date、mutate、mutiline 四个插件。

对于 filter 的各个插件执行流程，可以看下面这张图：

grok插件（通过grok插件实现对字段的分割，使用通配符）

这里就要用到 logstash 的 filter 中的 grok 插件。filebeat 发送给 logstash 的日志内容会放到message 字段里面，logstash 匹配这个 message 字段就可以了。

2.3 kibana 的DEV Tools 中 Grok Debugger工具的运用

在此我们以apache访问日志索引中的message字段为例，进行数据切片：

假设我们要进行的切片信息为messge中的

192.168.73.105 - - [13/Nov/2022:02:57:13 +0800] "GET /test1.html HTTP/1.1" 304 - "-" "Mozilla/5.0 (X11; Linux x86_64; rv:52.0) Gecko/20100101 Firefox/52.0"

数据切片的格式为：

匹配格式：(?<字段名>正则表达式)

切片案例一：分离访问用户的IP

(?<remote_addr>%{IPV6}|%{IPV4} )(?<other_info>.+)

切片案例二：以上面的为基础，将时间分离出来

(?<remote_addr>%{IPV4}|%{IPV6})[\s-]+\[(?<logTime>.+)](?<other_info>.+)

切片案例三：取出状态码和http方法

(?<remote_addr>%{IPV4}|%{IPV6})[\s-]+\[(?<logTime>.+)] "GET (?<http_method>.+")[\s](?<status_code>\d+) -\s"-"\s(?<other_info>.+)

2.4 数据切片的实战演练

切片前的操作

vim apache_log.conf
input {file{path => "/etc/httpd/logs/access_log"type => "access"start_position => "beginning"}
}output {if [type] == "access" {elasticsearch {hosts => ["192.168.73.105:9200","192.168.73.105:9200"]index => "apache_access-%{+YYYY.MM.dd}"}}
}logstash -f apache_log.conf

对apche服务进行访问

进行切片分离

修改logstash的控制conf

vim apache_log.conf
input {file{path => "/etc/httpd/logs/access_log"type => "access"start_position => "beginning"}
}
filter {grok {match => ["message","(?<remote_addr>%{IPV4}|%{IPV6})[\s-]+\[(?<logTime>.+)] \"GET (?<http_method>.+\")[\s](?<status_code>\d+) -\s\"-\"\s(?<other_info>.+)"]}}output {if [type] == "access" {elasticsearch {hosts => ["192.168.73.105:9200","192.168.73.105:9200"]index => "apache_access-%{+YYYY.MM.dd}"}}
}
logstash -f apache_log.conf

再次进行访问测试

三、filebeat中multiline 插件的引入

java错误日志一般都是一条日志很多行的，会把堆栈信息打印出来，当经过 logstash 解析后，每一行都会当做一条记录存放到 EslaticSearch中，那这种情况肯定是需要处理的。这里就需要使用 multiline 插件，对属于同一个条日志的记录进行拼接。

3.1 安装 multiline 插件

在线安装插件
cd /usr/share/logstash
bin/logstash-plugin install logstash-filter-multiline

离线安装插件
先在有网的机器上在线安装插件，然后打包，拷贝到服务器，执行安装命令
bin/logstash-plugin install file:///usr/share/logstash/logstash-offline-plugins-6.7.2.zip

检查下插件是否安装成功，可以执行以下命令查看插件列表
bin/logstash-plugin list

3.2 使用 multiline 插件进行日志合并

日志合并的过程：

第一步：每一条日志的第一行开头都是一个时间，可以用时间的正则表达式匹配到第一行。
第二步：然后将后面每一行的日志与第一行合并。
第三步：当遇到某一行的开头是可以匹配正则表达式的时间的，就停止第一条日志的合并，开始合并第二条日志。
第四步：重复第二步和第三步

filter {
multiline {
pattern => "^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}.\d{3}"
negate => true
what => "previous"
}
}

pattern: 用来匹配文本的表达式，也可以是grok表达式

what: 如果pattern匹配成功的话，那么匹配行是归属于上一个事件，还是归属于下一个事件。previous: 归属于上一个事件，向上合并。next: 归属于下一个事件，向下合并

negate:是否对 pattern 的结果取反。false: 不取反，是默认值。true: 取反。将多行事件扫描过程中的行匹配逻辑取反（如果pattern匹配失败，则认为当前行是多行事件的组成部分）

3.3 多行被拆分

Java 堆栈日志太长了，有 100 多行，被拆分了两部分，一部分被合并到了原来的那一条日志中，另外一部分被合并到了不相关的日志中。

为了解决这个问题，可以通过配置 filebeat 的 multiline 插件来截断日志。至于为什么不用 logstash 的 multiline 插件呢？因为在 filter 中使用 multiline 没有截断的配置项。filebeat 的 multiline 配置项如下：

multiline.type: pattern
multiline.pattern: '^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}.\d{3}'
multiline.negate: true
multiline.match: after
multiline.max_lines: 50

配置项说明：

multiline.pattern：希望匹配到的结果（正则表达式）
multiline.negate：值为 true 或 false。使用 false 代表匹配到的行合并到上一行；使用 true 代表不匹配的行合并到上一行
multiline.match：值为 after 或 before。after 代表合并到上一行的末尾；before 代表合并到下一行的开头
multiline.max_lines：合并的最大行数，默认 500
multiline.timeout：一次合并事件的超时时间，默认为 5s，防止合并消耗太多时间导致 filebeat 进程卡死