fluentd.conf

## File input
<source>
  @type tail
  path /tmp/fluentd/*/cell.log
  pos_file /tmp/fluentd/cell.log.pos
  tag cloudsql.cell
  read_from_head true
  <parse>
    @type json
  </parse>
</source>

## Filter
<filter cloudsql.cell>
  @type record_transformer
  enable_ruby
  <record>
    msg ${record["msg"].encode("UTF-8", invalid: :replace, replace: "__NON_ASCII_CHAR__")}
    timestamp ${record["ts"]}
  </record>
</filter>

## Elasticsearch
<match cloudsql.cell>
  @type elasticsearch
  host localhost
  port 9200
  index_name fluentd.${tag}
  <buffer>
    flush_interval 10s
  </buffer>
</match>