Skip to content

out_vivo_exporter: add versioned API endpoints #11040

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

Merged
edsiper merged 3 commits into from
Nov 1, 2025
Merged

out_vivo_exporter: add versioned API endpoints #11040

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b2b5ce0
out_vivo_exporter: change API endpoints to /api/v1/
edsiper Oct 4, 2025
d16acdf
out_vivo_exporter: add groups support
edsiper Oct 6, 2025
20db128
out_vivo_exporter: fix regression on buffer check
edsiper Nov 1, 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
271 changes: 216 additions & 55 deletions plugins/out_vivo_exporter/vivo.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,23 +20,61 @@
#include <fluent-bit/flb_output_plugin.h>
#include <fluent-bit/flb_kv.h>
#include <fluent-bit/flb_pack.h>
#include <fluent-bit/flb_mp.h>
#include <fluent-bit/flb_log_event_decoder.h>
#include <fluent-bit/flb_log_event_encoder.h>
#include <string.h>

#include "vivo.h"
#include "vivo_http.h"
#include "vivo_stream.h"

static flb_sds_t format_logs(struct flb_event_chunk *event_chunk, struct flb_config *config)
static msgpack_object *find_map_value(msgpack_object *map,
const char *key, size_t key_len)
{
struct flb_log_event_decoder log_decoder;
struct flb_log_event log_event;
size_t i;

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

for (i = 0; i < map->via.map.size; i++) {
if (map->via.map.ptr[i].key.type != MSGPACK_OBJECT_STR) {
continue;
}

if (map->via.map.ptr[i].key.via.str.size == key_len &&
strncmp(map->via.map.ptr[i].key.via.str.ptr, key, key_len) == 0) {
return &map->via.map.ptr[i].val;
}
}

return NULL;
}

static flb_sds_t format_logs(struct flb_input_instance *src_ins,
struct flb_event_chunk *event_chunk, struct flb_config *config)
{
int len;
int result;
int i;
char *name;
flb_sds_t out_js;
flb_sds_t out_buf = NULL;

@coderabbitai coderabbitai Bot Nov 1, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major

Remove unused variable out_buf.

The variable out_buf is allocated at line 87 but never written to or read from. It's only destroyed on error paths and at the end. This is dead code left over from the previous implementation that accumulated JSON per record.

Apply this diff to remove the unused variable:

     int result;
     char *name;
     flb_sds_t out_js;
-    flb_sds_t out_buf = NULL;
     msgpack_sbuffer tmp_sbuf;
     msgpack_packer tmp_pck;
         return NULL;
     }
 
-    out_buf = flb_sds_create_size((event_chunk->size * 2) / 4);
-    if (!out_buf) {
-        flb_errno();
-        return NULL;
-    }
-
     /* Create temporary msgpack buffer */
     msgpack_sbuffer_init(&tmp_sbuf);
     msgpack_sbuffer_destroy(&tmp_sbuf);
 
     if (!out_js) {
-        flb_sds_destroy(out_buf);
         return NULL;
     }
     /* append a newline */
     if (flb_sds_cat_safe(&out_js, "\n", 1) < 0) {
         flb_sds_destroy(out_js);
-        flb_sds_destroy(out_buf);
         return NULL;
     }
 
-    /* Replace out_buf with the complete JSON */
-    flb_sds_destroy(out_buf);
     return out_js;

Also applies to: 87-91, 281-281, 288-288, 293-293

msgpack_sbuffer tmp_sbuf;
msgpack_packer tmp_pck;
int group_mismatch = FLB_FALSE;
int is_otlp = FLB_FALSE;
struct flb_log_event log_event;
struct flb_log_event_decoder log_decoder;
struct flb_mp_map_header mh;
struct flb_mp_map_header root_map;
struct flb_mp_map_header otlp_map;
struct flb_mp_map_header group_map;
msgpack_object *group_metadata = NULL;
msgpack_object *group_attributes = NULL;
msgpack_object *schema_value = NULL;
msgpack_object *resource_value = NULL;
msgpack_object *scope_value = NULL;

result = flb_log_event_decoder_init(&log_decoder,
(char *) event_chunk->data,
Expand All @@ -56,90 +94,212 @@ static flb_sds_t format_logs(struct flb_event_chunk *event_chunk, struct flb_con
msgpack_sbuffer_init(&tmp_sbuf);
msgpack_packer_init(&tmp_pck, &tmp_sbuf, msgpack_sbuffer_write);

/*
* Here is an example of the packaging done for Logs
*
* {
* "source_type": "forward",
* "source_name": "forward.0",
* "tag": "dummy.0",
* "records": [
* {
* "timestamp": 1759591426808913765,
* "metadata": {
* "level": "info"
* },
* "message": "dummy"
* },
* {
* "timestamp": 1759591426908563348,
* "metadata": {
* "level": "debug",
* "service": "auth"
* },
* "message": "dummy"
* }
* ]
* }
*/

flb_mp_map_header_init(&root_map, &tmp_pck);

/* source_type: internal type of the plugin */
flb_mp_map_header_append(&root_map);
name = src_ins->p->name;
len = strlen(name);

msgpack_pack_str(&tmp_pck, 11);
msgpack_pack_str_body(&tmp_pck, "source_type", 11);
msgpack_pack_str(&tmp_pck, len);
msgpack_pack_str_body(&tmp_pck, name, len);

/* source_name: internal name or alias set by the user */
flb_mp_map_header_append(&root_map);
name = (char *) flb_input_name(src_ins);
len = strlen(name);
msgpack_pack_str(&tmp_pck, 11);
msgpack_pack_str_body(&tmp_pck, "source_name", 11);
msgpack_pack_str(&tmp_pck, len);
msgpack_pack_str_body(&tmp_pck, name, len);

/* tag */
flb_mp_map_header_append(&root_map);
msgpack_pack_str(&tmp_pck, 3);
msgpack_pack_str_body(&tmp_pck, "tag", 3);
msgpack_pack_str(&tmp_pck, flb_sds_len(event_chunk->tag));
msgpack_pack_str_body(&tmp_pck, event_chunk->tag, flb_sds_len(event_chunk->tag));

/* records */
flb_mp_map_header_append(&root_map);
msgpack_pack_str(&tmp_pck, 7);
msgpack_pack_str_body(&tmp_pck, "records", 7);

flb_mp_array_header_init(&mh, &tmp_pck);

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

if (log_event.group_metadata != NULL) {
if (group_metadata == NULL) {
group_metadata = log_event.group_metadata;
}
else if (group_metadata != log_event.group_metadata) {
group_mismatch = FLB_TRUE;
}
}

if (log_event.group_attributes != NULL) {
if (group_attributes == NULL) {
group_attributes = log_event.group_attributes;
}
else if (group_attributes != log_event.group_attributes) {
group_mismatch = FLB_TRUE;
}
}

flb_mp_array_header_append(&mh);

/*
* If the caller specified FLB_PACK_JSON_DATE_FLUENT, we format the data
* by using the following structure:
*
* [[TIMESTAMP, {"_tag": "...", ...MORE_METADATA}], {RECORD CONTENT}]
* [[TIMESTAMP, {"....": "...", ...MORE_METADATA}], {RECORD CONTENT}]
*/
msgpack_pack_array(&tmp_pck, 2);
msgpack_pack_array(&tmp_pck, 2);
msgpack_pack_uint64(&tmp_pck, flb_time_to_nanosec(&log_event.timestamp));

/* add tag only */
msgpack_pack_map(&tmp_pck, 1 + log_event.metadata->via.map.size);
/* pack metadata */
msgpack_pack_object(&tmp_pck, *log_event.metadata);

msgpack_pack_str(&tmp_pck, 4);
msgpack_pack_str_body(&tmp_pck, "_tag", 4);
/* pack the remaining content */
msgpack_pack_object(&tmp_pck, *log_event.body);
}

msgpack_pack_str(&tmp_pck, flb_sds_len(event_chunk->tag));
msgpack_pack_str_body(&tmp_pck, event_chunk->tag, flb_sds_len(event_chunk->tag));
flb_mp_array_header_end(&mh);

/* Append remaining keys/values */
for (i = 0;
i < log_event.metadata->via.map.size;
i++) {
msgpack_pack_object(&tmp_pck,
log_event.metadata->via.map.ptr[i].key);
msgpack_pack_object(&tmp_pck,
log_event.metadata->via.map.ptr[i].val);
if (group_mismatch == FLB_FALSE &&
(group_metadata != NULL || group_attributes != NULL)) {
if (group_metadata != NULL) {
schema_value = find_map_value(group_metadata, "schema", 6);
}

/* pack the remaining content */
msgpack_pack_map(&tmp_pck, log_event.body->via.map.size);

/* Append remaining keys/values */
for (i = 0;
i < log_event.body->via.map.size;
i++) {
msgpack_pack_object(&tmp_pck,
log_event.body->via.map.ptr[i].key);
msgpack_pack_object(&tmp_pck,
log_event.body->via.map.ptr[i].val);
if (schema_value &&
schema_value->type == MSGPACK_OBJECT_STR &&
schema_value->via.str.size == 4 &&
strncmp(schema_value->via.str.ptr, "otlp", 4) == 0) {
is_otlp = FLB_TRUE;
}

/* Concatenate by using break lines */
out_js = flb_msgpack_raw_to_json_sds(tmp_sbuf.data, tmp_sbuf.size,
config->json_escape_unicode);
if (!out_js) {
flb_sds_destroy(out_buf);
msgpack_sbuffer_destroy(&tmp_sbuf);
flb_log_event_decoder_destroy(&log_decoder);
return NULL;
if (is_otlp == FLB_TRUE) {
resource_value = NULL;
scope_value = NULL;

if (group_attributes != NULL &&
group_attributes->type == MSGPACK_OBJECT_MAP) {
resource_value = find_map_value(group_attributes, "resource", 8);
scope_value = find_map_value(group_attributes, "scope", 5);
}

flb_mp_map_header_append(&root_map);
msgpack_pack_str(&tmp_pck, 4);
msgpack_pack_str_body(&tmp_pck, "otlp", 4);

flb_mp_map_header_init(&otlp_map, &tmp_pck);

if (resource_value != NULL) {
flb_mp_map_header_append(&otlp_map);
msgpack_pack_str(&tmp_pck, 8);
msgpack_pack_str_body(&tmp_pck, "resource", 8);
msgpack_pack_object(&tmp_pck, *resource_value);
}

if (scope_value != NULL) {
flb_mp_map_header_append(&otlp_map);
msgpack_pack_str(&tmp_pck, 5);
msgpack_pack_str_body(&tmp_pck, "scope", 5);
msgpack_pack_object(&tmp_pck, *scope_value);
}

flb_mp_map_header_end(&otlp_map);
}
else {
flb_mp_map_header_append(&root_map);
msgpack_pack_str(&tmp_pck, 9);
msgpack_pack_str_body(&tmp_pck, "flb_group", 9);

flb_mp_map_header_init(&group_map, &tmp_pck);

if (group_metadata != NULL) {
flb_mp_map_header_append(&group_map);
msgpack_pack_str(&tmp_pck, 8);
msgpack_pack_str_body(&tmp_pck, "metadata", 8);
msgpack_pack_object(&tmp_pck, *group_metadata);
}

if (group_attributes != NULL) {
flb_mp_map_header_append(&group_map);
msgpack_pack_str(&tmp_pck, 4);
msgpack_pack_str_body(&tmp_pck, "body", 4);
msgpack_pack_object(&tmp_pck, *group_attributes);
}

flb_mp_map_header_end(&group_map);
}

/*
* One map record has been converted, now append it to the
* outgoing out_buf sds variable.
*/
flb_sds_cat_safe(&out_buf, out_js, flb_sds_len(out_js));
flb_sds_cat_safe(&out_buf, "\n", 1);

flb_sds_destroy(out_js);
msgpack_sbuffer_clear(&tmp_sbuf);
}

flb_mp_map_header_end(&root_map);

/* Release the unpacker */
flb_log_event_decoder_destroy(&log_decoder);

/* Convert the complete msgpack structure to JSON */
out_js = flb_msgpack_raw_to_json_sds(tmp_sbuf.data, tmp_sbuf.size,
config->json_escape_unicode);

msgpack_sbuffer_destroy(&tmp_sbuf);

return out_buf;
/* append a newline */
flb_sds_cat_safe(&out_js, "\n", 1);

if (!out_js) {
Comment on lines +274 to +280

@chatgpt-codex-connector chatgpt-codex-connector Bot Oct 18, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Guard newline append against failed JSON conversion

If flb_msgpack_raw_to_json_sds fails, out_js is NULL but flb_sds_cat_safe(&out_js, "\n", 1) is called before checking for the error. flb_sds_cat_safe dereferences the pointer, so this path will segfault instead of propagating a graceful error whenever the JSON conversion runs out of memory or receives invalid data. The previous implementation checked the return value before appending, so the crash is new to this change.

Useful? React with 👍 / 👎.

Comment on lines +274 to +280

@chatgpt-codex-connector chatgpt-codex-connector Bot Nov 1, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Guard null before appending newline to JSON output

After converting the msgpack buffer with flb_msgpack_raw_to_json_sds, the code immediately calls flb_sds_cat_safe(&out_js, "\n", 1) and only then checks if (!out_js). flb_sds_cat_safe dereferences the pointer passed in, so if flb_msgpack_raw_to_json_sds fails (e.g. out of memory or malformed data), this dereference will segfault instead of propagating the error. The previous version checked for NULL before operating on the buffer. Consider testing out_js for NULL before calling flb_sds_cat_safe so the function can return an error rather than crashing.

Useful? React with 👍 / 👎.

flb_sds_destroy(out_buf);
return NULL;
}

/* Replace out_buf with the complete JSON */
flb_sds_destroy(out_buf);
return out_js;
}

static int logs_event_chunk_append(struct vivo_exporter *ctx,
struct flb_input_instance *src_ins,
struct flb_event_chunk *event_chunk,
struct flb_config *config)
{
size_t len;
flb_sds_t json;
struct vivo_stream_entry *entry;


json = format_logs(event_chunk, config);
json = format_logs(src_ins, event_chunk, config);
if (!json) {
flb_plg_error(ctx->ins, "cannot convert logs chunk to JSON");
return -1;
Expand Down Expand Up @@ -207,6 +367,7 @@ static int cb_vivo_init(struct flb_output_instance *ins,
return -1;
}
ctx->ins = ins;
ctx->config = config;

ret = flb_output_config_map_set(ins, (void *) ctx);
if (ret == -1) {
Expand Down Expand Up @@ -272,7 +433,7 @@ static void cb_vivo_flush(struct flb_event_chunk *event_chunk,
}
#endif
if (event_chunk->type == FLB_EVENT_TYPE_LOGS) {
ret = logs_event_chunk_append(ctx, event_chunk, config);
ret = logs_event_chunk_append(ctx, ins, event_chunk, config);
}
else if (event_chunk->type == FLB_EVENT_TYPE_TRACES) {
ret = metrics_traces_event_chunk_append(ctx, ctx->stream_traces, event_chunk, config);
Expand Down
3 changes: 3 additions & 0 deletions plugins/out_vivo_exporter/vivo.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
#include <fluent-bit/flb_output_plugin.h>
#include <fluent-bit/flb_ring_buffer.h>

struct flb_config;

#define VIVO_RING_BUFFER_SIZE 10

/* Plugin context */
Expand All @@ -40,6 +42,7 @@ struct vivo_exporter {

/* instance context */
struct flb_output_instance *ins;
struct flb_config *config;
};

#endif
Loading
Loading