ELK学习笔记之配置logstash消费kafka多个topic并分别生成索引

0x00 filebeat配置多个topic

1filebeat.prospectors:
2
3- input_type: log
4  encoding: GB2312
5#  fields_under_root: true
6  fields:  ##添加字段
7    serverip: 192.168.1.10
8    logtopic: wap
9  enabled: True
10  paths:
11        - /app/wap/logs/catalina.out
12  multiline.pattern: '^\['  #java报错过滤
13  multiline.negate:  true
14  multiline.match: after
15  tail_files: false
16- input_type: log
17  encoding: GB2312
18 # fields_under_root: true
19  fields:  ##添加字段
20    serverip: 192.168.1.10
21    logtopic: api
22  enabled: True
23  paths:
24        - /app/api/logs/catalina.out
25  multiline.pattern: '^\['  #java报错过滤
26  multiline.negate:  true
27  multiline.match: after
28  tail_files: false
29#----------------------------- Logstash output --------------------------------
30output.kafka:
31  enabled: true
32  hosts: ["192.168.16.222:9092","192.168.16.237:9092","192.168.16.238:9092"]
33  topic: 'elk-%{[fields.logtopic]}' ##匹配fileds字段下的logtopic
34  partition.hash:
35    reachable_only: true
36  compression: gzip
37  max_message_bytes: 1000000
38  required_acks: 1
39logging.to_files: true

0x01 查看是否输出到kafka

1$  bin/kafka-topics.sh --list --zookeeper kafka-01:2181, kafka-02:2181,kafka-03:2181
2elk-wap
3elk-api

0x02 配置logstash集群

1input{
2  kafka{
3    bootstrap_servers => "kafka-01:9092,kafka-02:9092,kafka-03:9092"
4    topics_pattern  => "elk-.*"
5    consumer_threads => 5
6    decorate_events => true
7    codec => "json"
8    auto_offset_reset => "latest"
9    group_id => "logstash1"##logstash 集群需相同
10
11}
12}
13filter {
14        ruby {
15        code => "event.timestamp.time.localtime"
16      }
17
18
19        mutate {
20        remove_field => ["beat"]
21    }
22        grok {
23             match => {"message" => "\[(?<time>\d+-\d+-\d+\s\d+:\d+:\d+)\] \[(?<level>\w+)\] (?<thread>[\w|-]+) (?<class>[\w|\.]+) (?<lineNum>\d+):(?<msg>.+)"
24 }
25
26}
27}
28output {
29   elasticsearch {
30         hosts => ["192.168.16.221:9200","192.168.16.251:9200","192.168.16.252:9200"]
31   #      index => "%{[fields][logtopic}" ##直接在日志中匹配，索引会去掉elk
32         index =>  "%{[@metadata][topic]}-%{+YYYY-MM-dd}" 
33}    
34    stdout {
35        codec => rubydebug
36    }

0x03 Es查看是否创建索引

0x04 logstash集群配置

1# 一机多实例，同一个配置文件，启动时只需更改数据路径
2./bin/logstash -f test.conf --path.data=/usr/local/logdata/
3# 多台机器 logstash配置文件group_id 相同即可