out_websocket: Add ability to provide additional headers

[markusthoemmes]

This adds the ability to add additional headers to the websocket handshake, identical to how we specify headers in out_http. This allows users of the websocket output to handshake with an authenticated endpoint for example.

Fixes #8367

---

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

Example configuration

[INPUT]
    Name   dummy

[OUTPUT]
    Name websocket
    Match *
    Host foo.bar.com
    URI /my/path
    Port 443
    Format json_lines
    Header Authorization bearer my_token
    TLS on

Debug log

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

___________.__                        __    __________.__  __          ________  
\_   _____/|  |  __ __   ____   _____/  |_  \______   \__|/  |_  ___  _\_____  \ 
 |    __)  |  | |  |  \_/ __ \ /    \   __\  |    |  _/  \   __\ \  \/ / _(__  < 
 |     \   |  |_|  |  /\  ___/|   |  \  |    |    |   \  ||  |    \   / /       \
 \___  /   |____/____/  \___  >___|  /__|    |______  /__||__|     \_/ /______  /
     \/                     \/     \/               \/                        \/ 

[2024/03/01 11:30:22] [ info] Configuration:
[2024/03/01 11:30:22] [ info]  flush time     | 1.000000 seconds
[2024/03/01 11:30:22] [ info]  grace          | 5 seconds
[2024/03/01 11:30:22] [ info]  daemon         | 0
[2024/03/01 11:30:22] [ info] ___________
[2024/03/01 11:30:22] [ info]  inputs:
[2024/03/01 11:30:22] [ info]      cpu
[2024/03/01 11:30:22] [ info] ___________
[2024/03/01 11:30:22] [ info]  filters:
[2024/03/01 11:30:22] [ info] ___________
[2024/03/01 11:30:22] [ info]  outputs:
[2024/03/01 11:30:22] [ info]      websocket.0
[2024/03/01 11:30:22] [ info] ___________
[2024/03/01 11:30:22] [ info]  collectors:
[2024/03/01 11:30:22] [ info] [fluent bit] version=3.0.0, commit=190db4bf45, pid=43734
[2024/03/01 11:30:22] [debug] [engine] coroutine stack size: 196608 bytes (192.0K)
[2024/03/01 11:30:22] [ info] [storage] ver=1.5.1, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2024/03/01 11:30:22] [ info] [cmetrics] version=0.7.0
[2024/03/01 11:30:22] [ info] [ctraces ] version=0.4.0
[2024/03/01 11:30:22] [ info] [input:cpu:cpu.0] initializing
[2024/03/01 11:30:22] [ info] [input:cpu:cpu.0] storage_strategy='memory' (memory only)
[2024/03/01 11:30:22] [debug] [cpu:cpu.0] created event channels: read=25 write=26
[2024/03/01 11:30:22] [debug] [websocket:websocket.0] created event channels: read=27 write=28
[2024/03/01 11:30:22] [ info] [out_ws] we have following parameter <redacted>, <redacted>, 443, 25
[2024/03/01 11:30:22] [ info] [sp] stream processor started
[2024/03/01 11:30:23] [debug] [input chunk] update output instances with new chunk size diff=719, records=1, input=cpu.0
[2024/03/01 11:30:24] [debug] [task] created task=0xffffa0042830 id=0 OK
[2024/03/01 11:30:24] [debug] [input chunk] update output instances with new chunk size diff=719, records=1, input=cpu.0
[2024/03/01 11:30:24] [debug] [upstream] KA connection #35 to <redacted> is connected
[2024/03/01 11:30:24] [debug] [out_ws] interval is  2 and handshake is 1
[2024/03/01 11:30:24] [ info] [out_ws] handshake for ws
[2024/03/01 11:30:24] [debug] [http_client] not using http_proxy for header
[2024/03/01 11:30:24] [debug] [upstream] KA connection #35 to <redacted> is now available
[2024/03/01 11:30:24] [debug] [out flush] cb_destroy coro_id=0
[2024/03/01 11:30:24] [debug] [task] destroy task=0xffffa0042830 (task_id=0)
[2024/03/01 11:30:25] [debug] [task] created task=0xffffa0056fc0 id=0 OK
[2024/03/01 11:30:25] [debug] [input chunk] update output instances with new chunk size diff=719, records=1, input=cpu.0
[2024/03/01 11:30:25] [debug] [upstream] KA connection #35 to <redacted> has been assigned (recycled)
[2024/03/01 11:30:25] [debug] [out_ws] interval is  1 and handshake is 0
[2024/03/01 11:30:25] [debug] [upstream] KA connection #35 to <redacted> is now available
[2024/03/01 11:30:25] [debug] [out flush] cb_destroy coro_id=1
[2024/03/01 11:30:25] [debug] [task] destroy task=0xffffa0056fc0 (task_id=0)
[2024/03/01 11:30:26] [debug] [task] created task=0xffffa0057150 id=0 OK
[2024/03/01 11:30:26] [debug] [input chunk] update output instances with new chunk size diff=719, records=1, input=cpu.0
[2024/03/01 11:30:26] [debug] [upstream] KA connection #35 to <redacted> has been assigned (recycled)
[2024/03/01 11:30:26] [debug] [out_ws] interval is  1 and handshake is 0
[2024/03/01 11:30:26] [debug] [upstream] KA connection #35 to <redacted> is now available
[2024/03/01 11:30:26] [debug] [out flush] cb_destroy coro_id=2
[2024/03/01 11:30:26] [debug] [task] destroy task=0xffffa0057150 (task_id=0)
[2024/03/01 11:30:27] [debug] [task] created task=0xffffa0057290 id=0 OK
[2024/03/01 11:30:27] [debug] [input chunk] update output instances with new chunk size diff=719, records=1, input=cpu.0
[2024/03/01 11:30:27] [debug] [upstream] KA connection #35 to <redacted> has been assigned (recycled)
[2024/03/01 11:30:27] [debug] [out_ws] interval is  1 and handshake is 0
[2024/03/01 11:30:27] [debug] [upstream] KA connection #35 to <redacted> is now available
[2024/03/01 11:30:27] [debug] [out flush] cb_destroy coro_id=3
[2024/03/01 11:30:27] [debug] [task] destroy task=0xffffa0057290 (task_id=0)
[2024/03/01 11:30:28] [debug] [task] created task=0xffffa0055400 id=0 OK
[2024/03/01 11:30:28] [debug] [input chunk] update output instances with new chunk size diff=719, records=1, input=cpu.0
[2024/03/01 11:30:28] [debug] [upstream] KA connection #35 to <redacted> has been assigned (recycled)
[2024/03/01 11:30:28] [debug] [out_ws] interval is  1 and handshake is 0
[2024/03/01 11:30:28] [debug] [upstream] KA connection #35 to <redacted> is now available
[2024/03/01 11:30:28] [debug] [out flush] cb_destroy coro_id=4
[2024/03/01 11:30:28] [debug] [task] destroy task=0xffffa0055400 (task_id=0)
[2024/03/01 11:30:28] [engine] caught signal (SIGINT)
[2024/03/01 11:30:28] [debug] [task] created task=0xffffa0057340 id=0 OK
[2024/03/01 11:30:28] [ warn] [engine] service will shutdown in max 5 seconds
[2024/03/01 11:30:28] [ info] [input] pausing cpu.0
[2024/03/01 11:30:28] [debug] [upstream] KA connection #35 to <redacted> has been assigned (recycled)
[2024/03/01 11:30:28] [debug] [out_ws] interval is  0 and handshake is 0
[2024/03/01 11:30:28] [debug] [upstream] KA connection #35 to <redacted> is now available
[2024/03/01 11:30:28] [debug] [out flush] cb_destroy coro_id=5
[2024/03/01 11:30:28] [debug] [task] destroy task=0xffffa0057340 (task_id=0)
[2024/03/01 11:30:29] [ info] [engine] service has stopped (0 pending tasks)
[2024/03/01 11:30:29] [ info] [input] pausing cpu.0
[2024/03/01 11:30:29] [ info] [out_ws] flb_ws_conf_destroy 

Valgrind log

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

___________.__                        __    __________.__  __          ________  
\_   _____/|  |  __ __   ____   _____/  |_  \______   \__|/  |_  ___  _\_____  \ 
 |    __)  |  | |  |  \_/ __ \ /    \   __\  |    |  _/  \   __\ \  \/ / _(__  < 
 |     \   |  |_|  |  /\  ___/|   |  \  |    |    |   \  ||  |    \   / /       \
 \___  /   |____/____/  \___  >___|  /__|    |______  /__||__|     \_/ /______  /
     \/                     \/     \/               \/                        \/ 

[2024/03/01 11:38:56] [ info] [fluent bit] version=3.0.0, commit=190db4bf45, pid=45238
[2024/03/01 11:38:56] [ info] [storage] ver=1.5.1, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2024/03/01 11:38:56] [ info] [cmetrics] version=0.7.0
[2024/03/01 11:38:56] [ info] [ctraces ] version=0.4.0
[2024/03/01 11:38:56] [ info] [input:dummy:dummy.0] initializing
[2024/03/01 11:38:56] [ info] [input:dummy:dummy.0] storage_strategy='memory' (memory only)
[2024/03/01 11:38:56] [ info] [out_ws] we have following parameter <redacted>, <redacted>, 443, 25
[2024/03/01 11:38:56] [ info] [sp] stream processor started
[2024/03/01 11:38:58] [ info] [out_ws] handshake for ws
[2024/03/01 11:39:04] [engine] caught signal (SIGINT)
[2024/03/01 11:39:04] [ warn] [engine] service will shutdown in max 5 seconds
[2024/03/01 11:39:04] [ info] [input] pausing dummy.0
[2024/03/01 11:39:05] [ info] [engine] service has stopped (0 pending tasks)
[2024/03/01 11:39:05] [ info] [input] pausing dummy.0
[2024/03/01 11:39:05] [ info] [out_ws] flb_ws_conf_destroy 
==45238== 
==45238== HEAP SUMMARY:
==45238==     in use at exit: 0 bytes in 0 blocks
==45238==   total heap usage: 8,521 allocs, 8,521 frees, 4,986,514 bytes allocated
==45238== 
==45238== All heap blocks were freed -- no leaks are possible
==45238== 
==45238== For lists of detected and suppressed errors, rerun with: -s
==45238== 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.

• [N/A] Run local packaging test showing all targets (including any new ones) build.
• [N/A] 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.

[markusthoemmes]

Any chance to get a review on this one? @edsiper maybe?

[markusthoemmes]

Poke on this one again 🙏

[edsiper]

@markusthoemmes thanks for this contribution, adding it for next release. would you please also provide a PR for fluent-bit-docs with further docs ? (cc: @lecaros )