out_forward: support 'unix_path'

[nokute78]

Fixes #2181

This patch is to add 'unix_path' to forward via unix socket.

Key	Description	Default
Unix_Path	Specify the path to unix socket to send a Forward message. If set, Upstream is ignored.

---

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.

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

Documentation

• Documentation required for this feature

Backporting

• [N/A] Backport to latest stable release.

Configuration

1. fluent-bit -c server.conf
2. fluent-bit -c client.conf

server.conf:

[INPUT]
    Name forward
    unix_path hoge.sock

[OUTPUT]
    Name stdout

client.conf:

[INPUT]
    Name cpu

[OUTPUT]
    Name forward
    unix_path hoge.sock

Debug output

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

[2022/01/29 08:27:01] [ info] [engine] started (pid=5692)
[2022/01/29 08:27:01] [ info] [storage] version=1.1.5, initializing...
[2022/01/29 08:27:01] [ info] [storage] in-memory
[2022/01/29 08:27:01] [ info] [storage] normal synchronization mode, checksum disabled, max_chunks_up=128
[2022/01/29 08:27:01] [ info] [cmetrics] version=0.2.2
[2022/01/29 08:27:01] [ info] [input:forward:forward.0] listening on unix://hoge.sock
[2022/01/29 08:27:01] [ info] [sp] stream processor started
[0] cpu.0: [1643412427.909546068, {"cpu_p"=>1.000000, "user_p"=>1.000000, "system_p"=>0.000000, "cpu0.p_cpu"=>1.000000, "cpu0.p_user"=>1.000000, "cpu0.p_system"=>0.000000}]
[1] cpu.0: [1643412428.910456124, {"cpu_p"=>2.000000, "user_p"=>2.000000, "system_p"=>0.000000, "cpu0.p_cpu"=>2.000000, "cpu0.p_user"=>2.000000, "cpu0.p_system"=>0.000000}]
[2] cpu.0: [1643412429.910156104, {"cpu_p"=>2.000000, "user_p"=>2.000000, "system_p"=>0.000000, "cpu0.p_cpu"=>2.000000, "cpu0.p_user"=>2.000000, "cpu0.p_system"=>0.000000}]
[3] cpu.0: [1643412430.909589692, {"cpu_p"=>1.000000, "user_p"=>1.000000, "system_p"=>0.000000, "cpu0.p_cpu"=>1.000000, "cpu0.p_user"=>1.000000, "cpu0.p_system"=>0.000000}]
^C[2022/01/29 08:27:18] [engine] caught signal (SIGINT)
[2022/01/29 08:27:18] [ info] [input] pausing forward.0
[0] cpu.0: [1643412431.910034126, {"cpu_p"=>2.000000, "user_p"=>1.000000, "system_p"=>1.000000, "cpu0.p_cpu"=>2.000000, "cpu0.p_user"=>1.000000, "cpu0.p_system"=>1.000000}]
[1] cpu.0: [1643412432.909600487, {"cpu_p"=>0.000000, "user_p"=>0.000000, "system_p"=>0.000000, "cpu0.p_cpu"=>0.000000, "cpu0.p_user"=>0.000000, "cpu0.p_system"=>0.000000}]
[2] cpu.0: [1643412433.909575726, {"cpu_p"=>1.000000, "user_p"=>1.000000, "system_p"=>0.000000, "cpu0.p_cpu"=>1.000000, "cpu0.p_user"=>1.000000, "cpu0.p_system"=>0.000000}]
[3] cpu.0: [1643412434.909893060, {"cpu_p"=>29.000000, "user_p"=>28.000000, "system_p"=>1.000000, "cpu0.p_cpu"=>29.000000, "cpu0.p_user"=>28.000000, "cpu0.p_system"=>1.000000}]
[4] cpu.0: [1643412435.909501993, {"cpu_p"=>2.000000, "user_p"=>1.000000, "system_p"=>1.000000, "cpu0.p_cpu"=>2.000000, "cpu0.p_user"=>1.000000, "cpu0.p_system"=>1.000000}]
[2022/01/29 08:27:18] [ warn] [engine] service will shutdown in max 5 seconds
[2022/01/29 08:27:18] [ info] [engine] service has stopped (0 pending tasks)

Valgrind output

$ valgrind --leak-check=full bin/fluent-bit -c client.conf 
==5706== Memcheck, a memory error detector
==5706== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==5706== Using Valgrind-3.15.0 and LibVEX; rerun with -h for copyright info
==5706== Command: bin/fluent-bit -c client.conf
==5706== 
Fluent Bit v1.9.0
* Copyright (C) 2019-2021 The Fluent Bit Authors
* Copyright (C) 2015-2018 Treasure Data
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2022/01/29 08:29:19] [ info] [engine] started (pid=5706)
[2022/01/29 08:29:19] [ info] [storage] version=1.1.5, initializing...
[2022/01/29 08:29:19] [ info] [storage] in-memory
[2022/01/29 08:29:19] [ info] [storage] normal synchronization mode, checksum disabled, max_chunks_up=128
[2022/01/29 08:29:19] [ info] [cmetrics] version=0.2.2
[2022/01/29 08:29:19] [ info] [sp] stream processor started
^C[2022/01/29 08:29:29] [engine] caught signal (SIGINT)
[2022/01/29 08:29:29] [ info] [input] pausing cpu.0
[2022/01/29 08:29:29] [ warn] [engine] service will shutdown in max 5 seconds
[2022/01/29 08:29:29] [ info] [engine] service has stopped (0 pending tasks)
==5706== 
==5706== HEAP SUMMARY:
==5706==     in use at exit: 0 bytes in 0 blocks
==5706==   total heap usage: 1,154 allocs, 1,154 frees, 1,420,881 bytes allocated
==5706== 
==5706== All heap blocks were freed -- no leaks are possible
==5706== 
==5706== For lists of detected and suppressed errors, rerun with: -s
==5706== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

---

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.

[nokute78]

Doc PR: fluent/fluent-bit-docs#689

[nokute78]

Ping.

Cc @PettitWesley

[PettitWesley]

why create local wrappers for these?

[nokute78]

It is abstraction layer to hide network/socket API.

I think it exposes args too much since we need to set unused args (e.g. flb_upstream_conn for unix socket)
It may need to define a new struct to hide args and it can be exposed utility function.

[PettitWesley]

Thank you for this!

[nokute78]

@PettitWesley Thank you for reviewing.

[github-actions]

This PR is stale because it has been open 45 days with no activity. Remove stale label or comment or this will be closed in 10 days.