之前说要搭建帆软日志的解析，最近因为疫情原因终于有时间去做这件事情了，之前的日志查询还在使用tail+grep也确实是略显原始
节点是整了3台服务器装的3节点elasticsearch7.5+Filebeat+kibana进行的帆软日志解析
elastic集群的搭建是参考的https://www.jianshu.com/p/4bf5a8b743d2的文章多说一句，这没有使用ELK，直接使用filebeat进行日志采集监听后，使用es中的数据预处理IngestNode/Pipeline（https://www.jianshu.com/p/e8e0639c98f6）这里我们直接开始说明filebeat的配置以及如何设置es索引以及mapping，配置pipeline进行日志解析，提取关键词信息。
配置filebeat
目前我filebeat的配置是直接安装与节点机上，读取fr日志发送到es中进行的日志解析，帆软有自己的决策页面，但看不到查询但参数以及报错信息我觉得是没点屁用的。
#=========================== Filebeat inputs =============================
filebeat.inputs:
- type: log
enabled: true
paths:
- /嘿嘿嘿/data1/frLogs/FineReport.log
fields:
source: finereport
multiline.negate: true
multiline.match: after
#==================== Elasticsearch template setting ==========================
setup.template.name: "finereport"
setup.template.pattern: "finereport*"
setup.ilm.enabled: false
#setup.template.overwrite: true
setup.template.enabled: true
#
##============================== Kibana =====================================
setup.kibana:
host: "yourkibana:15021"
#================================ Outputs =====================================
output.elasticsearch:
enabled: true
hosts: ["http://yourelasticsearch:15020"]
pipelines:
- pipeline: fanruanlog
when.equals:
fields.source: finereport
indices:
- index: "finereport-%{+yyyy.MM.dd}"
when.equals:
fields.source: "finereport"
整个配置文件就是设定filebeat读哪，写哪，使用什么索引模版，使用那个pipeline这里坑了我的参数是
setup.ilm.enabled: false //不加这个你就用于都会是默认索引
setup.template.overwrite: true //加了这个你提前生成的索引模版会被覆盖
setup.template.enabled: true
启动命令如下：
nohup ./filebeat -e -c filebeat.yml > filebeat.log &
有什么问题可以查看filebeat.log定向解决
配置pipeline
如果我们不配置pipeline，因为我们的日志是杂乱未经过格式化的内容，传递到es中经过分词后会默认的存储在message这个字段里，也就是说所有的信息都会存在一起，当然这里如果使用模糊匹配也能过获取到你想要的信息，使用pipeline可以进行日志的解析渲染，我查了些资料，这里使用gock进行日志解析http://coder55.com/article/43041
我们在kibana的页面里可以进行这部分的测试工作grok Debugger
gock有封装好一些可以直接使用的正则匹配
将我们要解析的日志放到样例子数据中，在grok模式那一行编写正则表达式，点击模拟生成结构化数据：
这里包含部分业务数据所以测试的样例数据这里无法给出来了
%{TIME:time} %{NOTSPACE:THREAD} %{LOGLEVEL:loglevel} \[%{WORD:logrole}\] .*Name : %{NOTSPACE:reportName}\].*USERNAME : %{NOTSPACE:user}[\d\D](?<parameter>.*)}.*[\d\D](?<errorInfo>.*)
一开始我写这部分发现网上的例子大部分都是按照顺序进行解析，但帆软的日志打的是在是太多了，有很多的数据不是我们关心的的数据（网上能找到很多解析nginx日志的），可以看到前部分解析是的%{TIME:time} %{NOTSPACE:THREAD} %{LOGLEVEL:loglevel} \[%{WORD:logrole}\] .*Name : %{NOTSPACE:reportName}\]这部分我是按照顺序写的正则解析
但是后部分就不是我想要的，这里面的.*是匹配所有的字符（不包含换行符）这样就可以用来跳过我们不想要的字段.*USERNAME : %{NOTSPACE:user}这一步就直接跳到操作人名称来，中间会跳过大量的无用数据，同理[\d\D]是用来匹配所有字符(?<errorInfo>.*)把最后所有的信息都匹配为errorinfo,只要耐心够，这部分用正则就能解析出来
生成pipeline
curl -XPUT 'http://yourEs:15020/_ingest/pipeline/fanruanlog' -H 'Content-Type: application/json' -d'
{
"description" : "fanruan log pipeline",
"processors": [
{
"grok": {
"field": "message",
"patterns": ["%{TIME:time} %{NOTSPACE:THREAD} %{LOGLEVEL:loglevel} \\[%{WORD:logrole}\\] .*Name : %{NOTSPACE:reportName}\\].*USERNAME : %{NOTSPACE:user}[\\d\\D](?<parameter>.*)}.*[\\d\\D](?<errorInfo>.*)"]
}
}
]
}
'
注意这里的patterns里面的字符要经过一次转义
创建索引模版
将你提取的查询字段在里面列出来创建查询索引
创建mapping，这里的索引就创建完成啦
到kibana里创建kibana的索引
然后我们就可以到查询页面看看效果
可以看到报表名称，查询时间，查询用户，查询参数，errorinfo就提取完毕了
小结
顺序有点问题，应该是先pipeline，创建索引，filebeat配置
帆软这里我们进行了改造，编写了异步导出逻辑，然后发现异步的导出使用帆软的日志打印工具打印不出来，就很尬了，这一块又必须要整，那下一期就写咋埋点监控把，目前巨量数据的导出问题还在解决处理中，异步导出也解决不了的话，得开始考虑把导出独立出来走调度处理重新封装了（ps最近测试了下clickhouse的性能，很让人惊喜）
