Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Don't NACK empty updates.

Browse Source
pull/22293/head
Mark D. Roth 5 years ago
parent d8623ba548
commit 2520a81a66
4 changed files with 134 additions and 83 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 60
      src/core/ext/filters/client_channel/xds/xds_api.cc
  2. 8
      src/core/ext/filters/client_channel/xds/xds_api.h
  3. 119
      src/core/ext/filters/client_channel/xds/xds_client.cc
  4. 8
      src/core/ext/filters/client_channel/xds/xds_client.h

60
src/core/ext/filters/client_channel/xds/xds_api.cc
Unescape View File

@ -542,15 +542,12 @@ grpc_error* RouteConfigParse(
grpc_error* LdsResponseParse(const envoy_api_v2_DiscoveryResponse* response, grpc_error* LdsResponseParse(const envoy_api_v2_DiscoveryResponse* response,
const std::string& expected_server_name, const std::string& expected_server_name,
XdsApi::LdsUpdate* lds_update, upb_arena* arena) { absl::optional<XdsApi::LdsUpdate>* lds_update,
upb_arena* arena) {
// Get the resources from the response. // Get the resources from the response.
size_t size; size_t size;
const google_protobuf_Any* const* resources = const google_protobuf_Any* const* resources =
envoy_api_v2_DiscoveryResponse_resources(response, &size); envoy_api_v2_DiscoveryResponse_resources(response, &size);
if (size < 1) {
return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"LDS response contains 0 resource.");
}
for (size_t i = 0; i < size; ++i) { for (size_t i = 0; i < size; ++i) {
// Check the type_url of the resource. // Check the type_url of the resource.
const upb_strview type_url = google_protobuf_Any_type_url(resources[i]); const upb_strview type_url = google_protobuf_Any_type_url(resources[i]);
@ -593,10 +590,11 @@ grpc_error* LdsResponseParse(const envoy_api_v2_DiscoveryResponse* response,
grpc_error* error = grpc_error* error =
RouteConfigParse(route_config, expected_server_name, &rds_update); RouteConfigParse(route_config, expected_server_name, &rds_update);
if (error != GRPC_ERROR_NONE) return error; if (error != GRPC_ERROR_NONE) return error;
lds_update->rds_update.emplace(std::move(rds_update)); lds_update->emplace();
(*lds_update)->rds_update.emplace(std::move(rds_update));
const upb_strview route_config_name = const upb_strview route_config_name =
envoy_api_v2_RouteConfiguration_name(route_config); envoy_api_v2_RouteConfiguration_name(route_config);
lds_update->route_config_name = (*lds_update)->route_config_name =
std::string(route_config_name.data, route_config_name.size); std::string(route_config_name.data, route_config_name.size);
return GRPC_ERROR_NONE; return GRPC_ERROR_NONE;
} }
@ -613,26 +611,23 @@ grpc_error* LdsResponseParse(const envoy_api_v2_DiscoveryResponse* response,
const upb_strview route_config_name = const upb_strview route_config_name =
envoy_config_filter_network_http_connection_manager_v2_Rds_route_config_name( envoy_config_filter_network_http_connection_manager_v2_Rds_route_config_name(
rds); rds);
lds_update->route_config_name = lds_update->emplace();
(*lds_update)->route_config_name =
std::string(route_config_name.data, route_config_name.size); std::string(route_config_name.data, route_config_name.size);
return GRPC_ERROR_NONE; return GRPC_ERROR_NONE;
} }
return GRPC_ERROR_CREATE_FROM_STATIC_STRING( return GRPC_ERROR_NONE;
"No listener found for expected server name.");
} }
grpc_error* RdsResponseParse(const envoy_api_v2_DiscoveryResponse* response, grpc_error* RdsResponseParse(const envoy_api_v2_DiscoveryResponse* response,
const std::string& expected_server_name, const std::string& expected_server_name,
const std::string& expected_route_config_name, const std::string& expected_route_config_name,
XdsApi::RdsUpdate* rds_update, upb_arena* arena) { absl::optional<XdsApi::RdsUpdate>* rds_update,
upb_arena* arena) {
// Get the resources from the response. // Get the resources from the response.
size_t size; size_t size;
const google_protobuf_Any* const* resources = const google_protobuf_Any* const* resources =
envoy_api_v2_DiscoveryResponse_resources(response, &size); envoy_api_v2_DiscoveryResponse_resources(response, &size);
if (size < 1) {
return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"RDS response contains 0 resource.");
}
for (size_t i = 0; i < size; ++i) { for (size_t i = 0; i < size; ++i) {
// Check the type_url of the resource. // Check the type_url of the resource.
const upb_strview type_url = google_protobuf_Any_type_url(resources[i]); const upb_strview type_url = google_protobuf_Any_type_url(resources[i]);
@ -658,24 +653,20 @@ grpc_error* RdsResponseParse(const envoy_api_v2_DiscoveryResponse* response,
grpc_error* error = grpc_error* error =
RouteConfigParse(route_config, expected_server_name, &local_rds_update); RouteConfigParse(route_config, expected_server_name, &local_rds_update);
if (error != GRPC_ERROR_NONE) return error; if (error != GRPC_ERROR_NONE) return error;
*rds_update = std::move(local_rds_update); rds_update->emplace(std::move(local_rds_update));
return GRPC_ERROR_NONE; return GRPC_ERROR_NONE;
} }
return GRPC_ERROR_CREATE_FROM_STATIC_STRING( return GRPC_ERROR_NONE;
"No route config found for expected name.");
} }
grpc_error* CdsResponseParse(const envoy_api_v2_DiscoveryResponse* response, grpc_error* CdsResponseParse(
XdsApi::CdsUpdateMap* cds_update_map, const envoy_api_v2_DiscoveryResponse* response,
upb_arena* arena) { const std::set<StringView>& expected_cluster_names,
XdsApi::CdsUpdateMap* cds_update_map, upb_arena* arena) {
// Get the resources from the response. // Get the resources from the response.
size_t size; size_t size;
const google_protobuf_Any* const* resources = const google_protobuf_Any* const* resources =
envoy_api_v2_DiscoveryResponse_resources(response, &size); envoy_api_v2_DiscoveryResponse_resources(response, &size);
if (size < 1) {
return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"CDS response contains 0 resource.");
}
// Parse all the resources in the CDS response. // Parse all the resources in the CDS response.
for (size_t i = 0; i < size; ++i) { for (size_t i = 0; i < size; ++i) {
XdsApi::CdsUpdate cds_update; XdsApi::CdsUpdate cds_update;
@ -691,6 +682,13 @@ grpc_error* CdsResponseParse(const envoy_api_v2_DiscoveryResponse* response,
if (cluster == nullptr) { if (cluster == nullptr) {
return GRPC_ERROR_CREATE_FROM_STATIC_STRING("Can't decode cluster."); return GRPC_ERROR_CREATE_FROM_STATIC_STRING("Can't decode cluster.");
} }
// Ignore unexpected cluster names.
upb_strview cluster_name = envoy_api_v2_Cluster_name(cluster);
StringView cluster_name_strview(cluster_name.data, cluster_name.size);
if (expected_cluster_names.find(cluster_name_strview) ==
expected_cluster_names.end()) {
continue;
}
// Check the cluster_discovery_type. // Check the cluster_discovery_type.
if (!envoy_api_v2_Cluster_has_type(cluster)) { if (!envoy_api_v2_Cluster_has_type(cluster)) {
return GRPC_ERROR_CREATE_FROM_STATIC_STRING("DiscoveryType not found."); return GRPC_ERROR_CREATE_FROM_STATIC_STRING("DiscoveryType not found.");
@ -729,7 +727,6 @@ grpc_error* CdsResponseParse(const envoy_api_v2_DiscoveryResponse* response,
} }
cds_update.lrs_load_reporting_server_name.emplace(""); cds_update.lrs_load_reporting_server_name.emplace("");
} }
upb_strview cluster_name = envoy_api_v2_Cluster_name(cluster);
cds_update_map->emplace(std::string(cluster_name.data, cluster_name.size), cds_update_map->emplace(std::string(cluster_name.data, cluster_name.size),
std::move(cds_update)); std::move(cds_update));
} }
@ -856,10 +853,6 @@ grpc_error* EdsResponsedParse(
size_t size; size_t size;
const google_protobuf_Any* const* resources = const google_protobuf_Any* const* resources =
envoy_api_v2_DiscoveryResponse_resources(response, &size); envoy_api_v2_DiscoveryResponse_resources(response, &size);
if (size < 1) {
return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"EDS response contains 0 resource.");
}
for (size_t i = 0; i < size; ++i) { for (size_t i = 0; i < size; ++i) {
XdsApi::EdsUpdate eds_update; XdsApi::EdsUpdate eds_update;
// Check the type_url of the resource. // Check the type_url of the resource.
@ -934,8 +927,10 @@ grpc_error* EdsResponsedParse(
grpc_error* XdsApi::ParseAdsResponse( grpc_error* XdsApi::ParseAdsResponse(
const grpc_slice& encoded_response, const std::string& expected_server_name, const grpc_slice& encoded_response, const std::string& expected_server_name,
const std::string& expected_route_config_name, const std::string& expected_route_config_name,
const std::set<StringView>& expected_cluster_names,
const std::set<StringView>& expected_eds_service_names, const std::set<StringView>& expected_eds_service_names,
LdsUpdate* lds_update, RdsUpdate* rds_update, CdsUpdateMap* cds_update_map, absl::optional<LdsUpdate>* lds_update,
absl::optional<RdsUpdate>* rds_update, CdsUpdateMap* cds_update_map,
EdsUpdateMap* eds_update_map, std::string* version, std::string* nonce, EdsUpdateMap* eds_update_map, std::string* version, std::string* nonce,
std::string* type_url) { std::string* type_url) {
upb::Arena arena; upb::Arena arena;
@ -968,7 +963,8 @@ grpc_error* XdsApi::ParseAdsResponse(
expected_route_config_name, rds_update, expected_route_config_name, rds_update,
arena.ptr()); arena.ptr());
} else if (*type_url == kCdsTypeUrl) { } else if (*type_url == kCdsTypeUrl) {
return CdsResponseParse(response, cds_update_map, arena.ptr()); return CdsResponseParse(response, expected_cluster_names, cds_update_map,
arena.ptr());
} else if (*type_url == kEdsTypeUrl) { } else if (*type_url == kEdsTypeUrl) {
return EdsResponsedParse(response, expected_eds_service_names, return EdsResponsedParse(response, expected_eds_service_names,
eds_update_map, arena.ptr()); eds_update_map, arena.ptr());

8
src/core/ext/filters/client_channel/xds/xds_api.h
Unescape View File

@ -230,10 +230,12 @@ class XdsApi {
const grpc_slice& encoded_response, const grpc_slice& encoded_response,
const std::string& expected_server_name, const std::string& expected_server_name,
const std::string& expected_route_config_name, const std::string& expected_route_config_name,
const std::set<StringView>& expected_cluster_names,
const std::set<StringView>& expected_eds_service_names, const std::set<StringView>& expected_eds_service_names,
LdsUpdate* lds_update, RdsUpdate* rds_update, absl::optional<LdsUpdate>* lds_update,
CdsUpdateMap* cds_update_map, EdsUpdateMap* eds_update_map, absl::optional<RdsUpdate>* rds_update, CdsUpdateMap* cds_update_map,
std::string* version, std::string* nonce, std::string* type_url); EdsUpdateMap* eds_update_map, std::string* version, std::string* nonce,
std::string* type_url);
// Creates an LRS request querying \a server_name. // Creates an LRS request querying \a server_name.
grpc_slice CreateLrsInitialRequest(const std::string& server_name); grpc_slice CreateLrsInitialRequest(const std::string& server_name);

119
src/core/ext/filters/client_channel/xds/xds_client.cc
Unescape View File

@ -238,8 +238,8 @@ class XdsClient::ChannelState::AdsCallState
void SendMessageLocked(const std::string& type_url); void SendMessageLocked(const std::string& type_url);
void AcceptLdsUpdate(XdsApi::LdsUpdate lds_update); void AcceptLdsUpdate(absl::optional<XdsApi::LdsUpdate> lds_update);
void AcceptRdsUpdate(XdsApi::RdsUpdate rds_update); void AcceptRdsUpdate(absl::optional<XdsApi::RdsUpdate> rds_update);
void AcceptCdsUpdate(XdsApi::CdsUpdateMap cds_update_map); void AcceptCdsUpdate(XdsApi::CdsUpdateMap cds_update_map);
void AcceptEdsUpdate(XdsApi::EdsUpdateMap eds_update_map); void AcceptEdsUpdate(XdsApi::EdsUpdateMap eds_update_map);
@ -860,24 +860,32 @@ bool XdsClient::ChannelState::AdsCallState::HasSubscribedResources() const {
} }
void XdsClient::ChannelState::AdsCallState::AcceptLdsUpdate( void XdsClient::ChannelState::AdsCallState::AcceptLdsUpdate(
XdsApi::LdsUpdate lds_update) { absl::optional<XdsApi::LdsUpdate> lds_update) {
if (!lds_update.has_value()) {
gpr_log(GPR_INFO,
"[xds_client %p] LDS update does not include requested resource",
xds_client());
xds_client()->service_config_watcher_->OnError(
GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"LDS update does not include requested resource"));
return;
}
const std::string& cluster_name = const std::string& cluster_name =
lds_update.rds_update.has_value() lds_update->rds_update.has_value()
? lds_update.rds_update.value().cluster_name ? lds_update->rds_update.value().cluster_name
: ""; : "";
if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
"[xds_client %p] LDS update received: " "[xds_client %p] LDS update received: route_config_name=%s, "
"route_config_name=%s, "
"cluster_name=%s (empty if RDS is needed to obtain it)", "cluster_name=%s (empty if RDS is needed to obtain it)",
xds_client(), lds_update.route_config_name.c_str(), xds_client(), lds_update->route_config_name.c_str(),
cluster_name.c_str()); cluster_name.c_str());
} }
auto& lds_state = state_map_[XdsApi::kLdsTypeUrl]; auto& lds_state = state_map_[XdsApi::kLdsTypeUrl];
auto& state = lds_state.subscribed_resources[xds_client()->server_name_]; auto& state = lds_state.subscribed_resources[xds_client()->server_name_];
if (state != nullptr) state->Finish(); if (state != nullptr) state->Finish();
// Ignore identical update. // Ignore identical update.
if (xds_client()->route_config_name_ == lds_update.route_config_name && if (xds_client()->route_config_name_ == lds_update->route_config_name &&
xds_client()->cluster_name_ == cluster_name) { xds_client()->cluster_name_ == cluster_name) {
if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
@ -886,12 +894,15 @@ void XdsClient::ChannelState::AdsCallState::AcceptLdsUpdate(
} }
return; return;
} }
xds_client()->route_config_name_ = std::move(lds_update.route_config_name); if (!xds_client()->route_config_name_.empty()) {
if (lds_update.rds_update.has_value()) { Unsubscribe(XdsApi::kRdsTypeUrl, xds_client()->route_config_name_);
}
xds_client()->route_config_name_ = std::move(lds_update->route_config_name);
if (lds_update->rds_update.has_value()) {
// If cluster_name was found inlined in LDS response, notify the watcher // If cluster_name was found inlined in LDS response, notify the watcher
// immediately. // immediately.
xds_client()->cluster_name_ = xds_client()->cluster_name_ =
std::move(lds_update.rds_update.value().cluster_name); std::move(lds_update->rds_update.value().cluster_name);
RefCountedPtr<ServiceConfig> service_config; RefCountedPtr<ServiceConfig> service_config;
grpc_error* error = xds_client()->CreateServiceConfig( grpc_error* error = xds_client()->CreateServiceConfig(
xds_client()->cluster_name_, &service_config); xds_client()->cluster_name_, &service_config);
@ -908,19 +919,27 @@ void XdsClient::ChannelState::AdsCallState::AcceptLdsUpdate(
} }
void XdsClient::ChannelState::AdsCallState::AcceptRdsUpdate( void XdsClient::ChannelState::AdsCallState::AcceptRdsUpdate(
XdsApi::RdsUpdate rds_update) { absl::optional<XdsApi::RdsUpdate> rds_update) {
if (!rds_update.has_value()) {
gpr_log(GPR_INFO,
"[xds_client %p] RDS update does not include requested resource",
xds_client());
xds_client()->service_config_watcher_->OnError(
GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"RDS update does not include requested resource"));
return;
}
if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
"[xds_client %p] RDS update received: " "[xds_client %p] RDS update received: cluster_name=%s",
"cluster_name=%s", xds_client(), rds_update->cluster_name.c_str());
xds_client(), rds_update.cluster_name.c_str());
} }
auto& rds_state = state_map_[XdsApi::kRdsTypeUrl]; auto& rds_state = state_map_[XdsApi::kRdsTypeUrl];
auto& state = auto& state =
rds_state.subscribed_resources[xds_client()->route_config_name_]; rds_state.subscribed_resources[xds_client()->route_config_name_];
if (state != nullptr) state->Finish(); if (state != nullptr) state->Finish();
// Ignore identical update. // Ignore identical update.
if (xds_client()->cluster_name_ == rds_update.cluster_name) { if (xds_client()->cluster_name_ == rds_update->cluster_name) {
if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
"[xds_client %p] RDS update identical to current, ignoring.", "[xds_client %p] RDS update identical to current, ignoring.",
@ -928,7 +947,7 @@ void XdsClient::ChannelState::AdsCallState::AcceptRdsUpdate(
} }
return; return;
} }
xds_client()->cluster_name_ = std::move(rds_update.cluster_name); xds_client()->cluster_name_ = std::move(rds_update->cluster_name);
// Notify the watcher. // Notify the watcher.
RefCountedPtr<ServiceConfig> service_config; RefCountedPtr<ServiceConfig> service_config;
grpc_error* error = xds_client()->CreateServiceConfig( grpc_error* error = xds_client()->CreateServiceConfig(
@ -944,6 +963,7 @@ void XdsClient::ChannelState::AdsCallState::AcceptRdsUpdate(
void XdsClient::ChannelState::AdsCallState::AcceptCdsUpdate( void XdsClient::ChannelState::AdsCallState::AcceptCdsUpdate(
XdsApi::CdsUpdateMap cds_update_map) { XdsApi::CdsUpdateMap cds_update_map) {
auto& cds_state = state_map_[XdsApi::kCdsTypeUrl]; auto& cds_state = state_map_[XdsApi::kCdsTypeUrl];
std::set<std::string> eds_resource_names_seen;
for (auto& p : cds_update_map) { for (auto& p : cds_update_map) {
const char* cluster_name = p.first.c_str(); const char* cluster_name = p.first.c_str();
XdsApi::CdsUpdate& cds_update = p.second; XdsApi::CdsUpdate& cds_update = p.second;
@ -952,21 +972,22 @@ void XdsClient::ChannelState::AdsCallState::AcceptCdsUpdate(
if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
"[xds_client %p] CDS update (cluster=%s) received: " "[xds_client %p] CDS update (cluster=%s) received: "
"eds_service_name=%s, " "eds_service_name=%s, lrs_load_reporting_server_name=%s",
"lrs_load_reporting_server_name=%s",
xds_client(), cluster_name, cds_update.eds_service_name.c_str(), xds_client(), cluster_name, cds_update.eds_service_name.c_str(),
cds_update.lrs_load_reporting_server_name.has_value() cds_update.lrs_load_reporting_server_name.has_value()
? cds_update.lrs_load_reporting_server_name.value().c_str() ? cds_update.lrs_load_reporting_server_name.value().c_str()
: "(N/A)"); : "(N/A)");
} }
ClusterState& cluster_state = xds_client()->cluster_map_[cluster_name]; // Record the EDS resource names seen.
eds_resource_names_seen.insert(cds_update.eds_service_name.empty()
? cluster_name
: cds_update.eds_service_name);
// Ignore identical update. // Ignore identical update.
ClusterState& cluster_state = xds_client()->cluster_map_[cluster_name];
if (cluster_state.update.has_value() && if (cluster_state.update.has_value() &&
cds_update.eds_service_name == cds_update.eds_service_name == cluster_state.update->eds_service_name &&
cluster_state.update.value().eds_service_name && cds_update.lrs_load_reporting_server_name ==
cds_update.lrs_load_reporting_server_name.value() == cluster_state.update->lrs_load_reporting_server_name) {
cluster_state.update.value()
.lrs_load_reporting_server_name.value()) {
if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_client_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
"[xds_client %p] CDS update identical to current, ignoring.", "[xds_client %p] CDS update identical to current, ignoring.",
@ -975,12 +996,41 @@ void XdsClient::ChannelState::AdsCallState::AcceptCdsUpdate(
continue; continue;
} }
// Update the cluster state. // Update the cluster state.
cluster_state.update.emplace(std::move(cds_update)); cluster_state.update = std::move(cds_update);
// Notify all watchers. // Notify all watchers.
for (const auto& p : cluster_state.watchers) { for (const auto& p : cluster_state.watchers) {
p.first->OnClusterChanged(cluster_state.update.value()); p.first->OnClusterChanged(cluster_state.update.value());
} }
} }
// For any subscribed resource that is not present in the update,
// remove it from the cache and notify watchers of the error.
for (const auto& p : cds_state.subscribed_resources) {
const std::string& cluster_name = p.first;
if (cds_update_map.find(cluster_name) == cds_update_map.end()) {
ClusterState& cluster_state = xds_client()->cluster_map_[cluster_name];
cluster_state.update.reset();
for (const auto& p : cluster_state.watchers) {
p.first->OnError(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"Cluster not present in CDS update"));
}
}
}
// Also remove any EDS resources that are no longer referred to by any CDS
// resources.
auto& eds_state = state_map_[XdsApi::kEdsTypeUrl];
for (const auto& p : eds_state.subscribed_resources) {
const std::string& eds_resource_name = p.first;
if (eds_resource_names_seen.find(eds_resource_name) ==
eds_resource_names_seen.end()) {
EndpointState& endpoint_state =
xds_client()->endpoint_map_[eds_resource_name];
endpoint_state.update.reset();
for (const auto& p : endpoint_state.watchers) {
p.first->OnError(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"ClusterLoadAssignment resource removed due to CDS update"));
}
}
}
} }
void XdsClient::ChannelState::AdsCallState::AcceptEdsUpdate( void XdsClient::ChannelState::AdsCallState::AcceptEdsUpdate(
@ -1043,7 +1093,8 @@ void XdsClient::ChannelState::AdsCallState::AcceptEdsUpdate(
EndpointState& endpoint_state = EndpointState& endpoint_state =
xds_client()->endpoint_map_[eds_service_name]; xds_client()->endpoint_map_[eds_service_name];
// Ignore identical update. // Ignore identical update.
const XdsApi::EdsUpdate& prev_update = endpoint_state.update; if (endpoint_state.update.has_value()) {
const XdsApi::EdsUpdate& prev_update = endpoint_state.update.value();
const bool priority_list_changed = const bool priority_list_changed =
prev_update.priority_list_update != eds_update.priority_list_update; prev_update.priority_list_update != eds_update.priority_list_update;
const bool drop_config_changed = const bool drop_config_changed =
@ -1057,11 +1108,12 @@ void XdsClient::ChannelState::AdsCallState::AcceptEdsUpdate(
} }
continue; continue;
} }
}
// Update the cluster state. // Update the cluster state.
endpoint_state.update = std::move(eds_update); endpoint_state.update = std::move(eds_update);
// Notify all watchers. // Notify all watchers.
for (const auto& p : endpoint_state.watchers) { for (const auto& p : endpoint_state.watchers) {
p.first->OnEndpointChanged(endpoint_state.update); p.first->OnEndpointChanged(endpoint_state.update.value());
} }
} }
} }
@ -1135,8 +1187,8 @@ void XdsClient::ChannelState::AdsCallState::OnResponseReceivedLocked(
// mode. We will also need to cancel the timer when we receive a serverlist // mode. We will also need to cancel the timer when we receive a serverlist
// from the balancer. // from the balancer.
// Parse the response. // Parse the response.
XdsApi::LdsUpdate lds_update; absl::optional<XdsApi::LdsUpdate> lds_update;
XdsApi::RdsUpdate rds_update; absl::optional<XdsApi::RdsUpdate> rds_update;
XdsApi::CdsUpdateMap cds_update_map; XdsApi::CdsUpdateMap cds_update_map;
XdsApi::EdsUpdateMap eds_update_map; XdsApi::EdsUpdateMap eds_update_map;
std::string version; std::string version;
@ -1145,6 +1197,7 @@ void XdsClient::ChannelState::AdsCallState::OnResponseReceivedLocked(
// Note that ParseAdsResponse() also validates the response. // Note that ParseAdsResponse() also validates the response.
grpc_error* parse_error = xds_client->api_.ParseAdsResponse( grpc_error* parse_error = xds_client->api_.ParseAdsResponse(
response_slice, xds_client->server_name_, xds_client->route_config_name_, response_slice, xds_client->server_name_, xds_client->route_config_name_,
ads_calld->ClusterNamesForRequest(),
ads_calld->EdsServiceNamesForRequest(), &lds_update, &rds_update, ads_calld->EdsServiceNamesForRequest(), &lds_update, &rds_update,
&cds_update_map, &eds_update_map, &version, &nonce, &type_url); &cds_update_map, &eds_update_map, &version, &nonce, &type_url);
grpc_slice_unref_internal(response_slice); grpc_slice_unref_internal(response_slice);
@ -1811,8 +1864,8 @@ void XdsClient::WatchEndpointData(
endpoint_state.watchers[w] = std::move(watcher); endpoint_state.watchers[w] = std::move(watcher);
// If we've already received an EDS update, notify the new watcher // If we've already received an EDS update, notify the new watcher
// immediately. // immediately.
if (!endpoint_state.update.priority_list_update.empty()) { if (endpoint_state.update.has_value()) {
w->OnEndpointChanged(endpoint_state.update); w->OnEndpointChanged(endpoint_state.update.value());
} }
chand_->Subscribe(XdsApi::kEdsTypeUrl, eds_service_name_str); chand_->Subscribe(XdsApi::kEdsTypeUrl, eds_service_name_str);
} }

8
src/core/ext/filters/client_channel/xds/xds_client.h
Unescape View File

@ -189,7 +189,7 @@ class XdsClient : public InternallyRefCounted<XdsClient> {
std::map<ClusterWatcherInterface*, std::unique_ptr<ClusterWatcherInterface>> std::map<ClusterWatcherInterface*, std::unique_ptr<ClusterWatcherInterface>>
watchers; watchers;
// The latest data seen from CDS. // The latest data seen from CDS.
Optional<XdsApi::CdsUpdate> update; absl::optional<XdsApi::CdsUpdate> update;
}; };
struct EndpointState { struct EndpointState {
@ -197,7 +197,7 @@ class XdsClient : public InternallyRefCounted<XdsClient> {
std::unique_ptr<EndpointWatcherInterface>> std::unique_ptr<EndpointWatcherInterface>>
watchers; watchers;
// The latest data seen from EDS. // The latest data seen from EDS.
XdsApi::EdsUpdate update; absl::optional<XdsApi::EdsUpdate> update;
}; };
struct LoadReportState { struct LoadReportState {
@ -241,9 +241,9 @@ class XdsClient : public InternallyRefCounted<XdsClient> {
std::string route_config_name_; std::string route_config_name_;
std::string cluster_name_; std::string cluster_name_;
// All the received clusters are cached, no matter they are watched or not. // One entry for each watched CDS resource.
std::map<std::string /*cluster_name*/, ClusterState> cluster_map_; std::map<std::string /*cluster_name*/, ClusterState> cluster_map_;
// Only the watched EDS service names are stored. // One entry for each watched EDS resource.
std::map<std::string /*eds_service_name*/, EndpointState> endpoint_map_; std::map<std::string /*eds_service_name*/, EndpointState> endpoint_map_;
std::map< std::map<
std::pair<std::string /*cluster_name*/, std::string /*eds_service_name*/>, std::pair<std::string /*cluster_name*/, std::string /*eds_service_name*/>,

Write Preview
Loading…
Cancel
Save