Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

out_splunk: added otlp metadata support (backport 3.2) #10093

Merged
merged 5 commits into from
Mar 18, 2025
Merged

out_splunk: added otlp metadata support (backport 3.2) #10093

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
91506c4
out_splunk: added otlp metadata support
leonardo-albertovich Mar 3, 2025
273f7e2
out_splunk: fixed erroneous check that wasn't meant to make it to git
leonardo-albertovich Mar 3, 2025
70d20a7
log_event_decoder: improved how group metadata is exposed to client code
leonardo-albertovich Mar 3, 2025
2a9194c
log_event: added required fields to expose group metadata to client code
leonardo-albertovich Mar 3, 2025
312bb5f
tests: in_opentelemetry: updated log event decoder api usage
leonardo-albertovich Mar 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions include/fluent-bit/flb_log_event.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@
#define FLB_LOG_EVENT_GROUP_END (int32_t) -2

struct flb_log_event {
msgpack_object *group_attributes;
msgpack_object *group_metadata;
msgpack_object *raw_timestamp;
struct flb_time timestamp;
msgpack_object *metadata;
Expand Down
3 changes: 3 additions & 0 deletions include/fluent-bit/flb_log_event_decoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,10 @@
#define FLB_LOG_EVENT_EXPECTED_HEADER_ELEMENT_COUNT 2

struct flb_log_event_decoder {
msgpack_object *current_group_attributes;
msgpack_unpacked unpacked_group_record;
int dynamically_allocated;
msgpack_object *current_group_metadata;
msgpack_unpacked unpacked_empty_map;
size_t previous_offset;
msgpack_unpacked unpacked_event;
Expand Down
253 changes: 249 additions & 4 deletions plugins/out_splunk/splunk.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,215 @@ static int cb_splunk_init(struct flb_output_instance *ins,
return 0;
}

static msgpack_object *local_msgpack_map_lookup(
msgpack_object *map_object,
char *key)
{
size_t key_length;
size_t index;
msgpack_object_map *map;

if (key == NULL) {
return NULL;
}

if (map_object == NULL) {
return NULL;
}

if (map_object->type != MSGPACK_OBJECT_MAP) {
return NULL;
}

map = &map_object->via.map;

key_length = strlen(key);

for (index = 0; index < map->size ; index++) {
if (map->ptr[index].key.type == MSGPACK_OBJECT_STR) {
if (map->ptr[index].key.via.str.size == key_length) {
if (strncmp(map->ptr[index].key.via.str.ptr,
key,
key_length) == 0) {
return &map->ptr[index].val;
}
}
}
}

return NULL;
}

static int local_msgpack_map_string_lookup(
msgpack_object *map_object,
char *key,
char **value,
size_t *value_size)
{
msgpack_object *value_object;

value_object = local_msgpack_map_lookup(map_object, key);

if (value_object == NULL) {
return -1;
}

if (value_object->type != MSGPACK_OBJECT_STR) {
return -2;
}

*value = (char *) value_object->via.str.ptr;
*value_size = value_object->via.str.size;

return 0;
}

static int local_msgpack_map_string_extract(
msgpack_object *map_object,
char *key,
char *output_buffer,
size_t output_buffer_size)
{
size_t value_size;
int result;
char *value;

result = local_msgpack_map_string_lookup(map_object,
key,
&value,
&value_size);

if (result != 0) {
return -1;
}

if (value_size >= output_buffer_size) {
return -2;
}

strncpy(output_buffer,
value,
value_size);

output_buffer[value_size] = '\0';

return 0;
}

static inline void local_msgpack_pack_cstr(msgpack_packer *packer, char *value)
{
msgpack_pack_str(packer, strlen(value));
msgpack_pack_str_body(packer, value, strlen(value));
}

static int pack_otel_data(struct flb_splunk *ctx,
msgpack_packer *mp_pck,
struct flb_mp_map_header *mh_pck,
msgpack_object *group_metadata,
msgpack_object *group_attributes,
msgpack_object *record_attributes)
{
msgpack_object *source_map;
char schema[8];
int result;
struct flb_mp_map_header mh_tmp;
msgpack_object *value;
size_t index;

result = local_msgpack_map_string_extract(group_metadata,
"schema",
schema,
sizeof(schema));

if (result != 0) {
return 0;
}

if (strcmp(schema, "otlp") != 0) {
return 0;
}

source_map = local_msgpack_map_lookup(group_attributes, "resource");

if (source_map != NULL) {
source_map = local_msgpack_map_lookup(source_map, "attributes");

if (source_map != NULL) {
value = local_msgpack_map_lookup(source_map,
"host.name");

if (value != NULL) {
flb_mp_map_header_append(mh_pck);

local_msgpack_pack_cstr(mp_pck, "host");

msgpack_pack_object(mp_pck, *value);
}
else {
return -2;
}
}
}

flb_mp_map_header_append(mh_pck);

local_msgpack_pack_cstr(mp_pck, "fields");

flb_mp_map_header_init(&mh_tmp, mp_pck);

source_map = local_msgpack_map_lookup(record_attributes, "otlp");

if (source_map != NULL) {
value = local_msgpack_map_lookup(source_map,
"severity_number");

if (value != NULL &&
(value->type == MSGPACK_OBJECT_POSITIVE_INTEGER ||
value->type == MSGPACK_OBJECT_NEGATIVE_INTEGER)) {
flb_mp_map_header_append(&mh_tmp);

local_msgpack_pack_cstr(mp_pck, "otel.log.severity.number");

msgpack_pack_object(mp_pck, *value);
}
else {
return -2;
}

value = local_msgpack_map_lookup(source_map,
"severity_text");

if (value != NULL &&
value->type == MSGPACK_OBJECT_STR) {
flb_mp_map_header_append(&mh_tmp);

local_msgpack_pack_cstr(mp_pck, "otel.log.severity.text");

msgpack_pack_object(mp_pck, *value);
}
else {
return -3;
}

source_map = local_msgpack_map_lookup(source_map, "attributes");

if (source_map != NULL &&
source_map->type == MSGPACK_OBJECT_MAP) {

for (index = 0; index < source_map->via.map.size ; index++) {
flb_mp_map_header_append(&mh_tmp);

msgpack_pack_object(mp_pck, source_map->via.map.ptr[index].key);
msgpack_pack_object(mp_pck, source_map->via.map.ptr[index].val);
}
}
}

flb_mp_map_header_end(&mh_tmp);

return 0;
}

static int pack_map_meta(struct flb_splunk *ctx,
struct flb_mp_map_header *mh,
msgpack_packer *mp_pck,
Expand Down Expand Up @@ -202,9 +411,15 @@ static int pack_map_meta(struct flb_splunk *ctx,
}

static int pack_map(struct flb_splunk *ctx, msgpack_packer *mp_pck,
struct flb_time *tm, msgpack_object map,
char *tag, int tag_len)
struct flb_time *tm,
msgpack_object *group_metadata,
msgpack_object *group_attributes,
msgpack_object *record_attributes,
msgpack_object map,
char *tag,
int tag_len)
{
int result;
int i;
double t;
int map_size;
Expand Down Expand Up @@ -232,6 +447,20 @@ static int pack_map(struct flb_splunk *ctx, msgpack_packer *mp_pck,
/* Pack Splunk metadata */
pack_map_meta(ctx, &mh, mp_pck, map, tag, tag_len);

/* Pack Otel specific metadata */

result = pack_otel_data(ctx,
mp_pck,
&mh,
group_metadata,
group_attributes,
record_attributes);

if (result != 0) {
printf("ERROR %d\n", result);
return -1;
}

/* Add k/v pairs under the key 'event' instead of to the top level object */
flb_mp_map_header_append(&mh);
msgpack_pack_str(mp_pck, sizeof(FLB_SPLUNK_DEFAULT_EVENT) -1);
Expand Down Expand Up @@ -475,12 +704,28 @@ static inline int splunk_format(const void *in_buf, size_t in_bytes,
* record, we just warn the user and try to pack it
* as a normal map.
*/
ret = pack_map(ctx, &mp_pck, &log_event.timestamp, map, tag, tag_len);
ret = pack_map(ctx,
&mp_pck,
&log_event.timestamp,
log_event.group_metadata,
log_event.group_attributes,
log_event.metadata,
map,
tag,
tag_len);
}
}
else {
/* Pack as a map */
ret = pack_map(ctx, &mp_pck, &log_event.timestamp, map, tag, tag_len);
ret = pack_map(ctx,
&mp_pck,
&log_event.timestamp,
log_event.group_metadata,
log_event.group_attributes,
log_event.metadata,
map,
tag,
tag_len);
}

/* Validate packaging */
Expand Down
11 changes: 11 additions & 0 deletions plugins/out_stdout/stdout.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,17 @@ static void cb_stdout_flush(struct flb_event_chunk *event_chunk,

while (flb_log_event_decoder_next(&log_decoder,
&log_event) == FLB_EVENT_DECODER_SUCCESS) {

if (log_event.group_attributes != NULL) {
printf("GROUP METADATA : \n\n");
msgpack_object_print(stdout, *log_event.group_metadata);
printf("\n\n");

printf("GROUP ATTRIBUTES : \n\n");
msgpack_object_print(stdout, *log_event.group_attributes);
printf("\n\n");
}

printf("[%zd] %s: [[", cnt++, event_chunk->tag);

printf("%"PRId32".%09lu, ", (int32_t) log_event.timestamp.tm.tv_sec,
Expand Down
Loading
Loading