Fluentd數據採集到Kafka

• 安裝fluent

  下載安裝 fluentd (查看官方文檔，不詳細介紹)

• 安裝fluent-plugin-kafka

  a. 修改ruby 的gem源 

查看源：
$ gem source
  https://ruby.taobao.org/ (已經修改好了以後的) 
刪除默認源：
$ gem source -r <url>
添加新指向：
$ gem source -a https://ruby.taobao.org/
done

    b. 安裝fluent-plugin-kafka (github上有詳細介紹)     

• fluentd --setup /path/xxx

• 配置/path/xxx下的fluent.conf 

<source>
  @type forward
  @id forward_input
</source>

<source>
  @type http
  @id http_input
  port 8888
</source>

<source>
  @type    tail  #### tail方式採集日誌
  #format   none
  format   /^(?<all>.*)$/
  path     /tmp/app_log.txt
  pos_file /data/logs/td-agent/httpd-access.log.pos
  tag      app_log.tag
</source>

<match app_log.**>
  @type               kafka
  #brokers            localhost:9002,localhost:9003
  zookeeper           localhost:2181
  default_topic       app_log_raw  
  #刷新間隔
  flush_interval      30
  buffer_type         file
  ack_timeout_ms      2000
  output_data_type    attr:all
  compression_codec   none
</match>

<source>
  @type monitor_agent
  @id monitor_agent_input
  port 24220
</source>

<source>
  @type debug_agent
  @id debug_agent_input
  bind 127.0.0.1
  port 24230
</source>

<match debug.**>
  @type stdout
  @id stdout_output
</match>

<match system.**>
  @type forward
  @id forward_output

  <server>
    host 192.168.0.11
  </server>
  <secondary>
    <server>
      host 192.168.0.12
    </server>
  </secondary>
</match>

• 確保kafka已經啓動完成,點擊查看

• 啓動fluentd

$ fluentd -c fluent.conf

• 查看kafka中數據是否進入

$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic app_log_raw