diff --git a/src/client/etcd_state_client.cpp b/src/client/etcd_state_client.cpp index 32d2fef8..3dd2a6e1 100644 --- a/src/client/etcd_state_client.cpp +++ b/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 #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 callback) { std::string etcd_api_path; - bool ssl = etcd_address.substr(0, 8) == "https://"; - etcd_address = etcd_address.substr(ssl ? 8 : 7); + bool ssl = etcd_url.substr(0, 8) == "https://"; + 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 callback) { - if (!etcd_addresses.size() && !etcd_local.size()) + pick_next_etcd([=]() { - fprintf(stderr, "etcd_address is missing in Vitastor configuration\n"); - exit(1); - } - pick_next_etcd(); - 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); - etcd_address = etcd_address.substr(0, pos); - } + etcd_call_selected(api, payload, timeout, retries, interval, callback); + }); +} + +void etcd_state_client_t::etcd_call_selected(const std::string & api, json11::Json payload, int timeout, + int retries, int interval, std::function callback) +{ + const auto & url = selected_etcd_url; std::string req = payload.dump(); - req = "POST "+etcd_api_path+api+" HTTP/1.1\r\n" - "Host: "+etcd_address+"\r\n" + req = "POST "+url.path+api+" HTTP/1.1\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) - selected_etcd_address = ""; + if (cur_addr == selected_etcd_url.addr) + 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(), 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) - { - local = true; - break; - } + // Fill local_ips + for (auto & ip: getifaddr_list(std::vector(), true)) + local_ips.insert(ip); } - 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() -{ - 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 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() +void etcd_state_client_t::pick_next_etcd(std::function cb) { if (!etcd_addresses.size() && !etcd_local.size()) { fprintf(stderr, "etcd_address is missing in Vitastor configuration\n"); exit(1); } - pick_next_etcd(); - 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) + if (selected_etcd_url.addr != "") { - etcd_api_path = etcd_address.substr(pos); - etcd_address = etcd_address.substr(0, pos); + cb(); + 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& 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 }, - [this, cur_addr = selected_etcd_address](http_message_t *msg) + open_websocket(etcd_watch_ws, url.addr, url.hostname, url.path+"/watch", { .timeout = etcd_slow_timeout, .ssl = url.ssl }, + [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 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) - selected_etcd_address = ""; + if (cur_addr == selected_etcd_url.addr) + selected_etcd_url = (http_url_t){}; if (etcd_watches_initialised == 0) { // Connection not established, retry in @@ -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(); } diff --git a/src/client/etcd_state_client.h b/src/client/etcd_state_client.h index 5284dce9..d9b37f4e 100644 --- a/src/client/etcd_state_client.h +++ b/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 local_ips; - std::vector etcd_addresses; + std::set local_ips; std::vector etcd_local; - std::string selected_etcd_address; - std::vector addresses_to_try; + std::vector etcd_addresses; + std::vector etcd_local_addr_urls; + std::vector etcd_nonlocal_addr_urls; + std::vector etcd_name_urls; + size_t local_to_try = 0; + std::vector etcd_urls_to_try; + http_url_t selected_etcd_url; + size_t resolve_count = 0; std::vector watches; + std::vector> 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 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 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 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 callback); - void etcd_call(std::string api, json11::Json payload, int timeout, int retries, int interval, std::function callback); + void etcd_call_oneshot(const std::string & etcd_address, const std::string & api, json11::Json payload, int timeout, std::function callback); + void etcd_call(const std::string & api, json11::Json payload, int timeout, int retries, int interval, std::function callback); void etcd_txn(json11::Json txn, int timeout, int retries, int interval, std::function callback); void etcd_txn_slow(json11::Json txn, std::function callback); void start_etcd_watcher(); diff --git a/src/client/http_client.cpp b/src/client/http_client.cpp index 514ba029..7cf10c8c 100644 --- a/src/client/http_client.cpp +++ b/src/client/http_client.cpp @@ -199,6 +199,45 @@ init_err: return NULL; } +struct http_ctx_resolve_t +{ + std::function & 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 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 & 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 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; diff --git a/src/client/http_client.h b/src/client/http_client.h index 81db074d..457b27bf 100644 --- a/src/client/http_client.h +++ b/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 & 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 on_message); void http_request(http_co_t *handler, const std::string & host, const std::string & request, const http_options_t & options, std::function response_callback); diff --git a/src/util/addr_util.cpp b/src/util/addr_util.cpp index 8f07b413..6f59bd96 100644 --- a/src/util/addr_util.cpp +++ b/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)