upstream: network: mitigate tcp dns timeouts

include/fluent-bit/flb_network.h

@@ -137,6 +137,8 @@ void flb_net_setup_init(struct flb_net_setup *net);
 int flb_net_host_set(const char *plugin_name, struct flb_net_host *host, const char *address);
 
 /* DNS handling */
+int flb_net_getaddrinfo(const char *node, const char *service, struct addrinfo *hints,
+                       struct addrinfo **res, char *dns_mode_textual, int timeout);
 void flb_net_dns_lookup_context_cleanup(struct flb_net_dns *dns_ctx);
 
 /* TCP options */

src/flb_network.c

@@ -71,14 +71,12 @@ struct flb_dns_lookup_context {
     struct mk_event              response_event;                  /* c-ares socket event */
     int                          ares_socket_registered;
     struct ares_socket_functions ares_socket_functions;
-    int                         *udp_timeout_detected;
     int                          ares_socket_created;
-    int                          ares_socket_type;
     void                        *ares_channel;
     int                         *result_code;
     struct mk_event_loop        *event_loop;
     struct flb_coro             *coroutine;
-    struct flb_sched_timer      *udp_timer;
+    struct flb_sched_timer      *dns_timer;
     int                          finished;
     int                          dropped;
     struct flb_net_dns          *dns_ctx;
@@ -694,11 +692,11 @@ static void flb_net_dns_lookup_context_drop(struct flb_dns_lookup_context *looku
         mk_list_del(&lookup_context->_head);
         mk_list_add(&lookup_context->_head, &lookup_context->dns_ctx->lookups_drop);
 
-        if (lookup_context->udp_timer != NULL &&
-            lookup_context->udp_timer->active) {
-            flb_sched_timer_invalidate(lookup_context->udp_timer);
+        if (lookup_context->dns_timer != NULL &&
+            lookup_context->dns_timer->active) {
+            flb_sched_timer_invalidate(lookup_context->dns_timer);
 
-            lookup_context->udp_timer = NULL;
+            lookup_context->dns_timer = NULL;
         }
     }
 }
@@ -943,13 +941,12 @@ static void flb_net_getaddrinfo_timeout_handler(struct flb_config *config, void
         return;
     }
 
-    *(lookup_context->udp_timeout_detected) = FLB_TRUE;
     lookup_context->finished = FLB_TRUE;
-    lookup_context->udp_timer = NULL;
+    lookup_context->dns_timer = NULL;
 
-    /* We deliverately set udp_timer because we don't want flb_net_dns_lookup_context_drop
-     * to call flb_sched_timer_invalidate on the timer which was already disabled and
-     * is about to be destroyed after this this callback returns.
+    /* Clear dns_timer so flb_net_dns_lookup_context_drop does not invalidate
+     * the timer which was already disabled and will be destroyed after this
+     * callback returns.
      */
 
     ares_cancel(lookup_context->ares_channel);
@@ -996,7 +993,6 @@ static ares_socket_t flb_dns_ares_socket(int af, int type, int protocol, void *u
         return -1;
     }
 
-    lookup_context->ares_socket_type       = type;
     lookup_context->ares_socket_created    = FLB_TRUE;
 
     lookup_context->response_event.mask    = MK_EVENT_EMPTY;
@@ -1121,7 +1117,7 @@ static struct flb_dns_lookup_context *flb_net_dns_lookup_context_create(
     lookup_context->ares_socket_functions.asendv = flb_dns_ares_send;
     lookup_context->ares_socket_created = 0;
     lookup_context->event_loop = evl;
-    lookup_context->udp_timer = NULL;
+    lookup_context->dns_timer = NULL;
     lookup_context->coroutine = coroutine;
     lookup_context->finished = 0;
     lookup_context->dropped = 0;
@@ -1141,7 +1137,6 @@ static struct flb_dns_lookup_context *flb_net_dns_lookup_context_create(
 int flb_net_getaddrinfo(const char *node, const char *service, struct addrinfo *hints,
                         struct addrinfo **res, char *dns_mode_textual, int timeout)
 {
-    int                            udp_timeout_detected;
     struct flb_dns_lookup_context *lookup_context;
     int                            errno_backup;
     int                            result_code;
@@ -1179,7 +1174,6 @@ int flb_net_getaddrinfo(const char *node, const char *service, struct addrinfo *
         return result;
     }
 
-    lookup_context->udp_timeout_detected = &udp_timeout_detected;
     lookup_context->result_code = &result_code;
     lookup_context->result = &result_data;
 
@@ -1190,7 +1184,6 @@ int flb_net_getaddrinfo(const char *node, const char *service, struct addrinfo *
      */
     result_code = ARES_ESERVFAIL;
     result_data = NULL;
-    udp_timeout_detected = 0;
 
     /* The timeout we get is expressed in seconds so we need to convert it to
      * milliseconds
@@ -1215,42 +1208,44 @@ int flb_net_getaddrinfo(const char *node, const char *service, struct addrinfo *
                      flb_net_getaddrinfo_callback, lookup_context);
 
     if (!lookup_context->finished) {
-        if (lookup_context->ares_socket_created) {
-            if (lookup_context->ares_socket_type == SOCK_DGRAM) {
-                /* If the socket type created by c-ares is UDP then we need to create our
-                 * own timeout mechanism before yielding and cancel it if things go as
-                 * expected.
+        if (lookup_context->ares_socket_created && timeout > 0) {
+            /* Bound both UDP and TCP lookups, including a TCP connection to a
+             * DNS server that never completes. Cancellation schedules cleanup
+             * and resumes the waiting coroutine.
+             */
+            sched = flb_sched_ctx_get();
+            assert(sched != NULL);
+
+            result = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_ONESHOT,
+                                               timeout,
+                                               flb_net_getaddrinfo_timeout_handler,
+                                               lookup_context,
+                                               &lookup_context->dns_timer);
+            if (result == -1) {
+                /* Timer creation failed, it happen because of file descriptor or memory
+                 * exhaustion (ulimits usually)
                  */
 
-                sched = flb_sched_ctx_get();
-                assert(sched != NULL);
-
-                result = flb_sched_timer_cb_create(sched, FLB_SCHED_TIMER_CB_ONESHOT,
-                                                   timeout,
-                                                   flb_net_getaddrinfo_timeout_handler,
-                                                   lookup_context,
-                                                   &lookup_context->udp_timer);
-                if (result == -1) {
-                    /* Timer creation failed, it happen because of file descriptor or memory
-                     * exhaustion (ulimits usually)
-                     */
-
-                    result_code = ARES_ENOMEM;
+                /* Cancellation invokes the callback synchronously. Preserve the
+                 * allocation error instead of replacing it with ARES_ECANCELLED.
+                 */
+                lookup_context->finished = FLB_TRUE;
+                result_code = ARES_ENOMEM;
 
-                    ares_cancel(lookup_context->ares_channel);
+                ares_cancel(lookup_context->ares_channel);
 
-                    lookup_context->coroutine = NULL;
+                lookup_context->coroutine = NULL;
 
-                    flb_net_dns_lookup_context_drop(lookup_context);
-                }
-                else {
-                    flb_coro_yield(coroutine, FLB_FALSE);
-                }
+                flb_net_dns_lookup_context_drop(lookup_context);
             }
             else {
                 flb_coro_yield(coroutine, FLB_FALSE);
             }
         }
+        else if (lookup_context->ares_socket_created) {
+            /* A non-positive connect timeout disables the DNS deadline too. */
+            flb_coro_yield(coroutine, FLB_FALSE);
+        }
         else {
             /* Do we want to do anything special for this condition? */
         }

src/flb_upstream.c

@@ -1083,18 +1083,18 @@ int flb_upstream_conn_timeouts(struct mk_list *list)
             if (drop) {
                 if (!flb_upstream_is_shutting_down(u)) {
                     if (u->base.net.connect_timeout_log_error) {
-                        flb_error("[upstream] connection #%i to %s timed "
+                        flb_error("[upstream] connection #%i to %s:%i timed "
                                   "out after %i seconds (%s)",
                                   u_conn->fd,
-                                  flb_connection_get_remote_address(u_conn),
+                                  u->tcp_host, u->tcp_port,
                                   elapsed_time,
                                   reason);
                     }
                     else {
-                        flb_debug("[upstream] connection #%i to %s timed "
+                        flb_debug("[upstream] connection #%i to %s:%i timed "
                                   "out after %i seconds (%s)",
                                   u_conn->fd,
-                                  flb_connection_get_remote_address(u_conn),
+                                  u->tcp_host, u->tcp_port,
                                   elapsed_time,
                                   reason);
                     }

tests/integration/scenarios/networking/tests/test_upstream_timeout.py

@@ -0,0 +1,80 @@
+"""Exercise upstream timeout logging and retry completion against a silent peer."""
+
+from pathlib import Path
+import socketserver
+import threading
+
+import pytest
+import yaml
+
+from utils.test_service import FluentBitTestService
+
+
+class SilentPeer(socketserver.BaseRequestHandler):
+    def handle(self):
+        # Accept TCP but never answer the TLS handshake.
+        self.server.release.wait(30)
+
+
+class SilentServer(socketserver.ThreadingTCPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+
+@pytest.mark.parametrize("log_error", [True, False])
+def test_timeout_logs_configured_host_and_port(tmp_path, log_error):
+    server = SilentServer(("127.0.0.1", 0), SilentPeer)
+    server.release = threading.Event()
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    port = server.server_address[1]
+    config = {
+        "service": {
+            "flush": 0.2,
+            "grace": 1,
+            "log_level": "debug",
+            "http_server": "on",
+            "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}",
+        },
+        "pipeline": {
+            "inputs": [{"name": "dummy", "samples": 1, "tag": "timeout"}],
+            "outputs": [{
+                "name": "http",
+                "match": "*",
+                "host": "localhost",
+                "port": port,
+                "tls": "on",
+                "tls.verify": "off",
+                "net.dns.resolver": "LEGACY",
+                "net.connect_timeout": 2,
+                "net.connect_timeout_log_error": "on" if log_error else "off",
+                "retry_limit": "no_retries",
+            }],
+        },
+    }
+    config_path = tmp_path / "timeout.yaml"
+    config_path.write_text(yaml.safe_dump(config), encoding="utf-8")
+    service = FluentBitTestService(str(config_path))
+    try:
+        service.start()
+
+        def completed():
+            logs = Path(service.flb.log_file).read_text(encoding="utf-8", errors="replace")
+            return logs if "is not retried (no retry config)" in logs else None
+
+        logs = service.wait_for_condition(completed, timeout=30, description="timed-out flush")
+        timeout_lines = [line for line in logs.splitlines()
+                         if "[upstream]" in line and "timed out" in line]
+        assert timeout_lines, logs
+        assert any(f"to localhost:{port} timed out" in line for line in timeout_lines), logs
+        assert any(("[error]" if log_error else "[debug]") in line
+                   for line in timeout_lines), logs
+        assert "unavailable:0" not in logs
+    finally:
+        try:
+            service.stop()
+        finally:
+            server.release.set()
+            server.shutdown()
+            server.server_close()
+            thread.join(timeout=5)

tests/internal/CMakeLists.txt

@@ -17,6 +17,7 @@ set(UNIT_TESTS_FILES
   router.c
   router_config.c
   network.c
+  network_dns.c
   unit_sizes.c
   hashtable.c
   http_client.c