engine: added ring buffer flush request handler
#include <fluent-bit/flb_http_server.h>
#include <fluent-bit/flb_metrics.h>
#include <fluent-bit/flb_version.h>
+#include <fluent-bit/flb_ring_buffer.h>
#ifdef FLB_HAVE_METRICS
#include <fluent-bit/flb_metrics_exporter.h>
@@ -524,6 +525,14 @@ static int flb_engine_log_start(struct flb_config *config)
return 0;
}
+static void flb_engine_drain_ring_buffer_signal_channel(struct flb_ring_buffer *rb)
+{
+ static char signal_buffer[512];
+
+ flb_pipe_r(rb->signal_channels[0], signal_buffer, sizeof(signal_buffer));
+}
+
+
#ifdef FLB_HAVE_IN_STORAGE_BACKLOG
extern int sb_segregate_chunks(struct flb_config *config);
#else
@@ -538,6 +547,7 @@ int flb_engine_start(struct flb_config *config)
int ret;
uint64_t ts;
char tmp[16];
+ int rb_flush_flag;
struct flb_time t_flush;
struct mk_event *event;
struct mk_event_loop *evl;
@@ -771,13 +781,14 @@ int flb_engine_start(struct flb_config *config)
ret = sb_segregate_chunks(config);
- if (ret)
- {
+ if (ret) {
flb_error("[engine] could not segregate backlog chunks");
return -2;
}
while (1) {
+ rb_flush_flag = FLB_FALSE;
+
mk_event_wait(evl); /* potentially conditional mk_event_wait or mk_event_wait_2 based on bucket queue capacity for one shot events */
flb_event_priority_live_foreach(event, evl_bktq, evl, FLB_ENGINE_LOOP_MAX_ITER) {
if (event->type == FLB_ENGINE_EV_CORE) {
@@ -902,6 +913,15 @@ int flb_engine_start(struct flb_config *config)
ts = cmt_time_now();
handle_input_event(event->fd, ts, config);
}
+ else if(event->type == FLB_ENGINE_EV_THREAD_INPUT) {
+ flb_engine_drain_ring_buffer_signal_channel((struct flb_ring_buffer *) event->data);
+
+ rb_flush_flag = FLB_TRUE;
+ }
+ }
+
+ if (rb_flush_flag) {
+ flb_input_chunk_ring_buffer_collector(config, NULL);
}
/* Cleanup functions associated to events and timers */
