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

Merge pull request #22289 from markdroth/xds_test_cleanup

Browse Source 
Clean up xds_end2end_test code
pull/22309/head
Mark D. Roth 5 years ago committed by GitHub
parent 50619a562f b4d5901b5e
commit 35ee039c29
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 136 additions and 174 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 302
      test/cpp/end2end/xds_end2end_test.cc

302
test/cpp/end2end/xds_end2end_test.cc
Unescape View File

@ -400,7 +400,7 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
std::pair<std::string /* type url */, std::string /* resource name */>>;
// A struct representing a client's subscription to a particular resource.
struct SubscriberState {
struct SubscriptionState {
// Version that the client currently knows about.
int current_version = 0;
// The queue upon which to place updates when the resource is updated.
@ -408,23 +408,25 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
};
// A struct representing the a client's subscription to all the resources.
using SubscriptionNameMap =
std::map<std::string /* resource_name */, SubscriptionState>;
using SubscriptionMap =
std::map<std::string /* type_url */,
std::map<std::string /* resource_name */, SubscriberState>>;
std::map<std::string /* type_url */, SubscriptionNameMap>;
// A struct representing the current state for a resource:
// - the version of the resource that is set by the SetResource() methods.
// - a list of subscribers interested in this resource.
// - a list of subscriptions interested in this resource.
struct ResourceState {
int version = 0;
absl::optional<google::protobuf::Any> resource;
std::set<SubscriberState*> subscribers;
std::set<SubscriptionState*> subscriptions;
};
// A struct representing the current state for all resources:
// LDS, CDS, EDS, and RDS for the class as a whole.
using ResourcesMap =
std::map<std::string, std::map<std::string, ResourceState>>;
using ResourceNameMap =
std::map<std::string /* resource_name */, ResourceState>;
using ResourceMap = std::map<std::string /* type_url */, ResourceNameMap>;
AdsServiceImpl(bool enable_load_reporting) {
// Construct RDS response data.
@ -475,101 +477,61 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
}
// Checks whether the client needs to receive a newer version of
// the resource.
bool ClientNeedsResourceUpdate(const string& resource_type,
const string& name,
SubscriptionMap* subscription_map) {
auto subscriber_it = (*subscription_map)[resource_type].find(name);
if (subscriber_it == (*subscription_map)[resource_type].end()) {
gpr_log(GPR_INFO,
"ADS[%p]: Skipping an unsubscribed update for resource %s and "
"name %s",
this, resource_type.c_str(), name.c_str());
return false;
}
const auto& resource_state = resources_map_[resource_type][name];
if (subscriber_it->second.current_version < resource_state.version) {
subscriber_it->second.current_version = resource_state.version;
gpr_log(GPR_INFO,
"ADS[%p]: Need to process new %s update %s, bring current to %d",
this, resource_type.c_str(), name.c_str(),
subscriber_it->second.current_version);
// the resource. If so, updates subscription_state->current_version and
// returns true.
bool ClientNeedsResourceUpdate(const ResourceState& resource_state,
SubscriptionState* subscription_state) {
if (subscription_state->current_version < resource_state.version) {
subscription_state->current_version = resource_state.version;
return true;
} else {
gpr_log(GPR_INFO,
"ADS[%p]: Skipping an old %s update %s, current is at %d", this,
resource_type.c_str(), name.c_str(),
subscriber_it->second.current_version);
return false;
}
return false;
}
// Resource subscription:
// 1. inserting an entry into the subscription map indexed by resource
// type/name pair.
// 2. inserting or updating an entry into the resources map indexed
// by resource type/name pair about this subscription.
void ResourceSubscribe(const std::string& resource_type,
const std::string& name, UpdateQueue* update_queue,
SubscriptionMap* subscription_map) {
SubscriberState& subscriber_state =
(*subscription_map)[resource_type][name];
subscriber_state.update_queue = update_queue;
ResourceState& resource_state = resources_map_[resource_type][name];
resource_state.subscribers.emplace(&subscriber_state);
gpr_log(
GPR_INFO,
"ADS[%p]: subscribe to resource type %s name %s version %d state %p",
this, resource_type.c_str(), name.c_str(), resource_state.version,
&subscriber_state);
}
// Resource unsubscription:
// 1. update the entry in the resources map indexed
// by resource type/name pair to remove this subscription
// 2. remove this entry from the subscription map.
// 3. remove this resource type from the subscription map if there are no more
// resources subscribed for the resource type.
void ResourceUnsubscribe(const std::string& resource_type,
const std::string& name,
SubscriptionMap* subscription_map) {
auto subscription_by_type_it = subscription_map->find(resource_type);
if (subscription_by_type_it == subscription_map->end()) {
gpr_log(GPR_INFO, "ADS[%p]: resource type %s not subscribed", this,
resource_type.c_str());
return;
}
auto& subscription_by_type_map = subscription_by_type_it->second;
auto subscription_it = subscription_by_type_map.find(name);
if (subscription_it == subscription_by_type_map.end()) {
gpr_log(GPR_INFO, "ADS[%p]: resource name %s of type %s not subscribed",
this, name.c_str(), resource_type.c_str());
return;
// Subscribes to a resource if not already subscribed:
// 1. Sets the update_queue field in subscription_state.
// 2. Adds subscription_state to resource_state->subscriptions.
void MaybeSubscribe(const std::string& resource_type,
const std::string& resource_name,
SubscriptionState* subscription_state,
ResourceState* resource_state,
UpdateQueue* update_queue) {
if (subscription_state->update_queue != nullptr) return;
subscription_state->update_queue = update_queue;
resource_state->subscriptions.emplace(subscription_state);
gpr_log(GPR_INFO, "ADS[%p]: subscribe to resource type %s name %s state %p",
this, resource_type.c_str(), resource_name.c_str(),
&subscription_state);
}
// Removes subscriptions for resources no longer present in the
// current request.
void ProcessUnsubscriptions(
const std::string& resource_type,
const std::set<std::string>& resources_in_current_request,
SubscriptionNameMap* subscription_name_map,
ResourceNameMap* resource_name_map) {
for (auto it = subscription_name_map->begin();
it != subscription_name_map->end();) {
const std::string& resource_name = it->first;
SubscriptionState& subscription_state = it->second;
if (resources_in_current_request.find(resource_name) !=
resources_in_current_request.end()) {
++it;
continue;
}
gpr_log(GPR_INFO,
"ADS[%p]: Unsubscribe to resource type %s name %s state %p", this,
resource_type.c_str(), name.c_str(), &subscription_it->second);
auto resource_by_type_it = resources_map_.find(resource_type);
GPR_ASSERT(resource_by_type_it != resources_map_.end());
auto& resource_by_type_map = resource_by_type_it->second;
auto resource_it = resource_by_type_map.find(name);
GPR_ASSERT(resource_it != resource_by_type_map.end());
resource_it->second.subscribers.erase(&subscription_it->second);
if (resource_it->second.subscribers.empty() &&
!resource_it->second.resource.has_value()) {
gpr_log(GPR_INFO,
"ADS[%p]: Erasing resource type %s name %s from resource map "
"since there are no more subscribers for this unset resource",
this, resource_type.c_str(), name.c_str());
resource_by_type_map.erase(resource_it);
gpr_log(GPR_INFO, "ADS[%p]: Unsubscribe to type=%s name=%s state=%p",
this, resource_type.c_str(), resource_name.c_str(),
&subscription_state);
auto resource_it = resource_name_map->find(resource_name);
GPR_ASSERT(resource_it != resource_name_map->end());
auto& resource_state = resource_it->second;
resource_state.subscriptions.erase(&subscription_state);
if (resource_state.subscriptions.empty() &&
!resource_state.resource.has_value()) {
resource_name_map->erase(resource_it);
}
subscription_by_type_map.erase(subscription_it);
if (subscription_by_type_map.empty()) {
gpr_log(GPR_INFO,
"ADS[%p]: Erasing resource type %s from subscription_map", this,
resource_type.c_str());
subscription_map->erase(subscription_by_type_it);
it = subscription_name_map->erase(it);
}
}
@ -577,7 +539,7 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
// for all resources and by adding all subscribed resources for LDS and CDS.
void CompleteBuildingDiscoveryResponse(
const std::string& resource_type, const int version,
const SubscriptionMap& subscription_map,
const SubscriptionNameMap& subscription_name_map,
const std::set<std::string>& resources_added_to_response,
DiscoveryResponse* response) {
resource_type_response_state_[resource_type] = SENT;
@ -587,18 +549,15 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
if (resource_type == kLdsTypeUrl || resource_type == kCdsTypeUrl) {
// For LDS and CDS we must send back all subscribed resources
// (even the unchanged ones)
auto subscription_map_by_type_it = subscription_map.find(resource_type);
GPR_ASSERT(subscription_map_by_type_it != subscription_map.end());
for (const auto& subscription : subscription_map_by_type_it->second) {
if (resources_added_to_response.find(subscription.first) ==
for (const auto& p : subscription_name_map) {
const std::string& resource_name = p.first;
if (resources_added_to_response.find(resource_name) ==
resources_added_to_response.end()) {
absl::optional<google::protobuf::Any>& resource =
resources_map_[resource_type][subscription.first].resource;
if (resource.has_value()) {
response->add_resources()->CopyFrom(resource.value());
} else {
gpr_log(GPR_INFO, "ADS[%p]: Unknown resource type %s and name %s",
this, resource_type.c_str(), subscription.first.c_str());
const ResourceState& resource_state =
resource_map_[resource_type][resource_name];
if (resource_state.resource.has_value()) {
response->add_resources()->CopyFrom(
resource_state.resource.value());
}
}
}
@ -622,7 +581,6 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
// Resources that the client will be subscribed to keyed by resource type
// url.
SubscriptionMap subscription_map;
std::map<std::string, SubscriberState> subscriber_map;
// Current Version map keyed by resource type url.
std::map<std::string, int> resource_type_version;
// Creating blocking thread to read from stream.
@ -647,7 +605,8 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
DiscoveryRequest request = std::move(requests.front());
requests.pop_front();
did_work = true;
gpr_log(GPR_INFO, "ADS[%p]: Handling request %s with content %s",
gpr_log(GPR_INFO,
"ADS[%p]: Received request for type %s with content %s",
this, request.type_url().c_str(),
request.DebugString().c_str());
// Identify ACK and NACK by looking for version information and
@ -667,58 +626,51 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
// 3. unsubscribe if necessary
if (resource_types_to_ignore_.find(request.type_url()) ==
resource_types_to_ignore_.end()) {
auto& subscription_name_map =
subscription_map[request.type_url()];
auto& resource_name_map = resource_map_[request.type_url()];
std::set<std::string> resources_in_current_request;
std::set<std::string> resources_added_to_response;
for (const std::string& resource_name :
request.resource_names()) {
resources_in_current_request.emplace(resource_name);
auto subscriber_it =
subscription_map[request.type_url()].find(resource_name);
if (subscriber_it ==
subscription_map[request.type_url()].end()) {
ResourceSubscribe(request.type_url(), resource_name,
&update_queue, &subscription_map);
}
if (ClientNeedsResourceUpdate(request.type_url(), resource_name,
&subscription_map)) {
auto& subscription_state = subscription_name_map[resource_name];
auto& resource_state = resource_name_map[resource_name];
MaybeSubscribe(request.type_url(), resource_name,
&subscription_state, &resource_state,
&update_queue);
if (ClientNeedsResourceUpdate(resource_state,
&subscription_state)) {
gpr_log(
GPR_INFO,
"ADS[%p]: Sending update for type=%s name=%s version=%d",
this, request.type_url().c_str(), resource_name.c_str(),
resource_state.version);
resources_added_to_response.emplace(resource_name);
gpr_log(GPR_INFO,
"ADS[%p]: Handling resource type %s and name %s",
this, request.type_url().c_str(),
resource_name.c_str());
auto resource =
resources_map_[request.type_url()][resource_name];
GPR_ASSERT(resource.resource.has_value());
response.add_resources()->CopyFrom(resource.resource.value());
}
}
// Remove subscriptions no longer requested: build a list of
// unsubscriber names first while iterating the subscription_map
// and then erase from the subscription_map in
// ResourceUnsubscribe.
std::set<std::string> unsubscriber_list;
for (const auto& subscription :
subscription_map[request.type_url()]) {
if (resources_in_current_request.find(subscription.first) ==
resources_in_current_request.end()) {
unsubscriber_list.emplace(subscription.first);
if (resource_state.resource.has_value()) {
response.add_resources()->CopyFrom(
resource_state.resource.value());
}
}
for (const auto& name : unsubscriber_list) {
ResourceUnsubscribe(request.type_url(), name,
&subscription_map);
}
if (!response.resources().empty()) {
// Process unsubscriptions for any resource no longer
// present in the request's resource list.
ProcessUnsubscriptions(
request.type_url(), resources_in_current_request,
&subscription_name_map, &resource_name_map);
// Send response if needed.
if (!resources_added_to_response.empty()) {
CompleteBuildingDiscoveryResponse(
request.type_url(),
++resource_type_version[request.type_url()],
subscription_map, resources_added_to_response, &response);
subscription_name_map, resources_added_to_response,
&response);
}
}
}
}
if (!response.resources().empty()) {
gpr_log(GPR_INFO, "ADS[%p]: sending request response '%s'", this,
gpr_log(GPR_INFO, "ADS[%p]: Sending response: %s", this,
response.DebugString().c_str());
stream->Write(response);
}
@ -727,32 +679,40 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
{
grpc_core::MutexLock lock(&ads_mu_);
if (!update_queue.empty()) {
std::pair<std::string, std::string> update =
std::move(update_queue.front());
const std::string resource_type =
std::move(update_queue.front().first);
const std::string resource_name =
std::move(update_queue.front().second);
update_queue.pop_front();
did_work = true;
gpr_log(GPR_INFO, "ADS[%p]: Handling update type %s name %s", this,
update.first.c_str(), update.second.c_str());
auto subscriber_it =
subscription_map[update.first].find(update.second);
if (subscriber_it != subscription_map[update.first].end()) {
if (ClientNeedsResourceUpdate(update.first, update.second,
&subscription_map)) {
gpr_log(GPR_INFO,
"ADS[%p]: Updating resource type %s and name %s", this,
update.first.c_str(), update.second.c_str());
auto resource = resources_map_[update.first][update.second];
GPR_ASSERT(resource.resource.has_value());
response.add_resources()->CopyFrom(resource.resource.value());
gpr_log(GPR_INFO, "ADS[%p]: Received update for type=%s name=%s",
this, resource_type.c_str(), resource_name.c_str());
auto& subscription_name_map = subscription_map[resource_type];
auto& resource_name_map = resource_map_[resource_type];
auto it = subscription_name_map.find(resource_name);
if (it != subscription_name_map.end()) {
SubscriptionState& subscription_state = it->second;
ResourceState& resource_state = resource_name_map[resource_name];
if (ClientNeedsResourceUpdate(resource_state,
&subscription_state)) {
gpr_log(
GPR_INFO,
"ADS[%p]: Sending update for type=%s name=%s version=%d",
this, resource_type.c_str(), resource_name.c_str(),
resource_state.version);
if (resource_state.resource.has_value()) {
response.add_resources()->CopyFrom(
resource_state.resource.value());
CompleteBuildingDiscoveryResponse(
update.first, ++resource_type_version[update.first],
subscription_map, {update.second}, &response);
resource_type, ++resource_type_version[resource_type],
subscription_name_map, {resource_name}, &response);
}
}
}
}
}
if (!response.resources().empty()) {
gpr_log(GPR_INFO, "ADS[%p]: sending update response '%s'", this,
gpr_log(GPR_INFO, "ADS[%p]: Sending update response: %s", this,
response.DebugString().c_str());
stream->Write(response);
}
@ -808,13 +768,13 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
void SetResource(google::protobuf::Any resource, const std::string& type_url,
const std::string& name) {
grpc_core::MutexLock lock(&ads_mu_);
ResourceState& state = resources_map_[type_url][name];
ResourceState& state = resource_map_[type_url][name];
++state.version;
state.resource = std::move(resource);
gpr_log(GPR_INFO, "ADS[%p]: Updating %s resource %s to version %u", this,
type_url.c_str(), name.c_str(), state.version);
for (SubscriberState* subscriber : state.subscribers) {
subscriber->update_queue->emplace_back(type_url, name);
for (SubscriptionState* subscription : state.subscriptions) {
subscription->update_queue->emplace_back(type_url, name);
}
}
@ -873,15 +833,17 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
{
grpc_core::MutexLock lock(&ads_mu_);
NotifyDoneWithAdsCallLocked();
resources_map_.clear();
resource_map_.clear();
resource_type_response_state_.clear();
}
gpr_log(GPR_INFO, "ADS[%p]: shut down", this);
}
static ClusterLoadAssignment BuildEdsResource(const EdsResourceArgs& args) {
static ClusterLoadAssignment BuildEdsResource(
const EdsResourceArgs& args,
const char* cluster_name = kDefaultResourceName) {
ClusterLoadAssignment assignment;
assignment.set_cluster_name(kDefaultResourceName);
assignment.set_cluster_name(cluster_name);
for (const auto& locality : args.locality_list) {
auto* endpoints = assignment.add_endpoints();
endpoints->mutable_load_balancing_weight()->set_value(locality.lb_weight);
@ -946,8 +908,8 @@ class AdsServiceImpl : public AggregatedDiscoveryService::Service,
// Note that an entry will exist whenever either of the following is true:
// - The resource exists (i.e., has been created by SetResource() and has not
// yet been destroyed by UnsetResource()).
// - There is at least one subscriber for the resource.
ResourcesMap resources_map_;
// - There is at least one subscription for the resource.
ResourceMap resource_map_;
};
class LrsServiceImpl : public LrsService,

Write Preview
Loading…
Cancel
Save