filebeat+kafka+logstash+es 数据存储异常

Question

配置文件如下：

---

-------- filebeat -------

• type: log
  enabled: true
  paths:

  • /var/log/nginx/userapi.access.log

json.message_key: true
json.keys_under_root: true
json.overwrite_keys: true
tail_files: true
fields:

log_topics: nginx_access.userapi

• type: log
  enabled: true
  paths:

  • /var/log/nginx/openapi.access.log

json.message_key: true
json.keys_under_root: true
json.overwrite_keys: true
tail_files: true
fields:

log_topics: nginx_access.openapi

• type: log
  enabled: true
  paths:

  • /var/log/nginx/productapi.access.log

json.message_key: true
json.keys_under_root: true
json.overwrite_keys: true
tail_files: true
fields:

log_topics: nginx_access.productapi

---

-------- logstash -------

1、openapi.conf
[root@staus conf.d]# cat openapi.conf
input{

kafka {
    codec => "json"
    topics => ["nginx_access.openapi"]
    bootstrap_servers => "127.0.0.1:9092"
    enable_auto_commit => true
    auto_commit_interval_ms => "1000"
auto_offset_reset => "earliest"
}

}
output {

      elasticsearch {
            hosts => ["http://127.0.0.1:9200"]
            index => "nginx_access.openapi-%{+YYYY.MM.dd}"
      }

}

2、product.conf
[root@staus conf.d]# cat product.conf
input{

kafka {
    codec => "json"
    topics => ["nginx_access.productapi"]
    bootstrap_servers => "127.0.0.1:9092"
    consumer_threads => 2
    enable_auto_commit => true
    auto_commit_interval_ms => "1000"
}

}
output {

      elasticsearch {
            hosts => ["http://127.0.0.1:9200"]
            index =>"nginx_access.productapi-%{+YYYY.MM.dd}"
            template_overwrite => true
      }

}

3、userapi.conf
input{

kafka {
    codec => "json"
    topics => ["nginx_access.userapi"]
    bootstrap_servers => "127.0.0.1:9092"
    consumer_threads => 2
    enable_auto_commit => true
    auto_commit_interval_ms => "1000"
}

}
output {

      elasticsearch {
            hosts => ["http://127.0.0.1:9200"]
            index =>"nginx_access.userapi-%{+YYYY.MM.dd}"
            template_overwrite => true
      }

}

------- index 存储异常如下

------- logstash 报错信息如下