fluent.conf

<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

<filter nacelle.*>
  @type parser
  format json
  key_name log
  reserve_data true
  ignore_key_not_exist true
  suppress_parse_error_log true
</filter>

<filter nacelle.*>
  @type record_transformer

  <record>
    # Canned values
    hostname ${hostname}
    service_name ${tag_parts[1]}
    # Move timestamp field so ES will use it
    '@timestamp' ${record['timestamp']}
    # Ensure warning level is always "warn" (be safe about no such method on nil)
    level ${(record['level'] or '').sub('warning', 'warn')}
    # Convert level to a numeric value so we can sort by it
    syslog_level ${{'debug' => 7, 'info' => 6, 'warn' => 4, 'error' => 3, 'fatal' => 2}[record['level']]}
  </record>

  # Remove redundant fields
  remove_keys log, timestamp
  enable_ruby true
</filter>

<match nacelle.*>
  @type copy

  <store>
    @type elasticsearch
    host ${ES_HOST}
    port 9200
    logstash_format true
    logstash_prefix fluentd
    flush_interval 1s
  </store>

  <store>
    @type stdout
  </store>
</match>