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


    内容预知

     1.ELFK集群部署(Filebeat+ELK)

    步骤一:安装 Filebeat(在apache节点操作) 

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

    步骤三: 新建 Logstash 配置文件

    filebeat 功能的引入测试 

     (1)创建两个访问页面

    (2)启动filebeat和logstash 

    (3)进行访问测试 

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

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

    2.  Logstash的过滤模块

    2.1 Logstash配置文件中的模块

    (1)input {}

    (2)filter {}

    (3)ourput {}

    2.2 Filter(过滤模块)中的插件 

    grok插件(通过grok插件实现对字段的分割,使用通配符)

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

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

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

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

     2.4 数据切片的实战演练

    切片前的操作 

    对apche服务进行访问 

     进行切片分离

    再次进行访问测试 

     3. filebeat中multiline 插件的引入

    3.1  安装 multiline 插件

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

    3.3 多行被拆分

    4.  mutate 插件


     1.ELFK集群部署(Filebeat+ELK)

     

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

    服务器类型系统IP需要安装的组件硬件方面
    Apache节点CentOS7.4(64 位)192.168.73.107Logstash、Apache,2核4G
    node01节点CentOS7.4(64 位) 192.168.73.105Elasticsearch 、Kibana2核4G
    node02节点CentOS7.4(64 位) 192.168.73.106Elasticsearch,filebeat2核4G

    步骤一:安装 Filebeat(在apache节点操作) 

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

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

    1.  cd /usr/local/filebeat
    2.  ​
    3.  vim filebeat.yml
    4.  filebeat.prospectors:
    5.  - type: log         #指定 log 类型,从日志文件中读取消息
    6.   enabled: true
    7. paths:
    8. #指定监控的日志文件
    9. - /var/log/httpd/access_log
    10. tags: ["filebeat"] #设置索引标签
    11. fields: #可以使用 fields 配置选项设置一些参数字段添加到 output
    12. service_name: apache
    13. log_type: access
    14. from: 192.168.73.107
    15.  ​
    16.  --------------Elasticsearch output-------------------
    17.  (全部注释掉)
    18.  ​
    19.  ----------------Logstash output---------------------
    20.  output.logstash:
    21.   hosts: ["192.168.73.107:5044"]      #指定 logstash 的 IP 和端口
    22.  ​

    步骤三: 新建 Logstash 配置文件

    1. cd /etc/logstash/conf.d
    2. cp system.conf filebeat.conf ​
    3. vim filebeat.conf
    4. input {
    5. beats { port => "5044" }
    6. }
    7. output {
    8. elasticsearch {
    9. hosts => ["192.168.73.105:9200","192.168.73.106:9200"]
    10. index =>"%{[fields][service_name]}-%{+YYYY.MM.dd}"
    11. }
    12. stdout {
    13. codec => rubydebug
    14. }
    15. }
    16.  ​

    filebeat 功能的引入测试 

     (1)创建两个访问页面

    1. echo "

      this is filebeat test

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

    (2)启动filebeat和logstash 

    1.  #启动 filebeat
    2.  nohup ./filebeat -e -c filebeat.yml > filebeat.out &
    3.  #-e:输出到标准输出,禁用syslog/文件输出
    4.  #-c:指定配置文件
    5.  #nohup:在系统后台不挂断地运行命令,退出终端不会影响程序的运行
    6.  #启动 logstash
    7. cd /etc/logstash/conf.d
    8. logstash -f filebeat.conf

     

    (3)进行访问测试 

     

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

     

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

     

     

     

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

     

     

    2.  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 (?.+")[\s](?<status_code>\d+) -\s"-"\s(?<other_info>.+)

     

     

     

     2.4 数据切片的实战演练

    切片前的操作 

    1. vim apache_log.conf
    2. input {
    3. file{
    4. path => "/etc/httpd/logs/access_log"
    5. type => "access"
    6. start_position => "beginning"
    7. }
    8. }
    9. output {
    10. if [type] == "access" {
    11. elasticsearch {
    12. hosts => ["192.168.73.105:9200","192.168.73.105:9200"]
    13. index => "apache_access-%{+YYYY.MM.dd}"
    14. }
    15. }
    16. }
    17. logstash -f apache_log.conf

     

     

    对apche服务进行访问 

     

     

     

     

     

     进行切片分离

     

    修改logstash的控制conf 

    1. vim apache_log.conf
    2. input {
    3. file{
    4. path => "/etc/httpd/logs/access_log"
    5. type => "access"
    6. start_position => "beginning"
    7. }
    8. }
    9. filter {
    10. grok {
    11. match => ["message","(?%{IPV4}|%{IPV6})[\s-]+\[(?.+)] \"GET (?<http_method>.+\")[\s](?\d+) -\s\"-\"\s(?.+)"]
    12. }
    13. }
    14. output {
    15. if [type] == "access" {
    16. elasticsearch {
    17. hosts => ["192.168.73.105:9200","192.168.73.105:9200"]
    18. index => "apache_access-%{+YYYY.MM.dd}"
    19. }
    20. }
    21. }
    22. logstash -f apache_log.conf

     

     

    再次进行访问测试 

     

     

     

     3. filebeat中multiline 插件的引入

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

     

    3.1  安装 multiline 插件

    1. 在线安装插件
    2. cd /usr/share/logstash
    3. bin/logstash-plugin install logstash-filter-multiline
    4. 离线安装插件
    5. 先在有网的机器上在线安装插件,然后打包,拷贝到服务器,执行安装命令
    6. bin/logstash-plugin install file:///usr/share/logstash/logstash-offline-plugins-6.7.2.zip
    7. 检查下插件是否安装成功,可以执行以下命令查看插件列表
    8. bin/logstash-plugin list

     

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

     

    日志合并的过程: 

    • 第一步:每一条日志的第一行开头都是一个时间,可以用时间的正则表达式匹配到第一行。

    • 第二步:然后将后面每一行的日志与第一行合并。

    • 第三步:当遇到某一行的开头是可以匹配正则表达式的时间的,就停止第一条日志的合并,开始合并第二条日志。

    • 第四步:重复第二步和第三步

     

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

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

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

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

     

    3.3 多行被拆分

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

     

     

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

     

    1. multiline.type: pattern
    2. multiline.pattern: '^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}.\d{3}'
    3. multiline.negate: true
    4. multiline.match: after
    5. 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 进程卡死

     

    4.  mutate 插件

     当我们将日志解析出来后,Logstash 自身会传一些不相关的字段到 ES 中,这些字段对我们排查线上问题帮助不大。可以直接剔除掉。这里我们就要用到 mutate 插件了。它可以对字段进行转换,剔除等。

     

     

    一般会把把 log.offset 字段去掉,这个字段可能会包含很多无意义内容。

     

     

    1. filter{
    2. mutate {
    3.      remove_field => ["host" "[log][offset]"]
    4. }

     

     

     

     

  • 相关阅读:
    orb-slam2 从单目开始的简单学习(7):Optimizer
    IDEA 实用快捷键
    uni-app支持vue3(小程序、h5、app等)多端开发
    安装Cmake&ffmpeg&libssh
    上传本地包到私有maven仓库
    HIS医疗项目
    我也不想学之PHP系列(2)
    数据库-索引
    【CPP_Primer_Plus】C++ IDE推荐
    python绘制ROC曲线
  • 原文地址:https://blog.csdn.net/qq_62462797/article/details/127822723