diff --git a/AGENTS.md b/AGENTS.md index 088f57ecdaf..da99339563d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -71,6 +71,10 @@ Keep changes scoped: plugin logic in its plugin directory, shared behavior in `s - Run broader test coverage when changing shared lifecycle, routing, storage, or accounting code. - Validate both success and failure paths (invalid payloads, boundary sizes, null/missing fields). - You can also run specific binaries from `build/bin` (e.g., `./bin/flb-it-opentelemetry`). +- When changing code covered by `tests/integration`, agents must verify the + affected scenarios are valgrind-clean. Run the focused integration tests with + `tests/integration/run_tests.py --valgrind --valgrind-strict ...` and do not + stop at functional pass/fail if memory errors or leaks remain. - Keep generated integration artifacts out of git. Do not commit `.venv/`, `.pytest_cache/`, `results/`, or `__pycache__/` under `tests/integration`. diff --git a/include/fluent-bit/flb_input.h b/include/fluent-bit/flb_input.h index 552aed5d059..c2ca41bd75e 100644 --- a/include/fluent-bit/flb_input.h +++ b/include/fluent-bit/flb_input.h @@ -436,6 +436,9 @@ struct flb_input_instance { struct cmt_counter *cmt_ring_buffer_writes; struct cmt_counter *cmt_ring_buffer_retries; struct cmt_counter *cmt_ring_buffer_retry_failures; + struct cmt_counter *cmt_ingress_queue_busy; + struct cmt_gauge *cmt_ingress_queue_pending_events; + struct cmt_gauge *cmt_ingress_queue_pending_bytes; /* * Indexes for generated chunks: simple hash tables that keeps the latest diff --git a/include/fluent-bit/http_server/flb_http_server.h b/include/fluent-bit/http_server/flb_http_server.h index 435d2b77661..83bb10fb90b 100755 --- a/include/fluent-bit/http_server/flb_http_server.h +++ b/include/fluent-bit/http_server/flb_http_server.h @@ -65,12 +65,17 @@ typedef int (*flb_http_server_worker_callback)(struct flb_http_server *server, struct flb_input_instance; +#define FLB_HTTP_SERVER_INGRESS_QUEUE_EVENT_LIMIT 8192 +#define FLB_HTTP_SERVER_INGRESS_QUEUE_BYTE_LIMIT (256 * 1024 * 1024) + struct flb_http_server_config { int http2; size_t buffer_max_size; size_t buffer_chunk_size; size_t max_connections; int workers; + size_t ingress_queue_event_limit; + size_t ingress_queue_byte_limit; }; struct flb_http_server_options { diff --git a/src/flb_input.c b/src/flb_input.c index 54bc7457cf9..e990eee238c 100644 --- a/src/flb_input.c +++ b/src/flb_input.c @@ -466,8 +466,8 @@ struct flb_input_instance *flb_input_new(struct flb_config *config, instance->ingress_queue_signal_pending = FLB_FALSE; instance->ingress_queue_pending_events = 0; instance->ingress_queue_pending_bytes = 0; - instance->ingress_queue_event_limit = 8192; - instance->ingress_queue_byte_limit = 256 * 1024 * 1024; + instance->ingress_queue_event_limit = FLB_HTTP_SERVER_INGRESS_QUEUE_EVENT_LIMIT; + instance->ingress_queue_byte_limit = FLB_HTTP_SERVER_INGRESS_QUEUE_BYTE_LIMIT; /* Plugin use networking */ if (plugin->flags & (FLB_INPUT_NET | FLB_INPUT_NET_SERVER)) { @@ -975,6 +975,8 @@ void flb_input_instance_destroy(struct flb_input_instance *ins) struct mk_list *head; struct flb_input_collector *collector; + flb_input_ingress_destroy(ins); + if (ins->alias) { flb_sds_destroy(ins->alias); } @@ -1134,8 +1136,6 @@ void flb_input_instance_destroy(struct flb_input_instance *ins) mk_event_closesocket(ins->ingress_queue_channels[1]); } - flb_input_ingress_destroy(ins); - /* Collectors */ mk_list_foreach_safe(head, tmp, &ins->collectors) { collector = mk_list_entry(head, struct flb_input_collector, _head); @@ -1552,6 +1552,35 @@ int flb_input_instance_init(struct flb_input_instance *ins, 1, (char *[]) {"name"}); cmt_counter_set(ins->cmt_ring_buffer_retry_failures, ts, 0, 1, (char *[]) {name}); + if ((p->flags & FLB_INPUT_HTTP_SERVER) != 0) { + /* fluentbit_input_http_server_ingress_queue_busy_total */ + ins->cmt_ingress_queue_busy = \ + cmt_counter_create(ins->cmt, + "fluentbit", "input", + "http_server_ingress_queue_busy_total", + "Number of deferred HTTP server ingress queue busy events.", + 1, (char *[]) {"name"}); + cmt_counter_set(ins->cmt_ingress_queue_busy, ts, 0, 1, (char *[]) {name}); + + /* fluentbit_input_http_server_ingress_queue_pending_events */ + ins->cmt_ingress_queue_pending_events = \ + cmt_gauge_create(ins->cmt, + "fluentbit", "input", + "http_server_ingress_queue_pending_events", + "Current number of deferred HTTP server ingress queue events.", + 1, (char *[]) {"name"}); + cmt_gauge_set(ins->cmt_ingress_queue_pending_events, ts, 0, 1, (char *[]) {name}); + + /* fluentbit_input_http_server_ingress_queue_pending_bytes */ + ins->cmt_ingress_queue_pending_bytes = \ + cmt_gauge_create(ins->cmt, + "fluentbit", "input", + "http_server_ingress_queue_pending_bytes", + "Current number of deferred HTTP server ingress queue bytes.", + 1, (char *[]) {"name"}); + cmt_gauge_set(ins->cmt_ingress_queue_pending_bytes, ts, 0, 1, (char *[]) {name}); + } + /* OLD Metrics */ ins->metrics = flb_metrics_create(name); if (ins->metrics) { diff --git a/src/flb_input_ingest.c b/src/flb_input_ingest.c index 377c315b5ac..621ea6cc721 100644 --- a/src/flb_input_ingest.c +++ b/src/flb_input_ingest.c @@ -31,6 +31,7 @@ #include #include #include +#include #include @@ -157,6 +158,33 @@ static void flb_input_ingress_signal(struct flb_input_instance *ins) } } +static void flb_input_ingress_update_metrics(struct flb_input_instance *ins) +{ + uint64_t ts; + char *name; + + if (ins == NULL || ins->cmt == NULL) { + return; + } + + ts = cfl_time_now(); + name = (char *) flb_input_name(ins); + + if (ins->cmt_ingress_queue_pending_events != NULL) { + cmt_gauge_set(ins->cmt_ingress_queue_pending_events, + ts, + ins->ingress_queue_pending_events, + 1, (char *[]) {name}); + } + + if (ins->cmt_ingress_queue_pending_bytes != NULL) { + cmt_gauge_set(ins->cmt_ingress_queue_pending_bytes, + ts, + ins->ingress_queue_pending_bytes, + 1, (char *[]) {name}); + } +} + static int flb_input_ingress_enqueue(struct flb_input_instance *ins, struct flb_input_ingress_event *event) { @@ -165,6 +193,8 @@ static int flb_input_ingress_enqueue(struct flb_input_instance *ins, int should_signal; int wait_result; struct timespec deadline; + struct cmt_counter *cmt_ingress_queue_busy; + flb_sds_t input_name; if (ins == NULL || event == NULL || ins->ingress_queue_enabled != FLB_TRUE) { flb_input_ingress_event_destroy(event); @@ -210,8 +240,23 @@ static int flb_input_ingress_enqueue(struct flb_input_instance *ins, &ins->ingress_queue_lock, &deadline); if (wait_result == ETIMEDOUT) { + cmt_ingress_queue_busy = ins->cmt_ingress_queue_busy; + input_name = NULL; + + if (cmt_ingress_queue_busy != NULL) { + input_name = flb_sds_create(flb_input_name(ins)); + } + pthread_mutex_unlock(&ins->ingress_queue_lock); + if (cmt_ingress_queue_busy != NULL && input_name != NULL) { + cmt_counter_add(cmt_ingress_queue_busy, + cfl_time_now(), + 1, + 1, (char *[]) {input_name}); + flb_sds_destroy(input_name); + } + flb_input_ingress_event_destroy(event); return FLB_INPUT_INGRESS_BUSY; @@ -229,6 +274,7 @@ static int flb_input_ingress_enqueue(struct flb_input_instance *ins, mk_list_add(&event->_head, &ins->ingress_queue); ins->ingress_queue_pending_events++; ins->ingress_queue_pending_bytes += event_size; + flb_input_ingress_update_metrics(ins); if (ins->ingress_queue_signal_pending == FLB_FALSE) { ins->ingress_queue_signal_pending = FLB_TRUE; @@ -331,6 +377,7 @@ static int flb_input_ingress_collector(struct flb_input_instance *ins, ins->ingress_queue_pending_events = 0; ins->ingress_queue_pending_bytes = 0; ins->ingress_queue_signal_pending = FLB_FALSE; + flb_input_ingress_update_metrics(ins); if (queue_was_full == FLB_TRUE) { pthread_cond_broadcast(&ins->ingress_queue_space_available); @@ -409,8 +456,22 @@ int flb_input_ingress_enable(struct flb_input_instance *ins) return 0; } + if (ins->http_server_config != NULL) { + ins->ingress_queue_event_limit = + ins->http_server_config->ingress_queue_event_limit; + + if (ins->http_server_config->ingress_queue_byte_limit > 0) { + ins->ingress_queue_byte_limit = + ins->http_server_config->ingress_queue_byte_limit; + } + else { + ins->ingress_queue_byte_limit = ins->mem_buf_limit; + } + } + if (ins->mem_buf_limit > 0 && - ins->mem_buf_limit < ins->ingress_queue_byte_limit) { + (ins->ingress_queue_byte_limit == 0 || + ins->mem_buf_limit < ins->ingress_queue_byte_limit)) { ins->ingress_queue_byte_limit = ins->mem_buf_limit; } @@ -459,6 +520,7 @@ void flb_input_ingress_destroy(struct flb_input_instance *ins) } ins->ingress_queue_pending_events = 0; ins->ingress_queue_pending_bytes = 0; + flb_input_ingress_update_metrics(ins); pthread_mutex_unlock(&ins->ingress_queue_lock); } diff --git a/src/http_server/flb_http_server.c b/src/http_server/flb_http_server.c index b238767cff6..96bf24d149f 100644 --- a/src/http_server/flb_http_server.c +++ b/src/http_server/flb_http_server.c @@ -749,6 +749,8 @@ void flb_http_server_config_init(struct flb_http_server_config *config) config->buffer_chunk_size = HTTP_SERVER_INITIAL_BUFFER_SIZE; config->max_connections = 0; config->workers = 1; + config->ingress_queue_event_limit = FLB_HTTP_SERVER_INGRESS_QUEUE_EVENT_LIMIT; + config->ingress_queue_byte_limit = FLB_HTTP_SERVER_INGRESS_QUEUE_BYTE_LIMIT; } int flb_http_server_options_init_from_input(struct flb_http_server_options *options, diff --git a/src/http_server/flb_http_server_config_map.c b/src/http_server/flb_http_server_config_map.c index b251069a01c..8f1ed6b2908 100644 --- a/src/http_server/flb_http_server_config_map.c +++ b/src/http_server/flb_http_server_config_map.c @@ -47,6 +47,16 @@ struct flb_config_map flb_http_server_config_map[] = { 0, FLB_TRUE, offsetof(struct flb_http_server_config, workers), "Set the number of HTTP listener workers" }, + { + FLB_CONFIG_MAP_SIZE, "http_server.ingress_queue_event_limit", "8192", + 0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_event_limit), + "Set the maximum number of deferred ingress queue events. Applies only when http_server.workers > 1." + }, + { + FLB_CONFIG_MAP_SIZE, "http_server.ingress_queue_byte_limit", "256M", + 0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_byte_limit), + "Set the maximum number of deferred ingress queue bytes. Applies only when http_server.workers > 1." + }, { FLB_CONFIG_MAP_BOOL, "http2", "true", 0, FLB_TRUE, offsetof(struct flb_http_server_config, http2), diff --git a/tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yaml b/tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yaml new file mode 100644 index 00000000000..9547191d774 --- /dev/null +++ b/tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yaml @@ -0,0 +1,24 @@ +service: + flush: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + http2: off + tls: off + http_server.workers: 4 + http_server.ingress_queue_event_limit: 0 + http_server.ingress_queue_byte_limit: 1 + + outputs: + - name: stdout + match: '*' + + - name: opentelemetry + match: '*' + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} diff --git a/tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue_byte_limit.yaml b/tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue_byte_limit.yaml new file mode 100644 index 00000000000..75575fe1082 --- /dev/null +++ b/tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue_byte_limit.yaml @@ -0,0 +1,24 @@ +service: + flush: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: opentelemetry + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + http2: off + tls: off + http_server.workers: 4 + http_server.ingress_queue_event_limit: 1024 + http_server.ingress_queue_byte_limit: 2000 + + outputs: + - name: stdout + match: '*' + + - name: opentelemetry + match: '*' + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py index db2cc4357e4..957d59e9105 100644 --- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py +++ b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py @@ -62,6 +62,10 @@ "http2_tls": "otlp_http2_tls_workers.yaml", } +IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_CONFIG = "otlp_http1_cleartext_workers_tiny_ingress_queue.yaml" +IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_BYTE_LIMIT_CONFIG = \ + "otlp_http1_cleartext_workers_tiny_ingress_queue_byte_limit.yaml" + IN_OPENTELEMETRY_OAUTH2_PROTOCOL_CONFIGS = { "http1_cleartext": "otlp_http1_cleartext_oauth2.yaml", "http2_cleartext": "otlp_http2_cleartext_oauth2.yaml", @@ -225,6 +229,28 @@ def read_stdout_otlp_json_text(service, root_key, timeout=10, interval=0.25): raise TimeoutError(f"Timed out waiting for stdout OTLP JSON payload with root key {root_key}") + +def read_prometheus_metric_value(metrics_text, metric_name, input_name): + label = f'name="{input_name}"' + + for line in metrics_text.splitlines(): + if not line.startswith(metric_name): + continue + + if "{" in line and label not in line: + continue + + return float(line.rsplit(" ", 1)[-1]) + + raise AssertionError(f"Metric {metric_name} with {label} not found") + + +def maybe_read_prometheus_metric_value(metrics_text, metric_name, input_name): + try: + return read_prometheus_metric_value(metrics_text, metric_name, input_name) + except AssertionError: + return None + class Service: def __init__(self, config_file, *, use_auth_server=False): # Compose the absolute path for the Fluent Bit configuration file @@ -400,6 +426,22 @@ def wait_for_signal_count(self, signal_type, minimum_count, timeout=10, interval description=f"{minimum_count} OTLP {signal_type} payloads", ) + def scrape_prometheus_metrics(self): + url = f"http://127.0.0.1:{self.flb.http_monitoring_port}/api/v2/metrics/prometheus" + response = requests.get(url, timeout=5) + response.raise_for_status() + return response.text + + def wait_for_prometheus_metric(self, metric_name, timeout=10, interval=0.25): + return self.service.wait_for_condition( + lambda: ( + metrics if metric_name in metrics else None + ) if (metrics := self.scrape_prometheus_metrics()) else None, + timeout=timeout, + interval=interval, + description=f"prometheus metric {metric_name}", + ) + def stop(self): self.service.stop() @@ -925,3 +967,166 @@ def send_job(job): assert "/v1/logs" in paths_seen assert "/v1/metrics" in paths_seen assert "/v1/traces" in paths_seen + + +def test_in_opentelemetry_http_workers_export_ingress_queue_metrics(): + service = Service(IN_OPENTELEMETRY_WORKER_PROTOCOL_CONFIGS["http1_cleartext"]) + service.start() + service.wait_for_log_message("with 4 workers", timeout=10) + + response = service.send_raw_request( + "/v1/logs", + service.build_otel_payload("test_logs_001.in.json", "logs"), + ) + assert response.status_code == 201 + + service.wait_for_signal_count("logs", 1, timeout=10) + + metrics_text = service.service.wait_for_condition( + lambda: ( + metrics + if "fluentbit_input_http_server_ingress_queue_pending_events" in metrics + else None + ) if (metrics := service.scrape_prometheus_metrics()) else None, + timeout=10, + interval=0.25, + description="prometheus ingress queue metrics", + ) + + service.stop() + + assert "fluentbit_input_http_server_ingress_queue_pending_events" in metrics_text + assert "fluentbit_input_http_server_ingress_queue_pending_bytes" in metrics_text + assert "fluentbit_input_http_server_ingress_queue_busy_total" in metrics_text + + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_pending_events", + "opentelemetry.0", + ) == 0 + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_pending_bytes", + "opentelemetry.0", + ) == 0 + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_busy_total", + "opentelemetry.0", + ) == 0 + + +def test_in_opentelemetry_http_workers_respect_tiny_ingress_queue_limits(): + service = Service(IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_CONFIG) + service.start() + service.wait_for_log_message("with 4 workers", timeout=10) + logs_before = len(data_storage["logs"]) + + response = service.send_raw_request( + "/v1/logs", + service.build_otel_payload("test_logs_001.in.json", "logs"), + ) + + metrics_text = service.service.wait_for_condition( + lambda: ( + metrics if maybe_read_prometheus_metric_value( + metrics, + "fluentbit_input_http_server_ingress_queue_busy_total", + "opentelemetry.0", + ) is not None and maybe_read_prometheus_metric_value( + metrics, + "fluentbit_input_http_server_ingress_queue_busy_total", + "opentelemetry.0", + ) >= 1 else None + ) if (metrics := service.scrape_prometheus_metrics()) else None, + timeout=10, + interval=0.25, + description="ingress queue busy metric", + ) + + service.stop() + + assert response.status_code == 503 + assert "deferred ingress queue is full" in response.text + assert len(data_storage["logs"]) == logs_before + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_busy_total", + "opentelemetry.0", + ) >= 1 + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_pending_events", + "opentelemetry.0", + ) == 0 + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_pending_bytes", + "opentelemetry.0", + ) == 0 + + +def test_in_opentelemetry_http_workers_respect_ingress_queue_byte_limit(): + service = Service(IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_BYTE_LIMIT_CONFIG) + small_payload = service.build_otel_payload("test_logs_001.in.json", "logs") + + small_request = ExportLogsServiceRequest() + small_request.ParseFromString(small_payload) + + large_request = ExportLogsServiceRequest() + large_request.resource_logs.extend(small_request.resource_logs) + large_request.resource_logs.extend(small_request.resource_logs) + + large_payload = large_request.SerializeToString() + + assert len(small_payload) < len(large_payload) + + service.start() + service.wait_for_log_message("with 4 workers", timeout=10) + + logs_before = len(data_storage["logs"]) + + success_response = service.send_raw_request("/v1/logs", small_payload) + assert success_response.status_code == 201 + + service.wait_for_signal_count("logs", logs_before + 1, timeout=10) + + failure_response = service.send_raw_request("/v1/logs", large_payload) + assert failure_response.status_code == 503 + assert "deferred ingress queue is full" in failure_response.text + + metrics_text = service.service.wait_for_condition( + lambda: ( + metrics if maybe_read_prometheus_metric_value( + metrics, + "fluentbit_input_http_server_ingress_queue_busy_total", + "opentelemetry.0", + ) is not None and maybe_read_prometheus_metric_value( + metrics, + "fluentbit_input_http_server_ingress_queue_busy_total", + "opentelemetry.0", + ) >= 1 else None + ) if (metrics := service.scrape_prometheus_metrics()) else None, + timeout=10, + interval=0.25, + description="ingress queue byte-limit busy metric", + ) + + service.stop() + + assert len(data_storage["logs"]) == logs_before + 1 + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_busy_total", + "opentelemetry.0", + ) >= 1 + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_pending_events", + "opentelemetry.0", + ) == 0 + assert read_prometheus_metric_value( + metrics_text, + "fluentbit_input_http_server_ingress_queue_pending_bytes", + "opentelemetry.0", + ) == 0