in_forward: Accept metadata on encapsulated timestamps

[cosmo0920]

Just ensuring interoperability to accept the default format of forwarded logs which had been changed in Fluent Bit v5.0.5,
we need to extract the encapsulated timestamps inside of the second argument of the forwarded events from Fluent Bit.
With this minimal changes, we enable to process the new format of forwarded events from Fluent Bit.

Related to #5369.

What this PR does / why we need it:

Since Fluent Bit v5.0.5, Fluent Bit core team decided to send metadata within the forwarding events by default.
So, this could decrease the interoperability between Fluentd and Fluent Bit.
This changes covers to improve interoperability of forward protocol between Fluent Bit's out_forward plugin by default representation of msgpack and Fluentd's out_forward plugin accepts format variations.

This is because forward protocol v1.5 insists that the following BNF's 3rd part of the grammar requests to handle on forward plugins:

Metadata ::= integer | EventTime | [ EventTime, Hash* ] # We need to support the 3rd part of metadata BNF

Then, the event structure inherits the complexity of the changes of matadata information like:

Event ::= [ Metadata, Record ]

Docs Changes:
No.

Release Note:

Same as title.

Additional Context:

This PR is just ensuring to ingest metadata enabled logs type of events from Fluent Bit to Fluentd.
It's half pathway to handle complete processing of metadata in forward protocol.

[cosmo0920]

No sure the policy of backporting but this incompatibility change is needed to handle in maintained Fluentd versions as much as possible due to ensuring interoperability.

[kenhys]

In practical use case, in same event stream, are there mixed event coming from fluentbit?
I mean that [EventTime, Record], [[EventTime, Hash*], Record], [EventTime, Record], ...

I know that based on the specifications, of course, that case is a possible.

[cosmo0920]

No, it isn't possible to be contaminated. They'll be handled to switch with the retain_metadata_in_forward_mode parameter:
https://github.com/fluent/fluent-bit/blob/master/plugins/out_forward/forward.c#L1794C28-L1794C59

This should be set up with configuration file and not to be changed at runtime.

[Watson1978]

The Forward-mode path works as intended.
I put together a small end-to-end repro to check the other carrier modes, and found one path that still slips through: CompressedPackedForward received with skip_invalid_event false.

See: https://github.com/Watson1978/test-fluentd-pr-5411

(fluentd built from this PR + Fluent Bit v5.0.6 with Compress gzip, sending the same input to two in_forward ports.)

p24224.fbgzip       Array              # skip_invalid_event false -> wrapper NOT unwrapped
p24225skip.fbgzip   Fluent::EventTime  # skip_invalid_event true  -> unwrapped

Same input, same fluentd; only skip_invalid_event differs. The Array is the metadata wrapper leaking as the event time, e.g. _raw_time_inspect: "[2026-... +0000, {}]". (I confirmed via a receiver-side probe that every gzip message really does arrive as CompressedPackedForward with compressed="gzip", so this isn't compression silently falling back to Forward mode.)

extract_event_time is only reached from check_and_skip_invalid_event (which runs only when skip_invalid_event is true) and from the when Array branch.
The when String branch (PackedForward / CompressedPackedForward) with skip_invalid_event false emits the MessagePackEventStream unchanged, so the wrapper is never removed. The same shape exists in add_source_info and in the Message-mode path, where invalid_event? unwraps only locally and the unwrapped value never reaches the emitted time.

It need to apply the unwrap independently of skip_invalid_event, normalize the time for all carrier modes rather than only inside check_and_skip_invalid_event.

Also, the two new tests assert only tag and record, so they pass even when the time is wrong. Could we add an assertion on the emitted time? Something like:

assert_equal([time, time], d.events.map { |_tag, t, _record| t })

[cosmo0920]

I implemented to normalize all of the occurrences on forward protocol v1.5 formats. Could you check on it again?

[Watson1978]

I added benchmark scripts to confirm performance regression in
https://github.com/Watson1978/test-fluentd-pr-5411/tree/main/benchmark

Result

$ ./run.sh
fluentd master : 99c7074fcdcc05b89328516881564ef8d0a3dba0
fluentd pr-5411: fe52d269ab510f4958f83fa445c762a54b6170b4
ruby           : ruby 4.0.5 (2026-05-20 revision 64336ffd0e) +PRISM [x86_64-linux]

running: master skip_invalid_event=false ...
running: pr-5411 skip_invalid_event=false ...
running: master skip_invalid_event=true ...
running: pr-5411 skip_invalid_event=true ...

| target  | skip_invalid_event | chunks/s (±)        | allocated objs/chunk |
|---------|--------------------|---------------------|----------------------|
| master  | false              |    3018527.8 ±0.4% |                  5.0 |
| pr-5411 | false              |       1394.5 ±1.1% |              11024.0 |
| master  | true               |        922.1 ±0.2% |              14043.0 |
| pr-5411 | true               |        884.0 ±0.3% |              14043.0 |

main result   : skip=false  master/PR = 2164.6x (PR is the slower one if > 1)
sanity check  : skip=true   PR vs master = -4.1% (should be within ~10%)

When we configure skip_invalid_event: false, seems this PR introduces huge performance regression.

Note
To put the 2164x micro number in perspective: the baseline cell is nearly a no-op (binary passthrough, 5 objs/chunk), so the ratio is dominated by that. In an end-to-end fluentd→fluentd relay test (2M events, buffered out_forward), the relay's total process CPU went from 0.88s to 2.81s (3.19×) with skip_invalid_event false, and +7.5% with the default true. So the realistic impact is ~3× CPU on explicitly-lazy relay configs, not 2000×. The per-event cost is ~0.7–1µs + ~11 objects (eager unpack in normalize_event_stream; the PR already avoids re-materializing the stream, so buffer-side binary reuse survives).

[cosmo0920]

The last commit could be optimized for performance:

| target  | skip_invalid_event | chunks/s (±)        | allocated objs/chunk |
|---------|--------------------|---------------------|----------------------|
| master  | false              |    3615458.6 ±1.4% |                  2.0 |
| pr-5411 | false              |    3414307.0 ±1.0% |                  2.0 |
| master  | true               |       1123.5 ±1.0% |              15042.0 |
| pr-5411 | true               |       1040.3 ±1.2% |              15042.0 |

main result   : skip=false  master/PR = 1.1x (PR is the slower one if > 1)
sanity check  : skip=true   PR vs master = -7.4% (should be within ~10%)

[cosmo0920]

The latest result is:

| target  | skip_invalid_event | chunks/s (±)        | allocated objs/chunk |
|---------|--------------------|---------------------|----------------------|
| master  | false              |    3562309.3 ±1.1% |                  2.0 |
| pr-5411 | false              |    3370826.6 ±1.7% |                  2.0 |
| master  | true               |       1103.6 ±1.2% |              15042.0 |
| pr-5411 | true               |       1054.2 ±1.1% |              15042.0 |

main result   : skip=false  master/PR = 1.1x (PR is the slower one if > 1)
sanity check  : skip=true   PR vs master = -4.5% (should be within ~10%)

[cosmo0920]

The latest result is:

| target  | skip_invalid_event | chunks/s (±)        | allocated objs/chunk |
|---------|--------------------|---------------------|----------------------|
| master  | false              |    3506260.6 ±0.8% |                  2.0 |
| pr-5411 | false              |    3495848.0 ±1.6% |                  1.0 |
| master  | true               |       1085.3 ±1.0% |              15042.0 |
| pr-5411 | true               |       1023.8 ±1.1% |              15041.0 |

main result   : skip=false  master/PR = 1.0x (PR is the slower one if > 1)
sanity check  : skip=true   PR vs master = -5.7% (should be within ~10%)

Allocated object counts are reduced but the performance should be in-line.

[cosmo0920]

I refactored to put the right place to process the events which are enabled for metadata encapsulated timestamps.

The latest result is:

With Ruby 3.2.2

fluentd master : 99c7074fcdcc05b89328516881564ef8d0a3dba0
fluentd pr-5411: c2d251c66d4dcedf16db4e590c8e6456d80907b9
ruby           : ruby 3.2.2 (2023-03-30 revision e51014f9c0) [arm64-darwin23]

| target  | skip_invalid_event | chunks/s (±)        | allocated objs/chunk |
|---------|--------------------|---------------------|----------------------|
| master  | false              |    3607989.0 ±1.8% |                  2.0 |
| pr-5411 | false              |    3552383.1 ±1.6% |                  1.0 |
| master  | true               |       1049.7 ±4.9% |              15042.0 |
| pr-5411 | true               |       1059.6 ±1.3% |              15041.0 |

main result   : skip=false  master/PR = 1.0x (PR is the slower one if > 1)
sanity check  : skip=true   PR vs master = +0.9% (should be within ~10%)

With Ruby 4.0.5:

fluentd master : 99c7074fcdcc05b89328516881564ef8d0a3dba0
fluentd pr-5411: c2d251c66d4dcedf16db4e590c8e6456d80907b9
ruby           : ruby 4.0.5 (2026-05-20 revision 64336ffd0e) +PRISM [arm64-darwin25]

| target  | skip_invalid_event | chunks/s (±)        | allocated objs/chunk |
|---------|--------------------|---------------------|----------------------|
| master  | false              |    3111480.0 ±1.1% |                  5.0 |
| pr-5411 | false              |    3815331.2 ±0.9% |                  1.0 |
| master  | true               |       1203.4 ±0.9% |              14043.0 |
| pr-5411 | true               |       1129.3 ±1.1% |              14039.0 |

main result   : skip=false  master/PR = 0.8x (PR is the slower one if > 1)
sanity check  : skip=true   PR vs master = -6.2% (should be within ~10%)

With this PR, it sometimes became faster than w/o patched version.