tests: integration: cover downstream listener workers

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

Author: edsiper

tests/integration/scenarios/in_forward/config/in_forward_tls_workers.yaml

@@ -0,0 +1,24 @@
+service:
+  flush: 1
+  grace: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: forward
+      listen: 127.0.0.1
+      port: ${FLUENT_BIT_TEST_LISTENER_PORT}
+      tls: on
+      tls.crt_file: ${CERTIFICATE_TEST}
+      tls.key_file: ${PRIVATE_KEY_TEST}
+      workers: 4
+
+  outputs:
+    - name: http
+      match: test
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      uri: /data
+      format: json

tests/integration/scenarios/in_forward/config/in_forward_workers.yaml

@@ -0,0 +1,21 @@
+service:
+  flush: 1
+  grace: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: forward
+      listen: 127.0.0.1
+      port: ${FLUENT_BIT_TEST_LISTENER_PORT}
+      workers: 4
+
+  outputs:
+    - name: http
+      match: test
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      uri: /data
+      format: json

tests/integration/scenarios/in_forward/tests/test_in_forward_001.py

@@ -10,6 +10,7 @@
 import tempfile
 import time
 import uuid
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 import opentelemetry
@@ -104,6 +105,7 @@ def _stop_receiver(self, service):
 
     def start(self):
         self.service.start()
+        self.flb = self.service.flb
         self.flb_listener_port = self.service.flb_listener_port
 
     def stop(self):
@@ -136,6 +138,9 @@ def wait_for_log_contains(self, text, timeout=10):
             description=f"log text {text!r}",
         )
 
+    def wait_for_log_message(self, pattern, timeout=10):
+        return self.wait_for_log_contains(pattern, timeout)
+
 
 class StorageLimitService(Service):
     def __init__(self, config_file):
@@ -842,6 +847,62 @@ def test_in_forward_forward_mode_multiple_entries():
     assert [record["message"] for record in records[:2]] == ["entry-1", "entry-2"]
 
 
+def test_in_forward_workers_concurrent_message_mode_records():
+    total_records = 16
+    service = Service("in_forward_workers.yaml")
+    service.start()
+
+    try:
+        service.wait_for_log_message("with 4 workers", timeout=10)
+        payloads = [
+            _message_mode_payload(TEST_TAG, {"message": f"worker-{i}", "value": i})
+            for i in range(total_records)
+        ]
+        with ThreadPoolExecutor(max_workers=total_records) as executor:
+            list(executor.map(
+                lambda payload: _send_tcp_payload(service.flb_listener_port, payload),
+                payloads,
+            ))
+        records = service.wait_for_record_count(total_records, timeout=20)
+    finally:
+        service.stop()
+
+    values = sorted(record["value"] for record in records)
+    assert values == list(range(total_records))
+
+
+def test_in_forward_workers_drop_partial_connections_and_continue():
+    dropped_connections = 8
+    valid_records = 8
+    service = Service("in_forward_workers.yaml")
+    service.start()
+
+    try:
+        service.wait_for_log_message("with 4 workers", timeout=10)
+        # Send partial MessagePack bytes and close to exercise drop cleanup.
+        with ThreadPoolExecutor(max_workers=dropped_connections) as executor:
+            list(executor.map(
+                lambda _: _send_tcp_payload(service.flb_listener_port, b"\x93\xa4test"),
+                range(dropped_connections),
+            ))
+
+        payloads = [
+            _message_mode_payload(TEST_TAG, {"message": f"valid-{i}", "value": i})
+            for i in range(valid_records)
+        ]
+        with ThreadPoolExecutor(max_workers=valid_records) as executor:
+            list(executor.map(
+                lambda payload: _send_tcp_payload(service.flb_listener_port, payload),
+                payloads,
+            ))
+        records = service.wait_for_record_count(valid_records, timeout=20)
+    finally:
+        service.stop()
+
+    values = sorted(record["value"] for record in records)
+    assert values == list(range(valid_records))
+
+
 def test_in_forward_packed_forward_gzip():
     service = Service("in_forward.yaml")
     service.start()
@@ -1079,6 +1140,65 @@ def test_in_forward_tls_idle_connection_timeout_churn(monkeypatch):
     assert any(record.get("message") == "after-churn" for record in records)
 
 
+def test_in_forward_tls_workers_concurrent_message_mode_records():
+    total_records = 16
+    service = Service("in_forward_tls_workers.yaml")
+    service.start()
+
+    try:
+        service.wait_for_log_message("with 4 workers", timeout=10)
+        with ThreadPoolExecutor(max_workers=total_records) as executor:
+            list(executor.map(
+                lambda i: _send_tls_payload(service.flb_listener_port,
+                                            _message_mode_payload(
+                                                TEST_TAG,
+                                                {"message": f"tls-worker-{i}",
+                                                 "value": i}),
+                                            service.tls_crt_file),
+                range(total_records),
+            ))
+        records = service.wait_for_record_count(total_records, timeout=20)
+    finally:
+        service.stop()
+
+    values = sorted(record["value"] for record in records)
+    assert values == list(range(total_records))
+
+
+def test_in_forward_tls_workers_drop_bad_handshakes_and_continue():
+    dropped_connections = 8
+    valid_records = 8
+    service = Service("in_forward_tls_workers.yaml")
+    service.start()
+
+    try:
+        service.wait_for_log_message("with 4 workers", timeout=10)
+        # Send raw non-TLS bytes and close to exercise handshake cleanup.
+        with ThreadPoolExecutor(max_workers=dropped_connections) as executor:
+            list(executor.map(
+                lambda i: _send_tcp_payload(service.flb_listener_port,
+                                            f"not-tls-{i}".encode("utf-8")),
+                range(dropped_connections),
+            ))
+
+        with ThreadPoolExecutor(max_workers=valid_records) as executor:
+            list(executor.map(
+                lambda i: _send_tls_payload(service.flb_listener_port,
+                                            _message_mode_payload(
+                                                TEST_TAG,
+                                                {"message": f"valid-tls-{i}",
+                                                 "value": i}),
+                                            service.tls_crt_file),
+                range(valid_records),
+            ))
+        records = service.wait_for_record_count(valid_records, timeout=20)
+    finally:
+        service.stop()
+
+    values = sorted(record["value"] for record in records)
+    assert values == list(range(valid_records))
+
+
 def test_in_forward_secure_forward_auth_success():
     service = Service("in_forward_secure.yaml")
     service.start()

tests/integration/scenarios/in_http_max_connections/config/in_http_max_connections.yaml

@@ -12,6 +12,7 @@ pipeline:
       port: ${FLUENT_BIT_TEST_LISTENER_PORT}
       http_server.idle_timeout: 2s
       http_server.max_connections: 1
+      http_server.workers: 2
 
   outputs:
     - name: http

tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py

@@ -1565,3 +1565,37 @@ def test_in_opentelemetry_http_workers_respect_ingress_queue_byte_limit():
         "fluentbit_input_http_server_ingress_queue_pending_bytes",
         "opentelemetry.0",
     ) == 0
+
+
+def test_in_opentelemetry_http_workers_account_metrics_payload_bytes():
+    service = Service(IN_OPENTELEMETRY_TINY_INGRESS_QUEUE_BYTE_LIMIT_CONFIG)
+    small_payload = service.build_otel_payload("test_metrics_001.in.json", "metrics")
+
+    small_request = ExportMetricsServiceRequest()
+    small_request.ParseFromString(small_payload)
+
+    large_request = ExportMetricsServiceRequest()
+    for _ in range(16):
+        large_request.resource_metrics.extend(small_request.resource_metrics)
+
+    large_payload = large_request.SerializeToString()
+
+    assert len(small_payload) < 2000
+    assert len(large_payload) > 2000
+
+    service.start()
+    service.wait_for_log_message("with 4 workers", timeout=10)
+    metrics_before = len(data_storage["metrics"])
+
+    try:
+        success_response = service.send_raw_request("/v1/metrics", small_payload)
+        assert success_response.status_code == 201
+        service.wait_for_signal_count("metrics", metrics_before + 1, timeout=10)
+
+        failure_response = service.send_raw_request("/v1/metrics", large_payload)
+        assert failure_response.status_code == 503
+        assert "deferred ingress queue is full" in failure_response.text
+    finally:
+        service.stop()
+
+    assert len(data_storage["metrics"]) == metrics_before + 1

tests/integration/scenarios/in_tcp/config/in_tcp_parser_json_tls_workers.yaml

@@ -0,0 +1,29 @@
+service:
+  flush: 1
+  grace: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+  parsers_file: ${PARSERS_FILE_TEST}
+
+pipeline:
+  inputs:
+    - name: tcp
+      tag: target_input
+      listen: 127.0.0.1
+      port: ${FLUENT_BIT_TEST_LISTENER_PORT}
+      format: none
+      separator: "\\n"
+      parser: json
+      tls: on
+      tls.crt_file: ${CERTIFICATE_TEST}
+      tls.key_file: ${PRIVATE_KEY_TEST}
+      workers: 4
+
+  outputs:
+    - name: http
+      match: target_input
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      uri: /data
+      format: json

tests/integration/scenarios/in_tcp/config/in_tcp_parser_json_workers.yaml

@@ -0,0 +1,26 @@
+service:
+  flush: 1
+  grace: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+  parsers_file: ${PARSERS_FILE_TEST}
+
+pipeline:
+  inputs:
+    - name: tcp
+      tag: target_input
+      listen: 127.0.0.1
+      port: ${FLUENT_BIT_TEST_LISTENER_PORT}
+      format: none
+      separator: "\\n"
+      parser: json
+      workers: 4
+
+  outputs:
+    - name: http
+      match: target_input
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      uri: /data
+      format: json