logstash收集多个filebeat主机发送日志问题

 logstash服务器/etc/logstash/conf.d 下分别建立nginx.conf和mysql-slow.conf并启用不同端口：

logstash日志服务器配置
nginx.conf配置：
input {
beats {
port => 5044
}
}
filter {
if [fields][log_source] == "nginx" {
grok {
match => [ "message","%{COMBINEDAPACHELOG}+%{GREEDYDATA:extra_fields}"]
overwrite => [ "message"]
}
mutate {
convert => ["response","integer"]
convert => ["bytes","integer"]
convert => ["responsetime","float"]
}
geoip {
source=>"clientip"
target => "geoip"
}
date {
match => [ "timestamp","dd/MMM/YYYY:HH:mm:ss Z"]
remove_field => [ "timestamp"]
}
useragent {
source=>"agent"
}
}
}
output {
if [fields][log_source] == "nginx" {
elasticsearch {
hosts => ["192.168.100.196:9200"]
index => "access-%{+YYYY.MM.dd}"
}
}
}
mysql-slow.conf配置：
input {
beats {
port => 5045
}
#tcp {
# port => 1928
#}
}
filter {
if [fields][log_source] == "mysql-slow" {
grok {
match => [ "message", "(?m)^# User@Host: %{USER:query_user}\[[^\]]+\] @ (?:(?<query_host>\S*) )?\[(?:%{IP:query_ip})?\](?:\s*Id: %{NUMBER:id:int})?\s+# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*(?:use %{DATA:database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<query>(?<action>\w+)\s+.*)" ]
}
grok {
match => { "message" => "# Time: " }
add_tag => [ "drop" ]
tag_on_failure =>
}
if "drop" in [tags] {
drop {}
}
date {
match => ["mysql.slowlog.timestamp", "UNIX", "YYYY-MM-dd HH:mm:ss"]
target => "@timestamp"
timezone => "Asia/Chongqing"
}
ruby {
code => "event.set('[@metadata][today]', Time.at(event.get('@timestamp').to_i).localtime.strftime('%Y.%m.%d'))"
}
mutate {
#remove_field => [ "message" ]
remove_field => ["tags","beat"]
}
}
}
output {
if [fields][log_source] == "mysql-slow" {
#stdout { codec => rubydebug }
elasticsearch {
hosts => ["192.168.100.196:9200"]
index => "mysql-slow-%{[@metadata][today]}"
document_type => "mysql-slow"
template_overwrite => true
}
}
}
两个服务器filebeat配置：
mysql配置：



nginx的配置