filebeat收集日志输出到logstash然后合并输出到同名文件(7.7.1)

文章来源原创   作者:临窗旋墨   发布时间:2020-09-30   阅读:1403   标签:日志

[toc]

filebeat收集日志输出到logstash然后合并输出到同名文件(7.7.1)

Vic.xu 20200930

目标:多台集群机器日志收集到一起,且同名输出,能记录来源机器

  1. 多台机器输出的日志名是相同的,比如机器A 输出的日志包括err.logsql.logcommon.log
  2. 希望收集多台机器的日志,输入到同名日志文件,比如把A机器和B机器的err.logsql.logcommon.log收集到C机器的err.logsql.logcommon.log
  3. 在日志中标识来自哪一台机器;
  4. 多行日志需要合并,比如一个堆栈信息;

前言:项目有点老了(着急的话,直接调到最后看结论吧)

​ 一说分布式日志收集分析系统,网上都有一大推的关于ELK的文章。 不过咱的项目比较老了,只是最近才真正做了集群部署,所以才会涉及到分布式的日志收集。

​ 不过,项目中暂时也不会做日志分析之类的东西。所以就准备先弄一个filebeat收集日志到logstash,然后通过logstash定向输出到文件即可。加上之前因为开发环境和测试环境的隔离,所以写了个实时页面查看日志的小项目,也是勉强够用了。

版本:(因为之前使用过ElasticSearch的版本是7.7.1)

  • Filebeat 7.7.1
  • Logstash 7.7.1

一 为什么不只使用Logstash,而是额外引入了Filebeat

  1. 因为logstashjvm跑的,资源消耗比较大,
  2. filebeat更轻量,占用资源更少
  3. 一般结构都是filebeat采集日志,然后发送到消息队列,rediskafaka。然后logstash去获取,利用filter功能过滤分析,然后存储到elasticsearch

二 基本思路 Filebeat + Logstash

  1. 在各个需要收集日志的机器上FileBeat , 配置输出到远程的Logstash

  2. 在接收日志文件的机器上安装Logstash,收集各个Filebeat发送来到日志,输出到文件

三 windows下简单测试Filebeat

  1. 官网下载windows版本

  2. 修改filebeat.yml,具体参考官网

    1. filebeat.inputs:
    2. - type: log
    3. enabled: true
    4. fields:
    5. name: vic_first
    6. paths:
    7. # - /var/log/*.log
    8. - E:/space/idea/blog/logback/*.log
    9. output.file:
    10. path: d:/filebeat
    11. filename: filebeat_vic.log
    12. codec.format:
    13. string: '%{[fields][name]}%{[@timestamp]} %{[message]}'
    14. rotate_every_kb: 10240
    15. # 是否开启多行合并
    16. multiline.type: pattern
    17. #multiline.pattern: '^\['
    18. multiline.pattern: '^vic-blog'
    19. multiline.negate: false
    20. multiline.match: after
    21. #============================= Filebeat modules ===============================
    22. filebeat.config.modules:
    23. path: ${path.config}/modules.d/*.yml
    24. reload.enabled: false
    25. processors:
    26. - add_host_metadata: ~
    27. - add_cloud_metadata: ~
    28. - add_docker_metadata: ~
    29. - add_kubernetes_metadata: ~
    • 如上:
      1. 配置了需要收集的日志文件的位置。
      2. fields 下是携带的额外的数据,一般作为标识使用
      3. 配置了日志的输出的文件
      4. 配置合并多行:因为默认是读取一行就发送,但是其实有的时候多行才算一个日志,比如堆栈信息等
  3. 启动Filebeat,然后测试写入日志,会被写入到输出位置

    .\filebeat.exe -e -c filebeat.yml

  4. 其他

    1. 输出到es

      1. output.elasticsearch:
      2. # Array of hosts to connect to.
      3. hosts: ["localhost:9200"]
      4. # Protocol - either `http` (default) or `https`.
      5. #protocol: "https"
      6. # Authentication credentials - either API key or username/password.
      7. #api_key: "id:api_key"
      8. #username: "elastic"
      9. #password: "changeme"
    2. 输出到logstash

      1. output.logstash:
      2. # The Logstash hosts
      3. hosts: ["localhost:5044"]

四 windows下简单测试Logstash

  1. 官网下载zip解压,

  2. cofig下复制logstash.conf,然后配置成控制台输入输出:

    1. input {
    2. beats {
    3. port => 5044
    4. }
    5. }
    6. filter {
    7. ruby {
    8. code => "
    9. path = event.get('log')['file']['path']
    10. puts format('path = %<path>s', path: path)
    11. if (!path.nil?) && (!path.empty?)
    12. event.set('filename', path.split('/')[-1])
    13. end
    14. "
    15. }
    16. }
    17. output {
    18. file {
    19. # path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{+HH}-vic.log"
    20. path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{filename}"
    21. codec => line { format => "%{filename} %{[fields][name]}: %{message}"}
    22. }
    23. stdout {
    24. codec => rubydebug
    25. }
    26. }
  3. 启动Logstash

    bin\logstash -f config\logstash.conf

    • 启动成功后可访问localhost:9600,看到版本等信息,则表示启动成功
    • 测试:在启动控制台输入信息, 可看到相关输出信息

一些问题:

  1. 在logstash中获取Filebeat中的日志原文件名,然后输出到同名文件:通过filter实现

    如下,获取输入的日志文件的文件名filename,通过%{filename}获取

    1. filter {
    2. ruby {
    3. code => "
    4. path = event.get('log')['file']['path']
    5. puts format('path = %<path>s', path: path)
    6. if (!path.nil?) && (!path.empty?)
    7. event.set('filename', path.split('/')[-1])
    8. end
    9. "
    10. }
    11. }
  1. filebeat携带额外的信息到,标志来自哪台机器,通过fields携带(如下,携带的key是name,value是server1)

    1. filebeat.inputs:
    2. - type: log
    3. enabled: true
    4. fields:
    5. name: server1
  2. logstash获取filebeat携带的额外的信息

    1. %{[fields][name]}
  3. filebeat 不发送最后一行

    Filebeat使用换行符来检测事件的结束。 如果将行逐渐添加到正在采集的文件中,则在最后一行之后需要换行符,否则Filebeat将不会读取文件的最后一行。

五 总结: linux下的启动脚本和最终配置文件

linux下的安装和window下没什么区别,logstash需要确保安装了jdk1.8+;

启动也是一样的, 只是本人比较懒, 所以写了两个启动脚本start.sh ;运行时带参数的话则输出日志到catalina.out,不带参数则不输出日志

linux下的filebeat的启动脚本

  1. #!/bin/bash
  2. log="/dev/null"
  3. # if input paramter is not null ,then output log to catalina.out. else to /dev/null
  4. if [ $1 != '' ];then
  5. log="catalina.out"
  6. fi
  7. nohup ./filebeat -e -c filebeat.yml >$log 2>&1 &
  8. echo "start filebeat.......... ----> $log"
linux下的logstash的启动脚本
  1. #!/bin/bash
  2. log="/dev/null"
  3. # if input paramter is not null ,then output log to catalina.out. else to /dev/null
  4. if [ $1 != '' ];then
  5. log="catalina.out"
  6. fi
  7. nohup bin/./logstash -f config/logstash.conf >$log 2>&1 &
  8. echo "start logstash.......... ----> $log"
filebeat.yaml最终配置文件(见注释)
  1. # 输入日志
  2. filebeat.inputs:
  3. - type: log
  4. enabled: true
  5. # 携带的额外的数据,标识来自哪个机器
  6. fields:
  7. name: wsl-ubuntu
  8. # 日志的所在位置
  9. paths:
  10. - /data/share/logs/input/*.log
  11. - /data/share/logs/input2/*.log
  12. #=========================== 是否开启多行合并================================
  13. multiline.type: pattern
  14. #多行合并的正则,以[ 开头
  15. multiline.pattern: '^\['
  16. # 是否不合并多行合并: 负负得正 表示开启
  17. multiline.negate: false
  18. multiline.match: after
  19. #==========================输出日志到logstash==============================
  20. output.logstash:
  21. # The Logstash hosts
  22. hosts: ["localhost:5044"]
logstash.conf最终配置文件(见注释)
  1. #======从filebeat获取日志==============================
  2. input {
  3. beats {
  4. port => 5044
  5. }
  6. }
  7. #=====解析出来源日志的文件名===================================
  8. filter {
  9. ruby {
  10. code => "
  11. path = event.get('log')['file']['path']
  12. puts format('path = %<path>s', path: path)
  13. if (!path.nil?) && (!path.empty?)
  14. event.set('filename', path.split('/')[-1])
  15. end
  16. "
  17. }
  18. }
  19. #======输出日志==============================
  20. output {
  21. # 输出日志到文件,在日志的开头打印filebeat中携带的name属性,标识来自哪个文件
  22. file {
  23. path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{filename}"
  24. codec => line { format => "%{[fields][name]}: %{message}"}
  25. }
  26. # 输出日志到控制台 可删除掉
  27. stdout {
  28. codec => rubydebug
  29. }
  30. }

文章来源:临窗旋墨的博客,转载注明出处。

时间: 20200930


发表评论

目录