processor_metrics_selector: Implement native processor for metrics ma…

…nipulations * Prevent dangling pointer if cmt context is recreated on input_metric and processor. * processor_labels: Follow change of the signature of metrics callback * lib: Support setter for processor to make testing processors easily Signed-off-by: Hiroshi Hatake <[email protected]>
fluent · Mar 14, 2024 · a7b3d9b · a7b3d9b
1 parent cac75dc
commit a7b3d9b
Show file tree

Hide file tree

Showing 14 changed files with 1,127 additions and 11 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -283,7 +283,7 @@ option(FLB_FILTER_GEOIP2               "Enable geoip2 filter"
 option(FLB_FILTER_NIGHTFALL            "Enable Nightfall filter"                      Yes)
 option(FLB_FILTER_WASM                 "Enable WASM filter"                           Yes)
 option(FLB_PROCESSOR_LABELS            "Enable metrics label manipulation processor"  Yes)
-option(FLB_PROCESSOR_ATTRIBUTES        "Enable atributes manipulation processor"      Yes)
+option(FLB_PROCESSOR_METRICS_SELECTOR  "Enable metrics selector processor"            Yes)
 option(FLB_PROCESSOR_CONTENT_MODIFIER  "Enable content modifier processor"            Yes)
 
 if(DEFINED FLB_NIGHTLY_BUILD AND NOT "${FLB_NIGHTLY_BUILD}" STREQUAL "")

diff --git a/include/fluent-bit/flb_lib.h b/include/fluent-bit/flb_lib.h
@@ -46,11 +46,15 @@ struct flb_lib_out_cb {
 /* For Fluent Bit library callers, we only export the following symbols */
 typedef struct flb_lib_ctx         flb_ctx_t;
 
+struct flb_processor;
+
 FLB_EXPORT void flb_init_env();
 FLB_EXPORT flb_ctx_t *flb_create();
 FLB_EXPORT void flb_destroy(flb_ctx_t *ctx);
 FLB_EXPORT int flb_input(flb_ctx_t *ctx, const char *input, void *data);
+FLB_EXPORT int flb_input_set_processor(flb_ctx_t *ctx, int ffd, struct flb_processor *proc);
 FLB_EXPORT int flb_output(flb_ctx_t *ctx, const char *output, struct flb_lib_out_cb *cb);
+FLB_EXPORT int flb_output_set_processor(flb_ctx_t *ctx, int ffd, struct flb_processor *proc);
 FLB_EXPORT int flb_filter(flb_ctx_t *ctx, const char *filter, void *data);
 FLB_EXPORT int flb_input_set(flb_ctx_t *ctx, int ffd, ...);
 FLB_EXPORT int flb_input_property_check(flb_ctx_t *ctx, int ffd, char *key, char *val);

diff --git a/include/fluent-bit/flb_metrics.h b/include/fluent-bit/flb_metrics.h
@@ -39,6 +39,7 @@
 #include <cmetrics/cmt_encode_msgpack.h>
 #include <cmetrics/cmt_encode_splunk_hec.h>
 #include <cmetrics/cmt_encode_cloudwatch_emf.h>
+#include <cmetrics/cmt_filter.h>
 
 /* Metrics IDs for general purpose (used by core and Plugins */
 #define FLB_METRIC_N_RECORDS   0

diff --git a/include/fluent-bit/flb_processor.h b/include/fluent-bit/flb_processor.h
@@ -143,7 +143,8 @@ struct flb_processor_plugin {
                             int);
 
     int (*cb_process_metrics) (struct flb_processor_instance *,
-                               struct cmt *,
+                               struct cmt *, /* in */
+                               struct cmt **, /* out */
                                const char *,
                                int);
 

diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt
@@ -282,6 +282,7 @@ REGISTER_IN_PLUGIN("in_random")
 # PROCESSORS
 # ==========
 REGISTER_PROCESSOR_PLUGIN("processor_labels")
+REGISTER_PROCESSOR_PLUGIN("processor_metrics_selector")
 REGISTER_PROCESSOR_PLUGIN("processor_content_modifier")
 
 # OUTPUTS

diff --git a/plugins/processor_labels/labels.c b/plugins/processor_labels/labels.c
@@ -1696,15 +1696,23 @@ static int hash_labels(struct cmt *metrics_context,
 
 static int cb_process_metrics(struct flb_processor_instance *processor_instance,
                               struct cmt *metrics_context,
+                              struct cmt **out_context,
                               const char *tag,
                               int tag_len)
 {
+    struct cmt                        *out_cmt;
     struct internal_processor_context *processor_context;
     int                                result;
 
     processor_context =
         (struct internal_processor_context *) processor_instance->context;
 
+    out_cmt = cmt_create();
+    if (out_cmt == NULL) {
+        flb_plg_error(processor_instance, "could not create out_cmt context");
+        return FLB_PROCESSOR_FAILURE;
+    }
+
     result = delete_labels(metrics_context,
                            &processor_context->delete_labels);
 
@@ -1728,6 +1736,17 @@ static int cb_process_metrics(struct flb_processor_instance *processor_instance,
                              &processor_context->hash_labels);
     }
 
+    if (result == FLB_PROCESSOR_SUCCESS) {
+        result = cmt_cat(out_cmt, metrics_context);
+        if (result != 0) {
+            cmt_destroy(out_cmt);
+
+            return FLB_PROCESSOR_FAILURE;
+        }
+
+        *out_context = out_cmt;
+    }
+
     if (result != FLB_PROCESSOR_SUCCESS) {
         return FLB_PROCESSOR_FAILURE;
     }

diff --git a/plugins/processor_metrics_selector/CMakeLists.txt b/plugins/processor_metrics_selector/CMakeLists.txt
@@ -0,0 +1,4 @@
+set(src
+  selector.c)
+
+FLB_PLUGIN(processor_metrics_selector "${src}" "")