ELK filebeat->kafka->logstash->elasticsearch 架构

filebeat.inputs:

- type: log
  enabled: true
  paths:
    - /log/java001.log
  tags: ["java001"]
  exclude_lines: [‘^$‘]
  multiline:
    pattern: ‘^\[[0-9]{2}-[0-9]{2}‘
    negate: true
    match: after
  fields:
    service: java001

- type: log
  enabled: true
  paths:
    - /log/java002.log
  tags: ["java002"]
  exclude_lines: [‘^$‘]
  multiline:
    pattern: ‘^\[[0-9]{2}-[0-9]{2}‘
    negate: true
    match: after
  fields:
    service: java002


filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml

  reload.enabled: false


setup.template.settings:
  index.number_of_shards: 1


output.kafka:
# 指定kafka集群
  hosts: ["192.168.0.21:9092", "192.168.0.21:9093", "192.168.0.22:9092"]
  topic: ‘%{[fields.service]}‘
  partition.round_robin:
    reachable_only: false

  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

input {
  kafka {
    bootstrap_servers => "192.168.0.21:9092"
    topics => ["java001", "java002"]
    codec => "json"
    consumer_threads => 2
    enable_auto_commit => true
    auto_commit_interval_ms => "1000"
  }
}

output {
  elasticsearch {
# 指定es集群
    hosts => ["http://es-test-001:9200","http://es-test-002:9200","http://es-test-003:9201"]
    index => "%{[fields][service]}-%{+YYYY-MM}"
    #index => "%{[fields][service]}-%{+YYYY-MM-dd}"
  }

}