Backport #45629 to 22.8: Cancel c-ares failed requests and retry on system interrupts to prevent callbacks with dangling references and premature resolution failures

docs/en/sql-reference/functions/ip-address-functions.md

@@ -492,3 +492,41 @@ Result:
 │                                                                  0 │
 └────────────────────────────────────────────────────────────────────┘
 ```
+
+## reverseDNSQuery
+
+Performs a reverse DNS query to get the PTR records associated with the IP address.
+
+**Syntax**
+
+``` sql
+reverseDNSQuery(address)
+```
+
+This function performs reverse DNS resolutions on both IPv4 and IPv6.
+
+**Arguments**
+
+-   `address` — An IPv4 or IPv6 address. [String](../../sql-reference/data-types/string.md).
+
+**Returned value**
+
+-   Associated domains (PTR records).
+
+Type: Type: [Array(String)](../../sql-reference/data-types/array.md).
+
+**Example**
+
+Query:
+
+``` sql
+SELECT reverseDNSQuery('192.168.0.2');
+```
+
+Result:
+
+``` text
+┌─reverseDNSQuery('192.168.0.2')────────────┐
+│ ['test2.example.com','test3.example.com'] │
+└───────────────────────────────────────────┘
+```

src/Common/CaresPTRResolver.cpp

@@ -16,9 +16,9 @@ namespace DB
 
     static void callback(void * arg, int status, int, struct hostent * host)
     {
-        auto * ptr_records = static_cast<std::unordered_set<std::string>*>(arg);
-        if (ptr_records && status == ARES_SUCCESS)
+        if (status == ARES_SUCCESS)
         {
+            auto * ptr_records = static_cast<std::unordered_set<std::string>*>(arg);
             /*
              * In some cases (e.g /etc/hosts), hostent::h_name is filled and hostent::h_aliases is empty.
              * Thus, we can't rely solely on hostent::h_aliases. More info on:
@@ -81,7 +81,12 @@ namespace DB
         std::unordered_set<std::string> ptr_records;
 
         resolve(ip, ptr_records);
-        wait();
+
+        if (!wait_and_process())
+        {
+            cancel_requests();
+            throw DB::Exception(DB::ErrorCodes::DNS_ERROR, "Failed to complete reverse DNS query for IP {}", ip);
+        }
 
         return ptr_records;
     }
@@ -93,7 +98,12 @@ namespace DB
         std::unordered_set<std::string> ptr_records;
 
         resolve_v6(ip, ptr_records);
-        wait();
+
+        if (!wait_and_process())
+        {
+            cancel_requests();
+            throw DB::Exception(DB::ErrorCodes::DNS_ERROR, "Failed to complete reverse DNS query for IP {}", ip);
+        }
 
         return ptr_records;
     }
@@ -115,7 +125,7 @@ namespace DB
         ares_gethostbyaddr(channel, reinterpret_cast<const void*>(&addr), sizeof(addr), AF_INET6, callback, &response);
     }
 
-    void CaresPTRResolver::wait()
+    bool CaresPTRResolver::wait_and_process()
     {
         int sockets[ARES_GETSOCK_MAXNUM];
         pollfd pollfd[ARES_GETSOCK_MAXNUM];
@@ -128,7 +138,22 @@ namespace DB
             int number_of_fds_ready = 0;
             if (!readable_sockets.empty())
             {
-                number_of_fds_ready = poll(readable_sockets.data(), readable_sockets.size(), timeout);
+                number_of_fds_ready = poll(readable_sockets.data(), static_cast<nfds_t>(readable_sockets.size()), static_cast<int>(timeout));
+
+                bool poll_error = number_of_fds_ready < 0;
+                bool is_poll_error_an_interrupt = poll_error && errno == EINTR;
+
+                /*
+                 * Retry in case of interrupts and return false in case of actual errors.
+                 * */
+                if (is_poll_error_an_interrupt)
+                {
+                    continue;
+                }
+                else if (poll_error)
+                {
+                    return false;
+                }
             }
 
             if (number_of_fds_ready > 0)
@@ -141,6 +166,13 @@ namespace DB
                 break;
             }
         }
+
+        return true;
+    }
+
+    void CaresPTRResolver::cancel_requests()
+    {
+        ares_cancel(channel);
     }
 
     std::span<pollfd> CaresPTRResolver::get_readable_sockets(int * sockets, pollfd * pollfd)
@@ -149,15 +181,20 @@ namespace DB
 
         int number_of_sockets_to_poll = 0;
 
-        for (int i = 0; i < ARES_GETSOCK_MAXNUM; i++, number_of_sockets_to_poll++)
+        for (int i = 0; i < ARES_GETSOCK_MAXNUM; i++)
         {
             pollfd[i].events = 0;
             pollfd[i].revents = 0;
 
             if (ARES_GETSOCK_READABLE(sockets_bitmask, i))
             {
                 pollfd[i].fd = sockets[i];
-                pollfd[i].events = POLLIN;
+                pollfd[i].events = C_ARES_POLL_EVENTS;
+            }
+
+            if (pollfd[i].events)
+            {
+                number_of_sockets_to_poll++;
             }
             else
             {
@@ -192,7 +229,7 @@ namespace DB
     {
         for (auto readable_socket : readable_sockets)
         {
-            auto fd = readable_socket.revents & POLLIN ? readable_socket.fd : ARES_SOCKET_BAD;
+            auto fd = readable_socket.revents & C_ARES_POLL_EVENTS ? readable_socket.fd : ARES_SOCKET_BAD;
             ares_process_fd(channel, fd, ARES_SOCKET_BAD);
         }
     }

src/Common/CaresPTRResolver.h

@@ -23,6 +23,9 @@ namespace DB
          * Allow only DNSPTRProvider to instantiate this class
          * */
         struct provider_token {};
+
+        static constexpr auto C_ARES_POLL_EVENTS = POLLRDNORM | POLLIN;
+
     public:
         explicit CaresPTRResolver(provider_token);
         ~CaresPTRResolver() override;
@@ -32,7 +35,9 @@ namespace DB
         std::unordered_set<std::string> resolve_v6(const std::string & ip) override;
 
     private:
-        void wait();
+        bool wait_and_process();
+
+        void cancel_requests();
 
         void resolve(const std::string & ip, std::unordered_set<std::string> & response);
 

src/Common/tests/gtest_dns_reverse_resolve.cpp

@@ -0,0 +1,57 @@
+#include <gtest/gtest.h>
+#include <thread>
+#include <Common/DNSPTRResolverProvider.h>
+#include <Common/DNSResolver.h>
+#include <Poco/Net/IPAddress.h>
+#include <random>
+
+namespace DB
+{
+TEST(Common, ReverseDNS)
+{
+    auto addresses = std::vector<std::string>({
+        "8.8.8.8", "2001:4860:4860::8888", // dns.google
+        "142.250.219.35", // google.com
+        "157.240.12.35", // facebook
+        "208.84.244.116", "2600:1419:c400::214:c410", //www.terra.com.br,
+        "127.0.0.1", "::1"
+    });
+
+    auto func = [&]()
+    {
+        // Good random seed, good engine
+        auto rnd1 = std::mt19937(std::random_device{}());
+
+        for (int i = 0; i < 50; ++i)
+        {
+            auto & dns_resolver_instance = DNSResolver::instance();
+//            unfortunately, DNS cache can't be disabled because we might end up causing a DDoS attack
+//            dns_resolver_instance.setDisableCacheFlag();
+
+            auto addr_index = rnd1() % addresses.size();
+
+            [[maybe_unused]] auto result = dns_resolver_instance.reverseResolve(Poco::Net::IPAddress{ addresses[addr_index] });
+
+//            will not assert either because some of the IP addresses might change in the future and
+//            this test will become flaky
+//            ASSERT_TRUE(!result.empty());
+        }
+
+    };
+
+    auto number_of_threads = 200u;
+    std::vector<std::thread> threads;
+    threads.reserve(number_of_threads);
+
+    for (auto i = 0u; i < number_of_threads; i++)
+    {
+        threads.emplace_back(func);
+    }
+
+    for (auto & thread : threads)
+    {
+        thread.join();
+    }
+
+}
+}

src/Functions/reverseDNSQuery.cpp

@@ -0,0 +1,104 @@
+#include <Columns/ColumnString.h>
+#include <Columns/ColumnArray.h>
+#include <Functions/FunctionFactory.h>
+#include <Functions/IFunction.h>
+#include <DataTypes/DataTypeString.h>
+#include <DataTypes/DataTypeArray.h>
+#include <Common/DNSResolver.h>
+#include <Poco/Net/IPAddress.h>
+#include <Interpreters/Context.h>
+#include <Poco/Util/AbstractConfiguration.h>
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
+    extern const int BAD_ARGUMENTS;
+    extern const int FUNCTION_NOT_ALLOWED;
+}
+
+class ReverseDNSQuery : public IFunction
+{
+public:
+    static constexpr auto name = "reverseDNSQuery";
+    static constexpr auto allow_function_config_name = "allow_reverse_dns_query_function";
+
+    static FunctionPtr create(ContextPtr)
+    {
+        return std::make_shared<ReverseDNSQuery>();
+    }
+
+    String getName() const override
+    {
+        return name;
+    }
+
+    ColumnPtr executeImpl(const ColumnsWithTypeAndName & arguments, const DataTypePtr & data_type, size_t input_rows_count) const override
+    {
+        if (!Context::getGlobalContextInstance()->getConfigRef().getBool(allow_function_config_name, false))
+        {
+            throw Exception(ErrorCodes::FUNCTION_NOT_ALLOWED, "Function {} is not allowed because {} is not set", name, allow_function_config_name);
+        }
+
+        if (arguments.empty())
+        {
+            throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH, "Function {} requires at least one argument", name);
+        }
+
+        auto res_type = getReturnTypeImpl({data_type});
+
+        if (input_rows_count == 0u)
+        {
+            return res_type->createColumnConstWithDefaultValue(input_rows_count);
+        }
+
+        if (!isString(arguments[0].type))
+        {
+            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Function {} requires the input column to be of type String", name);
+        }
+
+        auto input_column = arguments[0].column;
+
+        auto ip_address = Poco::Net::IPAddress(input_column->getDataAt(0).toString());
+
+        auto ptr_records = DNSResolver::instance().reverseResolve(ip_address);
+
+        if (ptr_records.empty())
+            return res_type->createColumnConstWithDefaultValue(input_rows_count);
+
+        Array res;
+
+        for (const auto & ptr_record : ptr_records)
+        {
+            res.push_back(ptr_record);
+        }
+
+        return res_type->createColumnConst(input_rows_count, res);
+    }
+
+    bool isSuitableForShortCircuitArgumentsExecution(const DataTypesWithConstInfo & /*arguments*/) const override
+    {
+        return false;
+    }
+
+    size_t getNumberOfArguments() const override
+    {
+        return 1u;
+    }
+
+    DataTypePtr getReturnTypeImpl(const DataTypes & /*arguments*/) const override
+    {
+        return std::make_shared<DataTypeArray>(std::make_shared<DataTypeString>());
+    }
+
+};
+
+
+REGISTER_FUNCTION(ReverseDNSQuery)
+{
+    factory.registerFunction<ReverseDNSQuery>();
+}
+
+}

tests/config/config.d/reverse_dns_query_function.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <allow_reverse_dns_query_function>1</allow_reverse_dns_query_function>
+</clickhouse>

tests/config/install.sh

@@ -49,6 +49,7 @@ ln -sf $SRC_PATH/config.d/ssl_certs.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/filesystem_cache_log.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/session_log.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/system_unfreeze.xml $DEST_SERVER_PATH/config.d/
+ln -sf $SRC_PATH/config.d/reverse_dns_query_function.xml $DEST_SERVER_PATH/config.d/
 
 ln -sf $SRC_PATH/users.d/log_queries.xml $DEST_SERVER_PATH/users.d/
 ln -sf $SRC_PATH/users.d/readonly.xml $DEST_SERVER_PATH/users.d/

tests/integration/test_reverse_dns_query/configs/config.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <disable_internal_dns_cache>1</disable_internal_dns_cache>
+</clickhouse>

tests/integration/test_reverse_dns_query/configs/listen_host.xml

@@ -0,0 +1,5 @@
+<yandex>
+    <listen_host>::</listen_host>
+    <listen_host>0.0.0.0</listen_host>
+    <listen_try>1</listen_try>
+</yandex>

tests/integration/test_reverse_dns_query/configs/reverse_dns_function.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <allow_reverse_dns_query_function>1</allow_reverse_dns_query_function>
+</clickhouse>

tests/integration/test_reverse_dns_query/coredns_config/Corefile

@@ -0,0 +1,4 @@
+. {
+    forward . 127.0.0.11
+    log
+}