go: proxy: plugin: Implement multi-threaded Golang input plugin mechanism

[cosmo0920]

This PR provides Golang input plugin mechanism on Fluent Bit core.
This work is based on multi-threaded input plugin mechanism that is implemented on #4586.

The related issue is None.

---

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

fluent-bit.conf

[SERVICE]
    Flush        5
    Daemon       Off
    Log_Level    trace
    Plugins_File plugins.conf
    HTTP_Server  Off
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

[INPUT]
    Name gdummy
    Tag  gdummy.local

[OUTPUT]
    Name  stdout
    Match gdummy*

plugins.conf

[PLUGINS]
  Path /path/to/workdir/fluent-bit/build/in_gdummy.so

• Debug log output from testing the change
  This patch works with fluent-bit-go and corresponds a shared object that is built with Golang input plugin interface.
  This is also sent a PR there: input: Implement Golang Input plugin interface and added input example fluent-bit-go#52

$  bin/fluent-bit -c fluent-bit.conf
Fluent Bit v1.9.0
* Copyright (C) 2015-2021 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2022/03/14 10:36:55] [ info] Configuration:
[2022/03/14 10:36:55] [ info]  flush time     | 5.000000 seconds
[2022/03/14 10:36:55] [ info]  grace          | 5 seconds
[2022/03/14 10:36:55] [ info]  daemon         | 0
[2022/03/14 10:36:55] [ info] ___________
[2022/03/14 10:36:55] [ info]  inputs:
[2022/03/14 10:36:55] [ info]      gdummy
[2022/03/14 10:36:55] [ info] ___________
[2022/03/14 10:36:55] [ info]  filters:
[2022/03/14 10:36:55] [ info] ___________
[2022/03/14 10:36:55] [ info]  outputs:
[2022/03/14 10:36:55] [ info]      stdout.0
[2022/03/14 10:36:55] [ info] ___________
[2022/03/14 10:36:55] [ info]  collectors:
[2022/03/14 10:36:55] [ info] [engine] started (pid=1062825)
[2022/03/14 10:36:55] [debug] [engine] coroutine stack size: 24576 bytes (24.0K)
[2022/03/14 10:36:55] [debug] [storage] [cio stream] new stream registered: gdummy.0
[2022/03/14 10:36:55] [ info] [storage] version=1.1.6, initializing...
[2022/03/14 10:36:55] [ info] [storage] in-memory
[2022/03/14 10:36:55] [ info] [storage] normal synchronization mode, checksum disabled, max_chunks_up=128
[2022/03/14 10:36:55] [ info] [cmetrics] version=0.3.0
[2022/03/14 10:36:55] [debug] [gdummy:gdummy.0] created event channels: read=21 write=22
[flb-go] plugin parameter = ''
[2022/03/14 10:36:55] [debug] [stdout:stdout.0] created event channels: read=25 write=26
[2022/03/14 10:36:55] [debug] [router] match rule gdummy.0:stdout.0
[2022/03/14 10:36:55] [ info] [sp] stream processor started
[2022/03/14 10:36:55] [ info] [output:stdout:stdout.0] worker #0 started
[2022/03/14 10:36:56] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:36:56] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:36:57] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:36:57] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:36:58] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:36:58] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:36:59] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:36:59] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:37:00] [debug] [task] created task=0x7f94d401d5d0 id=0 OK
[2022/03/14 10:37:00] [debug] [output:stdout:stdout.0] task_id=0 assigned to thread #0
[0] gdummy.local: [1647221815.557805574, {"message"=>"dummy"}]
[1] gdummy.local: [1647221816.558982897, {"message"=>"dummy"}]
[2] gdummy.local: [1647221817.559238836, {"message"=>"dummy"}]
[3] gdummy.local: [1647221818.559523525, {"message"=>"dummy"}]
[2022/03/14 10:37:00] [debug] [out flush] cb_destroy coro_id=0
[2022/03/14 10:37:00] [debug] [task] destroy task=0x7f94d401d5d0 (task_id=0)
[2022/03/14 10:37:00] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:37:00] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:37:01] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:37:01] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:37:02] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:37:02] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:37:03] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:37:03] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:37:04] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:37:04] [debug] [input chunk] update output instances with new chunk size diff=26
[2022/03/14 10:37:05] [debug] [task] created task=0x7f94d405d830 id=0 OK
[2022/03/14 10:37:05] [debug] [output:stdout:stdout.0] task_id=0 assigned to thread #0
[0] gdummy.local: [1647221819.560334706, {"message"=>"dummy"}]
[1] gdummy.local: [1647221820.560606884, {"message"=>"dummy"}]
[2] gdummy.local: [1647221821.561860932, {"message"=>"dummy"}]
[3] gdummy.local: [1647221822.562044770, {"message"=>"dummy"}]
[4] gdummy.local: [1647221823.562335594, {"message"=>"dummy"}]
[2022/03/14 10:37:05] [debug] [out flush] cb_destroy coro_id=1
[2022/03/14 10:37:05] [debug] [task] destroy task=0x7f94d405d830 (task_id=0)
[2022/03/14 10:37:05] [trace] [input:gdummy:gdummy.0 at build/src/CMakeFiles/fluent-bit-static.dir/compiler_depend.ts:93] input thread read() = 26
[2022/03/14 10:37:05] [debug] [input chunk] update output instances with new chunk size diff=26
^C[2022/03/14 10:37:06] [engine] caught signal (SIGINT)
[2022/03/14 10:37:06] [debug] [task] created task=0x7f94d401d870 id=0 OK
[2022/03/14 10:37:06] [debug] [output:stdout:stdout.0] task_id=0 assigned to thread #0
[2022/03/14 10:37:06] [ warn] [engine] service will shutdown in max 5 seconds
[0] gdummy.local: [1647221824.562711972, {"message"=>"dummy"}]
[2022/03/14 10:37:06] [debug] [out flush] cb_destroy coro_id=2
[2022/03/14 10:37:06] [debug] [task] destroy task=0x7f94d401d870 (task_id=0)
[2022/03/14 10:37:06] [ info] [engine] service has stopped (0 pending tasks)
[2022/03/14 10:37:06] [debug] [GO] running exit callback
[2022/03/14 10:37:06] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2022/03/14 10:37:06] [ info] [output:stdout:stdout.0] thread worker #0 stopped

• [N/A] 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.

• [N/A] Attached local packaging test output showing all targets (including any new ones) build.

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]

Recently we merged a new approach to have native threaded input plugins; parts of the design consider:

• input plugins are not aware of threading. The engine runs them in a separate thread if the feature has been enabled.
• plugins ingest data as usual with the flb_input_chunk_append_raw() call, but that function detects if the caller comes from a thread or not; if so, it will use a ring buffer mechanism to ingest data.

There are two ways for an input plugin to specify that it needs to be run in a separate thread:

• configuration: if the input section enabled the flag threaded on, the plugin will run in a separate thread.

• plugin registration: plugins have a registration structure and a .flags field to hint the engine about its operational requirements. If the plugin desires to always run in a separate thread, the flag field needs to get appended with the value FLB_INPUT_THREADED.

As an example take a look at the new in_event_test plugin that sets the flag .flags FLB_INPUT_THREADED:

https://github.com/fluent/fluent-bit/blob/master/plugins/in_event_test/event_test.c#L404

The logic of this new proxy interface needs to be adapted, actually I think it will be simplified since no thread management will required, so we can make sure that any plugin that gets connected through the proxy interface by default runs in a separate thread.

[cosmo0920]

As an example take a look at the new in_event_test plugin that sets the flag .flags FLB_INPUT_THREADED:

https://github.com/fluent/fluent-bit/blob/master/plugins/in_event_test/event_test.c#L404

The logic of this new proxy interface needs to be adapted, actually I think it will be simplified since no thread management will required, so we can make sure that any plugin that gets connected through the proxy interface by default runs in a separate thread.

I already implemented this logic for plugin proxy.
@edsiper Any obstacles to start reviewing this PR?

[niedbalski]

@edsiper this PR needs your approval. I just +1 it.