out_prometheus_exporter: Handle multiply concatenated metrics type of events

[cosmo0920]

Like as ctraces, there is a possibility to handle concatenated metrics type of events payloads during processing #9119.
This is because in_statsd of metrics ingestion mode requests to handle cmetrics payloads which is ingested events more one times per a cycle.
Currently, we need to handle manually for multiply concatenated cmetrics payloads to handle multiple ingestion for metrics type of events.

---

Enter [N/A] in the box, if an item is not applicable to your change.

Testing
Before we can approve your change; please submit the following in a comment:

• Example configuration file for the change

$ bin/fluent-bit -i fluentbit_metrics -o prometheus_exporter -v

• Debug log output from testing the change

* Copyright (C) 2015-2024 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

______ _                  _    ______ _ _           _____  __  
|  ___| |                | |   | ___ (_) |         |____ |/  | 
| |_  | |_   _  ___ _ __ | |_  | |_/ /_| |_  __   __   / /`| | 
|  _| | | | | |/ _ \ '_ \| __| | ___ \ | __| \ \ / /   \ \ | | 
| |   | | |_| |  __/ | | | |_  | |_/ / | |_   \ V /.___/ /_| |_
\_|   |_|\__,_|\___|_| |_|\__| \____/|_|\__|   \_/ \____(_)___/

[2024/07/23 16:50:08] [ info] Configuration:
[2024/07/23 16:50:08] [ info]  flush time     | 1.000000 seconds
[2024/07/23 16:50:08] [ info]  grace          | 5 seconds
[2024/07/23 16:50:08] [ info]  daemon         | 0
[2024/07/23 16:50:08] [ info] ___________
[2024/07/23 16:50:08] [ info]  inputs:
[2024/07/23 16:50:08] [ info]      fluentbit_metrics
[2024/07/23 16:50:08] [ info] ___________
[2024/07/23 16:50:08] [ info]  filters:
[2024/07/23 16:50:08] [ info] ___________
[2024/07/23 16:50:08] [ info]  outputs:
[2024/07/23 16:50:08] [ info]      prometheus_exporter.0
[2024/07/23 16:50:08] [ info] ___________
[2024/07/23 16:50:08] [ info]  collectors:
[2024/07/23 16:50:08] [ info] [fluent bit] version=3.1.4, commit=82a222f60f, pid=3598557
[2024/07/23 16:50:08] [debug] [engine] coroutine stack size: 24576 bytes (24.0K)
[2024/07/23 16:50:08] [ info] [storage] ver=1.1.6, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2024/07/23 16:50:08] [ info] [cmetrics] version=0.9.1
[2024/07/23 16:50:08] [ info] [ctraces ] version=0.5.2
[2024/07/23 16:50:08] [ info] [input:fluentbit_metrics:fluentbit_metrics.0] initializing
[2024/07/23 16:50:08] [ info] [input:fluentbit_metrics:fluentbit_metrics.0] storage_strategy='memory' (memory only)
[2024/07/23 16:50:08] [debug] [fluentbit_metrics:fluentbit_metrics.0] created event channels: read=21 write=22
[2024/07/23 16:50:08] [debug] [prometheus_exporter:prometheus_exporter.0] created event channels: read=23 write=24
[2024/07/23 16:50:08] [ info] [output:prometheus_exporter:prometheus_exporter.0] listening iface=0.0.0.0 tcp_port=2021
[2024/07/23 16:50:08] [ info] [sp] stream processor started
[2024/07/23 16:50:11] [debug] [task] created task=0x5f1bf90 id=0 OK
[2024/07/23 16:50:11] [debug] [out flush] cb_destroy coro_id=0
[2024/07/23 16:50:11] [debug] [task] destroy task=0x5f1bf90 (task_id=0)
[2024/07/23 16:50:13] [debug] [task] created task=0x8c95a50 id=0 OK
[2024/07/23 16:50:13] [debug] [out flush] cb_destroy coro_id=1
[2024/07/23 16:50:13] [debug] [task] destroy task=0x8c95a50 (task_id=0)
^C[2024/07/23 16:50:14] [engine] caught signal (SIGINT)
[2024/07/23 16:50:14] [ warn] [engine] service will shutdown in max 5 seconds
[2024/07/23 16:50:14] [ info] [input] pausing fluentbit_metrics.0
[2024/07/23 16:50:14] [ info] [engine] service has stopped (0 pending tasks)
[2024/07/23 16:50:14] [ info] [input] pausing fluentbit_metrics.0

• Attached Valgrind output that shows no leaks or memory corruption was found

==3598557== 
==3598557== HEAP SUMMARY:
==3598557==     in use at exit: 0 bytes in 0 blocks
==3598557==   total heap usage: 8,750 allocs, 8,750 frees, 11,432,040 bytes allocated
==3598557== 
==3598557== All heap blocks were freed -- no leaks are possible
==3598557== 
==3598557== For lists of detected and suppressed errors, rerun with: -s
==3598557== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

If this is a change to packaging of containers or native binaries then please confirm it works for all targets.

• Run local packaging test showing all targets (including any new ones) build.
• Set ok-package-test label to test for all targets (requires maintainer to do).

Documentation

• Documentation required for this feature

Backporting

• Backport to latest stable release.

---

Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.

[edsiper]

I am wondering if this logic should be handled inside cmetrics in the prometheus encoder...

[cosmo0920]

I am wondering if this logic should be handled inside cmetrics in the prometheus encoder...

I don't think so. Because decoding msgpack should be placed in the msgpack decoder in cmetrics. And, it should be processing with offsets. Current implementation is only processing until reaching the first offset in msgpack decoder.
Prometheus encoder should handle entire cmetrics context but the current issue is only decoded for concatenated the first one in the msgpack decoder.

[cosmo0920]

FYI: out_prometheus_remote_write plugin uses similar pattern to decode multiply concatenated cmetrics context: https://github.com/fluent/fluent-bit/blob/master/plugins/out_prometheus_remote_write/remote_write.c#L319

[edsiper]

note: when submitting, please rework the commits to have a clean history