Fluentd, Kafka, InfluxDB, Grafanaでメトリクス監視環境を作る ref: http://qiita.com/kentarosasaki/items/f072773e08b8f2329b6e

file0.txt

<source>
  @type dstat
  tag raw_dstat
  option -cmldrn
  delay 10
  tmp_file /tmp/dstat_all.csv
</source>

<source>
  @type exec
  tag raw_df
  command df -TP | sed 1d | sed 's/%//' | sed 's/\s\+/\t/g'
  run_interval 10s
  format tsv
  keys device,type,size,used,available,capacity,mounted_on
</source>

<filter raw_df>
  @type record_transformer
  enable_ruby true
  <record>
    hostname ${hostname}
  </record>
</filter>

<match raw_dstat raw_df>
  @type kafka_buffered
  brokers kafkabroker001:9092,kafkabroker002:9092
  default_topic metrics-topic
  flush_interval 60
  buffer_type file
  buffer_path /tmp/td-agent.*.buffer
  output_data_type json
  output_include_tag true
  output_include_time true
</match>

file1.txt

<source>
  @type kafka_group
  brokers kafkabroker001:9092,kafkabroker002:9092
  consumer_group consumer_group_001
  topics metrics-topic
  format json
</source>

<match metrics-topic>
  @type record_reformer
  tag ${record['tag']}
  enable_ruby true
  auto_typecast true
</match>

<match raw_dstat>
  @type copy
  <store>
    @type record_reformer
    tag cpu
    enable_ruby true
    auto_typecast true
    renew_record true
    <record>
      host   ${record['hostname']}
      usr    ${record['dstat']['total_cpu_usage']['usr'].to_f}
      sys    ${record['dstat']['total_cpu_usage']['sys'].to_f}
      idl    ${record['dstat']['total_cpu_usage']['idl'].to_f}
      wai    ${record['dstat']['total_cpu_usage']['wai'].to_f}
      hiq    ${record['dstat']['total_cpu_usage']['hiq'].to_f}
      siq    ${record['dstat']['total_cpu_usage']['siq'].to_f}
      time   ${record['time']}
    </record>
  </store>
  <store>
    @type record_reformer
    tag mem
    enable_ruby true
    auto_typecast true
    renew_record true
    <record>
      host   ${record['hostname']}
      used   ${record['dstat']['memory_usage']['used'].to_f}
      buff   ${record['dstat']['memory_usage']['buff'].to_f}
      cach   ${record['dstat']['memory_usage']['cach'].to_f}
      free   ${record['dstat']['memory_usage']['free'].to_f}
      time   ${record['time']}
    </record>
  </store>
  <store>
    @type record_reformer
    tag load
    enable_ruby true
    auto_typecast true
    renew_record true
    <record>
      host   ${record['hostname']}
      1m     ${record['dstat']['load_avg']['1m'].to_f}
      5m     ${record['dstat']['load_avg']['5m'].to_f}
      15m    ${record['dstat']['load_avg']['15m'].to_f}
      time   ${record['time']}
    </record>
  </store>
  <store>
    @type record_reformer
    tag disk
    enable_ruby true
    auto_typecast true
    renew_record true
    <record>
      host   ${record['hostname']}
      read   ${record['dstat']['dsk/total']['read'].to_f}
      writ   ${record['dstat']['dsk/total']['writ'].to_f}
      time   ${record['time']}
    </record>
  </store>
  <store>
    @type record_reformer
    tag diskio
    enable_ruby true
    auto_typecast true
    renew_record true
    <record>
      host   ${record['hostname']}
      read   ${record['dstat']['io/total']['read'].to_f}
      writ   ${record['dstat']['io/total']['writ'].to_f}
      time   ${record['time']}
    </record>
  </store>
  <store>
    @type record_reformer
    tag net
    enable_ruby true
    auto_typecast true
    renew_record true
    <record>
      host   ${record['hostname']}
      recv   ${record['dstat']['net/total']['recv'].to_f}
      send   ${record['dstat']['net/total']['send'].to_f}
      time   ${record['time']}
    </record>
  </store>
</match>

<match raw_df>
  @type record_reformer
  tag df
  enable_ruby true
  auto_typecast true
  renew_record true
  <record>
    host       ${record['hostname']}
    size       ${record['size'].to_f}
    used       ${record['used'].to_f}
    available  ${record['available'].to_f}
    capacity   ${record['capacity'].to_f}
    device     ${record['device']}
    type       ${record['type']}
    mounted_on ${record['mounted_on']}
    time       ${record['time']}
  </record>
</match>

<filter df cpu mem load disk diskio net>
  @type record_transformer
  renew_time_key time
</filter>

<filter df cpu mem load disk diskio net>
  @type record_transformer
  remove_keys time
</filter>

<match df cpu mem load disk diskio net>
  @type copy
  <store>
    @type influxdb
    host influxdb001
    port 8086
    dbname metrics_db
    user kafka
    password kafka
    use_ssl false
    verify_ssl false
    tag_keys ["host", "device"]
    time_precision s
    flush_interval 10s
  </store>
  <store>
    @type influxdb
    host influxdb002
    port 8086
    dbname metrics_db
    user kafka
    password kafka
    use_ssl false
    verify_ssl false
    tag_keys ["host", "device", "type", "mounted_on",]
    time_precision s
    flush_interval 10s
  </store>
</match>