http_server: run internal server on worker thread safely

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

include/fluent-bit/http_server/flb_hs.h

@@ -26,12 +26,14 @@
 #include <fluent-bit/flb_sds.h>
 #include <fluent-bit/http_server/flb_http_server.h>
 #include <monkey/mk_core.h>
+#include <pthread.h>
 
 /*
  * HTTP buffers that contains certain cached data to be used
  * by end-points.
  */
 struct flb_hs_buf {
+    pthread_mutex_t lock;
     int users;
     int pending_free;
     flb_sds_t data;
@@ -81,6 +83,7 @@ struct flb_hs {
     struct flb_config *config;
     struct mk_list routes;
     struct mk_list health_metrics;
+    pthread_mutex_t health_metrics_lock;
     struct flb_health_check_metrics_counter health_counter;
 
     struct flb_hs_buf metrics;
@@ -102,6 +105,8 @@ int flb_hs_push_storage_metrics(struct flb_hs *hs, void *data, size_t size);
 int flb_hs_destroy(struct flb_hs *ctx);
 int flb_hs_start(struct flb_hs *hs);
 void flb_hs_cmt_buffer_destroy(void *data);
+int flb_hs_buf_acquire(struct flb_hs_buf *buffer, int require_data,
+                       int require_raw_data);
 void flb_hs_buf_release(struct flb_hs_buf *buffer, void (*raw_free)(void *));
 int flb_hs_register_endpoint(struct flb_hs *hs,
                              const char *path,

src/http_server/api/v1/health.c

@@ -48,12 +48,15 @@ int flb_hs_health_state_get(struct flb_hs *hs, struct flb_hs_health_state *state
     }
 
     memset(state, 0, sizeof(struct flb_hs_health_state));
+    pthread_mutex_lock(&hs->health_metrics_lock);
+
     state->error_limit = hs->health_counter.error_limit;
     state->retry_failure_limit = hs->health_counter.retry_failure_limit;
     state->period_limit = hs->health_counter.period_limit;
 
     if (mk_list_is_empty(&hs->health_metrics) == 0) {
         state->healthy = FLB_TRUE;
+        pthread_mutex_unlock(&hs->health_metrics_lock);
         return 0;
     }
 
@@ -79,10 +82,12 @@ int flb_hs_health_state_get(struct flb_hs *hs, struct flb_hs_health_state *state
     if (state->errors > hs->health_counter.error_limit ||
         state->retries_failed > hs->health_counter.retry_failure_limit) {
         state->healthy = FLB_FALSE;
+        pthread_mutex_unlock(&hs->health_metrics_lock);
         return 0;
     }
 
     state->healthy = FLB_TRUE;
+    pthread_mutex_unlock(&hs->health_metrics_lock);
 
     return 0;
 }

src/http_server/api/v1/metrics.c

@@ -37,9 +37,10 @@
 /* Return the newest metrics buffer */
 static struct flb_hs_buf *metrics_get_latest(struct flb_hs *hs)
 {
-    if (hs->metrics.data == NULL || hs->metrics.raw_data == NULL) {
+    if (flb_hs_buf_acquire(&hs->metrics, FLB_TRUE, FLB_TRUE) != 0) {
         return NULL;
     }
+
     return &hs->metrics;
 }
 
@@ -159,9 +160,6 @@ static int cb_metrics_prometheus(struct flb_hs *hs,
         return flb_http_response_commit(response);
     }
 
-    /* ref count */
-    buf->users++;
-
     /* Compose outgoing buffer string */
     sds = flb_sds_create_size(1024);
     if (!sds) {
@@ -416,8 +414,6 @@ static int cb_metrics(struct flb_hs *hs,
         return flb_http_response_commit(response);
     }
 
-    buf->users++;
-
     flb_hs_response_set_payload(response, 200,
                                 FLB_HS_CONTENT_TYPE_JSON,
                                 buf->data, flb_sds_len(buf->data));

src/http_server/api/v1/storage.c

@@ -30,9 +30,10 @@
 /* Return the newest storage metrics buffer */
 static struct flb_hs_buf *storage_metrics_get_latest(struct flb_hs *hs)
 {
-    if (hs->storage_metrics.data == NULL) {
+    if (flb_hs_buf_acquire(&hs->storage_metrics, FLB_TRUE, FLB_FALSE) != 0) {
         return NULL;
     }
+
     return &hs->storage_metrics;
 }
 
@@ -51,8 +52,6 @@ static int cb_storage(struct flb_hs *hs,
         return flb_http_response_commit(response);
     }
 
-    buf->users++;
-
     flb_hs_response_set_payload(response, 200,
                                 FLB_HS_CONTENT_TYPE_JSON,
                                 buf->data, flb_sds_len(buf->data));

src/http_server/api/v2/metrics.c

@@ -36,9 +36,10 @@
 /* Return the newest metrics buffer */
 static struct flb_hs_buf *metrics_get_latest(struct flb_hs *hs)
 {
-    if (hs->metrics_v2.raw_data == NULL) {
+    if (flb_hs_buf_acquire(&hs->metrics_v2, FLB_FALSE, FLB_TRUE) != 0) {
         return NULL;
     }
+
     return &hs->metrics_v2;
 }
 
@@ -59,7 +60,6 @@ static int cb_metrics_prometheus(struct flb_hs *hs,
         return flb_http_response_commit(response);
     }
 
-    buf->users++;
     cmt = (struct cmt *) buf->raw_data;
 
     /* convert CMetrics to text */
@@ -97,7 +97,6 @@ static int cb_metrics(struct flb_hs *hs,
         return flb_http_response_commit(response);
     }
 
-    buf->users++;
     cmt = (struct cmt *) buf->raw_data;
 
     /* convert CMetrics to text */