pipe: add flb_pipe_error

On Windows, the `flb_pipe_r` and `flb_pipe_w` macros do not set errno on
failure, meaning calling `flb_errno` in error scenarios is insufficient.
This PR adds a new macro that will check the correct place,
`WSAGetLastError`, and output a similar error message. On Linux this
will still be `flb_errno`, meaning messages should work the same as they
always did, but now on Windows we will get actual error messages.

Signed-off-by: braydonk <[email protected]>

Author: braydonk

include/fluent-bit/flb_input.h

@@ -671,7 +671,7 @@ static FLB_INLINE void flb_input_return(struct flb_coro *coro) {
     val = FLB_BITS_U64_SET(FLB_ENGINE_IN_CORO, ins->id);
     n = flb_pipe_w(ins->ch_events[1], (void *) &val, sizeof(val));
     if (n == -1) {
-        flb_errno();
+        flb_pipe_error();
     }
 
     flb_input_coro_prepare_destroy(input_coro);

include/fluent-bit/flb_log.h

@@ -232,11 +232,18 @@ static inline int flb_log_suppress_check(int log_suppress_interval, const char *
 int flb_log_worker_init(struct flb_worker *worker);
 int flb_log_worker_destroy(struct flb_worker *worker);
 int flb_errno_print(int errnum, const char *file, int line);
+int flb_WSAGetLastError_print(int errnum, const char *file, int line);
 
 #ifdef __FLB_FILENAME__
 #define flb_errno() flb_errno_print(errno, __FLB_FILENAME__, __LINE__)
+#ifdef WIN32
+#define flb_WSAGetLastError() flb_WSAGetLastError_print(WSAGetLastError(), __FLB_FILENAME__, __LINE__)
+#endif
 #else
 #define flb_errno() flb_errno_print(errno, __FILE__, __LINE__)
+#ifdef WIN32
+#define flb_WSAGetLastError() flb_WSAGetLastError_print(WSAGetLastError(), __FILE__, __LINE__)
+#endif
 #endif
 
 #endif

include/fluent-bit/flb_output.h

@@ -1199,7 +1199,7 @@ static inline void flb_output_return(int ret, struct flb_coro *co) {
     /* Notify the event loop about our return status */
     n = flb_pipe_w(pipe_fd, (void *) &val, sizeof(val));
     if (n == -1) {
-        flb_errno();
+        flb_pipe_error();
     }
 
     /*

include/fluent-bit/flb_pipe.h

@@ -28,12 +28,14 @@
 #define flb_sockfd_t evutil_socket_t
 #define flb_pipe_w(fd, buf, len) send(fd, buf, len, 0)
 #define flb_pipe_r(fd, buf, len) recv(fd, buf, len, 0)
+#define flb_pipe_error() flb_WSAGetLastError()
 #define FLB_PIPE_WOULDBLOCK() (WSAGetLastError() == WSAEWOULDBLOCK)
 #else
 #define flb_pipefd_t int
 #define flb_sockfd_t int
 #define flb_pipe_w(fd, buf, len) write(fd, buf, len)
 #define flb_pipe_r(fd, buf, len) read(fd, buf, len)
+#define flb_pipe_error() flb_errno()
 #define FLB_PIPE_WOULDBLOCK() (errno == EAGAIN || errno == EWOULDBLOCK)
 #endif
 
@@ -43,5 +45,6 @@ int flb_pipe_close(flb_pipefd_t fd);
 int flb_pipe_set_nonblocking(flb_pipefd_t fd);
 ssize_t flb_pipe_read_all(int fd, void *buf, size_t count);
 ssize_t flb_pipe_write_all(int fd, const void *buf, size_t count);
+void flb_pipe_log_last_error();
 
 #endif

include/fluent-bit/flb_scheduler.h

@@ -234,7 +234,7 @@ static FLB_INLINE void flb_sched_timer_cb_coro_return()
     val = FLB_BITS_U64_SET(FLB_SCHED_TIMER_CORO_RETURN, stc->id);
     n = flb_pipe_w(sched->ch_events[1], &val, sizeof(val));
     if (n == -1) {
-        flb_errno();
+        flb_pipe_error();
     }
 
     flb_coro_yield(coro, FLB_TRUE);

plugins/in_exec/in_exec.c

@@ -54,7 +54,7 @@ static int in_exec_collect(struct flb_input_instance *ins,
     if (ctx->oneshot == FLB_TRUE) {
         ret = flb_pipe_r(ctx->ch_manager[0], &val, sizeof(val));
         if (ret == -1) {
-            flb_errno();
+            flb_pipe_error();
             return -1;
         }
     }
@@ -256,7 +256,7 @@ static int in_exec_config_read(struct flb_exec *ctx,
         flb_plg_error(in, "unable to load configuration");
         return -1;
     }
-   
+
     /* filepath setting */
     if (ctx->cmd == NULL) {
         flb_plg_error(in, "no input 'command' was given");
@@ -418,7 +418,7 @@ static int in_exec_prerun(struct flb_input_instance *ins,
     /* Kick the oneshot execution */
     ret = flb_pipe_w(ctx->ch_manager[1], &val, sizeof(val));
     if (ret == -1) {
-        flb_errno();
+        flb_pipe_error();
         return -1;
     }
     return 0;

plugins/in_exec_wasi/in_exec_wasi.c

@@ -69,7 +69,7 @@ static int in_exec_wasi_collect(struct flb_input_instance *ins,
         ret = flb_pipe_r(ctx->ch_manager[0], &val, sizeof(val));
         if (ret == -1) {
             fclose(stdoutp);
-            flb_errno();
+            flb_pipe_error();
             return -1;
         }
     }
@@ -404,7 +404,7 @@ static int in_exec_wasi_prerun(struct flb_input_instance *ins,
     /* Kick the oneshot execution */
     ret = flb_pipe_w(ctx->ch_manager[1], &val, sizeof(val));
     if (ret == -1) {
-        flb_errno();
+        flb_pipe_error();
         return -1;
     }
     return 0;

plugins/in_lib/in_lib.c

@@ -69,6 +69,7 @@ static int in_lib_collect(struct flb_input_instance *ins,
     flb_plg_trace(ctx->ins, "in_lib read() = %i", bytes);
     if (bytes == -1) {
         perror("read");
+        flb_pipe_error();
         if (errno == -EPIPE) {
             return -1;
         }

plugins/in_tail/tail.c

@@ -50,7 +50,7 @@ static inline int consume_byte(flb_pipefd_t fd)
     /* We need to consume the byte */
     ret = flb_pipe_r(fd, (char *) &val, sizeof(val));
     if (ret <= 0) {
-        flb_errno();
+        flb_pipe_error();
         return -1;
     }
 

plugins/in_tail/tail_signal.h

@@ -45,7 +45,7 @@ static inline int tail_signal_manager(struct flb_tail_config *ctx)
     /* Insert a dummy event into the channel manager */
     n = flb_pipe_w(ctx->ch_manager[1], (const char *) &val, sizeof(val));
     if (n == -1) {
-        flb_errno();
+        flb_pipe_error();
         return -1;
     }
     else {
@@ -68,7 +68,7 @@ static inline int tail_signal_pending(struct flb_tail_config *ctx)
      * notification is already pending, it's safe to ignore.
      */
     if (n == -1 && !FLB_PIPE_WOULDBLOCK()) {
-        flb_errno();
+        flb_pipe_error();
         return -1;
     }
 
@@ -87,7 +87,7 @@ static inline int tail_consume_pending(struct flb_tail_config *ctx)
     do {
         ret = flb_pipe_r(ctx->ch_pending[0], (char *) &val, sizeof(val));
         if (ret <= 0 && !FLB_PIPE_WOULDBLOCK()) {
-            flb_errno();
+            flb_pipe_error();
             return -1;
         }
     } while (!FLB_PIPE_WOULDBLOCK());