Reproduce Fluentd fluent-plugin-flowcounter 0.4.2 error

Reproduce Fluent flowcounter error

This is intended to be executed like this:
./build.sh &
docker run --log-driver=fluentd --log-opt tag="docker.{{.ID}}" --log-opt fluentd-address="$(docker inspect -f '{{.NetworkSettings.IPAddress}}' custom-docker-fluent-logger):24224" python:alpine echo Hello

build.sh

#!/bin/bash
mkdir -p log
mkdir -p mkdir plugins
docker build -t custom-fluentd:latest ./

docker run -it --rm \
    --name custom-docker-fluent-logger \
    -v "$(pwd)/log:/fluentd/log" custom-fluentd:latest

Dockerfile

FROM fluent/fluentd:v0.14

MAINTAINER Steve Coffman <steve.coffman@ithaka.org>

WORKDIR /home/fluent

ENV PATH /home/fluent/.gem/ruby/2.3.0/bin:$PATH

# Run as root for backwards compatibility
USER root

# Do not split this into multiple RUN!
# Docker creates a layer for every RUN-Statement
# therefore an 'apk delete build*' has no effect
RUN apk --no-cache --update add \
                            build-base \
                            ruby-dev && \
    gem install fluent-plugin-record-modifier -v 0.5.0 && \
    gem install fluent-plugin-flowcounter -v 0.4.2 && \
    gem install fluent-plugin-graphite -v 0.0.6 && \
    apk del build-base ruby-dev && \
    rm -rf /tmp/* /var/tmp/* /var/cache/apk/* /usr/lib/ruby/gems/*/cache/*.gem && \
    chown -R fluent:fluent /fluentd
COPY fluent.conf /fluentd/etc/$FLUENTD_CONF
EXPOSE 24220 24224

CMD fluentd -c /fluentd/etc/$FLUENTD_CONF -p /fluentd/plugins $FLUENTD_OPT

fluent.conf

<source>
  @type forward
  @id  input1
  port  24224
</source>

<filter fluent.**>
 @type stdout
</filter>

<filter docker.**>
  @type stdout
</filter>

<match docker.**>
  @type copy
  <store>
    @type flowcounter
    tag flowcount
    aggregate tag
    output_style tagged
    count_keys *
    unit second
    delete_idle true
  </store>
</match>


#
# Match messages that came from the "flowcounter" plugin and duplicate them.
# The resulting messages should have different tags though, so that we can
# process them separately.
#
<match flowcount>
  @type copy

  #
  # Need to "deep copy", otherwise the <store>s below will share & modify
  # the same record. This is bad because we need to modify the new
  # "bytes"/"count" records separately.
  #
  deep_copy true

  <store>
    type record_modifier
    tag quantico.fluentd.messages.bytes
  </store>
  <store>
    type record_modifier
    tag quantico.fluentd.messages.count
  </store>
</match>

#
# Messages coming into here will have the form:
#
#     quantico.fluentd.messages.count: {
#         "count": 0,
#         "bytes": 0,
#         "count_rate": 0.0,
#         "bytes_rate": 0.0,
#         "tag": "syslog.messages"
#     }
#
# ... where metrics for both "bytes" and "count" are packed into the same
# message.
#
# Since the Datadog statsd plugin can only handle 1 metric/message, we must
# duplicate the flowcount message and tag the resulting messages for either
# "count" or "bytes", respectively.
#
# Those resulting duplicated events then come here for additional
# transformations:
#
#     * Add a new JSON key named "value", whose value is the respective
#       count/bytes value, taken from the key "count" or "bytes".
#     * Rename the key named "tag" into "fluentd-tag". This will result in a
#       Datadog metric-tag named "fluentd-tag" which we can then pivot on.
#       Ex: {"fluentd-tag": "system.messages"}
#     * Drop the original keys: "bytes,bytes_rate,count,count_rate". If not
#       removed, they would show up as distinct metrics tags in Datadog and
#       given that they would have values of varying sizes, this would cause
#       our number of tracked metrics (and associated $$ costs) in Datadog to
#       grow unbounded.
#
# The resulting messages will have the following format, which we can then
# <match> into the "dogstatsd" output plugin, for shipping off to the local
# "statsd":
#
#     quantico.fluentd.messages.count: {
#         "fluentd-tag": "system.messages",
#         "value": "20"
#     }
#
#
<filter quantico.fluentd.messages.*>
  @type record_transformer
  enable_ruby true
  remove_keys bytes,bytes_rate,count,count_rate,tag

  <record>
    value ${record[tag_parts[3]]}
  </record>
  <record>
    fluentd-tag ${record['tag']}
  </record>
</filter>
#
# Send messages like "quantico.fluentd.messages.count" to the local dogstatsd
# daemon.
#
<filter quantico.fluentd.**>
  @type stdout
</filter>

<match **>
  @type file
  @id   output1
  path         /fluentd/log/data.*.log
  symlink_path /fluentd/log/data.log
  append       true
  time_slice_format %Y%m%d
  time_slice_wait   10m
  time_format       %Y%m%dT%H%M%S%z
</match>

stacktrace.txt

  2017-04-14 20:54:07 +0000 [warn]: #0 emit transaction failed: error_class=RuntimeError error="BUG: use router.emit_stream instead of Engine.emit_stream" tag="flowcount"
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/engine.rb:167:in `emit_stream'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluent-plugin-record-modifier-0.5.0/lib/fluent/plugin/out_record_modifier.rb:79:in `emit'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/compat/output.rb:164:in `process'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/plugin/output.rb:716:in `emit_sync'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/plugin/out_copy.rb:42:in `block in process'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/plugin/out_copy.rb:41:in `each'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/plugin/out_copy.rb:41:in `process'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/plugin/multi_output.rb:147:in `emit_sync'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/event_router.rb:96:in `emit_stream'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluentd-0.14.14/lib/fluent/event_router.rb:87:in `emit'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluent-plugin-flowcounter-0.4.2/lib/fluent/plugin/out_flowcounter.rb:155:in `block in flush_emit'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluent-plugin-flowcounter-0.4.2/lib/fluent/plugin/out_flowcounter.rb:154:in `each'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluent-plugin-flowcounter-0.4.2/lib/fluent/plugin/out_flowcounter.rb:154:in `flush_emit'
  2017-04-14 20:54:07 +0000 [warn]: #0 /usr/lib/ruby/gems/2.3.0/gems/fluent-plugin-flowcounter-0.4.2/lib/fluent/plugin/out_flowcounter.rb:174:in `watch'