filebeat -> logstash -> elasticsearch ->kibana

 

1: 之前讲的模式是

filebeat(多台)

filebeat(多台)  -> logstash 通过正则提取有效数据 -> elastic ->kibana

filebeat(多台)

如果需要扩logstash的话，logstash前搭个负载均衡或者引入redis。

 

filebeat(多台)            logstash多台

filebeat(多台) -> 负载均衡(或者redis) -> logstash 通过正则提取有效数据 -> elastic ->kibana

filebeat(多台)                                          logstash多台

 

redis服务器说明：监听在127.0.0.1本机，设置密码为test1234

 

2: filebeat配置：/usr/local/filebeat-6.1.1/filebeat.yml

filebeat:

  prospectors:

  - input_type: log

    tail_files: true

    backoff: "1s" #every second check log file

    paths:

        - /usr/local/nginx/logs/kibana_access.log

 

output:

  redis:

      hosts: ["127.0.0.1"]

      port: 6379

      password: 'test1234'

      key: 'nginx_log'

 

3: logstash配置：/usr/local/logstash-6.1.1/config/logstash.conf

input {

  redis {

    host => '127.0.0.1'

    port => 6379

    key => "nginx_log"

    data_type => "list"

    password => 'test1234'

  }

}

 

filter {

    grok {

#125.119.2.71 - elk [17/Mar/2018:17:40:11 +0800] "POST /elasticsearch/_msearch HTTP/1.1" 200 9550 "http://144.202.123.228:5609/app/kibana" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.186 Safari/537.36" "-"

        match => {

            "message" => '(?<source_ip>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) - [a-zA-Z0-9-]+ \[(?<nginx_time>[^ ]+) \+\d+\] "(?<method>[A-Z]+) (?<request_url>[^ ]+) HTTP/\d.\d" (?<status>\d+) \d+ "(?<referer>[^"]+)" "(?<agent>[^"]+)" ".*"'

        }  

        remove_field => ["message"]

    }  

    date {

        match => ["nginx_time", "dd/MMM/yyyy:HH:mm:ss"]

 target => "@timestamp"

    }

}

 

output {

  elasticsearch {

    hosts => ["http://127.0.0.1:9200"]

  }

}