out_vivo_exporter: change API endpoints to /api/v1/

edsiper · edsiper · commit 652b3fb3d5e8 · 2025-11-01T00:18:46.000-06:00
Signed-off-by: Eduardo Silva &lt;eduardo@chronosphere.io&gt;
diff --git a/plugins/out_vivo_exporter/vivo.c b/plugins/out_vivo_exporter/vivo.c
@@ -20,23 +20,27 @@
 #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 "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 flb_sds_t format_logs(struct flb_input_instance *src_ins,
+                             struct flb_event_chunk *event_chunk, struct flb_config *config)
 {
-    struct flb_log_event_decoder log_decoder;
-    struct flb_log_event log_event;
+    int len;
     int result;
-    int i;
+    char *name;
     flb_sds_t out_js;
     flb_sds_t out_buf = NULL;
     msgpack_sbuffer tmp_sbuf;
     msgpack_packer tmp_pck;
+    struct flb_log_event log_event;
+    struct flb_log_event_decoder log_decoder;
+    struct flb_mp_map_header mh;
 
     result = flb_log_event_decoder_init(&log_decoder,
                                         (char *) event_chunk->data,
@@ -56,90 +60,118 @@ 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"
+     *      }
+     *    ]
+     * }
+     */
+
+    msgpack_pack_map(&tmp_pck, 4);
+
+    /* source_type: internal type of the plugin */
+    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 */
+    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 */
+    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 */
+    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) {
+
+        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);
-
-        msgpack_pack_str(&tmp_pck, 4);
-        msgpack_pack_str_body(&tmp_pck, "_tag", 4);
-
-        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));
-
-        /* 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);
-        }
+        /* pack metadata */
+        msgpack_pack_object(&tmp_pck, *log_event.metadata);
 
         /* 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);
-        }
-
-        /* 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;
-        }
-
-        /*
-         * 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);
+        msgpack_pack_object(&tmp_pck, *log_event.body);
     }
 
+    flb_mp_array_header_end(&mh);
+
     /* 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) {
+        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;
@@ -207,6 +239,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) {
@@ -272,7 +305,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);
diff --git a/plugins/out_vivo_exporter/vivo.h b/plugins/out_vivo_exporter/vivo.h
@@ -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 */
@@ -40,6 +42,7 @@ struct vivo_exporter {
 
     /* instance context */
     struct flb_output_instance *ins;
+    struct flb_config *config;
 };
 
 #endif
diff --git a/plugins/out_vivo_exporter/vivo_http.c b/plugins/out_vivo_exporter/vivo_http.c
@@ -19,6 +19,11 @@
 
 #include <fluent-bit/flb_output_plugin.h>
 #include <fluent-bit/flb_http_server.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_metrics_exporter.h>
+
+#include <cmetrics/cmt_encode_msgpack.h>
 
 #include "vivo.h"
 #include "vivo_http.h"
@@ -64,14 +69,8 @@ static int stream_get_uri_properties(mk_request_t *request,
     return 0;
 }
 
-static void headers_set(mk_request_t *request, struct vivo_stream *vs)
+static void headers_set_common(struct vivo_exporter *ctx, mk_request_t *request)
 {
-    struct vivo_exporter *ctx;
-
-
-    /* parent context */
-    ctx = vs->parent;
-
     /* content type */
     mk_http_header(request,
                    VIVO_CONTENT_TYPE, sizeof(VIVO_CONTENT_TYPE) - 1,
@@ -90,13 +89,25 @@ static void headers_set(mk_request_t *request, struct vivo_stream *vs)
                        sizeof("Access-Control-Allow-Headers") - 1,
                        "Origin, X-Requested-With, Content-Type, Accept",
                        sizeof("Origin, X-Requested-With, Content-Type, Accept") - 1);
+    }
+}
 
+static void headers_set(mk_request_t *request, struct vivo_stream *vs)
+{
+    struct vivo_exporter *ctx;
+
+
+    /* parent context */
+    ctx = vs->parent;
+
+    headers_set_common(ctx, request);
+
+    if (ctx->http_cors_allow_origin) {
         mk_http_header(request,
                        "Access-Control-Expose-Headers",
                        sizeof("Access-Control-Expose-Headers") - 1,
                        "vivo-stream-start-id, vivo-stream-end-id",
                        sizeof("vivo-stream-start-id, vivo-stream-end-id") - 1);
-
     }
 }
 
@@ -159,7 +170,7 @@ static void serve_content(mk_request_t *request, struct vivo_stream *vs)
     flb_sds_destroy(str_end);
 }
 
-/* HTTP endpoint: /logs */
+/* HTTP endpoint: /api/v1/logs */
 static void cb_logs(mk_request_t *request, void *data)
 {
     struct vivo_exporter *ctx;
@@ -170,7 +181,7 @@ static void cb_logs(mk_request_t *request, void *data)
     mk_http_done(request);
 }
 
-/* HTTP endpoint: /metrics */
+/* HTTP endpoint: /api/v1/metrics */
 static void cb_metrics(mk_request_t *request, void *data)
 {
     struct vivo_exporter *ctx;
@@ -181,6 +192,7 @@ static void cb_metrics(mk_request_t *request, void *data)
     mk_http_done(request);
 }
 
+/* HTTP endpoint: /api/v1/traces */
 static void cb_traces(mk_request_t *request, void *data)
 {
     struct vivo_exporter *ctx;
@@ -191,6 +203,53 @@ static void cb_traces(mk_request_t *request, void *data)
     mk_http_done(request);
 }
 
+/* HTTP endpoint: /api/v1/internal/metrics */
+static void cb_internal_metrics(mk_request_t *request, void *data)
+{
+    int ret;
+    char *mp_buf = NULL;
+    size_t mp_size = 0;
+    flb_sds_t json = NULL;
+    struct cmt *cmt = NULL;
+    struct vivo_exporter *ctx;
+
+    ctx = (struct vivo_exporter *) data;
+
+    cmt = flb_me_get_cmetrics(ctx->config);
+    if (!cmt) {
+        mk_http_status(request, 500);
+        mk_http_done(request);
+        return;
+    }
+
+    ret = cmt_encode_msgpack_create(cmt, &mp_buf, &mp_size);
+    if (ret != 0) {
+        cmt_destroy(cmt);
+        mk_http_status(request, 500);
+        mk_http_done(request);
+        return;
+    }
+
+    json = flb_msgpack_raw_to_json_sds(mp_buf, mp_size,
+                                       ctx->config->json_escape_unicode);
+
+    cmt_encode_msgpack_destroy(mp_buf);
+    cmt_destroy(cmt);
+
+    if (!json) {
+        mk_http_status(request, 500);
+        mk_http_done(request);
+        return;
+    }
+
+    mk_http_status(request, 200);
+    headers_set_common(ctx, request);
+    mk_http_send(request, json, flb_sds_len(json), NULL);
+    mk_http_done(request);
+
+    flb_sds_destroy(json);
+}
+
 /* HTTP endpoint: / (root) */
 static void cb_root(mk_request_t *request, void *data)
 {
@@ -236,9 +295,10 @@ struct vivo_http *vivo_http_server_create(struct vivo_exporter *ctx,
     ph->vid = vid;
 
     /* Set HTTP URI callbacks */
-    mk_vhost_handler(ph->ctx, vid, "/logs", cb_logs, ctx);
-    mk_vhost_handler(ph->ctx, vid, "/metrics", cb_metrics, ctx);
-    mk_vhost_handler(ph->ctx, vid, "/traces", cb_traces, ctx);
+    mk_vhost_handler(ph->ctx, vid, "/api/v1/logs", cb_logs, ctx);
+    mk_vhost_handler(ph->ctx, vid, "/api/v1/metrics", cb_metrics, ctx);
+    mk_vhost_handler(ph->ctx, vid, "/api/v1/traces", cb_traces, ctx);
+    mk_vhost_handler(ph->ctx, vid, "/api/v1/internal/metrics", cb_internal_metrics, ctx);
     mk_vhost_handler(ph->ctx, vid, "/", cb_root, NULL);
 
     return ph;
