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

# logstash 配置文件

$ egrep -v '^$|^#' logstash.yml 
pipeline.workers: 32
pipeline.batch.size: 1000
pipeline.batch.delay: 50

1
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

上次更新: 2023/04/21, 08:57:47

← es索引定期删除脚本 elk报错问题总结→