ELK日志分析系统安装和部署(4)

3.3 logstash配置
##接下来是重点，也是ELK的难点吧，就是将日志进行分析，使用正则匹配过滤日志中想要的内容。

[root@Elk_Server logstash]# cat /data/logs/test.log 120.79.189.51 - - [10/Jun/2018:12:58:12 +0800] "POST /wp-cron.php?doing_wp_cron=1528606692.3628709316253662109375 HTTP/1.1" 499 0 "-" "WordPress/4.8; "

##这是日志中的一行，logstash的正则匹配默认是匹配一行，以回车符作为分隔符。
##当然如果日志是json格式的那么就没有需要处理直接输出即可。

#我们到config目录中新建一个logstash.conf配置。 [root@Elk_Server logstash]# vim /etc/logstash/conf.d/logstash.conf input { file { path => "/data/logs/test.log" type => "test.log" start_position => "beginning" } } filter { grok { match => { "message" => "(?<ip_addr>\d+?\.\d+?\.\d+?\.\d+?)\s-\s-\s\[(?<log_date>\d+?/\w+?/\d+?):(?<log_time>\d+?:\d+?:\d+?)\s+" } } } output { stdout {codec => rubydebug} }

##这是logstash的日志分析文件，总共分为三个大部分，input、flter、output，其中input是输入日志，这里选择的是file文件，path是文件的路径，type是文件的类型，这个可以自定义，主要用来区分每个日志，start_position是设定为beginning是指从文件的开头开始读取。
##flter是处理日志的部分，使用grok这个强大的组件进行信息过滤，对于日志的正则匹配最总会以json的格式输出，所以正则匹配的格式是(?<字段名>正则表达式过滤将要匹配的内容)在每个括号内创建过滤的内容，将不需要的内容放在括号外，可以一直做匹配知道日志结束，这里我只是简单匹配日志前面的ip地址和日期时间。
针对正则匹配字段详细说明：
##正则匹配字段通过“(?<字段名>正则表达式)”来定义，我们看到“(?<ip_addr>\d+?.\d+?.\d+?.\d+?”匹配了日志中的IP地址，然后将不需要的内容放在括号外面匹配，接着是日期和时间，使用同样的方式，然后不需要的内容放在外面因为这里我只匹配了前面的一点内容，如果有需要提取，可以一直匹配，知道这一行结束，logstash默认只会匹配日志以行作为分割，下一行又是新的一次匹配了。
#output是处理输出部分，这里我只是简单输出到终端，先实验正则匹配没问题后，将输出指向elasticsearch。

[root@Elk_Server logstash]# logstash -r -f /etc/logstash/conf.d/logstash.conf .............. { "path" => "/data/logs/test.log", "host" => "Elk_Server", "type" => "test.log", "log_date" => "10/Jun/2018", "@version" => "1", "log_time" => "12:58:12", "@timestamp" => 2018-07-31T09:45:16.247Z, "ip_addr" => "120.79.189.51", "message" => "120.79.189.51 - - [10/Jun/2018:12:58:12 +0800] \"POST /wp-cron.php?doing_wp_cron=1528606692.3628709316253662109375 HTTP/1.1\" 499 0 \"-\" \"WordPress/4.8; \"" }

接下来将日志输出到elasticsearch，修改output里面的内容即可。

output { elasticsearch { ##指定输出到ES action => "index" ##建立索引 hosts => "192.168.6.108:9200" ##ES地址端口 index => "test_log-%{+yyyy.MM.dd}" ##索引的名称 } stdout {codec => rubydebug} }

3.4 验证配置，并报告错误

logstash -f /etc/logstash/conf.d/logstash.conf --config.test_and_exit \\或者 logstash -t -f /etc/logstash/conf.d/logstash.conf

3.5 服务启动

nohup logstash -f /etc/logstash/conf.d/logstash.conf --config.reload.automatic -l /data/logs/logstash & \\或者 nohup logstash -r -f /etc/logstash/conf.d/logstash.conf -l /data/logs/logstash &

--config.reload.automatic选项启用自动配置重新加载，这样您每次修改配置文件时都不必停止并重新启动Logstash。

3.6 结合filebeat案例
日志内容如下：

2018-12-20 17:40:50.665 INFO 1 --- [Thread-422] f.s.j.e.common.iml.AbstractJobWorker : no task in app f.s.job.executor.social.users's job 0 2018-12-22 11:23:55.947 INFO -- [cluster-ClusterId{value='5xxx001145200', description='null'}-dds-xxxc9e41.mongodb.rds.aliyuncs.com:3717] org.mongodb.xxx.cluster : Server xxx:3717 is no longer a member of the replica set. Removing from client view of cluster.

filebeat配置文件：

#=========================== Filebeat inputs ============================= filebeat.inputs: - type: log enabled: true paths: - /data/log/*.log fields: app: tomcat type: all fields_under_root: true #multiline.pattern: '^\[' multiline.pattern: '^\d+\-\d+\-\d+\s+\d+\:\d+\:\d+' multiline.negate: true multiline.match: after multiline.max_lines: 200 multiline.timeout: 10s #tags: ["filebeat"] #exclude_lines: ['^\d+\-\d+\-\d+\s+\d+\:\d+\:\d+\s+DEBUG'] #include_lines: ['^ERR', '^WARN'] #----------------------------- Logstash output -------------------------------- output.logstash: # enabled: true hosts: ["127.0.0.1:5044"] # index => "filebeat-%{+YYYY.MM.dd}" # worker: 4 # bulk_max_size: 1024 # compression_level: 6 # loadbalance: false # backoff.max: 120s