diff --git a/include/fluent-bit/http_server/flb_hs.h b/include/fluent-bit/http_server/flb_hs.h index 3e25de019c8..cfdf0cd9741 100644 --- a/include/fluent-bit/http_server/flb_hs.h +++ b/include/fluent-bit/http_server/flb_hs.h @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -32,6 +33,7 @@ * 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, diff --git a/src/http_server/api/v1/health.c b/src/http_server/api/v1/health.c index 17ffce5e215..2d1948ad2ec 100644 --- a/src/http_server/api/v1/health.c +++ b/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; } diff --git a/src/http_server/api/v1/metrics.c b/src/http_server/api/v1/metrics.c index af5dbad2680..9ca32cce9e7 100644 --- a/src/http_server/api/v1/metrics.c +++ b/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)); diff --git a/src/http_server/api/v1/storage.c b/src/http_server/api/v1/storage.c index 4e5dd2365a7..951f9367636 100644 --- a/src/http_server/api/v1/storage.c +++ b/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)); diff --git a/src/http_server/api/v2/metrics.c b/src/http_server/api/v2/metrics.c index f27cf48d826..9f55d9f720b 100644 --- a/src/http_server/api/v2/metrics.c +++ b/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 */ diff --git a/src/http_server/flb_hs.c b/src/http_server/flb_hs.c index f49d0fcb2e2..b5ad60fab21 100644 --- a/src/http_server/flb_hs.c +++ b/src/http_server/flb_hs.c @@ -110,8 +110,19 @@ static int flb_hs_request_handler(struct flb_http_request *request, return flb_http_response_commit(response); } -static void flb_hs_buf_cleanup(struct flb_hs_buf *buffer, - void (*raw_free)(void *)) +static int flb_hs_buf_init(struct flb_hs_buf *buffer) +{ + if (buffer == NULL) { + return -1; + } + + mk_list_init(&buffer->_head); + + return pthread_mutex_init(&buffer->lock, NULL); +} + +static void flb_hs_buf_cleanup_locked(struct flb_hs_buf *buffer, + void (*raw_free)(void *)) { if (buffer == NULL) { return; @@ -142,17 +153,86 @@ static void flb_hs_buf_cleanup(struct flb_hs_buf *buffer, buffer->users = 0; } +static void flb_hs_buf_destroy(struct flb_hs_buf *buffer, + void (*raw_free)(void *)) +{ + if (buffer == NULL) { + return; + } + + pthread_mutex_lock(&buffer->lock); + flb_hs_buf_cleanup_locked(buffer, raw_free); + pthread_mutex_unlock(&buffer->lock); + pthread_mutex_destroy(&buffer->lock); +} + +static int flb_hs_buf_set(struct flb_hs_buf *buffer, flb_sds_t data, + void *raw_data, size_t raw_size, + void (*raw_free)(void *)) +{ + if (buffer == NULL) { + return -1; + } + + pthread_mutex_lock(&buffer->lock); + if (buffer->users > 0) { + buffer->pending_free = FLB_TRUE; + pthread_mutex_unlock(&buffer->lock); + return -1; + } + + flb_hs_buf_cleanup_locked(buffer, raw_free); + + buffer->data = data; + buffer->raw_data = raw_data; + buffer->raw_size = raw_size; + + pthread_mutex_unlock(&buffer->lock); + + return 0; +} + +int flb_hs_buf_acquire(struct flb_hs_buf *buffer, int require_data, + int require_raw_data) +{ + if (buffer == NULL) { + return -1; + } + + pthread_mutex_lock(&buffer->lock); + + if ((require_data == FLB_TRUE && buffer->data == NULL) || + (require_raw_data == FLB_TRUE && buffer->raw_data == NULL)) { + pthread_mutex_unlock(&buffer->lock); + return -1; + } + + buffer->users++; + pthread_mutex_unlock(&buffer->lock); + + return 0; +} + void flb_hs_buf_release(struct flb_hs_buf *buffer, void (*raw_free)(void *)) { - if (buffer == NULL || buffer->users <= 0) { + if (buffer == NULL) { + return; + } + + pthread_mutex_lock(&buffer->lock); + + if (buffer->users <= 0) { + pthread_mutex_unlock(&buffer->lock); return; } buffer->users--; if (buffer->users == 0 && buffer->pending_free == FLB_TRUE) { - flb_hs_buf_cleanup(buffer, raw_free); + flb_hs_buf_cleanup_locked(buffer, raw_free); } + + pthread_mutex_unlock(&buffer->lock); } int flb_hs_register_endpoint(struct flb_hs *hs, @@ -182,6 +262,7 @@ int flb_hs_register_endpoint(struct flb_hs *hs, int flb_hs_push_health_metrics(struct flb_hs *hs, void *data, size_t size) { struct flb_hs_hc_buf *buf; + struct flb_hs_hc_buf *old_buf; int error_count; int retry_failure_count; @@ -191,19 +272,6 @@ int flb_hs_push_health_metrics(struct flb_hs *hs, void *data, size_t size) read_metrics(data, size, &error_count, &retry_failure_count); - hs->health_counter.period_counter++; - - while (hs->health_counter.period_counter > hs->health_counter.period_limit && - mk_list_size(&hs->health_metrics) > 0) { - buf = mk_list_entry_first(&hs->health_metrics, struct flb_hs_hc_buf, _head); - if (buf->users > 0) { - break; - } - mk_list_del(&buf->_head); - flb_free(buf); - hs->health_counter.period_counter--; - } - buf = flb_calloc(1, sizeof(struct flb_hs_hc_buf)); if (buf == NULL) { flb_errno(); @@ -213,17 +281,36 @@ int flb_hs_push_health_metrics(struct flb_hs *hs, void *data, size_t size) buf->error_count = error_count; buf->retry_failure_count = retry_failure_count; + pthread_mutex_lock(&hs->health_metrics_lock); + + hs->health_counter.period_counter++; + + while (hs->health_counter.period_counter > hs->health_counter.period_limit && + mk_list_size(&hs->health_metrics) > 0) { + old_buf = mk_list_entry_first(&hs->health_metrics, + struct flb_hs_hc_buf, _head); + if (old_buf->users > 0) { + break; + } + mk_list_del(&old_buf->_head); + flb_free(old_buf); + hs->health_counter.period_counter--; + } + hs->health_counter.error_counter = error_count; hs->health_counter.retry_failure_counter = retry_failure_count; mk_list_add(&buf->_head, &hs->health_metrics); + pthread_mutex_unlock(&hs->health_metrics_lock); + return 0; } /* Ingest pipeline metrics into the web service context */ int flb_hs_push_pipeline_metrics(struct flb_hs *hs, void *data, size_t size) { + int ret; flb_sds_t json_buffer; void *raw_buffer; @@ -245,17 +332,13 @@ int flb_hs_push_pipeline_metrics(struct flb_hs *hs, void *data, size_t size) memcpy(raw_buffer, data, size); - flb_hs_buf_cleanup(&hs->metrics, NULL); - if (hs->metrics.pending_free == FLB_TRUE) { + ret = flb_hs_buf_set(&hs->metrics, json_buffer, raw_buffer, size, NULL); + if (ret != 0) { flb_sds_destroy(json_buffer); flb_free(raw_buffer); return -1; } - hs->metrics.data = json_buffer; - hs->metrics.raw_data = raw_buffer; - hs->metrics.raw_size = size; - return 0; } @@ -275,20 +358,20 @@ int flb_hs_push_metrics(struct flb_hs *hs, void *data, size_t size) return -1; } - flb_hs_buf_cleanup(&hs->metrics_v2, flb_hs_destroy_cmt_buffer); - if (hs->metrics_v2.pending_free == FLB_TRUE) { + ret = flb_hs_buf_set(&hs->metrics_v2, NULL, cmt, 0, + flb_hs_destroy_cmt_buffer); + if (ret != 0) { flb_hs_destroy_cmt_buffer(cmt); return -1; } - hs->metrics_v2.raw_data = cmt; - return 0; } /* Ingest storage metrics into the web service context */ int flb_hs_push_storage_metrics(struct flb_hs *hs, void *data, size_t size) { + int ret; flb_sds_t json_buffer; void *raw_buffer; @@ -310,20 +393,70 @@ int flb_hs_push_storage_metrics(struct flb_hs *hs, void *data, size_t size) memcpy(raw_buffer, data, size); - flb_hs_buf_cleanup(&hs->storage_metrics, NULL); - if (hs->storage_metrics.pending_free == FLB_TRUE) { + ret = flb_hs_buf_set(&hs->storage_metrics, json_buffer, raw_buffer, size, + NULL); + if (ret != 0) { flb_sds_destroy(json_buffer); flb_free(raw_buffer); return -1; } - hs->storage_metrics.data = json_buffer; - hs->storage_metrics.raw_data = raw_buffer; - hs->storage_metrics.raw_size = size; + return 0; +} + +static int flb_hs_state_init(struct flb_hs *hs) +{ + int ret; + + ret = flb_hs_buf_init(&hs->metrics); + if (ret != 0) { + return -1; + } + + ret = flb_hs_buf_init(&hs->metrics_v2); + if (ret != 0) { + pthread_mutex_destroy(&hs->metrics.lock); + return -1; + } + + ret = flb_hs_buf_init(&hs->storage_metrics); + if (ret != 0) { + pthread_mutex_destroy(&hs->metrics_v2.lock); + pthread_mutex_destroy(&hs->metrics.lock); + return -1; + } + + ret = pthread_mutex_init(&hs->health_metrics_lock, NULL); + if (ret != 0) { + pthread_mutex_destroy(&hs->storage_metrics.lock); + pthread_mutex_destroy(&hs->metrics_v2.lock); + pthread_mutex_destroy(&hs->metrics.lock); + return -1; + } return 0; } +static void flb_hs_state_destroy(struct flb_hs *hs) +{ + struct mk_list *head; + struct mk_list *tmp; + struct flb_hs_hc_buf *health_buffer; + + flb_hs_buf_destroy(&hs->metrics, NULL); + flb_hs_buf_destroy(&hs->metrics_v2, flb_hs_destroy_cmt_buffer); + flb_hs_buf_destroy(&hs->storage_metrics, NULL); + + pthread_mutex_lock(&hs->health_metrics_lock); + mk_list_foreach_safe(head, tmp, &hs->health_metrics) { + health_buffer = mk_list_entry(head, struct flb_hs_hc_buf, _head); + mk_list_del(&health_buffer->_head); + flb_free(health_buffer); + } + pthread_mutex_unlock(&hs->health_metrics_lock); + pthread_mutex_destroy(&hs->health_metrics_lock); +} + /* Create ROOT endpoints */ struct flb_hs *flb_hs_create(const char *listen, const char *tcp_port, struct flb_config *config) @@ -343,6 +476,11 @@ struct flb_hs *flb_hs_create(const char *listen, const char *tcp_port, hs->config = config; mk_list_init(&hs->routes); mk_list_init(&hs->health_metrics); + ret = flb_hs_state_init(hs); + if (ret != 0) { + flb_free(hs); + return NULL; + } hs->health_counter.error_limit = config->hc_errors_count; hs->health_counter.retry_failure_limit = config->hc_retry_failure_count; @@ -363,6 +501,7 @@ struct flb_hs *flb_hs_create(const char *listen, const char *tcp_port, port <= 0 || port > 65535) { flb_error("[http_server] invalid monitoring tcp_port '%s'", tcp_port); flb_hs_endpoints_free(hs); + flb_hs_state_destroy(hs); flb_free(hs); return NULL; } @@ -370,13 +509,14 @@ struct flb_hs *flb_hs_create(const char *listen, const char *tcp_port, options.networking_flags = 0; flb_net_setup_init(&hs->net_setup); options.networking_setup = &hs->net_setup; - options.event_loop = config->evl; + options.event_loop = NULL; options.system_context = config; - options.use_caller_event_loop = FLB_TRUE; + options.use_caller_event_loop = FLB_FALSE; ret = flb_http_server_init_with_options(&hs->server, &options); if (ret != 0) { flb_hs_endpoints_free(hs); + flb_hs_state_destroy(hs); flb_free(hs); return NULL; } @@ -425,7 +565,6 @@ int flb_hs_destroy(struct flb_hs *hs) struct mk_list *head; struct mk_list *tmp; struct flb_hs_route *route; - struct flb_hs_hc_buf *health_buffer; if (!hs) { return 0; @@ -434,15 +573,7 @@ int flb_hs_destroy(struct flb_hs *hs) flb_hs_health_destroy(); flb_http_server_destroy(&hs->server); - flb_hs_buf_cleanup(&hs->metrics, NULL); - flb_hs_buf_cleanup(&hs->metrics_v2, flb_hs_destroy_cmt_buffer); - flb_hs_buf_cleanup(&hs->storage_metrics, NULL); - - mk_list_foreach_safe(head, tmp, &hs->health_metrics) { - health_buffer = mk_list_entry(head, struct flb_hs_hc_buf, _head); - mk_list_del(&health_buffer->_head); - flb_free(health_buffer); - } + flb_hs_state_destroy(hs); mk_list_foreach_safe(head, tmp, &hs->routes) { route = mk_list_entry(head, struct flb_hs_route, _head); diff --git a/src/http_server/flb_http_server_http1.c b/src/http_server/flb_http_server_http1.c index a84a9683628..df17758e1f5 100644 --- a/src/http_server/flb_http_server_http1.c +++ b/src/http_server/flb_http_server_http1.c @@ -419,6 +419,7 @@ int flb_http1_response_commit(struct flb_http_response *response) return -9; } + response_buffer = sds_result; if (response->body != NULL) { sds_result = cfl_sds_cat(response_buffer, diff --git a/tests/integration/scenarios/internal_http_server/config/internal_http_server_exec_deadlock.yaml b/tests/integration/scenarios/internal_http_server/config/internal_http_server_exec_deadlock.yaml new file mode 100644 index 00000000000..08037f82ae3 --- /dev/null +++ b/tests/integration/scenarios/internal_http_server/config/internal_http_server_exec_deadlock.yaml @@ -0,0 +1,21 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_listen: 127.0.0.1 + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + health_check: on + storage.metrics: on + +pipeline: + inputs: + - name: exec + tag: test + command: curl -s http://127.0.0.1:${FLUENT_BIT_HTTP_MONITORING_PORT}/api/v1/metrics/prometheus + interval_sec: 1 + buf_size: 128k + + outputs: + - name: null + match: "*" diff --git a/tests/integration/scenarios/internal_http_server/tests/test_internal_http_server_exec_deadlock_001.py b/tests/integration/scenarios/internal_http_server/tests/test_internal_http_server_exec_deadlock_001.py new file mode 100644 index 00000000000..6ded80ed3e4 --- /dev/null +++ b/tests/integration/scenarios/internal_http_server/tests/test_internal_http_server_exec_deadlock_001.py @@ -0,0 +1,90 @@ +import concurrent.futures +import os +import time + +from utils.http_matrix import run_curl_request +from utils.test_service import FluentBitTestService + + +class Service: + """Run Fluent Bit with an exec input that curls the monitoring server.""" + + def __init__(self): + config_dir = os.path.dirname(__file__) + self.config_file = os.path.abspath( + os.path.join(config_dir, "../config/internal_http_server_exec_deadlock.yaml") + ) + self.service = FluentBitTestService(self.config_file) + + def start(self): + """Start Fluent Bit and record the monitoring base URL.""" + try: + self.service.start() + except Exception: + self.service.stop() + raise + + self.flb = self.service.flb + self.base_url = f"http://127.0.0.1:{self.flb.http_monitoring_port}" + + def stop(self): + """Stop Fluent Bit and restore the test environment.""" + self.service.stop() + + def request(self, path, *, method="GET", http_mode="http1.1"): + """Issue a request against the internal monitoring server.""" + return run_curl_request( + f"{self.base_url}{path}", + method=method, + http_mode=http_mode, + ) + + +def test_http_server_stays_responsive_after_exec_self_request(): + """The monitoring server must not share the blocked exec collector loop.""" + service = Service() + service.start() + + try: + result = service.request("/api/v1/uptime") + assert result["status_code"] == 200 + assert "uptime_sec" in result["body"] + + time.sleep(2) + + endpoints = [ + ("/api/v1/uptime", "uptime_sec"), + ("/api/v1/health", "ok"), + ("/api/v1/metrics", "output"), + ("/api/v1/metrics/prometheus", "fluentbit_uptime"), + ("/api/v1/storage", "chunks"), + ("/api/v2/metrics", "fluentbit_uptime"), + ("/api/v2/metrics/prometheus", "fluentbit_uptime"), + ] + + for path, pattern in endpoints: + result = service.service.wait_for_condition( + lambda: ( + response + if response["status_code"] == 200 and pattern in response["body"] + else None + ) if (response := service.request(path)) else None, + timeout=10, + interval=0.5, + description=f"internal endpoint {path}", + ) + assert result["status_code"] == 200 + assert pattern in result["body"] + + def fetch(endpoint): + path, pattern = endpoint + response = service.request(path) + assert response["status_code"] == 200 + assert pattern in response["body"] + + with concurrent.futures.ThreadPoolExecutor(max_workers=7) as executor: + futures = [executor.submit(fetch, endpoint) for endpoint in endpoints * 4] + for future in futures: + future.result() + finally: + service.stop()