Prefer local etcd addresses and correctly cycle over them even when they need resolving

Seems slightly overcomplicated...

src/client/etcd_state_client.cpp

@@ -1,6 +1,7 @@
 // Copyright (c) Vitaliy Filippov, 2019+
 // License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
 
+#include <assert.h>
 #include "malloc_or_die.h"
 #include "osd_ops.h"
 #include "msgr_op.h"
@@ -94,12 +95,12 @@ http_context_t *etcd_state_client_t::get_http_ctx()
     return http_ctx;
 }
 
-void etcd_state_client_t::etcd_call_oneshot(std::string etcd_address, std::string api, json11::Json payload,
+void etcd_state_client_t::etcd_call_oneshot(const std::string & etcd_url, const std::string & api, json11::Json payload,
     int timeout, std::function<void(std::string, json11::Json)> callback)
 {
     std::string etcd_api_path;
-    bool ssl = etcd_address.substr(0, 8) == "https://";
+    bool ssl = etcd_url.substr(0, 8) == "https://";
-    etcd_address = etcd_address.substr(ssl ? 8 : 7);
+    auto etcd_address = etcd_url.substr(ssl ? 8 : 7);
     int pos = etcd_address.find('/');
     if (pos >= 0)
     {
@@ -125,28 +126,22 @@ void etcd_state_client_t::etcd_call_oneshot(std::string etcd_address, std::strin
     http_request(http_cli, etcd_address, req, { .timeout = timeout, .ssl = ssl }, cb);
 }
 
-void etcd_state_client_t::etcd_call(std::string api, json11::Json payload, int timeout,
+void etcd_state_client_t::etcd_call(const std::string & api, json11::Json payload, int timeout,
     int retries, int interval, std::function<void(std::string, json11::Json)> callback)
 {
-    if (!etcd_addresses.size() && !etcd_local.size())
+    pick_next_etcd([=]()
     {
-        fprintf(stderr, "etcd_address is missing in Vitastor configuration\n");
+        etcd_call_selected(api, payload, timeout, retries, interval, callback);
-        exit(1);
+    });
-    }
+}
-    pick_next_etcd();
+
-    std::string etcd_address = selected_etcd_address;
+void etcd_state_client_t::etcd_call_selected(const std::string & api, json11::Json payload, int timeout,
-    std::string etcd_api_path;
+    int retries, int interval, std::function<void(std::string, json11::Json)> callback)
-    bool ssl = etcd_address.substr(0, 8) == "https://";
+{
-    etcd_address = etcd_address.substr(ssl ? 8 : 7);
+    const auto & url = selected_etcd_url;
-    int pos = etcd_address.find('/');
-    if (pos >= 0)
-    {
-        etcd_api_path = etcd_address.substr(pos);
-        etcd_address = etcd_address.substr(0, pos);
-    }
     std::string req = payload.dump();
-    req = "POST "+etcd_api_path+api+" HTTP/1.1\r\n"
+    req = "POST "+url.path+api+" HTTP/1.1\r\n"
-        "Host: "+etcd_address+"\r\n"
+        "Host: "+url.hostname+"\r\n"
         "Content-Type: application/json\r\n"
         "Content-Length: "+std::to_string(req.size())+"\r\n"
         "Connection: keep-alive\r\n"
@@ -154,15 +149,15 @@ void etcd_state_client_t::etcd_call(std::string api, json11::Json payload, int t
         "\r\n"+req;
     retries--;
     auto cb = [this, api, payload, timeout, retries, interval, callback,
-        cur_addr = selected_etcd_address](http_message_t *response)
+        cur_addr = url.addr](http_message_t *response)
     {
         std::string err;
         json11::Json data;
         response->parse_json_response(err, data);
         if (err != "")
         {
-            if (cur_addr == selected_etcd_address)
+            if (cur_addr == selected_etcd_url.addr)
-                selected_etcd_address = "";
+                selected_etcd_url = (http_url_t){};
             if (retries > 0)
             {
                 if (this->log_level > 0)
@@ -191,53 +186,57 @@ void etcd_state_client_t::etcd_call(std::string api, json11::Json payload, int t
     };
     if (!keepalive_client)
         keepalive_client = http_init(get_http_ctx());
-    http_request(keepalive_client, etcd_address, req, { .timeout = timeout, .keepalive = true, .ssl = ssl }, cb);
+    http_request(keepalive_client, url.addr, req, { .timeout = timeout, .keepalive = true, .ssl = url.ssl }, cb);
 }
 
-void etcd_state_client_t::add_etcd_url(std::string addr)
+
+void etcd_state_client_t::add_etcd_url(std::string etcd_address)
 {
-    if (addr.length() > 0)
+    if (etcd_address.size() > 0)
     {
-        bool ssl = false;
-        if (strtolower(addr.substr(0, 7)) == "http://")
-            addr = addr.substr(7);
-        else if (strtolower(addr.substr(0, 8)) == "https://")
-        {
-            addr = addr.substr(8);
-            ssl = true;
-        }
         if (!local_ips.size())
-            local_ips = getifaddr_list(std::vector<addr_mask_t>(), true);
-        std::string check_addr;
-        int pos = addr.find('/');
-        int pos2 = addr.find(':');
-        if (pos2 >= 0)
-            check_addr = addr.substr(0, pos2);
-        else if (pos >= 0)
-            check_addr = addr.substr(0, pos);
-        else
-            check_addr = addr;
-        if (pos == std::string::npos)
-            addr += "/v3";
-        addr = (ssl ? "https://" : "http://") + addr;
-        bool local = false;
-        int i;
-        for (i = 0; i < local_ips.size(); i++)
         {
-            if (local_ips[i] == check_addr)
+            // Fill local_ips
-            {
+            for (auto & ip: getifaddr_list(std::vector<addr_mask_t>(), true))
-                local = true;
+                local_ips.insert(ip);
-                break;
-            }
         }
-        auto & to = local ? this->etcd_local : this->etcd_addresses;
+        std::string etcd_api_path;
+        bool ssl = false;
+        if (etcd_address.substr(0, 8) == "https://")
+        {
+            ssl = true;
+            etcd_address = etcd_address.substr(8);
+        }
+        else if (etcd_address.substr(0, 7) == "http://")
+            etcd_address = etcd_address.substr(7);
+        auto pos = etcd_address.find('/');
+        if (pos != std::string::npos)
+        {
+            etcd_api_path = etcd_address.substr(pos);
+            etcd_address = etcd_address.substr(0, pos);
+        }
+        else
+            etcd_api_path = "/v3";
+        pos = etcd_address.find(':');
+        auto check_addr = (pos != std::string::npos ? etcd_address.substr(0, pos) : etcd_address);
+        bool is_local = local_ips.find(check_addr) != local_ips.end();
+        auto & to = (is_local ? etcd_local : etcd_addresses);
+        check_addr = (ssl ? "https://" : "http://") + etcd_address + etcd_api_path;
+        size_t i;
         for (i = 0; i < to.size(); i++)
         {
-            if (to[i] == addr)
+            if (to[i] == check_addr)
                 break;
         }
         if (i >= to.size())
-            to.push_back(addr);
+        {
+            to.push_back(check_addr);
+            // Check if it's a domain name
+            sockaddr_storage ss;
+            bool is_name = !is_local && !string_to_addr(etcd_address, true, 0, &ss);
+            auto & to_addr = (is_local ? etcd_local_addr_urls : (is_name ? etcd_name_urls : etcd_nonlocal_addr_urls));
+            to_addr.push_back((http_url_t){ .ssl = ssl, .addr = etcd_address, .hostname = etcd_address, .path = etcd_api_path });
+        }
     }
 }
 
@@ -245,6 +244,9 @@ void etcd_state_client_t::parse_config(const json11::Json & config)
 {
     this->etcd_local.clear();
     this->etcd_addresses.clear();
+    this->etcd_local_addr_urls.clear();
+    this->etcd_nonlocal_addr_urls.clear();
+    this->etcd_name_urls.clear();
     if (config["etcd_address"].is_string())
     {
         std::string ea = config["etcd_address"].string_value();
@@ -328,67 +330,130 @@ void etcd_state_client_t::parse_config(const json11::Json & config)
     }
 }
 
-void etcd_state_client_t::pick_next_etcd()
+void etcd_state_client_t::pick_next_etcd(std::function<void()> cb)
-{
-    if (selected_etcd_address != "")
-        return;
-    if (addresses_to_try.size() == 0)
-    {
-        // Prefer local etcd, if any
-        for (int i = 0; i < etcd_local.size(); i++)
-            addresses_to_try.push_back(etcd_local[i]);
-        std::vector<int> ns;
-        for (int i = 0; i < etcd_addresses.size(); i++)
-            ns.push_back(i);
-        if (!rand_initialized)
-        {
-            timespec tv;
-            clock_gettime(CLOCK_REALTIME, &tv);
-            srand48(tv.tv_sec*1000000000 + tv.tv_nsec);
-            rand_initialized = true;
-        }
-        while (ns.size())
-        {
-            int i = lrand48() % ns.size();
-            addresses_to_try.push_back(etcd_addresses[ns[i]]);
-            ns.erase(ns.begin()+i, ns.begin()+i+1);
-        }
-    }
-    selected_etcd_address = addresses_to_try[0];
-    addresses_to_try.erase(addresses_to_try.begin(), addresses_to_try.begin()+1);
-}
-
-void etcd_state_client_t::start_etcd_watcher()
 {
     if (!etcd_addresses.size() && !etcd_local.size())
     {
         fprintf(stderr, "etcd_address is missing in Vitastor configuration\n");
         exit(1);
     }
-    pick_next_etcd();
+    if (selected_etcd_url.addr != "")
-    std::string etcd_address = selected_etcd_address;
-    std::string etcd_api_path;
-    bool ssl = etcd_address.substr(0, 8) == "https://";
-    etcd_address = etcd_address.substr(ssl ? 8 : 7);
-    int pos = etcd_address.find('/');
-    if (pos >= 0)
     {
-        etcd_api_path = etcd_address.substr(pos);
+        cb();
-        etcd_address = etcd_address.substr(0, pos);
+        return;
     }
+    if (etcd_urls_to_try.size() != 0)
+    {
+        selected_etcd_url = std::move(etcd_urls_to_try[0]);
+        etcd_urls_to_try.erase(etcd_urls_to_try.begin());
+        cb();
+        return;
+    }
+    on_resolve_queue.push_back(std::move(cb));
+    if (on_resolve_queue.size() > 1)
+    {
+        // Already resolving
+        return;
+    }
+    assert(!resolve_count);
+    local_to_try = 0;
+    for (auto & url: etcd_local_addr_urls)
+    {
+        // Prefer local IPs, if any
+        etcd_urls_to_try.push_back(url);
+        local_to_try++;
+    }
+    for (auto & url: etcd_nonlocal_addr_urls)
+    {
+        etcd_urls_to_try.push_back(url);
+    }
+    resolve_count++;
+    for (auto & url: etcd_name_urls)
+    {
+        resolve_count++;
+        http_resolve(get_http_ctx(), url.ssl, url.addr, [this, url](const std::string & error, const std::vector<std::string>& addresses)
+        {
+            if (error != "")
+                fprintf(stderr, "Error resolving %s: %s\n", url.addr.c_str(), error.c_str());
+            for (auto & addr: addresses)
+            {
+                auto url_copy = url;
+                url_copy.addr = addr;
+                if (local_ips.find(addr) != local_ips.end())
+                {
+                    etcd_urls_to_try.insert(etcd_urls_to_try.begin(), std::move(url_copy));
+                    local_to_try++;
+                }
+                else
+                    etcd_urls_to_try.push_back(std::move(url_copy));
+            }
+            resolve_count--;
+            if (!resolve_count)
+                pick_next_etcd_on_resolve();
+        });
+    }
+    resolve_count--;
+    if (!resolve_count)
+    {
+        pick_next_etcd_on_resolve();
+    }
+}
+
+void etcd_state_client_t::pick_next_etcd_on_resolve()
+{
+    if (!etcd_urls_to_try.size())
+    {
+        fprintf(stderr, "None of etcd_address could be resolved\n");
+        exit(1);
+    }
+    if (!rand_initialized)
+    {
+        timespec tv;
+        clock_gettime(CLOCK_REALTIME, &tv);
+        srand48(tv.tv_sec*1000000000 + tv.tv_nsec);
+        rand_initialized = true;
+    }
+    // Shuffle addresses
+    for (size_t i = etcd_urls_to_try.size()-1; i > local_to_try; i--)
+    {
+        size_t j = local_to_try + lrand48() % (i - local_to_try);
+        if (j != i)
+            std::swap(etcd_urls_to_try[i], etcd_urls_to_try[j]);
+    }
+    selected_etcd_url = std::move(etcd_urls_to_try[0]);
+    etcd_urls_to_try.erase(etcd_urls_to_try.begin());
+    auto cbs = std::move(on_resolve_queue);
+    for (auto cb: cbs)
+    {
+        cb();
+    }
+}
+
+void etcd_state_client_t::start_etcd_watcher()
+{
+    pick_next_etcd([this]()
+    {
+        start_etcd_watcher_selected();
+    });
+}
+
+void etcd_state_client_t::start_etcd_watcher_selected()
+{
+    const auto & url = selected_etcd_url;
     etcd_watches_initialised = 0;
     ws_alive = 1;
     if (this->log_level > 1)
     {
-        fprintf(stderr, "Trying to connect to etcd websocket at %s, watch from revision %ju/%ju/%ju\n", etcd_address.c_str(),
+        fprintf(stderr, "Trying to connect to etcd websocket at %s%s%s (hostname %s), watch from revision %ju/%ju/%ju\n",
+            url.ssl ? "https://" : "http://", url.addr.c_str(), url.path.c_str(), url.hostname.c_str(),
             etcd_watch_revision_config, etcd_watch_revision_osd, etcd_watch_revision_pg);
     }
     if (!etcd_watch_ws)
         etcd_watch_ws = http_init(get_http_ctx());
     else
         http_close(etcd_watch_ws);
-    open_websocket(etcd_watch_ws, etcd_address, etcd_api_path+"/watch", { .timeout = etcd_slow_timeout, .ssl = ssl },
+    open_websocket(etcd_watch_ws, url.addr, url.hostname, url.path+"/watch", { .timeout = etcd_slow_timeout, .ssl = url.ssl },
-        [this, cur_addr = selected_etcd_address](http_message_t *msg)
+        [this, cur_addr = url.addr](http_message_t *msg)
     {
         if (msg->body.length())
         {
@@ -475,7 +540,7 @@ void etcd_state_client_t::start_etcd_watcher()
                         etcd_watch_revision_pg = watch_rev;
                     else if (watch_id == ETCD_OSD_STATE_WATCH_ID)
                         etcd_watch_revision_osd = watch_rev;
-                    addresses_to_try.clear();
+                    etcd_urls_to_try.clear();
                 }
                 // First gather all changes into a hash to remove multiple overwrites
                 std::map<std::string, etcd_kv_t> changes;
@@ -505,8 +570,8 @@ void etcd_state_client_t::start_etcd_watcher()
         if (msg->eof)
         {
             fprintf(stderr, "Disconnected from etcd %s\n", cur_addr.c_str());
-            if (cur_addr == selected_etcd_address)
+            if (cur_addr == selected_etcd_url.addr)
-                selected_etcd_address = "";
+                selected_etcd_url = (http_url_t){};
             if (etcd_watches_initialised == 0)
             {
                 // Connection not established, retry in <etcd_quick_timeout>
@@ -581,7 +646,7 @@ void etcd_state_client_t::start_ws_keepalive()
             {
                 if (this->log_level > 0)
                 {
-                    fprintf(stderr, "Websocket ping failed, disconnecting from etcd %s\n", selected_etcd_address.c_str());
+                    fprintf(stderr, "Websocket ping failed, disconnecting from etcd %s\n", selected_etcd_url.addr.c_str());
                 }
                 start_etcd_watcher();
             }

src/client/etcd_state_client.h

@@ -97,24 +97,41 @@ struct inode_watch_t
     inode_config_t cfg = {};
 };
 
+struct http_url_t
+{
+    bool ssl;
+    std::string addr;
+    std::string hostname;
+    std::string path;
+};
+
 struct http_co_t;
 struct http_context_t;
 
 struct __attribute__((visibility("default"))) etcd_state_client_t
 {
 protected:
-    std::vector<std::string> local_ips;
+    std::set<std::string> local_ips;
-    std::vector<std::string> etcd_addresses;
     std::vector<std::string> etcd_local;
-    std::string selected_etcd_address;
+    std::vector<std::string> etcd_addresses;
-    std::vector<std::string> addresses_to_try;
+    std::vector<http_url_t> etcd_local_addr_urls;
+    std::vector<http_url_t> etcd_nonlocal_addr_urls;
+    std::vector<http_url_t> etcd_name_urls;
+    size_t local_to_try = 0;
+    std::vector<http_url_t> etcd_urls_to_try;
+    http_url_t selected_etcd_url;
+    size_t resolve_count = 0;
     std::vector<inode_watch_t*> watches;
+    std::vector<std::function<void()>> on_resolve_queue;
     bool new_pg_config = false;
     int ws_keepalive_timer = -1;
     int ws_alive = 0;
     bool rand_initialized = false;
     void add_etcd_url(std::string);
-    void pick_next_etcd();
+    void pick_next_etcd(std::function<void()> cb);
+    void pick_next_etcd_on_resolve();
+    void etcd_call_selected(const std::string & api, json11::Json payload, int timeout, int retries, int interval, std::function<void(std::string, json11::Json)> callback);
+    void start_etcd_watcher_selected();
 public:
     int etcd_keepalive_timeout = 30;
     int etcd_ws_keepalive_interval = 5;
@@ -170,8 +187,8 @@ public:
     etcd_kv_t parse_etcd_kv(const json11::Json & kv_json);
     std::vector<std::string> get_addresses();
     http_context_t *get_http_ctx();
-    void etcd_call_oneshot(std::string etcd_address, std::string api, json11::Json payload, int timeout, std::function<void(std::string, json11::Json)> callback);
+    void etcd_call_oneshot(const std::string & etcd_address, const std::string & api, json11::Json payload, int timeout, std::function<void(std::string, json11::Json)> callback);
-    void etcd_call(std::string api, json11::Json payload, int timeout, int retries, int interval, std::function<void(std::string, json11::Json)> callback);
+    void etcd_call(const std::string & api, json11::Json payload, int timeout, int retries, int interval, std::function<void(std::string, json11::Json)> callback);
     void etcd_txn(json11::Json txn, int timeout, int retries, int interval, std::function<void(std::string, json11::Json)> callback);
     void etcd_txn_slow(json11::Json txn, std::function<void(std::string, json11::Json)> callback);
     void start_etcd_watcher();

src/client/http_client.cpp

@@ -199,6 +199,45 @@ init_err:
     return NULL;
 }
 
+struct http_ctx_resolve_t
+{
+    std::function<void(const std::string & error, const std::vector<std::string> & addrs)> cb;
+};
+
+void http_resolve_ares_cb(void *data, int status, int timeouts, struct ares_addrinfo *result)
+{
+    http_ctx_resolve_t *obj = (http_ctx_resolve_t*)data;
+    if (status != ARES_SUCCESS)
+    {
+        obj->cb(ares_strerror(status), {});
+        delete obj;
+        return;
+    }
+    std::vector<std::string> addrs;
+    for (auto node = result->nodes; node; node = node->ai_next)
+    {
+        sockaddr_storage ss;
+        memset(&ss, 0, sizeof(ss));
+        memcpy(&ss, node->ai_addr, node->ai_addrlen);
+        addrs.push_back(addr_to_string(ss));
+    }
+    obj->cb("", addrs);
+    delete obj;
+}
+
+void http_resolve(http_context_t *ctx, bool ssl, std::string host,
+    std::function<void(const std::string & error, const std::vector<std::string> & addrs)> cb)
+{
+    auto obj = new http_ctx_resolve_t();
+    obj->cb = std::move(cb);
+    ares_addrinfo_hints hints = { .ai_flags = ARES_AI_NOSORT|ARES_AI_NUMERICSERV };
+    auto pos = host.rfind(':');
+    if (pos != std::string::npos)
+        host[pos] = 0;
+    ares_getaddrinfo(ctx->ares, host.c_str(),
+        pos != std::string::npos ? host.c_str()+pos+1 : (ssl ? "443" : "80"), &hints, http_resolve_ares_cb, obj);
+}
+
 void http_context_destroy(http_context_t *ctx)
 {
     delete ctx;
@@ -213,21 +252,21 @@ http_co_t *http_init(http_context_t *ctx)
     return handler;
 }
 
-void open_websocket(http_co_t *handler, const std::string & host, const std::string & path,
+void open_websocket(http_co_t *handler, const std::string & addr, const std::string & hostname, const std::string & path,
     const http_options_t & options, std::function<void(http_message_t *msg)> response_callback)
 {
-    if (handler->state == HTTP_CO_KEEPALIVE && (handler->connected_host != host || handler->ssl != options.ssl))
+    if (handler->state == HTTP_CO_KEEPALIVE && (handler->connected_host != addr || handler->ssl != options.ssl))
         handler->close_connection();
     if (handler->state != HTTP_CO_KEEPALIVE && handler->state != HTTP_CO_CLOSED)
         throw std::runtime_error("Attempt to open websocket on a keepalive stream");
     std::string request = "GET "+path+" HTTP/1.1\r\n"
-        "Host: "+host+"\r\n"
+        "Host: "+hostname+"\r\n"
         "Upgrade: websocket\r\n"
         "Connection: upgrade\r\n"
         "Sec-WebSocket-Key: x3JJHMbDL1EzLkh9GBhXDw==\r\n"
         "Sec-WebSocket-Version: 13\r\n"
         "\r\n";
-    handler->host = host;
+    handler->host = addr;
     handler->host_port = "";
     handler->request_timeout = options.timeout < 0 ? -1 : (options.timeout == 0 ? DEFAULT_TIMEOUT : options.timeout);
     handler->want_streaming = false;

src/client/http_client.h

@@ -48,9 +48,11 @@ struct http_co_t;
 
 http_context_t* http_context_init(timerfd_manager_t *tfd, const std::string & ssl_cert, const std::string & ssl_key,
     const std::string & ssl_ca, bool verify_peer, std::string & error);
+void http_resolve(http_context_t *ctx, bool ssl, std::string host,
+    std::function<void(const std::string & error, const std::vector<std::string> & addrs)> cb);
 void http_context_destroy(http_context_t *ctx);
 http_co_t* http_init(http_context_t *ctx = NULL);
-void open_websocket(http_co_t *handler, const std::string & host, const std::string & path,
+void open_websocket(http_co_t *handler, const std::string & addr, const std::string & hostname, const std::string & path,
     const http_options_t & options, std::function<void(http_message_t *msg)> on_message);
 void http_request(http_co_t *handler, const std::string & host, const std::string & request,
     const http_options_t & options, std::function<void(http_message_t *response)> response_callback);

src/util/addr_util.cpp

@@ -62,7 +62,8 @@ std::string addr_to_string(const sockaddr_storage &addr)
         throw std::runtime_error("Unknown address family "+std::to_string(addr.ss_family));
     if (!ok)
         throw std::runtime_error(std::string("inet_ntop: ") + strerror(errno));
-    return std::string(peer_str)+":"+std::to_string(port);
+    return (addr.ss_family == AF_INET6 ? "[" : "")+std::string(peer_str)+
+        (addr.ss_family == AF_INET6 ? "]:" : ":")+std::to_string(port);
 }
 
 bool cidr_match(const in_addr &addr, const in_addr &net, uint8_t bits)