Merge branch 'fluent:master' into master

Author: tanmaya-panda1

.github/workflows/cron-trivy.yaml

@@ -1,6 +1,8 @@
 cmake_minimum_required(VERSION 3.12)
 project(fluent-bit C)
 
+include(CheckCCompilerFlag)
+
 # CMP0069 ensures that LTO is enabled for all compilers
 cmake_policy(SET CMP0069 NEW)
 set(CMAKE_POLICY_DEFAULT_CMP0069 NEW)
@@ -207,7 +209,8 @@ option(FLB_SIGNV4              "Enable AWS Signv4 support"    Yes)
 option(FLB_AWS                 "Enable AWS support"           Yes)
 option(FLB_STATIC_CONF         "Build binary using static configuration")
 option(FLB_STREAM_PROCESSOR    "Enable Stream Processor"                    Yes)
-option(FLB_SIMD                "Enable SIMD support"                         No)
+set(FLB_SIMD "Auto" CACHE STRING "Enable SIMD support (On, Off, Auto)")
+set_property(CACHE FLB_SIMD PROPERTY STRINGS "On;Off;Auto")
 option(FLB_CORO_STACK_SIZE     "Set coroutine stack size")
 option(FLB_AVRO_ENCODER        "Build with Avro encoding support"            No)
 option(FLB_AWS_ERROR_REPORTER  "Build with aws error reporting support"      No)
@@ -258,7 +261,64 @@ option(FLB_EVENT_LOOP_SELECT    "Enable select(2) event loop backend" No)
 option(FLB_EVENT_LOOP_LIBEVENT  "Enable libevent event loop backend"  No)
 
 # SIMD support
-if(FLB_SIMD)
+# ------------
+# FLB_SIMD is a tri-state cache variable:
+#   Auto : enable SIMD only when the target architecture/toolchain is known-good
+#   On   : require SIMD support and fail during configure otherwise
+#   Off  : disable SIMD explicitly
+string(TOUPPER "${FLB_SIMD}" FLB_SIMD_MODE)
+
+if(FLB_SIMD_MODE MATCHES "^(ON|YES|TRUE|1)$")
+  set(FLB_SIMD_MODE "ON")
+elseif(FLB_SIMD_MODE MATCHES "^(OFF|NO|FALSE|0)$")
+  set(FLB_SIMD_MODE "OFF")
+elseif(NOT FLB_SIMD_MODE STREQUAL "AUTO")
+  message(FATAL_ERROR "FLB_SIMD must be one of: On, Off, Auto")
+endif()
+
+set(FLB_SIMD_ENABLED No)
+set(FLB_SIMD_RISCV_C_FLAGS "")
+
+if(NOT FLB_SIMD_MODE STREQUAL "OFF")
+  set(FLB_SIMD_SUPPORTED_ARCH No)
+
+  if(CMAKE_SYSTEM_PROCESSOR MATCHES "^(x86_64|AMD64|amd64)$" AND (CMAKE_SIZEOF_VOID_P GREATER 4))
+    set(FLB_SIMD_SUPPORTED_ARCH Yes)
+    set(FLB_SIMD_ENABLED Yes)
+  elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "^(arm64|aarch64|ARM64|AARCH64)$")
+    set(FLB_SIMD_SUPPORTED_ARCH Yes)
+    set(FLB_SIMD_ENABLED Yes)
+  elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "^(riscv64)")
+    set(FLB_SIMD_SUPPORTED_ARCH Yes)
+    check_c_compiler_flag("-march=rv64gcv_zba" FLB_SIMD_RISCV_COMPILER_FLAG)
+
+    if(FLB_SIMD_RISCV_COMPILER_FLAG)
+      set(FLB_SIMD_ENABLED Yes)
+      set(FLB_SIMD_RISCV_C_FLAGS "-march=rv64gcv_zba")
+    elseif(FLB_SIMD_MODE STREQUAL "ON")
+      message(FATAL_ERROR
+        "FLB_SIMD=On requires compiler support for -march=rv64gcv_zba on riscv64")
+    else()
+      message(WARNING
+        "FLB_SIMD=Auto disabled SIMD on riscv64 because the compiler does not "
+        "support -march=rv64gcv_zba")
+    endif()
+  endif()
+
+  if(NOT FLB_SIMD_SUPPORTED_ARCH)
+    if(FLB_SIMD_MODE STREQUAL "ON")
+      message(FATAL_ERROR
+        "FLB_SIMD=On is only supported on x86_64, aarch64/arm64, and "
+        "riscv64 with RVV compiler support")
+    else()
+      message(STATUS
+        "FLB_SIMD=Auto disabled SIMD on unsupported architecture: "
+        "${CMAKE_SYSTEM_PROCESSOR}")
+    endif()
+  endif()
+endif()
+
+if(FLB_SIMD_ENABLED)
   FLB_DEFINITION(FLB_HAVE_SIMD)
 endif()
 
@@ -665,6 +725,14 @@ add_subdirectory(${FLB_PATH_LIB_RING_BUFFER} EXCLUDE_FROM_ALL)
 
 # yyson
 add_subdirectory(${FLB_PATH_LIB_YYJSON} EXCLUDE_FROM_ALL)
+if (TARGET yyjson AND NOT FLB_COVERAGE AND
+    CMAKE_C_COMPILER_ID MATCHES "GNU|Clang|AppleClang|Intel")
+  # yyjson's O0 reader can exceed the default coroutine stack budget.
+  # Keep debug symbols, but force enough optimization to shrink the frame.
+  target_compile_options(yyjson PRIVATE
+    $<$<AND:$<COMPILE_LANGUAGE:C>,$<OR:$<CONFIG:Debug>,$<CONFIG:RelWithDebInfo>>>:-Og>
+  )
+endif()
 
 # Avro
 if(FLB_AVRO_ENCODER)

CMakeLists.txt

@@ -1,11 +1,13 @@
 if(CMAKE_SYSTEM_PROCESSOR MATCHES "^(riscv64)")
   message(STATUS "Forcing characters to be signed, as on x86_64.")
   set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -fsigned-char")
+
   if(FLB_LUAJIT)
     message(WARNING "LuaJIT is disabled, this platform does not support built-in LuaJIT and system provided one neither.")
     set(FLB_LUAJIT OFF)
   endif()
-  if(FLB_SIMD)
-    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -march=rv64gcv_zba")
+
+  if(FLB_SIMD_ENABLED AND NOT "${FLB_SIMD_RISCV_C_FLAGS}" STREQUAL "")
+    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${FLB_SIMD_RISCV_C_FLAGS}")
   endif()
 endif ()

cmake/riscv64.cmake

@@ -26,6 +26,7 @@
 #define FLB_AWS_COMPRESS_ARROW   2
 #define FLB_AWS_COMPRESS_PARQUET 3
 #define FLB_AWS_COMPRESS_ZSTD    4
+#define FLB_AWS_COMPRESS_SNAPPY  5
 
 /*
  * Get compression type from compression keyword. The return value is used to identify

include/fluent-bit/aws/flb_aws_compress.h

@@ -109,7 +109,7 @@ static inline char* basename(const char *path)
     char dir[_MAX_DIR];
     char fname[_MAX_FNAME];
     char ext[_MAX_EXT];
-    static char buf[_MAX_PATH];
+    static __declspec(thread) char buf[_MAX_PATH];
 
     _splitpath_s(path, drive, _MAX_DRIVE, dir, _MAX_DIR,
                        fname, _MAX_FNAME, ext, _MAX_EXT);

include/fluent-bit/flb_compat.h

@@ -450,6 +450,7 @@ struct flb_input_instance {
     /* TLS settings */
     int use_tls;                         /* bool, try to use TLS for I/O */
     int tls_verify;                      /* Verify certs (default: true) */
+    int tls_verify_client;               /* Verify client certs (default: false) */
     int tls_verify_hostname;             /* Verify hostname (default: false) */
     int tls_debug;                       /* mbedtls debug level          */
     char *tls_vhost;                     /* Virtual hostname for SNI     */

include/fluent-bit/flb_input.h

@@ -27,6 +27,11 @@
 #define FLB_MP_ARRAY      MSGPACK_OBJECT_ARRAY
 
 int flb_mp_count(const void *data, size_t bytes);
+int flb_mp_count_log_records(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(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);
 int flb_mp_validate_log_chunk(const void *data, size_t bytes,
                               int *out_records, size_t *processed_bytes);

include/fluent-bit/flb_mp.h

@@ -71,6 +71,8 @@ struct flb_mp_chunk_cobj *flb_mp_chunk_cobj_create(struct flb_log_event_encoder
 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_count_log_records(struct flb_mp_chunk_cobj *chunk_cobj);
+int flb_mp_chunk_cobj_normalize_groups(struct flb_mp_chunk_cobj *chunk_cobj);
 
 
 

include/fluent-bit/flb_mp_chunk.h

@@ -48,6 +48,7 @@
 #include <fluent-bit/flb_upstream_ha.h>
 #include <fluent-bit/flb_event.h>
 #include <fluent-bit/flb_processor.h>
+#include <fluent-bit/flb_mp.h>
 
 #include <cfl/cfl.h>
 #include <cmetrics/cmetrics.h>
@@ -468,6 +469,8 @@ struct flb_output_instance {
     struct cmt_gauge   *cmt_chunk_available_capacity_percent;
     /* m: output_latency_seconds */
     struct cmt_histogram *cmt_latency;
+    /* m: output_backpressure_wait_seconds */
+    struct cmt_histogram *cmt_backpressure_wait;
 
     /* OLD Metrics API */
 #ifdef FLB_HAVE_METRICS
@@ -783,6 +786,9 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
 
     if (flb_processor_is_active(o_ins->processor)) {
         if (evc->type == FLB_EVENT_TYPE_LOGS) {
+            char *normalized_buf;
+            size_t normalized_size;
+
             /* run the processor */
             ret = flb_processor_run(o_ins->processor,
                                     0,
@@ -796,7 +802,22 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
                 return NULL;
             }
 
-            records = flb_mp_count(p_buf, p_size);
+            normalized_buf = NULL;
+            normalized_size = 0;
+
+            ret = flb_mp_normalize_log_buffer_groups_msgpack(p_buf, p_size,
+                                                             &normalized_buf,
+                                                             &normalized_size);
+            if (ret == 0) {
+                if (p_buf != evc->data) {
+                    flb_free(p_buf);
+                }
+
+                p_buf = normalized_buf;
+                p_size = normalized_size;
+            }
+
+            records = flb_mp_count_log_records(p_buf, p_size);
             tmp = flb_event_chunk_create(evc->type, records, evc->tag, flb_sds_len(evc->tag), p_buf, p_size);
             if (!tmp) {
                 flb_coro_destroy(coro);
@@ -1196,10 +1217,13 @@ struct flb_output_flush *flb_output_flush_create(struct flb_task *task,
  */
 static inline void flb_output_return(int ret, struct flb_coro *co) {
     int n;
+    int records;
     int pipe_fd;
     uint32_t set;
     uint64_t val;
+    size_t bytes;
     struct flb_task *task;
+    struct flb_event_chunk *counted_event_chunk;
     struct flb_output_flush *out_flush;
     struct flb_output_instance *o_ins;
     struct flb_out_thread_instance *th_ins = NULL;
@@ -1208,10 +1232,22 @@ static inline void flb_output_return(int ret, struct flb_coro *co) {
     o_ins = out_flush->o_ins;
     task = out_flush->task;
 
-    flb_task_acquire_lock(task);
+    if (out_flush->processed_event_chunk) {
+        counted_event_chunk = out_flush->processed_event_chunk;
+    }
+    else {
+        counted_event_chunk = task->event_chunk;
+    }
 
-    flb_task_deactivate_route(task, o_ins);
+    records = task->event_chunk->total_events;
+    if (counted_event_chunk->type == FLB_EVENT_TYPE_LOGS) {
+        records = counted_event_chunk->total_events;
+    }
+    bytes = counted_event_chunk->size;
 
+    flb_task_acquire_lock(task);
+    flb_task_set_route_data(task, o_ins, records, bytes);
+    flb_task_deactivate_route(task, o_ins);
     flb_task_release_lock(task);
 
 #ifdef FLB_HAVE_CHUNK_TRACE

include/fluent-bit/flb_output.h

@@ -111,8 +111,8 @@ flb_sds_t flb_msgpack_raw_to_json_sds(const void *in_buf, size_t in_size, int es
 
 int flb_pack_time_now(msgpack_packer *pck);
 int flb_msgpack_expand_map(char *map_data, size_t map_size,
-                           msgpack_object_kv **obj_arr, int obj_arr_len,
-                           char** out_buf, int* out_size);
+                           msgpack_object_kv **obj_arr, size_t obj_arr_len,
+                           char** out_buf, size_t *out_size);
 
 struct flb_gelf_fields {
     flb_sds_t timestamp_key;