messagepack bug in http (in) plugin

[Mizunanari]

Describe the bug

The http (in) plugin will not accept a messagepack request that should succeed if it accepts a messagepack request that fails.

To Reproduce

Create

mkdir -p /containers/fluentd
cd /containers/fluentd
mkdir ./output
ls -lnd ./output

compose.yaml

/containers/fluentd/compose.yaml

services:
 fluentd:
   restart: always
   image: fluent/fluentd:edge
   command: fluentd -c /fluentd/etc/fluent.conf
   user: 1000:1000
   ports:
     - 8888:8888
   volumes:
     - ./fluent.conf:/fluentd/etc/fluent.conf:ro
     - ./output:/var/fluentd/output

fluent.conf

/containers/fluentd/fluent.conf

# http input
<source>
  @type http
  port 8888
  bind 0.0.0.0
</source>

# file output
<match myapp.access>
  @type file
  path /var/fluentd/output/myapp.access
  <buffer>
    timekey 86400
    timekey_wait 600
    timekey_use_utc true
  </buffer>
</match>

compose up

docker compose up -d

Check

1. Successful Requests

msgpack=`echo -e "\x81\xa3bar\xa3bar"`

curl \
-X POST \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "msgpack=$msgpack" \
http://localhost:8888/myapp.access

2. Failure Requests

msgpack=`echo -e "\x81\xa3bar\xa3bar"`

curl \
-X POST \
-H "Content-Type: application/msgpack" \
-d "msgpack=$msgpack" \
http://localhost:8888/myapp.access

400 Bad Request
undefined method `delete' for 109:Integer

      if t = @default_keep_time_key ? record[@parser_time_key] : record.delete(@parser_time_key)
                                                                       ^^^^^^

3. Same request as "1."

msgpack=`echo -e "\x81\xa3bar\xa3bar"`

curl \
-X POST \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "msgpack=$msgpack" \
http://localhost:8888/myapp.access

400 Bad Request
undefined method `delete' for 115:Integer

      if t = @default_keep_time_key ? record[@parser_time_key] : record.delete(@parser_time_key)
                                                                       ^^^^^^^

Expected behavior

3. Same request as "1." succeeds

Your Environment

- Fluentd version: 1.16.2
- TD Agent version: 
- Operating system: Ubuntu 22.04.3 LTS
- Kernel version: 5.15.0-87-generic

Your Configuration

# http input
<source>
  @type http
  port 8888
  bind 0.0.0.0
</source>

# file output
<match myapp.access>
  @type file
  path /var/fluentd/output/myapp.access
  <buffer>
    timekey 86400
    timekey_wait 600
    timekey_use_utc true
  </buffer>
</match>

Your Error Log

fluentd -c /fluentd/etc/fluent.conf
2023-11-20 01:27:06 +0000 [info]: init supervisor logger path=nil rotate_age=nil rotate_size=nil
2023-11-20 01:27:06 +0000 [info]: parsing config file is succeeded path="/fluentd/etc/fluent.conf"
2023-11-20 01:27:06 +0000 [info]: gem 'fluentd' version '1.16.2'
2023-11-20 01:27:06 +0000 [info]: using configuration file: <ROOT>
  <source>
    @type http
    port 8888
    bind "0.0.0.0"
  </source>
  <match myapp.access>
    @type file
    path "/var/fluentd/output/myapp.access"
    <buffer>
      timekey 86400
      timekey_wait 600
      timekey_use_utc true
      path "/var/fluentd/output/myapp.access"
    </buffer>
  </match>
</ROOT>
2023-11-20 01:27:06 +0000 [info]: starting fluentd-1.16.2 pid=7 ruby="3.1.4"
2023-11-20 01:27:06 +0000 [info]: spawn command to main:  cmdline=["/usr/bin/ruby", "-Eascii-8bit:ascii-8bit", "/usr/bin/fluentd", "-c", "/fluentd/etc/fluent.conf", "--plugin", "/fluentd/plugins", "--under-supervisor"]
2023-11-20 01:27:07 +0000 [info]: #0 init worker0 logger path=nil rotate_age=nil rotate_size=nil
2023-11-20 01:27:07 +0000 [info]: adding match pattern="myapp.access" type="file"
2023-11-20 01:27:07 +0000 [info]: adding source type="http"
2023-11-20 01:27:07 +0000 [info]: #0 starting fluentd worker pid=16 ppid=7 worker=0
2023-11-20 01:27:07 +0000 [info]: #0 fluentd worker is now running worker=0
2023-11-20 01:27:24 +0000 [error]: #0 failed to process request error_class=NoMethodError error="undefined method `delete' for 109:Integer\n\n      if t = @default_keep_time_key ? record[@parser_time_key] : record.delete(@parser_time_key)\n                                                                       ^^^^^^^"
2023-11-20 01:27:53 +0000 [error]: #0 failed to process request error_class=NoMethodError error="undefined method `delete' for 115:Integer\n\n      if t = @default_keep_time_key ? record[@parser_time_key] : record.delete(@parser_time_key)\n

Additional context

No response

[ashie]

Thanks for you report, I can reproduce it.
In this case, the request body including msgpack= in the second request is passed to msgpack parser:

fluentd/lib/fluent/plugin/in_http.rb

Line 577 in 74db947

params['msgpack'] = @body

Because it's invalid msgpack sequence, broken data is remained in

fluentd/lib/fluent/plugin/parser_msgpack.rb

Line 35 in 74db947

@unpacker.feed_each(data) do |obj|

It will break also following data.

Someone should reset @unpakcer in msgpack parser on each request.