使用Splunk Connect for Kafka采集Kafka数据

Splunk Connect for Kafka安装

参考链接:
https://docs.splunk.com/Documentation/KafkaConnect/1.1.0/User/InstallSplunkKafkaConnect

配置参数介绍：
https://github.com/splunk/kafka-connect-splunk/blob/develop/README.md

下载Splunk Connect for Kafka jar包:

https://github.com/splunk/kafka-connect-splunk/releases

启动Kafka群集并确认kafka正在运行

curl http://localhost:8083

(可选)创建用于存储Kafka Connect connectors jar的目录。将用于保存plugin.path的设置

4修改$KAFKA_HOME/config/connect-distributed.properties 文件

bootstrap.servers=10.122.22.70:9092
//连接到kafka的ip、端口
key.converter=org.apache.kafka.connect.storage.StringConverter
//可以控制从kafka读取的消息中的键的格式
value.converter=org.apache.kafka.connect.storage.StringConverter
//可以控制从kafka读取的消息中的值的格式
plugin.path=/home/renss1/kafka/kafka_2.12-1.1.1/libs
//存储Kafka Connect connectors的目录
consumer.auto.offset.reset=latest
//从最新日志开始收集

将下载的Splunk Connect for Kafka的jar文件放在plugin.path所有Splunk Connect主机的目录中。

启动kafka connect

./bin/connect-distributed.sh config/connect-distributed.propertie

检查是否正确安装和配置了Splunk Connect of Kafka

如果返回的可用连接器列表包含com.splunk.kafka.connect.SplunkSinkConnector表示Splunk Connect for Kafka安装成功！

curl  http://localhost:8083/connector-plugins

Splunk 端配置

因为kafka端目前暂不支持SSL协议发送数据,在管理节点上配置的HEC(http event collector)，接收数据。

登录到管理节点web。

点击 设置 > 数据输入 > HTTP事件收集器，

点击所有的标记中的已启用 ，并点击保存。

设置 > 数据输入 > HTTP事件收集器，点击页面右上角的新建令牌。新建一个token。

定义名称，启用索引器确认

选择sourcetype、index，选择应用(默认都选择search。规范)。最后就是保存。

标记值为token

注: 最终kafka端配置会引用到token、开启HEC的splunk服务器等配置

Kafka端配置

参考链接:
https://docs.splunk.com/Documentation/KafkaConnect/1.1.0/User/ConfigureSplunkKafkaConnect

在每台主机上启用Kafka Connect后，Kafka Connect实例会自动形成一个群集。对其中一个集群实例执行REST调用，配置将在整个集群中生效。

启动kafka Connect

.$KAFKA_HOME/bin/connect-distributed.sh config/connect-distributed.properties

创建连接器任务

curl 10.122.22.70:8083/connectors -X POST -H "Content-Type: application/json" -d '{
    "name": "kafka-connect-splunk",
    "config": {
      "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector    ",
      "tasks.max": "1",
      "topics":"log_message",
      "splunk.indexes":"main",
      "splunk.sourcetypes":"kafka_plm",
      "splunk.hec.uri": "http://blsplunkf01:8088",
      "splunk.hec.token": "d4b9788b-9fdd-4d83-8a94-7625296527ac ",
      "splunk.hec.ssl.validate.certs":"false",
      "splunk.hec.raw": "true",
      "splunk.hec.ack.enabled":"true"
    }
  }'

在splunk搜索数据，验证传输正常