input: fix crash caused by incorrectly initializing input coroutine params.

[pwhelan]

Summary

Refactor the flb_input coroutine parameters to follow the same pattern as flb_output and use the FLB_TLS set of macros.

This is not only to standardize the code but to fix a bug where fluent-bit can crash when using input coroutines. Input coroutines can initialize fast enough to reset the libco key which leads to a NULL dereference in input_pre_cb_collect leading to a crash in co_switch.

Issues

This fixes a bug found in #7954.

---

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
• Debug log output from testing the change

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

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.

[pwhelan]

Here is a valgrind log:

valgrind --leak-check=full ./bin/fluent-bit -c ke.conf
==600057== Memcheck, a memory error detector
==600057== Copyright (C) 2002-2022, and GNU GPL'd, by Julian Seward et al.
==600057== Using Valgrind-3.21.0 and LibVEX; rerun with -h for copyright info
==600057== Command: ./bin/fluent-bit -c ke.conf
==600057== 
Fluent Bit v2.1.10
* Copyright (C) 2015-2022 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2023/09/27 10:44:11] [ info] [fluent bit] version=2.1.10, commit=19a8a3056e, pid=600057
[2023/09/27 10:44:11] [ info] [storage] ver=1.5.1, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.0] API server: http://127.0.0.1:8000
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.1] API server: http://127.0.0.1:8000
[2023/09/27 10:44:11] [ info] [cmetrics] version=0.6.3
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.2] API server: http://127.0.0.1:8000
[2023/09/27 10:44:11] [ info] [ctraces ] version=0.3.1
[2023/09/27 10:44:11] [ info] [output:stdout:stdout.0] worker #0 started
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.0] initializing
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.0] storage_strategy='memory' (memory only)
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.0] thread instance initialized
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.1] initializing
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.1] storage_strategy='memory' (memory only)
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.1] thread instance initialized
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.2] initializing
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.2] storage_strategy='memory' (memory only)
[2023/09/27 10:44:11] [ info] [input:kubernetes_events:kubernetes_events.2] thread instance initialized
[2023/09/27 10:44:11] [ info] [sp] stream processor started
[2023/09/27 10:44:11] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:11] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:11] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:11] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:11] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:11] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:12] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:12] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:12] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:12] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:12] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:12] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:12] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:12] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:12] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:12] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:12] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:12] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:13] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:13] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:13] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:13] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:13] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:13] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:13] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:13] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:13] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:13] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:13] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:13] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:14] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:14] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:14] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:14] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:14] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:14] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:14] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:14] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:14] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:14] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:14] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:14] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:15] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:15] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:15] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:15] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:15] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:15] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:15] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:15] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:15] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:15] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:15] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:15] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:16] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:16] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:16] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:16] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:16] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:16] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:16] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:16] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
[2023/09/27 10:44:16] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:16] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:16] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:16] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
^C[2023/09/27 10:44:16] [engine] caught signal (SIGINT)
[2023/09/27 10:44:16] [ warn] [engine] service will shutdown in max 5 seconds
[2023/09/27 10:44:16] [ info] [engine] service has stopped (0 pending tasks)
[2023/09/27 10:44:17] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2023/09/27 10:44:17] [ info] [output:stdout:stdout.0] thread worker #0 stopped
[2023/09/27 10:44:17] [ warn] [input:kubernetes_events:kubernetes_events.1] cannot open ./kube.token.txt
[2023/09/27 10:44:17] [ warn] [input:kubernetes_events:kubernetes_events.2] cannot open ./kube.token.txt
[2023/09/27 10:44:17] [error] [input:kubernetes_events:kubernetes_events.2] failed to refresh token
[2023/09/27 10:44:17] [ warn] [input:kubernetes_events:kubernetes_events.0] cannot open ./kube.token.txt
[2023/09/27 10:44:17] [error] [input:kubernetes_events:kubernetes_events.1] failed to refresh token
[2023/09/27 10:44:17] [error] [input:kubernetes_events:kubernetes_events.0] failed to refresh token
==600057== 
==600057== HEAP SUMMARY:
==600057==     in use at exit: 0 bytes in 0 blocks
==600057==   total heap usage: 3,551 allocs, 3,551 frees, 4,874,153 bytes allocated
==600057== 
==600057== All heap blocks were freed -- no leaks are possible
==600057== 
==600057== For lists of detected and suppressed errors, rerun with: -s
==600057== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

This was run with the following fluent-bit configuration:

[INPUT]
    name                kubernetes_events
    tag                 k8s_events.*
    kube_retention_time 5m
    kube_request_limit  100
    kube_namespace      tenant-svc
    kube_token_file     ./kube.token.txt
    threaded            on
    kube_url            http://127.0.0.1:8000

[INPUT]
    name                kubernetes_events
    tag                 k8s_events.*
    kube_retention_time 5m
    kube_request_limit  100
    kube_namespace      platform-svc
    kube_token_file     ./kube.token.txt
    threaded            on
    kube_url            http://127.0.0.1:8000

[INPUT]
    name                kubernetes_events
    tag                 k8s_events.*
    kube_retention_time 5m
    kube_request_limit  100
    kube_namespace      timeseries-svc
    kube_token_file     ./kube.token.txt
    threaded            on
    kube_url            http://127.0.0.1:8000

[OUTPUT]
    Name   stdout
    Match  *
    Format json_lines

It was also tested with similar results with all of the inputs with threaded off.

[pwhelan]

I ran the code with the health input plugin (which uses input coroutines) and ran it under valgrind, by default, with threaded and threaded off:

~/P/w/f/p/build   pwhelan-fix-input-coroutine-params ±  valgrind ./bin/fluent-bit -i health -p threaded=off -p host=localhost -p port=22 -o stdout -f 1
==358494== Memcheck, a memory error detector
==358494== Copyright (C) 2002-2022, and GNU GPL'd, by Julian Seward et al.
==358494== Using Valgrind-3.21.0 and LibVEX; rerun with -h for copyright info
==358494== Command: ./bin/fluent-bit -i health -p threaded=off -p host=localhost -p port=22 -o stdout -f 1
==358494== 
Fluent Bit v2.2.0
* Copyright (C) 2015-2023 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2023/11/06 17:16:52] [ info] [fluent bit] version=2.2.0, commit=5ab7388dfe, pid=358494
[2023/11/06 17:16:52] [ info] [storage] ver=1.5.1, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2023/11/06 17:16:52] [ info] [cmetrics] version=0.6.4
[2023/11/06 17:16:52] [ info] [output:stdout:stdout.0] worker #0 started
[2023/11/06 17:16:52] [ info] [ctraces ] version=0.3.1
[2023/11/06 17:16:52] [ info] [input:health:health.0] initializing
[2023/11/06 17:16:52] [ info] [input:health:health.0] storage_strategy='memory' (memory only)
[2023/11/06 17:16:52] [ info] [sp] stream processor started
[0] health.0: [[1699301813.101271036, {}], {"alive"=>true}]
[0] health.0: [[1699301814.068157081, {}], {"alive"=>true}]
[0] health.0: [[1699301815.037417128, {}], {"alive"=>true}]
[0] health.0: [[1699301816.039248362, {}], {"alive"=>true}]
^C[2023/11/06 17:16:57] [engine] caught signal (SIGINT)
[0] health.0: [[1699301817.049944818, {}], {"alive"=>true}]
[2023/11/06 17:16:57] [ warn] [engine] service will shutdown in max 5 seconds
[2023/11/06 17:16:57] [ info] [engine] service has stopped (0 pending tasks)
[2023/11/06 17:16:57] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2023/11/06 17:16:57] [ info] [output:stdout:stdout.0] thread worker #0 stopped
==358494== 
==358494== HEAP SUMMARY:
==358494==     in use at exit: 0 bytes in 0 blocks
==358494==   total heap usage: 1,945 allocs, 1,945 frees, 2,596,597 bytes allocated
==358494== 
==358494== All heap blocks were freed -- no leaks are possible
==358494== 
==358494== For lists of detected and suppressed errors, rerun with: -s
==358494== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)
 hydra  ~/P/w/f/p/build   pwhelan-fix-input-coroutine-params ±  less -SR ../plugins/in_health/health.c 
 hydra  ~/P/w/f/p/build   pwhelan-fix-input-coroutine-params ±  valgrind ./bin/fluent-bit -i health -p host=localhost -p port=22 -o stdout -f 1
==358762== Memcheck, a memory error detector
==358762== Copyright (C) 2002-2022, and GNU GPL'd, by Julian Seward et al.
==358762== Using Valgrind-3.21.0 and LibVEX; rerun with -h for copyright info
==358762== Command: ./bin/fluent-bit -i health -p host=localhost -p port=22 -o stdout -f 1
==358762== 
Fluent Bit v2.2.0
* Copyright (C) 2015-2023 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2023/11/06 17:17:23] [ info] [fluent bit] version=2.2.0, commit=5ab7388dfe, pid=358762
[2023/11/06 17:17:23] [ info] [storage] ver=1.5.1, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2023/11/06 17:17:23] [ info] [cmetrics] version=0.6.4
[2023/11/06 17:17:23] [ info] [output:stdout:stdout.0] worker #0 started
[2023/11/06 17:17:23] [ info] [ctraces ] version=0.3.1
[2023/11/06 17:17:23] [ info] [input:health:health.0] initializing
[2023/11/06 17:17:23] [ info] [input:health:health.0] storage_strategy='memory' (memory only)
[2023/11/06 17:17:23] [ info] [sp] stream processor started
[0] health.0: [[1699301844.096657684, {}], {"alive"=>true}]
[0] health.0: [[1699301845.087902127, {}], {"alive"=>true}]
[0] health.0: [[1699301846.046412698, {}], {"alive"=>true}]
[0] health.0: [[1699301847.046901973, {}], {"alive"=>true}]
^C[2023/11/06 17:17:28] [engine] caught signal (SIGINT)
[2023/11/06 17:17:28] [ warn] [engine] service will shutdown in max 5 seconds
[0] health.0: [[1699301848.036293303, {}], {"alive"=>true}]
[2023/11/06 17:17:28] [ info] [engine] service has stopped (0 pending tasks)
[2023/11/06 17:17:28] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2023/11/06 17:17:29] [ info] [output:stdout:stdout.0] thread worker #0 stopped
==358762== 
==358762== HEAP SUMMARY:
==358762==     in use at exit: 0 bytes in 0 blocks
==358762==   total heap usage: 1,931 allocs, 1,931 frees, 2,596,292 bytes allocated
==358762== 
==358762== All heap blocks were freed -- no leaks are possible
==358762== 
==358762== For lists of detected and suppressed errors, rerun with: -s
==358762== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)
 hydra  ~/P/w/f/p/build   pwhelan-fix-input-coroutine-params ±  valgrind ./bin/fluent-bit -i health -p host=localhost -p threaded=on -p port=22 -o stdout -f 1
==358925== Memcheck, a memory error detector
==358925== Copyright (C) 2002-2022, and GNU GPL'd, by Julian Seward et al.
==358925== Using Valgrind-3.21.0 and LibVEX; rerun with -h for copyright info
==358925== Command: ./bin/fluent-bit -i health -p host=localhost -p threaded=on -p port=22 -o stdout -f 1
==358925== 
Fluent Bit v2.2.0
* Copyright (C) 2015-2023 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2023/11/06 17:17:36] [ info] [fluent bit] version=2.2.0, commit=5ab7388dfe, pid=358925
[2023/11/06 17:17:36] [ info] [storage] ver=1.5.1, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2023/11/06 17:17:36] [ info] [output:stdout:stdout.0] worker #0 started
[2023/11/06 17:17:36] [ info] [cmetrics] version=0.6.4
[2023/11/06 17:17:36] [ info] [ctraces ] version=0.3.1
[2023/11/06 17:17:36] [ info] [input:health:health.0] initializing
[2023/11/06 17:17:36] [ info] [input:health:health.0] storage_strategy='memory' (memory only)
[2023/11/06 17:17:36] [ info] [input:health:health.0] thread instance initialized
[2023/11/06 17:17:36] [ info] [sp] stream processor started
[0] health.0: [[1699301857.139485202, {}], {"alive"=>true}]
[0] health.0: [[1699301858.090059771, {}], {"alive"=>true}]
[0] health.0: [[1699301859.026312287, {}], {"alive"=>true}]
^C[2023/11/06 17:17:40] [engine] caught signal (SIGINT)
[2023/11/06 17:17:40] [ warn] [engine] service will shutdown in max 5 seconds
[0] health.0: [[1699301860.026443692, {}], {"alive"=>true}]
[2023/11/06 17:17:41] [ info] [engine] service has stopped (0 pending tasks)
[2023/11/06 17:17:41] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2023/11/06 17:17:41] [ info] [output:stdout:stdout.0] thread worker #0 stopped
==358925== 
==358925== HEAP SUMMARY:
==358925==     in use at exit: 0 bytes in 0 blocks
==358925==   total heap usage: 1,975 allocs, 1,975 frees, 2,294,696 bytes allocated
==358925== 
==358925== All heap blocks were freed -- no leaks are possible
==358925== 
==358925== For lists of detected and suppressed errors, rerun with: -s
==358925== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

[pwhelan]

I was finally able to fix the test on macos. It required calling the initialization functions for both input and output.