ELK结合redis或kafka收集日志

filebeat+redis+logstash

• filebeat配置

# filebeat-redis.yml
filebeat.inputs:
- type: log
  enabled: true
  backoff: "1s"
  tail_files: false
  paths:
    - /var/log/nginx/access.log
  fields:
    filetype: nginx1
  fields_under_root: true

output.redis:
  enabled: true
  hosts: ["10.90.25.20:6379"]
  key: nginx
  db: 0
  password: xxzx@789
  datatype: list

• logstash配置

# logstash-redis.conf
input {
   redis {
     host => "43.254.44.156"
     port => 6379
     password => "xxzx@789"
     key => "nginx"
     data_type => "list"
     db => 0
   }
}

filter {
    grok {  
      match => { "message" => "%{HTTPD_COMBINEDLOG}" }
    }

    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
    }
}

output {
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "nginx-redis-%{+YYYY.MM.dd}"
    }
}

filebeat+kafka+logstash

• filebeat配置

filebeat.inputs:
- type: log
  enabled: true
  backoff: "1s"
  tail_files: false
  paths:
    - /var/log/messages
  fields:
    filetype: kafka
  fields_under_root: true

output.kafka:
  enabled: true
  hosts: ["10.57.22.170:9092"]
  topic: liyk

• logstash

input {
  kafka {
    bootstrap_servers => "10.57.22.170:9092"
    topics => ["liyk"]
    group_id => "liyk_id"
  }
}

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200"]
    index => "kafka-liyk"
  }
}