in sdtin: add buffer_size parameter

Author: dojci

plugins/in_stdin/in_stdin.c

@@ -25,6 +25,7 @@
 #include <fluent-bit/flb_time.h>
 #include <fluent-bit/flb_parser.h>
 #include <fluent-bit/flb_error.h>
+#include <fluent-bit/flb_utils.h>
 
 #include <msgpack.h>
 
@@ -106,7 +107,7 @@ static int in_stdin_collect(struct flb_input_instance *ins,
 
     bytes = read(ctx->fd,
                  ctx->buf + ctx->buf_len,
-                 sizeof(ctx->buf) - ctx->buf_len - 1);
+                 ctx->buf_size - ctx->buf_len - 1);
     flb_plg_trace(ctx->ins, "stdin read() = %i", bytes);
 
     if (bytes == 0) {
@@ -210,45 +211,91 @@ static int in_stdin_collect(struct flb_input_instance *ins,
     return 0;
 }
 
+/* Read config file and*/
+static int in_stdin_config_init(struct flb_in_stdin_config *ctx,
+                               struct flb_input_instance *in,
+                               struct flb_config *config)
+{
+    const char *pval = NULL;
+
+    /* parser settings */
+    pval = flb_input_get_property("parser", in);
+    if (pval) {
+        ctx->parser = flb_parser_get(pval, config);
+        if (!ctx->parser) {
+            flb_plg_error(ctx->ins, "requested parser '%s' not found", pval);
+        }
+    }
+    else {
+        ctx->parser = NULL;
+    }
+
+    /* buffer size setting */
+    pval = flb_input_get_property("buffer_size", in);
+    if (pval != NULL && flb_utils_size_to_bytes(pval) > 0) {
+        ctx->buf_size = flb_utils_size_to_bytes(pval);
+    }
+    else {
+        ctx->buf_size = DEFAULT_BUF_SIZE;
+    }
+
+    flb_plg_debug(ctx->ins, "buf_size=%zu", ctx->buf_size);
+    return 0;
+}
+
+static void in_stdin_config_destroy(struct flb_in_stdin_config *ctx)
+{
+    if (!ctx) {
+        return;
+    }
+
+    /* release buffer */
+    if (ctx->buf) {
+        flb_free(ctx->buf);
+    }
+    flb_free(ctx);
+}
+
 /* Initialize plugin */
 static int in_stdin_init(struct flb_input_instance *in,
                          struct flb_config *config, void *data)
 {
     int fd;
     int ret;
-    const char *tmp;
     struct flb_in_stdin_config *ctx;
     (void) data;
 
-    /* Allocate space for the configuration */
+    /* Allocate space for the configuration context */
     ctx = flb_malloc(sizeof(struct flb_in_stdin_config));
     if (!ctx) {
         return -1;
     }
+
+    ctx->buf = NULL;
     ctx->buf_len = 0;
     ctx->ins = in;
 
+    /* Initialize stdin config */
+    ret = in_stdin_config_init(ctx, in, config);
+    if (ret < 0) {
+        goto init_error;
+    }
+
+    ctx->buf = flb_malloc(ctx->buf_size);
+    if (!ctx->buf) {
+        flb_errno();
+        goto init_error;
+    }
+
     /* Clone the standard input file descriptor */
     fd = dup(STDIN_FILENO);
     if (fd == -1) {
         flb_errno();
         flb_plg_error(ctx->ins, "Could not open standard input!");
-        flb_free(ctx);
-        return -1;
+        goto init_error;
     }
     ctx->fd = fd;
 
-    tmp = flb_input_get_property("parser", in);
-    if (tmp) {
-        ctx->parser = flb_parser_get(tmp, config);
-        if (!ctx->parser) {
-            flb_plg_error(ctx->ins, "requested parser '%s' not found", tmp);
-        }
-    }
-    else {
-        ctx->parser = NULL;
-    }
-
     /* Always initialize built-in JSON pack state */
     flb_pack_state_init(&ctx->pack_state);
     ctx->pack_state.multiple = FLB_TRUE;
@@ -263,12 +310,16 @@ static int in_stdin_init(struct flb_input_instance *in,
                                         config);
     if (ret == -1) {
         flb_plg_error(ctx->ins, "Could not set collector for STDIN input plugin");
-        flb_free(ctx);
-        return -1;
+        goto init_error;
     }
     ctx->coll_fd = ret;
 
     return 0;
+
+init_error:
+    in_stdin_config_destroy(ctx);
+
+    return -1;
 }
 
 /* Cleanup serial input */
@@ -284,7 +335,7 @@ static int in_stdin_exit(void *in_context, struct flb_config *config)
         close(ctx->fd);
     }
     flb_pack_state_reset(&ctx->pack_state);
-    flb_free(ctx);
+    in_stdin_config_destroy(ctx);
 
     return 0;
 }

plugins/in_stdin/in_stdin.h

@@ -24,12 +24,15 @@
 #include <fluent-bit/flb_config.h>
 #include <fluent-bit/flb_input.h>
 
+#define DEFAULT_BUF_SIZE 16384
+
 /* STDIN Input configuration & context */
 struct flb_in_stdin_config {
-    int fd;                           /* stdin file descriptor */
-    int coll_fd;                      /* collector fd          */
-    int buf_len;                      /* read buffer length    */
-    char buf[8192 * 2];               /* read buffer: 16Kb max */
+    int    fd;                /* stdin file descriptor */
+    int    coll_fd;           /* collector fd          */
+    size_t buf_size;          /* size of a buffer      */
+    int    buf_len;           /* read buffer length    */
+    char   *buf;              /* read buffer           */
 
     /* Parser / Format */
     struct flb_parser *parser;