in_statsd: Implement metrics type of ingestion way

[cosmo0920]

---

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 statsd -p metrics=On -o stdout -v

• Debug log output from testing the change

Fluent Bit v3.1.6
* 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/08/14 17:18:01] [ info] Configuration:
[2024/08/14 17:18:01] [ info]  flush time     | 1.000000 seconds
[2024/08/14 17:18:01] [ info]  grace          | 5 seconds
[2024/08/14 17:18:01] [ info]  daemon         | 0
[2024/08/14 17:18:01] [ info] ___________
[2024/08/14 17:18:01] [ info]  inputs:
[2024/08/14 17:18:01] [ info]      statsd
[2024/08/14 17:18:01] [ info] ___________
[2024/08/14 17:18:01] [ info]  filters:
[2024/08/14 17:18:01] [ info] ___________
[2024/08/14 17:18:01] [ info]  outputs:
[2024/08/14 17:18:01] [ info]      stdout.0
[2024/08/14 17:18:01] [ info] ___________
[2024/08/14 17:18:01] [ info]  collectors:
[2024/08/14 17:18:01] [ info] [fluent bit] version=3.1.6, commit=5b663fde5e, pid=2043159
[2024/08/14 17:18:01] [debug] [engine] coroutine stack size: 24576 bytes (24.0K)
[2024/08/14 17:18:01] [ info] [storage] ver=1.1.6, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2024/08/14 17:18:01] [ info] [cmetrics] version=0.9.4
[2024/08/14 17:18:01] [ info] [ctraces ] version=0.5.5
[2024/08/14 17:18:01] [ info] [input:statsd:statsd.0] initializing
[2024/08/14 17:18:01] [ info] [output:stdout:stdout.0] worker #0 started
[2024/08/14 17:18:01] [ info] [input:statsd:statsd.0] storage_strategy='memory' (memory only)
[2024/08/14 17:18:01] [debug] [statsd:statsd.0] created event channels: read=21 write=22
[2024/08/14 17:18:01] [ info] [input:statsd:statsd.0] start UDP server on 0.0.0.0:8125
[2024/08/14 17:18:01] [debug] [stdout:stdout.0] created event channels: read=24 write=25
[2024/08/14 17:18:01] [ info] [sp] stream processor started
2024-08-14T08:18:03.562424694Z click{incremental="true",hello="tag"} = 1000
2024-08-14T08:18:03.591477424Z active{incremental="true"} = 9900
2024-08-14T08:18:03.593118033Z inactive{hi="from_fluent-bit"} = 2320
[2024/08/14 17:18:04] [debug] [task] created task=0x5f114e0 id=0 OK
[2024/08/14 17:18:04] [debug] [output:stdout:stdout.0] task_id=0 assigned to thread #0
[2024/08/14 17:18:04] [debug] [output:stdout:stdout.0] cmt decode msgpack returned : 1
[2024/08/14 17:18:04] [debug] [out flush] cb_destroy coro_id=0
[2024/08/14 17:18:04] [debug] [task] destroy task=0x5f114e0 (task_id=0)
^C[2024/08/14 17:18:05] [engine] caught signal (SIGINT)
[2024/08/14 17:18:05] [ warn] [engine] service will shutdown in max 5 seconds
[2024/08/14 17:18:05] [ info] [input] pausing statsd.0
[2024/08/14 17:18:06] [ info] [engine] service has stopped (0 pending tasks)
[2024/08/14 17:18:06] [ info] [input] pausing statsd.0
[2024/08/14 17:18:06] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2024/08/14 17:18:06] [ info] [output:stdout:stdout.0] thread worker #0 stopped

Sending Multiple Payloads from Another Terminal

$ echo "click:+10|c|@0.01|#hello:tag" | nc -q0 -u 127.0.0.1 8125
  echo "active:+99|g|@0.01"     | nc -q0 -u 127.0.0.1 8125
 echo "inactive:29|g|@0.0125|#hi:from_fluent-bit"     | nc -q0 -u 127.0.0.1 8125

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

==2043159== 
==2043159== HEAP SUMMARY:
==2043159==     in use at exit: 0 bytes in 0 blocks
==2043159==   total heap usage: 3,142 allocs, 3,142 frees, 853,189 bytes allocated
==2043159== 
==2043159== All heap blocks were freed -- no leaks are possible
==2043159== 
==2043159== For lists of detected and suppressed errors, rerun with: -s
==2043159== 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

TBD

Backporting

• Backport to latest stable release.

Additional Note

This PR needs to ingest multiple times per a cycle.
So, the metrics contexts of buffers sometimes concatenated and to be needed to loop to consume EOF.
This should be handled in #9118 and #9122.

This PR also depends on fluent/cmetrics#210.

---

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]

note: I've cancelled Appveyor tests since they are taking too long, windows was validated by github