整合实战使用FileBeats将日志发送到Logstash配置Logstash接收FileBeat收集的数据并打印Logstash输出数据到Elasticsearch利用Logstash过滤器解析日志Grok插件Grok语法用法

输出到Elasticsearch指定索引

com.hs.single.controller.UserController

请求参数为1

[org.apache.ibatis.session.defaults.DefaultSqlSession4196ffc2]

was

[com.alibaba.druid.proxy.jdbc.ConnectionProxyImpl790df42b]

will

id,username,password,name,description,status,create_time,update_time

FROM

[org.apache.ibatis.session.defaults.DefaultSqlSession4196ffc2]使用FileBeats将日志发送到Logstash

创建配置文件filebeat-logstash.yml配置FileBeats将数据发送到Logstash

vim

filebeat-logstash.yml#因为Tomcat的web

log日志都是以IP地址开头的我们应用日志一般是以日期时间开头。

所以我们需要修改下匹配字段。

filebeat.inputs:

/root/application/nohup.outmultiline.pattern:

^\d{4}-\d{2}-\d{2}multiline.negate:

truemultiline.match:

false默认是false匹配pattern的行合并到上一行true不匹配pattern的行合并到上一行matchafter

chmod

filebeat-logstash.yml启动FileBeat并指定使用指定的配置文件

前台运行

/home/es/filebeat-7.17.3-linux-x86_64/filebeat-logstash.yml)因为安全原因不要其他用户写的权限去掉写的权限就可以了

Failed

backoff(async(tcp://192.168.75.65:5044)):

dial

refusedFileBeat将尝试建立与Logstash监听的IP和端口号进行连接。

但此时我们并没有开启并配置Logstash所以FileBeat是无法连接到Logstash的。

输入的使用5044端口接收输入输出是直接在控制台打印查看是否能正常接收数据

vim

config/filebeat-console.conf调用接口filebeat成功将数据发送至logstash中

如果我们需要将数据输出值ES而不是控制台的话我们修改Logstash的output配置。

vim

config/filebeat-elasticSearch.conf

input

[http://192.168.75.65:9200]user

elasticpassword

config/filebeat-elasticSearch.conf

[rooths-es-node1

config/filebeat-elasticSearch.confES中会生成一个以logstash开头的索引测试日志是否保存到了ES。

思考日志信息都保证在message字段中是否可以把日志进行解析一个个的字段例如时间、日志级别、哪个类打印的日志、日志具体内容。

在Logstash中可以配置过滤器Filter对采集到的数据进行过滤处理Logstash中有大量的插件可以供我们使用。

查看Logstash已经安装的插件,默认已经安装了grok插件

[rooths-es-node1

Grok是一种将非结构化日志解析为结构化的插件。

这个工具非常适合用来解析系统日志、Web服务器日志、MySQL或者是任意其他的日志格式。

https://www.elastic.co/guide/en/logstash/7.17/plugins-filters-grok.html

Grok语法

Grok是通过模式匹配的方式来识别日志中的数据,可以把Grok插件简单理解为升级版本的正则表达式。

它拥有更多的模式默认Logstash拥有120个模式。

如果这些模式不满足我们解析日志的需求我们可以直接使用正则表达式来进行匹配。

SYNTAX语法指的是Grok模式名称SEMANTIC语义是给模式匹配到的文本字段名。

注意这中间不要有空格

duration表示匹配一个数字client表示匹配一个IP地址。

%{NUMBER:duration}

%{IP:client}默认在Grok中所有匹配到的的数据类型都是字符串如果要转换成int类型目前只支持int和float可以这样

https://help.aliyun.com/document_detail/129387.html?scm20140722.184.2.173

比如tomacat日志

/docs/images/docs-stylesheet.css

HTTP/1.1

IP浏览器端IPdate请求的时间戳method请求方式GET/POSTuri请求的链接地址status服务器端响应状态length响应的数据长度

%{IP:ip}

com.hs.single.controller.UserController

请求参数为1%{TIMESTAMP_ISO8601:timestamp}

%{LOGLEVEL:level}

%{GREEDYDATA:logResult}修改Logstash配置文件

vim

config/filebeat-elasticSearch.confinput

{beats

因为我们的消息是保存在message字段中所以下面主要就是匹配message字段中的值

filter

--config.reload.automatic修改了配置文件能热加载

bin/logstash

config/filebeat-elasticSearch.conf

--config.reload.automatic使用mutate插件过滤掉不需要的字段

mutate

}如果要将日期格式进行转换我们可以使用Date插件来实现。

该插件专门用来解析字段中的日期官方说明文档https://www.elastic.co/guide/en/logstash/7.17/plugins-filters-date.html

用法如下

时分秒」格式。

默认字段经过date插件处理后会输出到timestamp字段所以我们可以通过修改target属性来重新定义输出字段。

date

index来指定索引名称默认输出的index名称为logstash-%{yyyy.MM.dd}。

但注意要在index中使用时间格式化filter的输出必须包含

output

[http://192.168.75.65:9200]index

elasticpassword

%{GREEDYDATA:logResult}}}mutate

{enable_metric

[http://192.168.75.65:9200]index

elasticpassword