processor: new enhancements, API changes and new processor

CMakeLists.txt

@@ -284,6 +284,7 @@ option(FLB_FILTER_NIGHTFALL            "Enable Nightfall filter"
 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_CONTENT_MODIFIER  "Enable content modifier processor"            Yes)
 
 if(DEFINED FLB_NIGHTLY_BUILD AND NOT "${FLB_NIGHTLY_BUILD}" STREQUAL "")
   FLB_DEFINITION_VAL(FLB_NIGHTLY_BUILD ${FLB_NIGHTLY_BUILD})

include/fluent-bit/flb_log_event.h

@@ -20,9 +20,9 @@
 #ifndef FLB_LOG_EVENT_H
 #define FLB_LOG_EVENT_H
 
+#include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_time.h>
 #include <fluent-bit/flb_sds.h>
-#include <fluent-bit/flb_mp.h>
 
 #include <msgpack.h>
 

include/fluent-bit/flb_mp.h

@@ -21,6 +21,7 @@
 #define FLB_MP_H
 
 #include <msgpack.h>
+#include <cfl/cfl.h>
 
 #define FLB_MP_MAP        MSGPACK_OBJECT_MAP
 #define FLB_MP_ARRAY      MSGPACK_OBJECT_ARRAY
@@ -67,6 +68,7 @@ struct flb_mp_accessor {
     struct mk_list ra_list;
 };
 
+
 int flb_mp_map_header_init(struct flb_mp_map_header *mh, msgpack_packer *mp_pck);
 int flb_mp_map_header_append(struct flb_mp_map_header *mh);
 void flb_mp_map_header_end(struct flb_mp_map_header *mh);
@@ -88,4 +90,7 @@ int flb_mp_accessor_set_active_by_pattern(struct flb_mp_accessor *mpa,
 struct cfl_object *flb_mp_object_to_cfl(msgpack_object *o);
 int flb_mp_cfl_to_msgpack(struct cfl_object *obj, char **out_buf, size_t *out_size);
 
+
+
+
 #endif

include/fluent-bit/flb_mp_chunk.h

@@ -0,0 +1,63 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2024 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_MP_CHUNK_H
+#define FLB_MP_CHUNK_H
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_log_event.h>
+#include <cfl/cfl.h>
+
+#define FLB_MP_CHUNK_RECORD_ERROR -1  /* Error while retrieving content */
+#define FLB_MP_CHUNK_RECORD_OK     0  /* Content retrieved successfully */
+#define FLB_MP_CHUNK_RECORD_EOF    1  /* No more content to retrieve */
+
+struct flb_mp_chunk_record {
+    int modified;
+    struct flb_log_event event;
+    struct cfl_object *cobj_metadata;
+    struct cfl_object *cobj_record;
+    struct cfl_list _head;
+};
+
+struct flb_mp_chunk_cobj {
+    int total_records;
+    struct flb_log_event_encoder *log_encoder;
+    struct flb_log_event_decoder *log_decoder;
+
+    struct flb_mp_chunk_record *record_pos;
+    struct cfl_list records;
+};
+
+
+struct flb_mp_chunk_record *flb_mp_chunk_record_create(struct flb_mp_chunk_cobj *chunk_cobj);
+int flb_mp_chunk_cobj_record_next(struct flb_mp_chunk_cobj *chunk_cobj,
+                                  struct flb_mp_chunk_record **out_record);
+
+struct flb_mp_chunk_cobj *flb_mp_chunk_cobj_create(struct flb_log_event_encoder *log_encoder,
+                                                   struct flb_log_event_decoder *log_decoder);
+int flb_mp_chunk_cobj_destroy(struct flb_mp_chunk_cobj *chunk_cobj);
+
+int flb_mp_chunk_cobj_encode(struct flb_mp_chunk_cobj *chunk_cobj, char **out_buf, size_t *out_size);
+
+int flb_mp_chunk_cobj_record_next(struct flb_mp_chunk_cobj *chunk_cobj,
+                                  struct flb_mp_chunk_record **out_record);
+
+
+#endif

include/fluent-bit/flb_processor.h

@@ -138,8 +138,7 @@ struct flb_processor_plugin {
                     struct flb_config *);
 
     int (*cb_process_logs) (struct flb_processor_instance *,
-                            struct flb_log_event_encoder *,
-                            struct flb_log_event *,
+                            void *,       /* struct flb_mp_chunk_cobj_create */
                             const char *,
                             int);
 
@@ -161,6 +160,7 @@ struct flb_processor_plugin {
 struct flb_processor_instance {
     int id;                                /* instance id              */
     int log_level;                         /* instance log level       */
+    int event_type;                        /* event type               */
     char name[32];                         /* numbered name            */
     char *alias;                           /* alias name               */
     void *context;                         /* Instance local context   */
@@ -215,6 +215,7 @@ int flb_processors_load_from_config_format_group(struct flb_processor *proc, str
 
 struct flb_processor_instance *flb_processor_instance_create(
                                     struct flb_config *config,
+                                    int event_type,
                                     const char *name,
                                     void *data);
 

plugins/CMakeLists.txt

@@ -282,7 +282,7 @@ REGISTER_IN_PLUGIN("in_random")
 # PROCESSORS
 # ==========
 REGISTER_PROCESSOR_PLUGIN("processor_labels")
-REGISTER_PROCESSOR_PLUGIN("processor_attributes")
+REGISTER_PROCESSOR_PLUGIN("processor_content_modifier")
 
 # OUTPUTS
 # =======