Is there any problem with using Fluent::MessagePackFactory#msgpack_packer.pack in format of output plugin ??

[cm-kazup0n]

I experienced frequent SEGV crashes (double-free/use after free) with the out-cloudwatch-logs plugin at high log traffic (around 70k logs per minute or more). After trying various solutions, I found that modifying the CloudwatchLogsOutput#format method fixed the crash issue.

    def format(tag, time, record)
      record = inject_values_to_record(tag, time, record)
      # original code
      # Fluent::MessagePackFactory.msgpack_packer.pack([tag, time, record]).to_s

      # my modification preventing SEGV
      [tag, time, record].to_msgpack
    end

The modification involved replacing the original Fluent::MessagePackFactory.msgpack_packer.pack([tag, time, record]).to_s with [tag, time, record].to_msgpack.

Though there is not much difference between the two codes, the original code uses a cached factory while my modification uses a local one.

Is there any potential problem in the original code?

---

my configs and crash logs is here.

FROM amazonlinux:2.0.20221210.0
ENV RUBYOPT -EUTF-8

# Use root account to use apk
USER root

COPY td.repo /etc/yum.repos.d/td.repo
RUN rpm --import https://packages.treasuredata.com/GPG-KEY-td-agent \ 
    && yum check-update \
    && yes | yum install -y td-agent \
    && td-agent-gem install fluent-plugin-cloudwatch-logs -v "0.14.3" \
    && td-agent-gem install fluent-plugin-datadog -v "0.14.2" \
    && td-agent-gem sources --clear-all \
    && rm -rf /tmp/* /var/tmp/* /opt/td-agent/lib/ruby/gems/*/cache/*.gem

COPY fluent.conf.min /etc/td-agent/td-agent.conf

USER td-agent
CMD td-agent

<system>
  log_level info
  <log>
    format json
  </log>
</system>

# expose http endpoint for monitoring
<source>
  @type monitor_agent
  @id monitor_agent
  bind 0.0.0.0
  port 24220
</source>

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

<match myapp>
  @type cloudwatch_logs
    region ap-northeast-1
    log_group_name my_log_group
    auto_create_stream true
    use_tag_as_stream true
    concurrency 1
    json_handler json
    <buffer tag>
      @type file
      path /var/log/td-agent/buffer/myapp/cwl
      chunk_limit_size 128m
      flush_mode interval
      flush_interval 30s
      flush_at_shutdown true
    </buffer>
</match>

[cm-kazup0n]

here's a example of crash log . crash_log.log

[ashie]

The modification involved replacing the original Fluent::MessagePackFactory.msgpack_packer.pack([tag, time, record]).to_s with [tag, time, record].to_msgpack.

You mean like this?:

    def format(tag, time, record)
      record = inject_values_to_record(tag, time, record)
-      Fluent::MessagePackFactory.msgpack_packer.pack([tag, time, record]).to_s
+      [tag, time, record].to_msgpack
    end

https://github.com/fluent-plugins-nursery/fluent-plugin-cloudwatch-logs/blob/7287d1ae78b24e3fb74aee8d3830a65ecd89f65d/lib/fluent/plugin/out_cloudwatch_logs.rb#L178-L181

[cm-kazup0n]

yes, that's actually i did

[ashie]

How about replacing it with Fluent::MessagePackFactory.thread_local_msgpack_packer?

[cm-kazup0n]

@ashie Thank you for your quick reply.

I also attempted to replace it with #thread_local_msgpack_packer, which did fix the crash. However, it led to a significant increase in memory usage and the worker process was frequently terminated by the OOM killer.

[cm-kazup0n]

finally, I found that my fix does not fix the crash issue.
it seems that the actual cause of crash is double free in garbage collecting C ext objects, e.g. msgpack buffer, yajl.
I couldn't inspect in detail yet, but my fix is not relevant.