logstash线上配置文件
线上logstash配置文件,特此记录。
# pipelines.yml 文件
$ egrep -v '^$|^#' pipelines.yml
- pipeline.id: feature
path.config: "/data/elk/logstash-7.5.0/conf.d/kafka-feature.conf"
- pipeline.id: feature-log
path.config: "/data/elk/logstash-7.5.0/conf.d/kafka-feature-log.conf"
- pipeline.id: report
path.config: "/data/elk/logstash-7.5.0/conf.d/kafka-report.conf"
- pipeline.id: device
path.config: "/data/elk/logstash-7.5.0/conf.d/kafka-devicelog.conf"
- pipeline.id: main
path.config: "/data/elk/logstash-7.5.0/conf.d/kafka-reqlog.conf"
1
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
# logstash 配置文件
$ egrep -v '^$|^#' logstash.yml
pipeline.workers: 32
pipeline.batch.size: 1000
pipeline.batch.delay: 50
1
2
3
4
2
3
4
# 任意一个.conf 文件
cat kafka-devicelog.conf
input{
kafka {
bootstrap_servers => "kafka01:9092,kafka02:9092,kafka03:9092"
auto_offset_reset => "latest"
topics => ["deviceRequestLog"]
client_id => "dev-no01"
group_id => "logstash-devlog"
decorate_events => true
}
}
filter{
json {
source => "message"
}
mutate {
remove_field => ["message"]
add_field => { "@kafka_timestamp" => "" }
}
date {
match => ["[@metadata][kafka][timestamp]","UNIX_MS"]
target => "@kafka_timestamp"
}
# 解决8小时时差问题
ruby {
code => "event.set('timestamp', event.get('req_time') + 8*60*60*1000)"
}
date {
match => ["timestamp", "UNIX_MS"]
target => "@timestamp"
}
mutate {
remove_field => ["timestamp"]
}
}
output {
elasticsearch {
hosts => ["es01:9200","es02:9200","es03:9200"]
index => "device-log-%{+YYYY-MM-dd}"
user => "elastic"
password => "ppasswd"
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
上次更新: 2023/04/21, 08:57:47