in_emitter: Fix single record chunks and respect mem_buf_limit pause

plugins/filter_multiline/ml.c

@@ -176,7 +176,7 @@ static int flush_callback(struct flb_ml_parser *parser,
         /* Emit record with original tag */
         flb_plg_trace(ctx->ins, "emitting from %s to %s", stream->input_name, stream->tag);
         ret = in_emitter_add_record(stream->tag, flb_sds_len(stream->tag), buf_data, buf_size,
-                                    ctx->ins_emitter);
+                                    ctx->ins_emitter, ctx->i_ins);
 
         return ret;
     }
@@ -526,7 +526,8 @@ static void partial_timer_cb(struct flb_config *config, void *data)
             ret = in_emitter_add_record(packer->tag, flb_sds_len(packer->tag),
                                         packer->log_encoder.output_buffer,
                                         packer->log_encoder.output_length,
-                                        ctx->ins_emitter);
+                                        ctx->ins_emitter,
+                                        ctx->i_ins);
             if (ret < 0) {
                 /* this shouldn't happen in normal execution */
                 flb_plg_warn(ctx->ins,
@@ -741,6 +742,15 @@ static int cb_ml_filter(const void *data, size_t bytes,
         return FLB_FILTER_NOTOUCH;
     }
 
+    if (ctx->i_ins == NULL){
+        ctx->i_ins = i_ins;
+    }
+    if (ctx->i_ins != i_ins) {
+        flb_plg_trace(ctx->ins, "input instance changed from %s to %s",
+                     ctx->i_ins->name, i_ins->name);
+        ctx->i_ins = i_ins;
+    }
+
     /* 'partial_message' mode */
     if (ctx->partial_mode == FLB_TRUE) {
         return ml_filter_partial(data, bytes, tag, tag_len,

plugins/filter_multiline/ml.h

@@ -73,6 +73,7 @@ struct ml_ctx {
     size_t emitter_mem_buf_limit;           /* Emitter buffer limit */
     struct flb_input_instance *ins_emitter; /* emitter input plugin instance */
     struct flb_config *config;              /* Fluent Bit context */
+    struct flb_input_instance *i_ins;       /* Fluent Bit input instance (last used)*/
 
 #ifdef FLB_HAVE_METRICS
     struct cmt_counter *cmt_emitted;
@@ -82,6 +83,7 @@ struct ml_ctx {
 /* Register external function to emit records, check 'plugins/in_emitter' */
 int in_emitter_add_record(const char *tag, int tag_len,
                           const char *buf_data, size_t buf_size,
-                          struct flb_input_instance *in);
+                          struct flb_input_instance *in,
+                          struct flb_input_instance *i_ins);
 
 #endif

plugins/filter_rewrite_tag/rewrite_tag.c

@@ -355,7 +355,8 @@ static int ingest_inline(struct flb_rewrite_tag *ctx,
  */
 static int process_record(const char *tag, int tag_len, msgpack_object map,
                           const void *buf, size_t buf_size, int *keep,
-                          struct flb_rewrite_tag *ctx, int *matched)
+                          struct flb_rewrite_tag *ctx, int *matched,
+                          struct flb_input_instance *i_ins)
 {
     int ret;
     flb_sds_t out_tag;
@@ -404,7 +405,7 @@ static int process_record(const char *tag, int tag_len, msgpack_object map,
     if (!ret) {
         /* Emit record with new tag */
         ret = in_emitter_add_record(out_tag, flb_sds_len(out_tag), buf, buf_size,
-                                    ctx->ins_emitter);
+                                    ctx->ins_emitter, i_ins);
     }
     else {
         ret = 0;
@@ -489,7 +490,7 @@ static int cb_rewrite_tag_filter(const void *data, size_t bytes,
          * If a record was emitted, the variable 'keep' will define if the record must
          * be preserved or not.
          */
-        is_emitted = process_record(tag, tag_len, map, (char *) data + pre, off - pre, &keep, ctx, &is_matched);
+        is_emitted = process_record(tag, tag_len, map, (char *) data + pre, off - pre, &keep, ctx, &is_matched, i_ins);
         if (is_emitted == FLB_TRUE) {
             /* A record with the new tag was emitted */
             emitted_num++;

plugins/filter_rewrite_tag/rewrite_tag.h

@@ -57,7 +57,8 @@ struct flb_rewrite_tag {
 /* Register external function to emit records, check 'plugins/in_emitter' */
 int in_emitter_add_record(const char *tag, int tag_len,
                           const char *buf_data, size_t buf_size,
-                          struct flb_input_instance *in);
+                          struct flb_input_instance *in,
+                          struct flb_input_instance *i_ins);
 int in_emitter_get_collector_id(struct flb_input_instance *in);
 
 

plugins/in_emitter/emitter.c

@@ -31,18 +31,28 @@
 
 #define DEFAULT_EMITTER_RING_BUFFER_FLUSH_FREQUENCY 2000
 
+/* return values */
+#define FLB_EMITTER_BUSY     -2
+
 struct em_chunk {
     flb_sds_t tag;
     struct msgpack_sbuffer mp_sbuf;  /* msgpack sbuffer        */
     struct msgpack_packer mp_pck;    /* msgpack packer         */
     struct mk_list _head;
 };
 
+struct input_ref {
+    struct flb_input_instance *i_ins;
+    struct mk_list _head;
+};
+
 struct flb_emitter {
+    int coll_fd;                        /* collector id */
     struct mk_list chunks;              /* list of all pending chunks */
     struct flb_input_instance *ins;     /* input instance */
     struct flb_ring_buffer *msgs;       /* ring buffer for cross-thread messages */
     int ring_buffer_size;               /* size of the ring buffer */
+    struct mk_list i_ins_list;          /* instance list of linked/sending inputs */
 };
 
 struct em_chunk *em_chunk_create(const char *tag, int tag_len,
@@ -85,6 +95,12 @@ int static do_in_emitter_add_record(struct em_chunk *ec,
     struct flb_emitter *ctx = (struct flb_emitter *) in->context;
     int ret;
 
+    if (flb_input_buf_paused(ctx->ins) == FLB_TRUE) {
+        flb_plg_debug(ctx->ins, "_emitter %s paused. Not processing records.",
+                         ctx->ins->name);
+        return FLB_EMITTER_BUSY;
+    }
+
     /* Associate this backlog chunk to this instance into the engine */
     ret = flb_input_log_append(in,
                                ec->tag, flb_sds_len(ec->tag),
@@ -97,7 +113,6 @@ int static do_in_emitter_add_record(struct em_chunk *ec,
         em_chunk_destroy(ec);
         return -1;
     }
-    /* Release the echunk */
     em_chunk_destroy(ec);
     return 0;
 }
@@ -108,15 +123,51 @@ int static do_in_emitter_add_record(struct em_chunk *ec,
  */
 int in_emitter_add_record(const char *tag, int tag_len,
                           const char *buf_data, size_t buf_size,
-                          struct flb_input_instance *in)
+                          struct flb_input_instance *in,
+                          struct flb_input_instance *i_ins)
 {
     struct em_chunk temporary_chunk;
     struct mk_list *head;
+    struct input_ref *i_ref;
+    bool ref_found;
+    struct mk_list *tmp;
+
     struct em_chunk *ec;
     struct flb_emitter *ctx;
 
     ctx = (struct flb_emitter *) in->context;
     ec = NULL;
+    /* Iterate over list of already known (source) inputs */
+    /* If new, add it to the list to be able to pause it later on */
+    ref_found = false;
+    mk_list_foreach_safe(head, tmp, &ctx->i_ins_list) {
+        i_ref = mk_list_entry(head, struct input_ref, _head);
+        if(i_ref->i_ins == i_ins){
+            ref_found = true;
+            break;
+        }
+    }
+    if (!ref_found) {
+        i_ref = flb_malloc(sizeof(struct input_ref));
+        if (!i_ref) {
+            flb_errno();
+            return FLB_FILTER_NOTOUCH;
+        }
+        i_ref->i_ins = i_ins;
+        mk_list_add(&i_ref->_head, &ctx->i_ins_list);
+        /* If in_emitter is paused, but new input plugin is not paused, pause it */
+        if (flb_input_buf_paused(ctx->ins) == FLB_TRUE && 
+                flb_input_buf_paused(i_ins) == FLB_FALSE) {
+            flb_input_pause(i_ins);
+        }
+    }
+
+
+    /* Restricted by mem_buf_limit */
+    if (flb_input_buf_paused(ctx->ins) == FLB_TRUE) {
+        flb_plg_debug(ctx->ins, "emitter memory buffer limit reached. Not accepting record.");
+        return FLB_EMITTER_BUSY;
+    }
 
     /* Use the ring buffer first if it exists */
     if (ctx->msgs) {
@@ -161,8 +212,7 @@ int in_emitter_add_record(const char *tag, int tag_len,
 
     /* Append raw msgpack data */
     msgpack_sbuffer_write(&ec->mp_sbuf, buf_data, buf_size);
-
-    return do_in_emitter_add_record(ec, in);
+    return 0;
 }
 
 /*
@@ -191,6 +241,34 @@ static int in_emitter_ingest_ring_buffer(struct flb_input_instance *in,
     return ret;
 }
 
+static int cb_queue_chunks(struct flb_input_instance *in,
+                           struct flb_config *config, void *data)
+{
+    int ret;
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct em_chunk *echunk;
+    struct flb_emitter *ctx;
+
+    /* Get context */
+    ctx = (struct flb_emitter *) data;
+
+    /* Try to enqueue chunks under our limits */
+    mk_list_foreach_safe(head, tmp, &ctx->chunks) {
+        echunk = mk_list_entry(head, struct em_chunk, _head);
+
+        /* Associate this backlog chunk to this instance into the engine */
+        ret = do_in_emitter_add_record(echunk, in);
+        if (ret == -1) {
+            flb_error("[in_emitter] error registering chunk with tag: %s",
+                      echunk->tag);
+            continue;
+        }
+    }
+
+    return 0;
+}
+
 static int in_emitter_start_ring_buffer(struct flb_input_instance *in, struct flb_emitter *ctx)
 {
     if (ctx->ring_buffer_size <= 0) {
@@ -232,6 +310,8 @@ static int cb_emitter_init(struct flb_input_instance *in,
     ctx->ins = in;
     mk_list_init(&ctx->chunks);
 
+    mk_list_init(&ctx->i_ins_list);
+
 
     ret = flb_input_config_map_set(in, (void *) ctx);
     if (ret == -1) {
@@ -257,23 +337,62 @@ static int cb_emitter_init(struct flb_input_instance *in,
             return -1;
         }
     }
+    else{
+        ret = flb_input_set_collector_time(in, cb_queue_chunks, 0, 25000000, config);
+        if (ret < 0) {
+            flb_error("[in_emitter] could not create collector");
+            flb_free(ctx);
+            return -1;
+        }
+        ctx->coll_fd = ret;
+    }
 
     /* export plugin context */
     flb_input_set_context(in, ctx);
 
     return 0;
 }
 
+static void cb_emitter_pause(void *data, struct flb_config *config)
+{
+    struct flb_emitter *ctx = data;
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct input_ref *i_ref;
+
+    /* Pause all known senders */
+    flb_input_collector_pause(ctx->coll_fd, ctx->ins);
+    mk_list_foreach_safe(head, tmp, &ctx->i_ins_list) {
+        i_ref = mk_list_entry(head, struct input_ref, _head);
+        flb_input_pause(i_ref->i_ins);
+    }
+}
+
+static void cb_emitter_resume(void *data, struct flb_config *config)
+{
+    struct flb_emitter *ctx = data;
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct input_ref *i_ref;
+
+    /* Resume all known senders */
+    flb_input_collector_resume(ctx->coll_fd, ctx->ins);
+    mk_list_foreach_safe(head, tmp, &ctx->i_ins_list) {
+        i_ref = mk_list_entry(head, struct input_ref, _head);
+        flb_input_resume(i_ref->i_ins);
+    }
+}
+
 static int cb_emitter_exit(void *data, struct flb_config *config)
 {
     struct mk_list *tmp;
     struct mk_list *head;
     struct flb_emitter *ctx = data;
     struct em_chunk *echunk;
     struct em_chunk ec;
+    struct input_ref *i_ref;
     int ret;
 
-
     mk_list_foreach_safe(head, tmp, &ctx->chunks) {
         echunk = mk_list_entry(head, struct em_chunk, _head);
         mk_list_del(&echunk->_head);
@@ -289,6 +408,13 @@ static int cb_emitter_exit(void *data, struct flb_config *config)
         flb_ring_buffer_destroy(ctx->msgs);
     }
 
+    mk_list_foreach_safe(head,tmp, &ctx->i_ins_list) {
+        i_ref = mk_list_entry(head, struct input_ref, _head);
+        mk_list_del(&i_ref->_head);
+        flb_free(i_ref);
+    }
+
+
     flb_free(ctx);
     return 0;
 }
@@ -312,8 +438,8 @@ struct flb_input_plugin in_emitter_plugin = {
     .cb_ingest    = NULL,
     .cb_flush_buf = NULL,
     .config_map   = config_map,
-    .cb_pause     = NULL,
-    .cb_resume    = NULL,
+    .cb_pause     = cb_emitter_pause,
+    .cb_resume    = cb_emitter_resume,
     .cb_exit      = cb_emitter_exit,
 
     /* This plugin can only be configured and invoked by the Engine only */

src/flb_input.c

@@ -1729,6 +1729,7 @@ int flb_input_resume(struct flb_input_instance *ins)
             flb_input_thread_instance_resume(ins);
         }
         else {
+            flb_info("[input] resume %s", flb_input_name(ins));
             ins->p->cb_resume(ins->context, ins->config);
         }
     }