fluent · edsiper · Jul 13, 2026 · May 26, 2026 · May 26, 2026 · May 26, 2026
@@ -816,6 +816,7 @@ add_subdirectory(${FLB_PATH_LIB_JSMN})
 # Runtime Tests (filter_kubernetes) requires HTTP Server
 if(FLB_TESTS_RUNTIME)
   FLB_OPTION(FLB_HTTP_SERVER  ON)
+  FLB_OPTION(FLB_IN_EVENT_TEST ON)
 endif()
 
 # Monkey Core Library

@@ -95,7 +95,7 @@ if(FLB_WINDOWS_DEFAULTS)
   set(FLB_OUT_CHRONICLE         Yes)
   set(FLB_OUT_DATADOG           Yes)
   set(FLB_OUT_ES                Yes)
-  set(FLB_OUT_EXIT               No)
+  set(FLB_OUT_EXIT              Yes)
   set(FLB_OUT_FORWARD           Yes)
   set(FLB_OUT_GELF              Yes)
   set(FLB_OUT_HTTP              Yes)

@@ -58,8 +58,13 @@ static int file_to_buffer(const char *path,
     ssize_t bytes;
     FILE *fp;
     struct stat st;
+    const char *file_mode = "r";
 
-    if (!(fp = fopen(path, "r"))) {
+#ifdef FLB_SYSTEM_WINDOWS
+    file_mode = "rb";
+#endif
+
+    if (!(fp = fopen(path, file_mode))) {
         return -1;
     }
 
@@ -337,8 +342,13 @@ static int get_meta_file_info(struct flb_kube *ctx, const char *namespace,
     struct stat sb;
     int packed = -1;
     int ret;
+    int open_flags = O_RDONLY;
     char uri[1024];
 
+#ifdef FLB_SYSTEM_WINDOWS
+    open_flags |= O_BINARY;
+#endif
+
     if (ctx->meta_preload_cache_dir && namespace) {
 
         if (podname && strlen(podname) > 0) {
@@ -350,7 +360,7 @@ static int get_meta_file_info(struct flb_kube *ctx, const char *namespace,
                     ctx->meta_preload_cache_dir, namespace);
         }
         if (ret > 0) {
-            fd = open(uri, O_RDONLY, 0);
+            fd = open(uri, open_flags, 0);
             if (fd != -1) {
                 if (fstat(fd, &sb) == 0) {
                     payload = flb_malloc(sb.st_size);

@@ -92,8 +92,9 @@ void *time_ticker(void *args)
                          ctx->hash->total / ctx->hash->size);
         }
         pthread_mutex_unlock(&throttle_mut);
-        /* sleep is a cancelable function */
+        /* Windows sleep is not a pthread cancellation point. */
         sleep(ctx->ticker_data.seconds);
+        pthread_testcancel();
-        /* Windows sleep is not a pthread cancellation point. */
-        sleep(ctx->ticker_data.seconds);
-        pthread_testcancel();
+        /* Windows sleep is not a pthread cancellation point. */
+        {
+            int remaining = ctx->ticker_data.seconds;
+
+            while (remaining > 0) {
+                int chunk = remaining > 5 ? 5 : remaining;
+
+                sleep(chunk);
+                pthread_testcancel();
+                remaining -= chunk;
+            }
+        }
-        /* Windows sleep is not a pthread cancellation point. */
-        sleep(ctx->ticker_data.seconds);
-        pthread_testcancel();
+        /* Windows sleep is not a pthread cancellation point. */
+        {
+            int remaining = ctx->ticker_data.seconds;
+
+            while (remaining > 0) {
+                int chunk = remaining > 5 ? 5 : remaining;
+
+                sleep(chunk);
+                pthread_testcancel();
+                remaining -= chunk;
+            }
+        }
     }
 }
 

@@ -33,8 +33,8 @@
 #define STATUS_PENDING  -1
 #define CALLBACK_TIME    2 /* 2 seconds */
 
-#define SERVER_PORT      "9092"
-#define SERVER_IFACE     "0.0.0.0"
+#define SERVER_PORT      "0"
+#define SERVER_IFACE     "127.0.0.1"
 
 struct unit_test {
     int id;
@@ -55,7 +55,8 @@ struct unit_test tests[] = {
 
 struct event_test {
     flb_pipefd_t pipe[2];
-    int server_fd;
+    flb_sockfd_t server_fd;
+    int server_port;
     int client_coll_id;
     struct flb_upstream *upstream;
     struct unit_test *tests;
@@ -127,9 +128,9 @@ static int cb_collector_time(struct flb_input_instance *ins,
      * to our local pipe.
      */
     val = 1;
-    ret = write(ctx->pipe[1], &val, sizeof(val));
+    ret = flb_pipe_w(ctx->pipe[1], &val, sizeof(val));
     if (ret == -1) {
-        flb_errno();
+        flb_pipe_error();
         set_unit_test_status(ctx, 0, STATUS_ERROR);
         flb_engine_exit(config);
     }
@@ -143,13 +144,13 @@ static int cb_collector_fd(struct flb_input_instance *ins,
                            struct flb_config *config, void *in_context)
 {
     uint64_t val = 0;
-    size_t bytes;
+    ssize_t bytes;
     struct unit_test *ut;
     struct event_test *ctx = (struct event_test *) in_context;
 
-    bytes = read(ctx->pipe[0], &val, sizeof(val));
+    bytes = flb_pipe_r(ctx->pipe[0], &val, sizeof(val));
     if (bytes <= 0) {
-        flb_errno();
+        flb_pipe_error();
         set_unit_test_status(ctx, 1, STATUS_ERROR);
         flb_engine_exit(config);
     }
@@ -242,12 +243,28 @@ static struct event_test *config_create(struct flb_input_instance *ins)
     return ctx;
 }
 
+static int get_server_port(flb_sockfd_t fd)
+{
+    int ret;
+    socklen_t len;
+    struct sockaddr_in addr;
+
+    len = sizeof(addr);
+    ret = getsockname(fd, (struct sockaddr *) &addr, &len);
+    if (ret == -1) {
+        return -1;
+    }
+
+    return ntohs(addr.sin_port);
+}
+
 /* Initialize plugin */
 static int cb_event_test_init(struct flb_input_instance *ins,
                               struct flb_config *config, void *data)
 {
-    int fd;
     int ret;
+    int port;
+    flb_sockfd_t fd;
     struct unit_test *ut;
     struct event_test *ctx = NULL;
     struct flb_upstream *upstream;
@@ -297,7 +314,15 @@ static int cb_event_test_init(struct flb_input_instance *ins,
         return -1;
     }
     flb_net_socket_nonblocking(fd);
-        ctx->server_fd = fd;
+    ctx->server_fd = fd;
+
+    port = get_server_port(ctx->server_fd);
+    if (port <= 0) {
+        flb_errno();
+        config_destroy(ctx);
+        return -1;
+    }
+    ctx->server_port = port;
 
     /* socket server */
     ret = flb_input_set_collector_socket(ins,
@@ -321,7 +346,7 @@ static int cb_event_test_init(struct flb_input_instance *ins,
     ctx->client_coll_id = ret;
 
     /* upstream context for socket client */
-    upstream = flb_upstream_create(config, "127.0.0.1", atoi(SERVER_PORT),
+    upstream = flb_upstream_create(config, SERVER_IFACE, ctx->server_port,
                                    FLB_IO_TCP, NULL);
     if (!upstream) {
         config_destroy(ctx);

@@ -54,8 +54,13 @@ static int file_to_buffer(const char *path,
     ssize_t bytes;
     FILE *fp;
     struct stat st;
+    const char *file_mode = "r";
 
-    if (!(fp = fopen(path, "r"))) {
+#ifdef FLB_SYSTEM_WINDOWS
+    file_mode = "rb";
+#endif
+
+    if (!(fp = fopen(path, file_mode))) {
         return -1;
     }
 
@@ -303,7 +308,7 @@ static bool check_event_is_filtered(struct k8s_events *ctx, msgpack_object *obj,
     flb_sds_t uid;
     uint64_t resource_version;
 
-    outdated = cfl_time_now() - (ctx->retention_time * 1000000000L);
+    outdated = cfl_time_now() - ((uint64_t) ctx->retention_time * 1000000000ULL);
     if (flb_time_to_nanosec(event_time) < outdated) {
         flb_plg_debug(ctx->ins, "Item is older than retention_time: %" PRIu64 " < %" PRIu64,
                       flb_time_to_nanosec(event_time), outdated);
@@ -655,7 +660,8 @@ static int k8s_events_cleanup_db(struct flb_input_instance *ins,
         FLB_INPUT_RETURN(0);
     }
 
-    retention_time_ago = cfl_time_now() - (ctx->retention_time * 1000000000L);
+    retention_time_ago = cfl_time_now() -
+                         ((uint64_t) ctx->retention_time * 1000000000ULL);
     sqlite3_bind_int64(ctx->stmt_delete_old_kubernetes_events,
                         1, (int64_t)retention_time_ago);
     ret = sqlite3_step(ctx->stmt_delete_old_kubernetes_events);

@@ -48,7 +48,8 @@ static int tail_is_excluded(char *path, struct flb_tail_config *ctx)
 
     mk_list_foreach(head, ctx->exclude_list) {
         pattern = mk_list_entry(head, struct flb_slist_entry, _head);
-        if (PathMatchSpecA(path, pattern->str)) {
+        if (PathMatchSpecA(path, pattern->str) ||
+            PathMatchSpecA(PathFindFileNameA(path), pattern->str)) {
             return FLB_TRUE;
         }
     }

@@ -759,6 +759,11 @@ static int read_label_map_path_file(struct flb_output_instance *ins, flb_sds_t p
     struct stat st;
     size_t file_size;
     size_t ret_size;
+    const char *file_mode = "r";
+
+#ifdef FLB_SYSTEM_WINDOWS
+    file_mode = "rb";
+#endif
 
     ret = access(path, R_OK);
     if (ret < 0) {
@@ -775,7 +780,7 @@ static int read_label_map_path_file(struct flb_output_instance *ins, flb_sds_t p
     }
     file_size = st.st_size;
 
-    fp = fopen(path, "r");
+    fp = fopen(path, file_mode);
     if (fp == NULL) {
         flb_plg_error(ins, "can't open %s", path);
         return -1;

@@ -57,7 +57,11 @@ FLB_TLS_DEFINE(struct worker_info, s3_worker_info);
 #ifdef FLB_SYSTEM_WINDOWS
 static int setenv(const char *name, const char *value, int overwrite)
 {
-    return SetEnvironmentVariableA(name, value);
+    if (overwrite == 0 && getenv(name) != NULL) {
+        return 0;
+    }
+
+    return _putenv_s(name, value);
 }
 #endif
 
@@ -391,6 +395,37 @@ static flb_sds_t concat_path(char *p1, char *p2)
     return dir;
 }
 
+static flb_sds_t create_buffer_path(struct flb_s3 *ctx)
+{
+#ifdef FLB_SYSTEM_WINDOWS
+    char *temp_dir;
+    flb_sds_t dir;
+    flb_sds_t tmp;
+
+    if (strcmp(ctx->store_dir, "/tmp/fluent-bit/s3") == 0) {
+        temp_dir = getenv("TEMP");
+        if (temp_dir == NULL) {
+            temp_dir = getenv("TMP");
+        }
+
+        if (temp_dir != NULL) {
+            dir = flb_sds_create_size(64);
+            tmp = flb_sds_printf(&dir, "%s/fluent-bit/s3/%s",
+                                 temp_dir, ctx->bucket);
+            if (tmp == NULL) {
+                flb_errno();
+                flb_sds_destroy(dir);
+                return NULL;
+            }
+
+            return tmp;
+        }
+    }
+#endif
+
+    return concat_path(ctx->store_dir, ctx->bucket);
+}
+
 /* Reads in index value from metadata file and sets seq_index to value */
 static int read_seq_index(char *seq_index_file, uint64_t *seq_index)
 {
@@ -769,7 +804,7 @@ static int cb_s3_init(struct flb_output_instance *ins,
      * We append the bucket name to the dir, to support multiple instances
      * of this plugin using the same buffer dir
      */
-    tmp_sds = concat_path(ctx->store_dir, ctx->bucket);
+    tmp_sds = create_buffer_path(ctx);
     if (!tmp_sds) {
         flb_plg_error(ctx->ins, "Could not construct buffer path");
         return -1;
@@ -1103,8 +1138,14 @@ static int cb_s3_init(struct flb_output_instance *ins,
     ctx->provider->provider_vtable->init(ctx->provider);
 
     ctx->timer_created = FLB_FALSE;
-    ctx->timer_ms = (int) (ctx->upload_timeout / 6) * 1000;
-    if (s3_plugin_under_test() == FLB_FALSE) {
+    if (s3_plugin_under_test() == FLB_TRUE) {
+        ctx->timer_ms = (int) (ctx->upload_timeout * 1000 / 6);
+        if (ctx->timer_ms < 100) {
+            ctx->timer_ms = 100;
+        }
+    }
+    else {
+        ctx->timer_ms = (int) (ctx->upload_timeout / 6) * 1000;
         if (ctx->timer_ms > UPLOAD_TIMER_MAX_WAIT) {
             ctx->timer_ms = UPLOAD_TIMER_MAX_WAIT;
         }

@@ -969,8 +969,8 @@ static int cb_syslog_exit(void *data, struct flb_config *config)
         flb_upstream_destroy(ctx->u);
     }
 
-    if (ctx->fd > 0) {
-        close(ctx->fd);
+    if (ctx->fd != FLB_INVALID_SOCKET) {
+        flb_socket_close(ctx->fd);
     }
 
     flb_syslog_config_destroy(ctx);

@@ -67,12 +67,21 @@
 #define AWS_CLIENT_MOCK_H
 
 /* Variadic Argument Counter, Counts up to 64 variadic args */
+#ifdef _MSC_VER
+#define FLB_AWS_CLIENT_MOCK_COUNT64(...)                                                 \
+    _FLB_AWS_CLIENT_MOCK_COUNT64(dummy __VA_OPT__(,) __VA_ARGS__, 63, 62, 61, 60, 59,    \
+                                 58, 57, 56, 55, 54, 53, 52, 51, 50, 49, 48, 47, 46, 45, \
+                                 44, 43, 42, 41, 40, 39, 38, 37, 36, 35, 34, 33, 32, 31, \
+                                 30, 29, 28, 27, 26, 25, 24, 23, 22, 21, 20, 19, 18, 17, \
+                                 16, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0)
+#else
 #define FLB_AWS_CLIENT_MOCK_COUNT64(...)                                                 \
     _FLB_AWS_CLIENT_MOCK_COUNT64(dummy, ##__VA_ARGS__, 63, 62, 61, 60, 59, 58, 57, 56,   \
                                  55, 54, 53, 52, 51, 50, 49, 48, 47, 46, 45, 44, 43, 42, \
                                  41, 40, 39, 38, 37, 36, 35, 34, 33, 32, 31, 30, 29, 28, \
                                  27, 26, 25, 24, 23, 22, 21, 20, 19, 18, 17, 16, 15, 14, \
                                  13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0)
+#endif
 #define _FLB_AWS_CLIENT_MOCK_COUNT64(                                                    \
     x0, x1, x2, x3, x4, x5, x6, x7, x8, x9, x10, x11, x12, x13, x14, x15, x16, x17, x18, \
     x19, x20, x21, x22, x23, x24, x25, x26, x27, x28, x29, x30, x31, x32, x33, x34, x35, \