out_es: check if host contains port number

[nokute78]

Some cloud id format is <deployment_region>$<elasticsearch_hostname>:<port>$<kibana_hostname>.
e.g. https://github.com/elastic/beats/blob/v8.4.1/libbeat/cloudid/cloudid_test.go#L60

Current out_es fails to parse it.

[2022/09/17 09:11:25] [debug] [output:es:es.0] extracted cloud_host: 'ac31ebb90241773157043c34fd26fd46:9243.us-central1.gcp.cloud.es.io'
[2022/09/17 09:11:25] [debug] [output:es:es.0] extracted cloud_port_char: '9243.us-central1.gcp.cloud.es.io'
[2022/09/17 09:11:25] [debug] [output:es:es.0] converted cloud_port_char to port int: '9243'
[2022/09/17 09:11:25] [debug] [output:es:es.0] checked whether extracted port was null and set it to default https port or not. Outcome: '9243' and cloud_host: 'ac31ebb90241773157043c34fd26fd46'.
[2022/09/17 09:11:25] [debug] [output:es:es.0] host=ac31ebb90241773157043c34fd26fd46 port=9243 uri=/_bulk index=fluent-bit type=_doc
(snip)
[2022/09/17 09:11:27] [ warn] [net] getaddrinfo(host='ac31ebb90241773157043c34fd26fd46', err=11): Could not contact DNS servers
[2022/09/17 09:11:27] [debug] [upstream] connection #-1 failed to ac31ebb90241773157043c34fd26fd46:9243

This patch is to fix it.

---

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

• [N/A] Documentation required for this feature

Backporting

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

Configuration

[INPUT]
    Name dummy

[OUTPUT]
    Name es
# https://ac31ebb90241773157043c34fd26fd46.us-central1.gcp.cloud.es.io:9243
    cloud_id different-es-kb-port:dXMtY2VudHJhbDEuZ2NwLmNsb3VkLmVzLmlvJGFjMzFlYmI5MDI0MTc3MzE1NzA0M2MzNGZkMjZmZDQ2OjkyNDMkYTRjMDYyMzBlNDhjOGZjZTdiZTg4YTA3NGEzYmIzZTA6OTI0NA==

Debug / Valgrind output

Parsed result:

[2022/09/17 09:05:13] [debug] [output:es:es.0] host=ac31ebb90241773157043c34fd26fd46.us-central1.gcp.cloud.es.io port=9243 uri=/_bulk index=fluent-bit type=_doc

Full log:

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

[2022/09/17 09:05:12] [ info] Configuration:
[2022/09/17 09:05:12] [ info]  flush time     | 1.000000 seconds
[2022/09/17 09:05:12] [ info]  grace          | 5 seconds
[2022/09/17 09:05:12] [ info]  daemon         | 0
[2022/09/17 09:05:12] [ info] ___________
[2022/09/17 09:05:12] [ info]  inputs:
[2022/09/17 09:05:12] [ info]      dummy
[2022/09/17 09:05:12] [ info] ___________
[2022/09/17 09:05:12] [ info]  filters:
[2022/09/17 09:05:12] [ info] ___________
[2022/09/17 09:05:12] [ info]  outputs:
[2022/09/17 09:05:12] [ info]      es.0
[2022/09/17 09:05:12] [ info] ___________
[2022/09/17 09:05:12] [ info]  collectors:
[2022/09/17 09:05:13] [ info] [fluent bit] version=2.0.0, commit=e1a58dfc80, pid=24798
[2022/09/17 09:05:13] [debug] [engine] coroutine stack size: 24576 bytes (24.0K)
[2022/09/17 09:05:13] [ info] [storage] version=1.2.0, type=memory-only, sync=normal, checksum=disabled, max_chunks_up=128
[2022/09/17 09:05:13] [ info] [cmetrics] version=0.4.0
[2022/09/17 09:05:13] [debug] [dummy:dummy.0] created event channels: read=21 write=22
[2022/09/17 09:05:13] [debug] [es:es.0] created event channels: read=23 write=24
[2022/09/17 09:05:13] [debug] [output:es:es.0] extracted cloud_host: 'ac31ebb90241773157043c34fd26fd46.us-central1.gcp.cloud.es.io:9243'
[2022/09/17 09:05:13] [debug] [output:es:es.0] extracted cloud_port_char: '9243'
[2022/09/17 09:05:13] [debug] [output:es:es.0] converted cloud_port_char to port int: '9243'
[2022/09/17 09:05:13] [debug] [output:es:es.0] checked whether extracted port was null and set it to default https port or not. Outcome: '9243' and cloud_host: 'ac31ebb90241773157043c34fd26fd46.us-central1.gcp.cloud.es.io'.
[2022/09/17 09:05:13] [debug] [output:es:es.0] host=ac31ebb90241773157043c34fd26fd46.us-central1.gcp.cloud.es.io port=9243 uri=/_bulk index=fluent-bit type=_doc
[2022/09/17 09:05:14] [ info] [output:es:es.0] worker #0 started
[2022/09/17 09:05:14] [ info] [output:es:es.0] worker #1 started
[2022/09/17 09:05:14] [debug] [router] default match rule dummy.0:es.0
[2022/09/17 09:05:14] [ info] [sp] stream processor started
^C[2022/09/17 09:05:14] [engine] caught signal (SIGINT)
[2022/09/17 09:05:14] [ warn] [engine] service will shutdown in max 5 seconds
[2022/09/17 09:05:14] [ info] [input] pausing dummy.0
[2022/09/17 09:05:14] [ info] [engine] service has stopped (0 pending tasks)
[2022/09/17 09:05:14] [ info] [input] pausing dummy.0
[2022/09/17 09:05:14] [ info] [output:es:es.0] thread worker #0 stopping...
[2022/09/17 09:05:15] [ info] [output:es:es.0] thread worker #0 stopped
[2022/09/17 09:05:15] [ info] [output:es:es.0] thread worker #1 stopping...
[2022/09/17 09:05:15] [ info] [output:es:es.0] thread worker #1 stopped
==24798== 
==24798== HEAP SUMMARY:
==24798==     in use at exit: 0 bytes in 0 blocks
==24798==   total heap usage: 1,210 allocs, 1,210 frees, 228,340 bytes allocated
==24798== 
==24798== All heap blocks were freed -- no leaks are possible
==24798== 
==24798== For lists of detected and suppressed errors, rerun with: -s
==24798== 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.

[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.

[nokute78]

Ping.

[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.

[edsiper]

can this also apply to out_opensearch ?

[nokute78]

@PettitWesley Does opensearch support cloud id like elasticsearch ?
This patch is to modify handling cloud id for elasticsearch.

There is no configuration to change cloud id for opensearch.
https://docs.fluentbit.io/manual/pipeline/outputs/opensearch

[PettitWesley]

@nokute78 AFAIK, opensearch only supports AWS Sigv4 IAM auth, and does not have anything called "cloud ID". I think that's an elastic only feature. @edsiper can ask his contacts on opensearch team, to double confirm if needed.