This is an automated email from the ASF dual-hosted git repository. zhangstar333 pushed a commit to branch doris_master_new in repository https://gitbox.apache.org/repos/asf/doris.git
commit 182a9ffce652aa65c158cd3b661e88c5c1cd5f9f Author: zhangstar333 <[email protected]> AuthorDate: Fri May 22 20:34:39 2026 +0800 brpc dns cache --- be/src/cloud/cloud_backend_service.cpp | 2 +- be/src/cloud/cloud_warm_up_manager.cpp | 4 +- be/src/common/config.cpp | 3 + be/src/common/config.h | 10 ++ be/src/exec/operator/exchange_sink_buffer.cpp | 24 +-- be/src/exec/operator/exchange_sink_buffer.h | 4 +- be/src/io/cache/peer_file_cache_reader.cpp | 2 +- be/src/util/brpc_client_cache.h | 218 ++++++++++++++++++-------- be/src/util/dns_cache.cpp | 14 +- be/src/util/dns_cache.h | 11 ++ be/src/util/proto_util.h | 3 +- 11 files changed, 207 insertions(+), 88 deletions(-) diff --git a/be/src/cloud/cloud_backend_service.cpp b/be/src/cloud/cloud_backend_service.cpp index 5f09f194e37..d66e2a94bae 100644 --- a/be/src/cloud/cloud_backend_service.cpp +++ b/be/src/cloud/cloud_backend_service.cpp @@ -219,7 +219,7 @@ void CloudBackendService::_warm_up_cache(const TWarmUpCacheAsyncRequest& request } std::string brpc_addr = get_host_port(host, request.brpc_port); std::shared_ptr<PBackendService_Stub> brpc_stub = - _exec_env->brpc_internal_client_cache()->get_new_client_no_cache(brpc_addr); + _exec_env->brpc_internal_client_cache()->get_new_client_no_cache(brpc_addr, "", "", "", request.host); if (!brpc_stub) { LOG(WARNING) << "warm_up_cache_async: failed to get brpc_stub for addr " << brpc_addr; return; diff --git a/be/src/cloud/cloud_warm_up_manager.cpp b/be/src/cloud/cloud_warm_up_manager.cpp index e6fad053f07..5633794ddf9 100644 --- a/be/src/cloud/cloud_warm_up_manager.cpp +++ b/be/src/cloud/cloud_warm_up_manager.cpp @@ -815,7 +815,7 @@ Status CloudWarmUpManager::_do_warm_up_rowset(RowsetMeta& rs_meta, int64_t table Status st = Status::OK(); std::shared_ptr<PBackendService_Stub> brpc_stub = ExecEnv::GetInstance()->brpc_internal_client_cache()->get_new_client_no_cache( - brpc_addr); + brpc_addr, "", "", "", info.replica.host); if (!brpc_stub) { st = Status::RpcError("Address {} is wrong", brpc_addr); add_failure(info, target, st); @@ -975,7 +975,7 @@ void CloudWarmUpManager::_recycle_cache(int64_t tablet_id, Status st = Status::OK(); std::shared_ptr<PBackendService_Stub> brpc_stub = ExecEnv::GetInstance()->brpc_internal_client_cache()->get_new_client_no_cache( - brpc_addr); + brpc_addr, "", "", "", replica.replica.host); if (!brpc_stub) { st = Status::RpcError("Address {} is wrong", brpc_addr); continue; diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index b2d51ecbb6d..5c193ec603c 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -885,6 +885,9 @@ DEFINE_Int64(brpc_max_body_size, "3147483648"); DEFINE_Int64(brpc_socket_max_unwritten_bytes, "-1"); DEFINE_mBool(brpc_usercode_in_pthread, "false"); +DEFINE_mBool(enable_brpc_get_client_handshake, "true"); // test it +DEFINE_mInt32(brpc_get_client_handshake_max_retries, "3"); + // TODO(zxy): expect to be true in v1.3 // Whether to embed the ProtoBuf Request serialized string together with Tuple/Block data into // Controller Attachment and send it through http brpc when the length of the Tuple/Block data diff --git a/be/src/common/config.h b/be/src/common/config.h index 25c7a8ef89d..670f0284a83 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -943,6 +943,16 @@ DECLARE_Int64(brpc_max_body_size); DECLARE_Int64(brpc_socket_max_unwritten_bytes); // Whether to set FLAGS_usercode_in_pthread to true in brpc DECLARE_mBool(brpc_usercode_in_pthread); + +// Whether BrpcClientCache::get_client() should handshake-validate a freshly +// rebuilt stub before caching/returning it, and retry on failure. Costs 1 RTT +// per cache-miss / rebuild (the warm-cache fast path is untouched). Default +// false to preserve current behavior; flip on (or wire a session variable into +// get_client) to harden rolling-restart scenarios. +DECLARE_mBool(enable_brpc_get_client_handshake); +// Max attempts in get_client when handshake validation is enabled. Each +// failed attempt evicts the stub and marks DNSCache dirty for re-resolution. +DECLARE_mInt32(brpc_get_client_handshake_max_retries); // TODO(zxy): expect to be true in v1.3 // Whether to embed the ProtoBuf Request serialized string together with Tuple/Block data into // Controller Attachment and send it through http brpc when the length of the Tuple/Block data diff --git a/be/src/exec/operator/exchange_sink_buffer.cpp b/be/src/exec/operator/exchange_sink_buffer.cpp index 2c564891da0..19dc480cb2f 100644 --- a/be/src/exec/operator/exchange_sink_buffer.cpp +++ b/be/src/exec/operator/exchange_sink_buffer.cpp @@ -322,7 +322,7 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) { } // attach task for memory tracker and query id when core SCOPED_ATTACH_TASK(_state); - _failed(ins->id, err); + _failed(ins->id, Status::RpcError(err)); }); send_callback->start_rpc_time = GetCurrentTimeNanos(); send_callback->addSuccessHandler([&, weak_task_ctx = weak_task_exec_ctx()]( @@ -345,8 +345,8 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) { if (s.is<ErrorCode::END_OF_FILE>()) { _set_receiver_eof(ins); } else if (!s.ok()) { - _failed(ins.id, - fmt::format("exchange req success but status isn't ok: {}", s.to_string())); + _failed(ins.id, Status::Cancelled("exchange req success but status isn't ok: {}", + s.to_string())); return; } else if (eos) { _ended(ins); @@ -356,8 +356,8 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) { // `_send_rpc` must be the LAST operation in this function, because it may reuse the callback! s = _send_rpc(ins); if (!s) { - _failed(ins.id, - fmt::format("exchange req success but status isn't ok: {}", s.to_string())); + _failed(ins.id, Status::Cancelled("exchange req success but status isn't ok: {}", + s.to_string())); } }); { @@ -457,7 +457,7 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) { } // attach task for memory tracker and query id when core SCOPED_ATTACH_TASK(_state); - _failed(ins->id, err); + _failed(ins->id, Status::RpcError(err)); }); send_callback->start_rpc_time = GetCurrentTimeNanos(); send_callback->addSuccessHandler([&, weak_task_ctx = weak_task_exec_ctx()]( @@ -479,8 +479,8 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) { if (s.is<ErrorCode::END_OF_FILE>()) { _set_receiver_eof(ins); } else if (!s.ok()) { - _failed(ins.id, - fmt::format("exchange req success but status isn't ok: {}", s.to_string())); + _failed(ins.id, Status::Cancelled("exchange req success but status isn't ok: {}", + s.to_string())); return; } else if (eos) { _ended(ins); @@ -490,8 +490,8 @@ Status ExchangeSinkBuffer::_send_rpc(RpcInstance& instance_data) { // `_send_rpc` must be the LAST operation in this function, because it may reuse the callback! s = _send_rpc(ins); if (!s) { - _failed(ins.id, - fmt::format("exchange req success but status isn't ok: {}", s.to_string())); + _failed(ins.id, Status::Cancelled("exchange req success but status isn't ok: {}", + s.to_string())); } }); { @@ -522,11 +522,11 @@ void ExchangeSinkBuffer::_ended(RpcInstance& ins) { } } -void ExchangeSinkBuffer::_failed(InstanceLoId id, const std::string& err) { +void ExchangeSinkBuffer::_failed(InstanceLoId id, Status err) { _is_failed = true; LOG(INFO) << "send rpc failed, instance id: " << id << ", _dest_node_id: " << _dest_node_id << ", node id: " << _node_id << ", err: " << err; - _context->cancel(Status::Cancelled(err)); + _context->cancel(std::move(err)); } void ExchangeSinkBuffer::_set_receiver_eof(RpcInstance& ins) { diff --git a/be/src/exec/operator/exchange_sink_buffer.h b/be/src/exec/operator/exchange_sink_buffer.h index 7478c63f4e8..fb95445cca7 100644 --- a/be/src/exec/operator/exchange_sink_buffer.h +++ b/be/src/exec/operator/exchange_sink_buffer.h @@ -310,13 +310,13 @@ private: #ifndef BE_TEST inline void _ended(RpcInstance& ins); - inline void _failed(InstanceLoId id, const std::string& err); + inline void _failed(InstanceLoId id, Status err); inline void _set_receiver_eof(RpcInstance& ins); inline void _turn_off_channel(RpcInstance& ins, std::unique_lock<std::mutex>& with_lock); #else virtual void _ended(RpcInstance& ins); - virtual void _failed(InstanceLoId id, const std::string& err); + virtual void _failed(InstanceLoId id, Status err); virtual void _set_receiver_eof(RpcInstance& ins); virtual void _turn_off_channel(RpcInstance& ins, std::unique_lock<std::mutex>& with_lock); #endif diff --git a/be/src/io/cache/peer_file_cache_reader.cpp b/be/src/io/cache/peer_file_cache_reader.cpp index ace961ede06..4febb7c97ac 100644 --- a/be/src/io/cache/peer_file_cache_reader.cpp +++ b/be/src/io/cache/peer_file_cache_reader.cpp @@ -212,7 +212,7 @@ Status PeerFileCacheReader::fetch_blocks(const std::vector<FileBlockSPtr>& block Status st = Status::OK(); std::shared_ptr<PBackendService_Stub> brpc_stub = ExecEnv::GetInstance()->brpc_internal_client_cache()->get_new_client_no_cache( - brpc_addr); + brpc_addr, "", "", "", _host); if (!brpc_stub) { peer_cache_reader_failed_counter << 1; LOG(WARNING) << "failed to get brpc stub " << brpc_addr; diff --git a/be/src/util/brpc_client_cache.h b/be/src/util/brpc_client_cache.h index 993838699e8..b4ebb10acea 100644 --- a/be/src/util/brpc_client_cache.h +++ b/be/src/util/brpc_client_cache.h @@ -30,6 +30,7 @@ #include <parallel_hashmap/phmap.h> #include <stddef.h> +#include <algorithm> #include <functional> #include <memory> #include <mutex> @@ -65,25 +66,73 @@ using StubMap = phmap::parallel_flat_hash_map< std::allocator<std::pair<const std::string, StubEntry<T>>>, 8, std::mutex>; namespace doris { + +// True if `errcode` from brpc::Controller::ErrorCode() represents an actual +// network-level fault (peer Pod terminating, mid-flight reset, port not +// listening, etc.) — i.e. cases where we should mark the cached channel bad +// and invalidate DNS so a re-resolve picks up the new Pod IP. +// +// IMPORTANT: keep this a strict white-list of network errnos. Do NOT add +// EOVERCROWDED / ELIMIT / EREJECT (those are *business* overload signals — the +// peer is alive and we must NOT trash the cache / DNS for them). +inline bool is_brpc_network_fault(int errcode) { + switch (errcode) { + case EHOSTDOWN: // 112 - Pod terminating (graceful shutdown SIGTERM path) + case ETIMEDOUT: // connect / send timeout + case ECONNREFUSED: // peer port not yet listening (Pod just (re)started) + case ECONNRESET: // peer closed connection mid-flight + case EHOSTUNREACH: + case ENETUNREACH: + case ENOTCONN: + case EPIPE: + return true; + default: + return false; + } +} + +// Common handler invoked on every detected brpc network fault. Two side effects: +// 1) marks the cached channel bad so BrpcClientCache evicts it next get_client(); +// 2) synchronously marks `hostname` dirty in DNSCache so the next resolve goes +// through getaddrinfo() instead of returning the stale IP. Step 2 is the +// cure for the cross-node rolling-restart "stuck on old IP" deadlock. +// `hostname` may be empty (e.g. for IP-only callers); in that case step 2 is a no-op. +// Never spawns threads — invalidate() is just a set::insert. +inline void on_brpc_network_fault(const std::shared_ptr<AtomicStatus>& channel_st, + const std::string& hostname, brpc::Controller* cntl) { + Status error_st = Status::NetworkError( + "Failed to send brpc, error={}, error_text={}, client: {}, latency = {}", + berror(cntl->ErrorCode()), cntl->ErrorText(), BackendOptions::get_localhost(), + cntl->latency_us()); + LOG(WARNING) << error_st; + channel_st->update(error_st); + + if (!hostname.empty() && !is_valid_ip(hostname)) { + auto* env = ExecEnv::GetInstance(); + auto* dns_cache = (env != nullptr) ? env->dns_cache() : nullptr; + if (dns_cache != nullptr) { + dns_cache->invalidate(hostname); + } + } +} + class FailureDetectClosure : public ::google::protobuf::Closure { public: - FailureDetectClosure(std::shared_ptr<AtomicStatus>& channel_st, + FailureDetectClosure(std::shared_ptr<AtomicStatus>& channel_st, std::string hostname, ::google::protobuf::RpcController* controller, ::google::protobuf::Closure* done) - : _channel_st(channel_st), _controller(controller), _done(done) {} + : _channel_st(channel_st), + _hostname(std::move(hostname)), + _controller(controller), + _done(done) {} void Run() override { Defer defer {[&]() { delete this; }}; // All brpc related API will use brpc::Controller, so that it is safe // to do static cast here. auto* cntl = static_cast<brpc::Controller*>(_controller); - if (cntl->Failed() && cntl->ErrorCode() == EHOSTDOWN) { - Status error_st = Status::NetworkError( - "Failed to send brpc, error={}, error_text={}, client: {}, latency = {}", - berror(cntl->ErrorCode()), cntl->ErrorText(), BackendOptions::get_localhost(), - cntl->latency_us()); - LOG(WARNING) << error_st; - _channel_st->update(error_st); + if (cntl->Failed() && is_brpc_network_fault(cntl->ErrorCode())) { + on_brpc_network_fault(_channel_st, _hostname, cntl); } // Sometimes done == nullptr, for example hand_shake API. if (_done != nullptr) { @@ -95,6 +144,7 @@ public: private: std::shared_ptr<AtomicStatus> _channel_st; + std::string _hostname; ::google::protobuf::RpcController* _controller; ::google::protobuf::Closure* _done; }; @@ -108,6 +158,10 @@ public: FailureDetectChannel() : ::brpc::Channel() { _channel_st = std::make_shared<AtomicStatus>(); // default OK } + // Original FQDN this channel was built for. Used to invalidate DNSCache on + // network faults so the next get() re-resolves instead of returning stale IP. + void set_hostname(std::string hostname) { _hostname = std::move(hostname); } + void CallMethod(const google::protobuf::MethodDescriptor* method, google::protobuf::RpcController* controller, const google::protobuf::Message* request, google::protobuf::Message* response, @@ -116,19 +170,15 @@ public: if (done != nullptr) { // If done == nullptr, then it means the call is sync call, so that should not // gen a failure detect closure for it. Or it will core. - failure_detect_closure = new FailureDetectClosure(_channel_st, controller, done); + failure_detect_closure = + new FailureDetectClosure(_channel_st, _hostname, controller, done); } ::brpc::Channel::CallMethod(method, controller, request, response, failure_detect_closure); // Done == nullptr, it is a sync call, should also deal with the bad channel. if (done == nullptr) { auto* cntl = static_cast<brpc::Controller*>(controller); - if (cntl->Failed() && cntl->ErrorCode() == EHOSTDOWN) { - Status error_st = Status::NetworkError( - "Failed to send brpc, error={}, error_text={}, client: {}, latency = {}", - berror(cntl->ErrorCode()), cntl->ErrorText(), - BackendOptions::get_localhost(), cntl->latency_us()); - LOG(WARNING) << error_st; - _channel_st->update(error_st); + if (cntl->Failed() && is_brpc_network_fault(cntl->ErrorCode())) { + on_brpc_network_fault(_channel_st, _hostname, cntl); } } } @@ -137,6 +187,7 @@ public: private: std::shared_ptr<AtomicStatus> _channel_st; + std::string _hostname; // original FQDN, may be empty for IP-only callers }; template <class T> @@ -166,66 +217,93 @@ public: } std::shared_ptr<T> get_client(const std::string& host, int port) { - std::string realhost = host; - auto dns_cache = ExecEnv::GetInstance()->dns_cache(); - if (dns_cache == nullptr) { - LOG(WARNING) << "DNS cache is not initialized, skipping hostname resolve"; - } else if (!is_valid_ip(host)) { - Status status = dns_cache->get(host, &realhost); - if (!status.ok()) { - LOG(WARNING) << "failed to get ip from host:" << status.to_string(); - return nullptr; - } - } - - // Use original host:port as key (like Java's TNetworkAddress address) - // This allows us to detect IP changes when DNS resolution changes - std::string host_port = fmt::format("{}:{}", host, port); + // Default: read the global config switch. Callers with session context + // (e.g. a RuntimeState query option) should call the 3-arg overload + // below and pass an explicit bool — that's the integration point for a + // future session-level "enable handshake on get_client" variable. + return get_client(host, port, config::enable_brpc_get_client_handshake); + } - std::shared_ptr<T> stub_ptr; - bool need_remove = false; - - auto check_entry = [&](const auto& v) { - const StubEntry<T>& entry = v.second; - // Check if cached IP matches current resolved IP - if (entry.real_ip != realhost) { - // IP changed (DNS resolution changed) - LOG(WARNING) << "Cached ip changed for " << host << ", before ip: " << entry.real_ip - << ", current ip: " << realhost; - need_remove = true; - } else if (!static_cast<FailureDetectChannel*>(entry.stub->channel()) - ->channel_status() - ->ok()) { - // Client is not in normal state, need to recreate - // At this point we cannot judge the progress of reconnecting the underlying channel. - // In the worst case, it may take two minutes. But we can't stand the connection refused - // for two minutes, so rebuild the channel directly. - need_remove = true; - } else { - // Cache hit: IP matches and client is healthy - stub_ptr = entry.stub; + // Explicit-override entry point. Pass `enable_handshake=true` to force the + // handshake-and-retry path (paid only on cache miss / rebuild — warm cache + // hits stay free). Pass `false` to disable. Max attempts is governed by + // config::brpc_get_client_handshake_max_retries (default 3). + std::shared_ptr<T> get_client(const std::string& host, int port, bool enable_handshake) { + const int max_attempts = enable_handshake + ? std::max(1, config::brpc_get_client_handshake_max_retries) + : 1; + std::string host_port; + for (int attempt = 1; attempt <= max_attempts; ++attempt) { + std::string realhost = host; + auto dns_cache = ExecEnv::GetInstance()->dns_cache(); + if (dns_cache == nullptr) { + LOG(WARNING) << "DNS cache is not initialized, skipping hostname resolve"; + } else if (!is_valid_ip(host)) { + Status status = dns_cache->get(host, &realhost); + if (!status.ok()) { + LOG(WARNING) << "failed to get ip from host:" << status.to_string(); + return nullptr; + } + } + // Keep the original hostname as the cache key so a DNS answer + // change can replace the cached channel for the same logical peer. + host_port = fmt::format("{}:{}", host, port); + + std::shared_ptr<T> stub_ptr; + bool need_remove = false; + auto check_entry = [&](const auto& v) { + const StubEntry<T>& entry = v.second; + if (entry.real_ip != realhost) { + LOG(WARNING) << "Cached ip changed for " << host << ", before ip: " + << entry.real_ip << ", current ip: " << realhost; + need_remove = true; + } else if (!static_cast<FailureDetectChannel*>(entry.stub->channel()) + ->channel_status() + ->ok()) { + need_remove = true; + } else { + stub_ptr = entry.stub; + } + }; + if (LIKELY(_stub_map.if_contains(host_port, check_entry))) { + if (stub_ptr != nullptr) { + return stub_ptr; + } + DCHECK(need_remove); + _stub_map.erase(host_port); } - }; - if (LIKELY(_stub_map.if_contains(host_port, check_entry))) { - if (stub_ptr != nullptr) { - return stub_ptr; + const std::string real_host_port = get_host_port(realhost, port); + // Rebuild path. + auto stub = get_new_client_no_cache(real_host_port, "", "", "", host); + if (stub == nullptr) { + LOG(WARNING) << "failed to build brpc stub to " << real_host_port + << ", attempt=" << attempt << "/" << max_attempts; + if (enable_handshake) { + continue; + } + return nullptr; } - // IP changed or client unhealthy, need to remove old entry - if (need_remove) { - _stub_map.erase(host_port); + if (enable_handshake && !available(stub, real_host_port)) { + LOG(WARNING) << "handshake failed to " << real_host_port + << ", attempt=" << attempt << "/" << max_attempts; + // The rebuilt stub talked to a dead peer (or stale IP). Mark + // DNS dirty so the next iteration re-resolves via getaddrinfo, + // and drop the freshly-built stub. + if (dns_cache != nullptr && !is_valid_ip(host)) { + dns_cache->invalidate(host); + } + continue; } - } - // Create new stub using resolved IP for actual connection - std::string real_host_port = get_host_port(realhost, port); - auto stub = get_new_client_no_cache(real_host_port); - if (stub != nullptr) { StubEntry<T> entry {realhost, stub}; _stub_map.try_emplace_l( host_port, [&stub](const auto& v) { stub = v.second.stub; }, entry); + return stub; } - return stub; + LOG(WARNING) << "get_client gave up after " << max_attempts + << " handshake attempts to " << host_port; + return nullptr; } std::shared_ptr<T> get_client(const std::string& host_port) { @@ -244,7 +322,8 @@ public: std::shared_ptr<T> get_new_client_no_cache(const std::string& host_port, const std::string& protocol = "", const std::string& connection_type = "", - const std::string& connection_group = "") { + const std::string& connection_group = "", + const std::string& original_hostname = "") { brpc::ChannelOptions options; Status status = doris::client::configure_brpc_channel_options(&options); if (!status.ok()) { @@ -272,6 +351,9 @@ public: options.max_retry = 10; std::unique_ptr<FailureDetectChannel> channel(new FailureDetectChannel()); + if (!original_hostname.empty()) { + channel->set_hostname(original_hostname); + } int ret_code = 0; if (host_port.find("://") == std::string::npos) { ret_code = channel->Init(host_port.c_str(), &options); diff --git a/be/src/util/dns_cache.cpp b/be/src/util/dns_cache.cpp index 0ea794872c1..b3912b96306 100644 --- a/be/src/util/dns_cache.cpp +++ b/be/src/util/dns_cache.cpp @@ -37,7 +37,8 @@ Status DNSCache::get(const std::string& hostname, std::string* ip) { { std::shared_lock<std::shared_mutex> lock(mutex); auto it = cache.find(hostname); - if (it != cache.end()) { + // dirty hostnames must NOT take the fast path: force re-resolve below. + if (it != cache.end() && dirty.find(hostname) == dirty.end()) { *ip = it->second; return Status::OK(); } @@ -84,6 +85,16 @@ std::string DNSCache::_resolve_hostname(const std::string& hostname) { return resolved_ip; } +void DNSCache::invalidate(const std::string& hostname) { + if (hostname.empty()) { + return; + } + std::unique_lock<std::shared_mutex> lock(mutex); + if (dirty.insert(hostname).second) { + LOG(INFO) << "DNSCache: mark hostname dirty, will re-resolve on next get: " << hostname; + } +} + Status DNSCache::_update(const std::string& hostname) { std::string real_ip = _resolve_hostname(hostname); if (real_ip.empty()) { @@ -97,6 +108,7 @@ Status DNSCache::_update(const std::string& hostname) { cache[hostname] = real_ip; LOG(INFO) << "update hostname " << hostname << "'s ip to " << real_ip; } + dirty.erase(hostname); // here myabe need lock all update function. return Status::OK(); } diff --git a/be/src/util/dns_cache.h b/be/src/util/dns_cache.h index 51ffb6567ec..76b877bdf20 100644 --- a/be/src/util/dns_cache.h +++ b/be/src/util/dns_cache.h @@ -23,6 +23,7 @@ #include <string> #include <thread> #include <unordered_map> +#include <unordered_set> #include "common/status.h" @@ -38,6 +39,14 @@ public: // get ip by hostname Status get(const std::string& hostname, std::string* ip); + // Synchronously mark `hostname` as dirty so that the next get() is + // guaranteed to re-resolve via _update() instead of returning the cached + // (potentially stale) IP. Idempotent under failure storms: multiple + // invalidate() calls for the same hostname are O(1) and harmless. + // Never blocks on getaddrinfo(); the actual re-resolution happens lazily + // inside the next get() (which is the cache-miss path anyway). + void invalidate(const std::string& hostname); + private: // Resolve hostname to IP address. // If resolution fails, falls back to cached IP if available. @@ -54,6 +63,8 @@ private: private: // hostname -> ip std::unordered_map<std::string, std::string> cache; + // hostnames marked dirty by invalidate(); next get() must re-resolve. + std::unordered_set<std::string> dirty; mutable std::shared_mutex mutex; std::thread refresh_thread; bool stop_refresh = false; diff --git a/be/src/util/proto_util.h b/be/src/util/proto_util.h index aec345688a9..3e89746a200 100644 --- a/be/src/util/proto_util.h +++ b/be/src/util/proto_util.h @@ -87,7 +87,8 @@ Status transmit_block_httpv2(ExecEnv* exec_env, std::unique_ptr<Closure> closure std::string brpc_url = get_brpc_http_url(brpc_dest_addr.hostname, brpc_dest_addr.port); std::shared_ptr<PBackendService_Stub> brpc_http_stub = - exec_env->brpc_internal_client_cache()->get_new_client_no_cache(brpc_url, "http"); + exec_env->brpc_internal_client_cache()->get_new_client_no_cache( + brpc_url, "http", "", "", brpc_dest_addr.hostname); if (brpc_http_stub == nullptr) { return Status::InternalError("failed to open brpc http client to {}", brpc_url); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
