EFK实战二 - 日志集成

前言

在EFK基础架构中，咱们须要在客户端部署Filebeat，经过Filebeat将日志收集并传到LogStash中。在LogStash中对日志进行解析后再将日志传输到ElasticSearch中，最后经过Kibana查看日志。

上文EFK实战一 - 基础环境搭建已经搭建好了EFK的基础环境，本文咱们经过真实案例打通三者之间的数据传输以及解决EFK在使用过程当中的一些常见问题。

首先看一下实际的业务日志

2020-01-09 10:03:26,719 INFO ========GetCostCenter Start===============
2020-01-09 10:03:44,267 WARN 成本中心编码少于10位！{"deptId":"D000004345","companyCode":"01"}
2020-01-09 10:22:37,193 ERROR java.lang.IllegalStateException: SessionImpl[abcpI7fK-WYnW4nzXrv7w,]: can't call getAttribute() when session is no longer valid.
    at com.caucho.server.session.SessionImpl.getAttribute(SessionImpl.java:283)
    at weaver.filter.PFixFilter.doFilter(PFixFilter.java:73)
    at com.caucho.server.dispatch.FilterFilterChain.doFilter(FilterFilterChain.java:87)
    at weaver.filter.MonitorXFixIPFilter.doFilter(MonitorXFixIPFilter.java:30)
    at weaver.filter.MonitorForbiddenUrlFilter.doFilter(MonitorForbiddenUrlFilter.java:133)复制代码

日志组成格式为：时间 日志级别 日志详情主要任务就是将这段日志正常写入EFK中。

filebeat安装配置

• 下载filebeat7.5.1
• 将下载后的文件上传至服务器并解压
  tar -zxvf filebeat-7.5.1-linux-x86_64.tar.gz
• 修改filebeat.yml,

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /app/weaver/Resin/log/xxx.log复制代码

此段配置日志输入，指定日志存储路径

output.logstash:
  # The Logstash hosts
  hosts: ["172.31.0.207:5044"]复制代码

此段配置日志输出，指定Logstash存储路径

• 启动filebeat
  ./filebeat -e -c filebeat.yml
  若是须要静默启动，则使用nohup ./filebeat -e -c filebeat.yml & 命令启动便可

logstash配置

logstash的配置主要分为三段input，filter，output。
input用于指定输入，主要是开放端口给Filebeat用于接收日志
filter用于指定过滤，对日志内容进行解析过滤。
output用于指定输出，直接配置ES的地址便可

input {
  beats {
    port => 5044
  }
}

output {
  elasticsearch {
    hosts => ["http://172.31.0.127:9200"]
    index => "myindex-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "xxxxxx"
  }
}复制代码

咱们配置好logstash后经过命令重启logstash
docker-compose -f elk.yml restart logstash

通过上述两步配置后应用程序往日志文件写入日志，filebeat会将日志写入logstash。在kibana查看写入的日志结果以下：

日志显示有2个问题：

• 因为错误日志堆栈信息有多行，在kibana中展现成了多行，数据查看很乱。须要将堆栈异常整理成一行显示。
• 须要对日志进行解析，拆成“时间 日志级别 日志详情”的显示格式。

优化升级

• 在filebeat中设置合并行
  filebeat默认是行传输的，可是咱们的日志确定是多行一个日志，咱们要把多行合并到一块儿就要找到日志的规律。好比咱们的日志格式全都是以时间格式开头，因此咱们在filebeat中filebeat.inputs区域添加以下几行配置

# 以日期做为前缀
  multiline.pattern: ^\d{4}-\d{1,2}-\d{1,2}
  # 开启多行合并
  multiline.negate: true
  # 合并到上一行以后
  multiline.match: after复制代码

• 在logstash中设置对日志的解析
  将日志解析成“时间 日志级别 日志详情”的展现格式，因此咱们须要在logstash配置文件中添加filter段

filter {
    grok{
        match => {
            "message" => "(?<date>\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}),\d{3} %{LOGLEVEL:loglevel} (?<des>.*)"
        }
    }
}复制代码

这里主要是使用grok语法对日志进行解析，经过正则表达式对日志进行过滤。你们能够经过kibana里的grok调试工具进行调试

配置完成后咱们从新打开kibana Discover界面查看日志，符合预期，完美！

常见问题

kibana 乱码

这个主要缘由仍是客户端日志文件格式有问题，你们能够经过file xxx.log查看日志文件的编码格式，若是是ISO8859的编码基本都会乱码，咱们能够在filebeat配置文件中经过encoding指定日志编码进行传输。

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /app/weaver/Resin/log/xxx.log
  encoding: GB2312复制代码

kibana 提取字段出错

如上所示，打开kibana Discover面板时出现此异常，你们只要删除ES中的.kibana_1索引而后从新访问Kibana便可。

查看周围文件

咱们在终端查看日志某关键字时通常会查上下文信息便于排查问题，如常常用到的指令cat xxx.log | grep -C50 keyword，那么在Kibana中如何实现这功能呢。

在Kibana中搜索关键字，而后找到具体日志记录，点击左边向下箭头，而后再点击“查看周围文档”便可实现。

动态索引

咱们日志平台可能须要对接多个业务系统，须要根据业务系统创建不一样的索引。

• 在filebeat中给日志打上标记

- type: log
  ......
  fields:
    logType: oabusiness复制代码

• 在logstash中根据标记生成索引

input {
  beats {
    port => 5044
  }
}
filter {
  if [fields][logType] == "oabusiness" {
        grok{
            match => {
                "message" => "(?<date>\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}),\d{3} %{LOGLEVEL:loglevel} (?<des>.*)"
            }
        }
    }
}
output {
    elasticsearch {
        hosts => ["http://172.31.0.207:9200"]
        index => "%{[fields][logType]}-%{+YYYY.MM.dd}"
        user => "elastic"
        password => "elastic"
    }
}复制代码

好了，各位朋友们，本期的内容到此就所有结束啦，能看到这里的同窗都是优秀的同窗，下一个升职加薪的就是你了！若是以为这篇文章对你有所帮助的话请扫描下面二维码加个关注。"转发" 加 "在看"，养成好习惯！我们下期再见！

欢迎扫码关注微信公众号或 我的博客