proxy: input plugins can now emit metrics and traces event types

Read event_type from the proxy definition and dispatch to the
appropriate append function (flb_input_metrics_append or
flb_input_trace_append) in the collect callback, mirroring how
the output proxy side already handles event_type.

Metrics and traces are decoded from msgpack using cmt_decode_msgpack
and ctr_decode_msgpack respectively before being appended. Logs remain
the default (event_type == 0) to preserve backwards compatibility.

Error handling mirrors the pattern in fw_prot.c:append_log: return
values are checked, resources are freed on failure, and ctr ownership
is not assumed on the traces success path since flb_input_trace_append
takes ownership of the context.

Fixes: #11914

Author: seblaz

src/flb_plugin_proxy.c

@@ -33,7 +33,12 @@
 #include <fluent-bit/flb_utils.h>
 #include <fluent-bit/flb_plugin_proxy.h>
 #include <fluent-bit/flb_input_log.h>
+#include <fluent-bit/flb_input_metric.h>
+#include <fluent-bit/flb_input_trace.h>
+#include <fluent-bit/flb_input_event.h>
 #include <fluent-bit/flb_custom.h>
+#include <cmetrics/cmt_decode_msgpack.h>
+#include <ctraces/ctr_decode_msgpack.h>
 
 /* Proxies */
 #include "proxy/go/go.h"
@@ -75,8 +80,12 @@ static int flb_proxy_input_cb_collect(struct flb_input_instance *ins,
                                       struct flb_config *config, void *in_context)
 {
     int ret = FLB_OK;
+    int event_type;
     size_t len = 0;
+    size_t offset = 0;
     void *data = NULL;
+    struct cmt *cmt = NULL;
+    struct ctrace *ctr = NULL;
     struct flb_plugin_input_proxy_context *ctx = (struct flb_plugin_input_proxy_context *) in_context;
 
 #ifdef FLB_HAVE_PROXY_GO
@@ -85,7 +94,7 @@ static int flb_proxy_input_cb_collect(struct flb_input_instance *ins,
         ret = proxy_go_input_collect(ctx, &data, &len);
 
         if (len == 0) {
-            flb_trace("[GO] No logs are ingested");
+            flb_trace("[GO] No data ingested");
             return -1;
         }
 
@@ -94,17 +103,53 @@ static int flb_proxy_input_cb_collect(struct flb_input_instance *ins,
             return -1;
         }
 
-        flb_input_log_append(ins, NULL, 0, data, len);
+        event_type = ctx->proxy->def->event_type;
+
+        if (event_type == FLB_INPUT_METRICS) {
+            ret = cmt_decode_msgpack_create(&cmt, (char *) data, len, &offset);
+            if (ret != CMT_DECODE_MSGPACK_SUCCESS) {
+                flb_error("[proxy] failed to decode metrics msgpack (error %d)", ret);
+                proxy_go_input_cleanup(ctx, data);
+                return -1;
+            }
+            ret = flb_input_metrics_append(ins, NULL, 0, cmt);
+            if (ret != 0) {
+                flb_error("[proxy] could not append metrics, ret=%d", ret);
+                cmt_decode_msgpack_destroy(cmt);
+                proxy_go_input_cleanup(ctx, data);
+                return -1;
+            }
+            cmt_decode_msgpack_destroy(cmt);
+        }
+        else if (event_type == FLB_INPUT_TRACES) {
+            ret = ctr_decode_msgpack_create(&ctr, (char *) data, len, &offset);
+            if (ret != CTR_DECODE_MSGPACK_SUCCESS) {
+                flb_error("[proxy] failed to decode traces msgpack (error %d)", ret);
+                proxy_go_input_cleanup(ctx, data);
+                return -1;
+            }
+            ret = flb_input_trace_append(ins, NULL, 0, ctr);
+            if (ret != 0) {
+                flb_error("[proxy] could not append traces, ret=%d", ret);
+                ctr_decode_msgpack_destroy(ctr);
+                proxy_go_input_cleanup(ctx, data);
+                return -1;
+            }
+            /* flb_input_trace_append takes ownership of ctr and destroys it on success */
+        }
+        else {
+            /* default: logs */
+            ret = flb_input_log_append(ins, NULL, 0, data, len);
+        }
 
-        ret = proxy_go_input_cleanup(ctx, data);
-        if (ret == -1) {
+        if (proxy_go_input_cleanup(ctx, data) == -1) {
             flb_errno();
             return -1;
         }
     }
 #endif
 
-    return 0;
+    return ret;
 }
 
 static int flb_proxy_input_cb_init(struct flb_input_instance *ins,