# This is a combination of 4 commits.

# This is the 1st commit message: processor: input_metric: Prevent dangling pointer if cmt context is recreated Signed-off-by: Hiroshi Hatake <[email protected]> # This is the commit message #2: processor_labels: Follow change of the signature of metrics callback Signed-off-by: Hiroshi Hatake <[email protected]> # This is the commit message #3: processor_selector: Implement selector processor for metrics For future extensibility, we use "selector" as a name for this processor. Signed-off-by: Hiroshi Hatake <[email protected]> # This is the commit message #4: lib: Support setter for processor Signed-off-by: Hiroshi Hatake <[email protected]>
fluent · Mar 14, 2024 · 350c9b5 · 350c9b5
1 parent 9a173d0
commit 350c9b5
Show file tree

Hide file tree

Showing 11 changed files with 680 additions and 10 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -285,6 +285,8 @@ option(FLB_FILTER_WASM                 "Enable WASM filter"
 option(FLB_PROCESSOR_LABELS            "Enable metrics label manipulation processor"  Yes)
 option(FLB_PROCESSOR_ATTRIBUTES        "Enable atributes manipulation processor"      Yes)
 option(FLB_PROCESSOR_CONTENT_MODIFIER  "Enable content modifier processor"            Yes)
+option(FLB_PROCESSOR_SELECTOR          "Enable selector processor"                    Yes)
+
 
 if(DEFINED FLB_NIGHTLY_BUILD AND NOT "${FLB_NIGHTLY_BUILD}" STREQUAL "")
   FLB_DEFINITION_VAL(FLB_NIGHTLY_BUILD ${FLB_NIGHTLY_BUILD})

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_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
@@ -283,6 +283,7 @@ REGISTER_IN_PLUGIN("in_random")
 # ==========
 REGISTER_PROCESSOR_PLUGIN("processor_labels")
 REGISTER_PROCESSOR_PLUGIN("processor_content_modifier")
+REGISTER_PROCESSOR_PLUGIN("processor_selector")
 
 # 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_selector/CMakeLists.txt b/plugins/processor_selector/CMakeLists.txt
@@ -0,0 +1,4 @@
+set(src
+  selector.c)
+
+FLB_PLUGIN(processor_selector "${src}" "")