Envoy 原始碼分析－－LDS
阿新 • • 發佈：2019-07-25
Envoy 原始碼分析－－LDS
LDS 是 Envoy 用來自動獲取 listener 的 API。 Envoy 通過 API 可以增加、修改或刪除 listener。
先來總結下 listener 的更新語義如下：
- 每個 listener 必須有一個唯一的名稱。如果沒有提供名稱，Envoy 會生成一個 UUID 來作為它的名字。要動態更新 listener，管理服務必須提供一個唯一名稱。
- 當 listener 被新增，在接收流量之前，會先進入 “預熱” 階段。
- 一旦 listener 被建立，就會保持不變。因此，listener 更新時，會建立一個全新的 listener（同一個偵聽套接字）。這個新增加的 listener 同樣需要一個 “預熱” 過程。
- 當更新或刪除 listener 時，舊的 listener 將被置於 “draining（驅逐）” 狀態，和整個服務重新啟動時一樣。在刪除偵聽器並關閉任何其餘連線之前，偵聽器擁有的連線將在一段時間內優雅關閉（如果可能的話）。逐出時間通過
--drain-time-s設定。 - 相同名稱的 listener 必須要有相同的配置地址。
接下來是對 lds 進行原始碼分析，對各種語義的情況都可以在原始碼中看到。
初始化
在初始化 bootstrap 配置時，如果有 lds 配置會進行初始化。
// Instruct the listener manager to create the LDS provider if needed. This must be done later // because various items do not yet exist when the listener manager is created. if (bootstrap_.dynamic_resources().has_lds_config()) { listener_manager_->createLdsApi(bootstrap_.dynamic_resources().lds_config()); }
在 ListenerManagerImpl 類中通過 ProdListenerComponentFactory 配置類建立 lds。
// Server::ListenerManager void createLdsApi(const envoy::api::v2::core::ConfigSource& lds_config) override { ASSERT(lds_api_ == nullptr); lds_api_ = factory_.createLdsApi(lds_config); } // Server::ListenerComponentFactory LdsApiPtr createLdsApi(const envoy::api::v2::core::ConfigSource& lds_config) override { return std::make_unique<LdsApiImpl>(lds_config, server_.clusterManager(), server_.dispatcher(), server_.random(), server_.initManager(), server_.localInfo(), server_.stats(), server_.listenerManager(), server_.api()); }
新建 lds 時，會建立一個通道。
LdsApiImpl::LdsApiImpl(const envoy::api::v2::core::ConfigSource& lds_config,
Upstream::ClusterManager& cm, Event::Dispatcher& dispatcher,
Runtime::RandomGenerator& random, Init::Manager& init_manager,
const LocalInfo::LocalInfo& local_info, Stats::Scope& scope,
ListenerManager& lm, Api::Api& api)
: listener_manager_(lm), scope_(scope.createScope("listener_manager.lds.")), cm_(cm),
init_target_("LDS", [this]() { subscription_->start({}, *this); }) {
//建立 subscription 物件
subscription_ = Envoy::Config::SubscriptionFactory::subscriptionFromConfigSource(
lds_config, local_info, dispatcher, cm, random, *scope_,
"envoy.api.v2.ListenerDiscoveryService.FetchListeners",
"envoy.api.v2.ListenerDiscoveryService.StreamListeners",
Grpc::Common::typeUrl(envoy::api::v2::Listener().GetDescriptor()->full_name()), api);
Config::Utility::checkLocalInfo("lds", local_info);
init_manager.add(init_target_);
}更新
有資料下發時，通道呼叫 callback 觸發 onConfigUpdate。
// the configuration update targets.
callbacks_->onConfigUpdate(resources, version_info);配置更新過程中如何判斷哪些是移除的 listener？。首先建立一個移除的列表，將現有的 listener 加入，然後去除資料下發的 listener，剩下的就是需要移除的 listener。移除的 listener 會列印 lds: remove listener '{}'。
// We build the list of listeners to be removed and remove them before
// adding new listeners. This allows adding a new listener with the same
// address as a listener that is to be removed. Do not change the order.
for (const auto& listener : listener_manager_.listeners()) {
listeners_to_remove.emplace(listener.get().name(), listener);
}
for (const auto& listener : listeners) {
listeners_to_remove.erase(listener.name());
}
for (const auto& listener : listeners_to_remove) {
if (listener_manager_.removeListener(listener.first)) {
ENVOY_LOG(info, "lds: remove listener '{}'", listener.first);
}
}移除 listener 時，如果是在 “預熱” 階段的 listener 直接刪除，如果是在活動的 listener 將其置於驅逐中，過一段時間關閉。
// Destroy a warming listener directly.
if (existing_warming_listener != warming_listeners_.end()) {
(*existing_warming_listener)->debugLog("removing warming listener");
warming_listeners_.erase(existing_warming_listener);
}
// If there is an active listener it needs to be moved to draining.
if (existing_active_listener != active_listeners_.end()) {
drainListener(std::move(*existing_active_listener));
active_listeners_.erase(existing_active_listener);
}
// ListenerManagerImpl::drainListener
// 關閉accept，不再接收新連線
draining_it->listener_->debugLog("draining listener");
for (const auto& worker : workers_) {
worker->stopListener(*draining_it->listener_);
}
// 等待一段時間
draining_it->listener_->localDrainManager().startDrainSequence([this, draining_it]() -> void {
draining_it->listener_->debugLog("removing listener");
for (const auto& worker : workers_) {
// 移除 listener
worker->removeListener(*draining_it->listener_, [this, draining_it]() -> void {
// 通知主執行緒（移除是在工作執行緒移除的，主執行緒無法知道）
server_.dispatcher().post([this, draining_it]() -> void {
if (--draining_it->workers_pending_removal_ == 0) {
draining_it->listener_->debugLog("listener removal complete");
draining_listeners_.erase(draining_it);
stats_.total_listeners_draining_.set(draining_listeners_.size());
}
});
});
}
});接下來是對配置內的 listener 進行增加或更新。增加或更新 listener 列印日誌 lds: add/update listener '{}'。
for (const auto& listener : listeners) {
const std::string& listener_name = listener.name();
try {
if (listener_manager_.addOrUpdateListener(listener, version_info, true)) {
ENVOY_LOG(info, "lds: add/update listener '{}'", listener_name);
} else {
ENVOY_LOG(debug, "lds: add/update listener '{}' skipped", listener_name);
}
} catch (const EnvoyException& e) {
exception_msgs.push_back(fmt::format("{}: {}", listener_name, e.what()));
}
}增加 listener 時，如果沒有名稱，Envoy 生成一個 UUID。更新 listener 如果沒有名稱，Envoy 生成 UUID，無法通過名稱進行關聯，因此不帶名稱進行更新，只會變成新增。
std::string name;
if (!config.name().empty()) {
name = config.name();
} else {
name = server_.random().uuid();
}不論是增加還是更新 listener 都是直接新建，舊的 listener 會被 “draining（驅逐）” 。
ListenerImplPtr new_listener(
new ListenerImpl(config, version_info, *this, name, modifiable, workers_started_, hash));
ListenerImpl& new_listener_ref = *new_listener;判斷相同名稱的 listener 必須有相同的配置地址。
if ((existing_warming_listener != warming_listeners_.end() &&
*(*existing_warming_listener)->address() != *new_listener->address()) ||
(existing_active_listener != active_listeners_.end() &&
*(*existing_active_listener)->address() != *new_listener->address())) {
const std::string message = fmt::format(
"error updating listener: '{}' has a different address '{}' from existing listener", name,
new_listener->address()->asString());
ENVOY_LOG(warn, "{}", message);
throw EnvoyException(message);
}如果更新的 listener 在 “預熱” 階段的 listener 中，直接內部替換。
if (existing_warming_listener != warming_listeners_.end()) {
ASSERT(workers_started_);
new_listener->debugLog("update warming listener");
new_listener->setSocket((*existing_warming_listener)->getSocket());
*existing_warming_listener = std::move(new_listener);
}如果更新的 listener 在活動中的 listener 中。當前工作執行緒已經啟動的話，加入 “預熱” 階段的 listener，同時後繼會把舊的 listener 置於 “draining（驅逐）” 狀態。當前工作執行緒未啟動狀態，直接內部替換。
if (existing_active_listener != active_listeners_.end()) {
new_listener->setSocket((*existing_active_listener)->getSocket());
if (workers_started_) {
new_listener->debugLog("add warming listener");
warming_listeners_.emplace_back(std::move(new_listener));
} else {
new_listener->debugLog("update active listener");
*existing_active_listener = std::move(new_listener);
}
}新增 listener 時，會先去 “draining（驅逐）” 狀態的 listener 中查詢是否已有相同地址的 listener，有相同的重新拉回來，並根據當前工作執行緒是否已就緒加入對應的 listener。
Network::SocketSharedPtr draining_listener_socket;
auto existing_draining_listener = std::find_if(
draining_listeners_.cbegin(), draining_listeners_.cend(),
[&new_listener](const DrainingListener& listener) {
return *new_listener->address() == *listener.listener_->socket().localAddress();
});
if (existing_draining_listener != draining_listeners_.cend()) {
draining_listener_socket = existing_draining_listener->listener_->getSocket();
}
new_listener->setSocket(draining_listener_socket
? draining_listener_socket
: factory_.createListenSocket(new_listener->address(),
new_listener->socketType(),
new_listener->listenSocketOptions(),
new_listener->bindToPort()));
if (workers_started_) {
new_listener->debugLog("add warming listener");
warming_listeners_.emplace_back(std::move(new_listener));
} else {
new_listener->debugLog("add active listener");
active_listeners_.emplace_back(std::move(new_listener));
}最後把舊的 listener 置於 “draining（驅逐）” 狀態。
// 呼叫 initialize()
new_listener_ref.initialize();
void ListenerImpl::initialize() {
last_updated_ = timeSource().systemTime();
if (workers_started_) {
// init_watcher
dynamic_init_manager_.initialize(*init_watcher_);
}
}
// init_watcher 在建構函式時建立
init_watcher_(std::make_unique<Init::WatcherImpl>(
"ListenerImpl", [this] { parent_.onListenerWarmed(*this); })),
void ListenerManagerImpl::onListenerWarmed(ListenerImpl& listener) {
auto existing_active_listener = getListenerByName(active_listeners_, listener.name());
auto existing_warming_listener = getListenerByName(warming_listeners_, listener.name());
(*existing_warming_listener)->debugLog("warm complete. updating active listener");
if (existing_active_listener != active_listeners_.end()) {
// 舊的 listener 置於 “draining（驅逐）” 狀態。
drainListener(std::move(*existing_active_listener));
// 預熱的 listener 放到活動的 listener
*existing_active_listener = std::move(*existing_warming_listener);
} else {
active_listeners_.emplace_back(std::move(*existing_warming_listener));
}
}