Logstash从kafka获取数据写入ES失败

背景
搭建好了6.2.4版本的ELK集群，从另外一个kafka集群中获取topic中的数据，写入到ES中，出现了两个问题，第一：从kafka集群获取数据失败；第二：能够获取到数据，但是写入ES，一直报错如下：

Attempted to send a bulk request to Elasticsearch configured at '["http://localhost:9200/"]', but an error occurred and it failed! Are you sure you can reach elasticsearch from this machine using the configuration provided? {:error_message=>"[406] {"error":"Content-Type header [text/plain; charset=ISO-8859-1] is not supported","status":406}", :error_class=>"Elasticsearch::Transport::Transport::Errors::NotAcceptable"

分析

问题1：从kafka集群获取数据失败
现象是：logstash启动成功，控制台没有打印message的日志，去Kibana上查看index，发现index也没有创建成功。排查方式如下：

配置项是参考kafka-input-plugin对应版本配置的，check了下，没有错。
输出改为控制台输出，仍没有数据，说明和ES无关。
修改bootstrap_servers从另外一个kafka集群获取数据，发现能够正常获取数据，说明kafka兼容性问题。
logstash查看插件版本 ./bin/logstash-plugin list --verbose，发现内嵌的kafka插件是logstash-input-kafka (8.0.6) ，对应的客户端是1.0.0；而kafka集群的版本为0.9.0.1。<u>https://www.elastic.co/guide/en/logstash-versioned-plugins/current/v8.0.2-plugins-inputs-kafka.html</u>
logstash官网很明确的说明了，对于0.9的kafka集群，是不兼容的，因此没法获取到数据

image.png

<<既然是kafka插件不兼容的问题，查看了官网，支持0.9的，下载了logstash2.4版本安装，配置如下>>

input {
     kafka {
        zk_connect => "datastream0.lt.org:2181,datastream1.lt.org:2181,datastream2.lt.org:2181/kafka_online"
        auto_offset_reset => "smallest"
        group_id => "logstash23"
        topic_id => "datastream.novel"
        codec => "json"
        reset_beginning => "true"
        }
}
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
         stdout {  }
         elasticsearch {
                hosts => "esIp:port"
                index => "exception-%{YYYY.MM.dd}"
        } 
}

问题2：写入ES，报错406，NotAcceptable
现象是：使用了Logstash2.4的版本号，从0.9版本的kafka集群可以获取到数据，控制台有打印，但是写入ES一直报错。注：本地ES是6.2.4的，已有业务运行其上。
出现这个错误的原因是因为Logstash2.4和ES6.2.4无法兼容。也曾尝试将logstash6.2.4版本的kafka插件降为2.0.9(logstash2.4使用的kafka插件)，但是因为在更新插件过程中，会和logstash-core-plugin-api不兼容，出现很多冲突，也不能解决这个问题。

综上所述：

降低logstash2.4.0可以获取到0.9kafka集群的数据，但是写入ES6.2.4有兼容性问题。
基于Logstash6.2.4，降低logstash-input-kafka插件为2.0.9，会出现API兼容性问题。
正如官网所说，ELK版本尽可能保持和kafka等输入源版本一致。

解决方案：

升级kafka集群的版本
迁移kafka集群的数据至高版本kafka集群
ELK全部降级为匹配到kafka0.9的版本

经验总结：

搭建ELK首先需要确定数据来源有哪些，尤其是涉及到kafka这种组件的，需要提前了解kafka集群的版本信息，确定你需要下载的ELK版本。否则数据读取不到。
ELK这三个组件，版本尽可能一致，否则即使能够读取到输入的数据，但是写入不到ES中，也是很蛋疼的。
ELK的配置问题，对于不同的版本，配置项很多不一样，不能仅仅在网上查到一个配置就填上去，需到官网查询对应的配置项。例如，kafka插件早期版本logstash的配置项是zk_connect,但是稍微新一点的版本都是bootstrap_servers；另外需要注意配置项的类型，有些是string的，有些是array的，不要乱填。
Kibana的关键配置项

server.port: 5601 
server.host: "10.171.xxx.yyy"   //这个是外部浏览器访问的地址
elasticsearch.url: "http://esIp:9200" //这个是集群内部通信的

最后编辑于：2018.09.07 10:53:37