out_kafka: Introduce raw_log_key to write a single value to kafka #8655
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
zecke
requested review from
edsiper,
leonardo-albertovich,
fujimotos and
koleini
as code owners
|
Example config used for testing this: Using kcat to read from kafka: output: |
|
Fixed an unrelated issue found by valgrind (and added a dedicated commit). The reachable bits belong to global state created by curl (and then two dependencies down...) |
zecke
force-pushed
the
freyth/message-key
branch
from
7785d5f to
411463e
Compare
|
I have rebased this against the master branch. |
|
@cosmo0920 would you have some time to take a look |
|
@zecke can you link the docs PR for this as well? It's a new parameter I believe is what you're saying so should be documented here (ideally with an example): https://github.com/fluent/fluent-bit-docs/blob/master/pipeline/outputs/kafka.md I see you've ticked the backport option as well, that typically means also linking a PR to the branch for the backport version. |
zecke
force-pushed
the
freyth/message-key
branch
from
411463e to
3235b1d
Compare
I opened fluent/fluent-bit-docs#1397 Not sure how to link these two together.
I have unticked the box. Getting this into any future release will be great. |
cosmo0920
approved these changes
Jun 20, 2024
|
The doc PR has been merged (fluent/fluent-bit-docs#1397) |
edsiper
requested changes
Aug 27, 2024
plugins/out_kafka/kafka.c
Outdated
| if (ctx->raw_log_key && !raw_key && val.type == MSGPACK_OBJECT_STR) { | ||
| if (key.via.str.size == ctx->raw_log_key_len && | ||
| strncmp(key.via.str.ptr, ctx->raw_log_key, ctx->raw_log_key_len) == 0) { | ||
| raw_key = (char *) val.via.str.ptr; |
There was a problem hiding this comment.
this routine is inside a loop, if I am not wrong the values might be set for one message but if the next message does not match they will have the value from the previous one (cc: @cosmo0920 )
There was a problem hiding this comment.
Ohh, I've overlooked one. Thanks for catching this.
There was a problem hiding this comment.
How is this different to the message_key handling a few lines of above? What is the lifetime of val.via.str.ptr? Shouldn't it be valid until after the msgpack is freed?
There was a problem hiding this comment.
Normally, we preserve the temporary reference as a heap memory with flb_sds_create_len.
In this case, just referencing via val.via.str.ptr would be replaced with the latter line in the loop.
So, we need to create a heap region with:
raw_key = flb_sds_create_len(val.via.str.ptr, val.via.str.size);and in the finalizing function:
flb_sds_destroy(raw_key);Plus, this heap should be used flb_sds_t type for it.
There was a problem hiding this comment.
Took a copy of that memory and freed it on the exit paths. flb_sds_destroy(NULL) seems legit (like free(NULL))
Allow to write the value of a single key instead of the entire message to kafka. This allows to use a part of the message as the message_key and another part as the payload. This is similar to other output plugins. The value of the key must be a string and the format must be set to raw. Signed-off-by: Holger Hans Peter Freyther <[email protected]>
Create a simple test for the out_kafka plugin that creates the context and tears it down. Signed-off-by: Holger Hans Peter Freyther <[email protected]>
The value is owned by the properties of the output and we don't need to free this. ==57818== Invalid free() / delete / delete[] / realloc() ==57818== at 0x4887B40: free (vg_replace_malloc.c:872) ==57818== by 0x8EE5F3: flb_free (flb_mem.h:127) ==57818== by 0x8F27A7: flb_out_kafka_destroy (kafka_config.c:243) ==57818== by 0x8EE3F7: cb_kafka_exit (kafka.c:558) ==57818== by 0x4D6D5B: flb_output_exit (flb_output.c:474) ==57818== by 0x4FF65F: flb_engine_shutdown (flb_engine.c:1119) ==57818== by 0x4FF333: flb_engine_start (flb_engine.c:1017) ==57818== by 0x49F20B: flb_lib_worker (flb_lib.c:674) ==57818== by 0x4F1EE57: start_thread (pthread_create.c:442) ==57818== by 0x4F87F9B: thread_start (clone.S:79) ==57818== Address 0x5d22850 is 16 bytes inside a block of size 24 alloc'd ==57818== at 0x48850C8: malloc (vg_replace_malloc.c:381) ==57818== by 0x4AD267: flb_malloc (flb_mem.h:80) ==57818== by 0x4AD4C3: sds_alloc (flb_sds.c:41) ==57818== by 0x4AD60F: flb_sds_create_size (flb_sds.c:93) ==57818== by 0x5B0D3F: flb_env_var_translate (flb_env.c:180) ==57818== by 0x4D76BF: flb_output_set_property (flb_output.c:753) ==57818== by 0x4E869F: configure_plugins_type (flb_config.c:833) ==57818== by 0x4E8BFF: flb_config_load_config_format (flb_config.c:941) ==57818== by 0x489843: service_configure (fluent-bit.c:765) ==57818== by 0x48A95B: flb_main (fluent-bit.c:1298) ==57818== by 0x48AD47: main (fluent-bit.c:1456) Signed-off-by: Holger Hans Peter Freyther <[email protected]>
zecke
force-pushed
the
freyth/message-key
branch
from
3235b1d to
1d22978
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Implement a new option called raw_log_key that allows to write a single value to kafka. This allows to use message_key to put some fields into the Kafka message_key and another field as the payload. Make this work as an additional format.
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:
If this is a change to packaging of containers or native binaries then please confirm it works for all targets.
ok-package-testlabel to test for all targets (requires maintainer to do).Documentation
Backporting
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.