processor_fts: add Fluent Typed Scripting processor

cmake/plugins_options.cmake

@@ -78,6 +78,7 @@ DEFINE_OPTION(FLB_IN_EBPF                     "Enable Linux eBPF input plugin"
 # ==========
 DEFINE_OPTION(FLB_PROCESSOR_CONTENT_MODIFIER  "Enable content modifier processor"            ON)
 DEFINE_OPTION(FLB_PROCESSOR_CUMULATIVE_TO_DELTA "Enable cumulative to delta metrics processor" ON)
+DEFINE_OPTION(FLB_PROCESSOR_FTS               "Enable Fluent Typed Scripting processor"      ON)
 DEFINE_OPTION(FLB_PROCESSOR_LABELS            "Enable metrics label manipulation processor"  ON)
 DEFINE_OPTION(FLB_PROCESSOR_METRICS_SELECTOR  "Enable metrics selector processor"            ON)
 DEFINE_OPTION(FLB_PROCESSOR_OPENTELEMETRY_ENVELOPE "Enable OpenTelemetry envelope processor" ON)

include/fluent-bit/flb_mp.h

@@ -27,9 +27,28 @@
 #define FLB_MP_ARRAY      MSGPACK_OBJECT_ARRAY
 
 int flb_mp_count(const void *data, size_t bytes);
+
+/*
+ * Count log records without building msgpack objects; allocation-free in
+ * non-tracking builds. Rejects unsupported or deep shapes, so callers can fall
+ * back to the event decoder.
+ */
+int flb_mp_scan_log_records(const void *data, size_t bytes, int *records);
 int flb_mp_count_log_records(const void *data, size_t bytes);
+int flb_mp_log_window_next(const void *data, size_t bytes, size_t offset,
+                           size_t limit_bytes, int limit_records,
+                           size_t *window_end, int *window_records);
+/* group markers that belong to neighbouring appends, see flb_mp_log_group_boundaries() */
+#define FLB_MP_GROUPS_KEEP_LEADING_END   1
+#define FLB_MP_GROUPS_KEEP_TRAILING_OPEN 2
+
+int flb_mp_log_group_boundaries(const void *data, size_t bytes);
+int flb_mp_log_group_starts(const void *data, size_t bytes);
 int flb_mp_normalize_log_buffer_groups_msgpack(const void *in_buf, size_t in_size,
                                                char **out_buf, size_t *out_size);
+int flb_mp_normalize_log_buffer_groups_msgpack_ex(const void *in_buf, size_t in_size,
+                                                  char **out_buf, size_t *out_size,
+                                                  int flags);
 int flb_mp_normalize_log_buffer_groups(const void *in_buf, size_t in_size,
                                        char **out_buf, size_t *out_size);
 int flb_mp_count_remaining(const void *data, size_t bytes, size_t *remaining_bytes);

include/fluent-bit/flb_mp_chunk.h

@@ -30,6 +30,7 @@
 
 /* Forward declaration to avoid circular dependencies */
 struct flb_condition;
+struct flb_mp_record_block;
 
 struct flb_mp_chunk_record {
     int modified;
@@ -41,6 +42,9 @@ struct flb_mp_chunk_record {
     int owns_group_metadata;
     int owns_group_attributes;
     struct cfl_list _head;
+
+    /* Pool block holding this record, NULL when allocated individually. */
+    struct flb_mp_record_block *storage_block;
 };
 
 struct flb_mp_chunk_cobj {
@@ -56,6 +60,15 @@ struct flb_mp_chunk_cobj {
 
     /* Condition for filtering records during processing */
     struct flb_condition *condition;
+
+    /* Opt-in, bounded native value storage for the final serialization stage. */
+    struct cfl_arena *value_pool;
+    size_t value_pool_bytes;
+
+    /* Reference-counted wrapper blocks remain valid when records move. */
+    int record_pool_enabled;
+    size_t record_pool_blocks;
+    struct flb_mp_record_block *record_pool;
 };
 
 
@@ -70,9 +83,17 @@ struct flb_mp_chunk_cobj *flb_mp_chunk_cobj_create(struct flb_log_event_encoder
                                                    struct flb_log_event_decoder *log_decoder);
 int flb_mp_chunk_cobj_destroy(struct flb_mp_chunk_cobj *chunk_cobj);
 
+/* Only enable when CFL values will be read and serialized, without mutation by
+ * a later native processor. The chunk owns the arena until destruction. */
+void flb_mp_chunk_cobj_value_pool_enable(struct flb_mp_chunk_cobj *chunk_cobj);
+struct cfl_arena *flb_mp_chunk_cobj_value_pool_get(struct flb_mp_chunk_cobj *chunk_cobj,
+                                                  size_t estimated_bytes);
+
+int flb_mp_chunk_cobj_encode_append(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_count_log_records(struct flb_mp_chunk_cobj *chunk_cobj);
 int flb_mp_chunk_cobj_normalize_groups(struct flb_mp_chunk_cobj *chunk_cobj);
+int flb_mp_chunk_cobj_normalize_groups_ex(struct flb_mp_chunk_cobj *chunk_cobj, int flags);
 
 
 

include/fluent-bit/flb_output.h

@@ -902,9 +902,10 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
             normalized_buf = NULL;
             normalized_size = 0;
 
-            ret = flb_mp_normalize_log_buffer_groups_msgpack(p_buf, p_size,
-                                                             &normalized_buf,
-                                                             &normalized_size);
+            /* keep only the group markers that span chunks in the stored data */
+            ret = flb_mp_normalize_log_buffer_groups_msgpack_ex(
+                    p_buf, p_size, &normalized_buf, &normalized_size,
+                    flb_mp_log_group_boundaries(evc->data, evc->size));
             if (ret == 0) {
                 if (p_buf != evc->data) {
                     flb_free(p_buf);
@@ -1151,6 +1152,8 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
                 return NULL;
             }
 
+            /* Capacity can exceed the serialized trace bytes after processing. */
+            p_size = serialization_buffer_offset;
             out_flush->processed_event_chunk = flb_event_chunk_create(
                                                 evc->type,
                                                 0,
@@ -1251,6 +1254,8 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
                 return NULL;
             }
 
+            /* Capacity can exceed the serialized profile bytes after processing. */
+            p_size = serialization_buffer_offset;
             out_flush->processed_event_chunk = flb_event_chunk_create(
                                                 evc->type,
                                                 0,

include/fluent-bit/flb_processor.h

@@ -44,6 +44,15 @@
 #define FLB_PROCESSOR_UNIT_NATIVE    0
 #define FLB_PROCESSOR_UNIT_FILTER    1
 
+/*
+ * Plugin flags
+ *
+ * FLB_PROCESSOR_FLAG_LOG_WINDOWS: the logs callback only depends on each
+ * record and its group, so the processor chain can run it over large buffers
+ * in group aligned windows.
+ */
+#define FLB_PROCESSOR_FLAG_LOG_WINDOWS 1
+
 
 /* The current values mean the processor stack will
  * wait for 2 seconds at most in 50 millisecond increments
@@ -148,7 +157,7 @@ struct flb_processor {
 };
 
 struct flb_processor_plugin {
-    int flags;             /* Flags (not available at the moment */
+    int flags;             /* FLB_PROCESSOR_FLAG_* capabilities  */
     char *name;            /* Processor short name               */
     char *description;     /* Description                        */
 
@@ -240,6 +249,12 @@ int flb_processor_run(struct flb_processor *proc,
                       void **out_buf, size_t *out_size);
 
 
+/* Carry known logical log counts through the pipeline. -1 means unknown. */
+int flb_processor_run_with_count(struct flb_processor *proc, size_t starting_stage, int type,
+                                 const char *tag, size_t tag_len, void *data, size_t data_size,
+                                 void **out_buf, size_t *out_size,
+                                 int input_records, int *output_records);
+
 struct flb_processor_unit *flb_processor_unit_create(struct flb_processor *proc,
                                                      int event_type,
                                                      char *unit_name);

lib/cfl/ARENA.md

@@ -259,8 +259,19 @@ Setting the limit to zero disables external-buffer caching. Reducing the limit
 immediately trims the cache. A larger limit can improve throughput for repeated
 large payloads but may increase retained memory and RSS.
 
-The arena also maintains size classes for commonly-sized owned SDS values.
-Callers do not need to select a class.
+The arena also maintains size classes for owned SDS values up to 1 KiB.
+Callers do not need to select a class. Classes below 128 bytes grow in
+alignment-sized steps of the complete allocation, so short strings such as map
+keys use at most one alignment unit more than they need; larger classes grow
+by about 25 percent. A released string is recycled for the next string of the
+same class. `cfl_sds_alloc()` reports the class capacity, which can exceed the
+requested length.
+
+Variants, kvpairs, key/value lists, arrays, and SDS headers are aligned for the
+pointers, sizes, 64-bit integers, and doubles they contain. This can be less
+strict than the raw allocation alignment, which also covers `long double`, so
+CFL objects pack more densely in arena chunks. Raw allocations keep the public
+alignment guarantee.
 
 ## Memory statistics
 

lib/cfl/CHANGELOG.md

@@ -6,6 +6,15 @@ This file records the notable changes in each CFL release.
 
 - Added public request-lifetime arena allocation, duplication helpers,
   allocator callbacks, and optional bounded geometric chunk growth.
+- Reduced arena memory for CFL object graphs: objects use their natural
+  alignment instead of the raw allocation alignment, and owned SDS strings use
+  finer size classes. Decoded log records use 27 to 31 percent less arena
+  memory. Owned SDS blocks are no longer zeroed before initialization.
+- Added a bump-allocation fast path for the current arena chunk.
+- Fixed the alignment of external arena allocations, which returned storage
+  at a misaligned offset for large owned SDS strings.
+- Added the `cfl-benchmark-arena-records` benchmark for decoded record
+  batches.
 
 ## 1.0.0 - 2026-07-11
 

lib/cfl/benchmarks/CMakeLists.txt

@@ -3,3 +3,6 @@ target_link_libraries(cfl-benchmark-variant-arena cfl-static)
 
 add_executable(cfl-benchmark-variant-mutable variant_mutable.c)
 target_link_libraries(cfl-benchmark-variant-mutable cfl-static)
+
+add_executable(cfl-benchmark-arena-records records.c)
+target_link_libraries(cfl-benchmark-arena-records cfl-static)

lib/cfl/benchmarks/README.md

@@ -88,3 +88,20 @@ document is still alive. Compare them with `arena_initial_used` and
 `arena_used`. `max_rss_kb` is process-level high-water RSS and can remain the
 same for small workloads because it is page-granular and includes executable,
 library, and allocator-retained pages.
+
+## Decoded record batches
+
+The records benchmark builds batches of small log records, each with short
+keys, a nested map, an array, and string, integer, and double values, as
+produced by decoding a buffer of telemetry records. Arena batches are reset
+between iterations; heap batches destroy every record:
+
+```sh
+build-bench/benchmarks/cfl-benchmark-arena-records arena 512 3000
+build-bench/benchmarks/cfl-benchmark-arena-records heap 512 3000
+```
+
+Optional arguments select `referenced` or `owned` values, enable record edits
+(`1` removes, inserts, and renames keys), and set the arena chunk size. The
+tool reports nanoseconds per record and, for arenas, used and reserved bytes
+per record.