config: ADS implementation. #1621
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,161 @@ | ||
| #include "common/config/ads_api_impl.h" | ||
|
|
||
| #include "common/config/utility.h" | ||
| #include "common/protobuf/protobuf.h" | ||
|
|
||
| namespace Envoy { | ||
| namespace Config { | ||
|
|
||
| // TODO(htuch): describe the ADS protocol | ||
|
|
||
| AdsApiImpl::AdsApiImpl(const envoy::api::v2::Node& node, | ||
| const envoy::api::v2::ApiConfigSource& ads_config, | ||
| Upstream::ClusterManager& cluster_manager, Event::Dispatcher& dispatcher) | ||
| : node_(node), service_method_(*Protobuf::DescriptorPool::generated_pool()->FindMethodByName( | ||
| "envoy.api.v2.AggregatedDiscoveryService.StreamAggregatedResources")) { | ||
| if (ads_config.cluster_name().empty()) { | ||
| ENVOY_LOG(debug, "No ADS clusters defined, ADS will not be initialized."); | ||
| return; | ||
| } | ||
| retry_timer_ = dispatcher.createTimer([this]() -> void { establishNewStream(); }); | ||
| if (ads_config.cluster_name().size() != 1) { | ||
| // TODO(htuch): Add support for multiple clusters, #1170. | ||
| throw EnvoyException( | ||
| "envoy::api::v2::ApiConfigSource must have a singleton cluster name specified"); | ||
| } | ||
| async_client_.reset(new Grpc::AsyncClientImpl<envoy::api::v2::DiscoveryRequest, | ||
| envoy::api::v2::DiscoveryResponse>( | ||
| cluster_manager, ads_config.cluster_name()[0])); | ||
| } | ||
|
|
||
| AdsApiImpl::~AdsApiImpl() { | ||
| for (auto watches : watches_) { | ||
| for (auto watch : watches.second) { | ||
| watch->inserted_ = false; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| void AdsApiImpl::start() { | ||
| if (async_client_) { | ||
| establishNewStream(); | ||
| } | ||
| } | ||
|
|
||
| void AdsApiImpl::setRetryTimer() { | ||
| retry_timer_->enableTimer(std::chrono::milliseconds(RETRY_DELAY_MS)); | ||
| } | ||
|
|
||
| void AdsApiImpl::establishNewStream() { | ||
| // TODO(htuch): stats | ||
| ENVOY_LOG(debug, "Establishing new gRPC bidi stream for {}", service_method_.DebugString()); | ||
| stream_ = async_client_->start(service_method_, *this); | ||
| if (stream_ == nullptr) { | ||
| ENVOY_LOG(warn, "Unable to establish new stream"); | ||
| handleFailure(); | ||
| return; | ||
| } | ||
|
|
||
| for (const auto type_url : subscriptions_) { | ||
| sendDiscoveryRequest(requests_[type_url]); | ||
| } | ||
| } | ||
|
|
||
| void AdsApiImpl::sendDiscoveryRequest(const envoy::api::v2::DiscoveryRequest& request) { | ||
| if (stream_ == nullptr) { | ||
| return; | ||
| } | ||
| stream_->sendMessage(request, false); | ||
| } | ||
|
|
||
| void AdsApiImpl::handleFailure() { | ||
| for (auto watches : watches_) { | ||
| for (auto watch : watches.second) { | ||
| watch->callbacks_.onConfigUpdateFailed(nullptr); | ||
| } | ||
| } | ||
| setRetryTimer(); | ||
| } | ||
|
|
||
| AdsWatchPtr AdsApiImpl::subscribe(const std::string& type_url, | ||
| const std::vector<std::string>& resources, | ||
| AdsCallbacks& callbacks) { | ||
| auto watch = | ||
| std::unique_ptr<AdsWatch>(new AdsWatchImpl(resources, callbacks, watches_[type_url])); | ||
| ENVOY_LOG(debug, "ADS subscribe for " + type_url); | ||
|
|
||
| // Lazily kick off the requests based on first subscription. This has the | ||
| // convenient side-effect that we order messages on the channel based on | ||
| // Envoy's internal dependency ordering. | ||
| if (requests_.count(type_url) == 0) { | ||
| requests_[type_url].set_type_url(type_url); | ||
| requests_[type_url].mutable_node()->CopyFrom(node_); | ||
| subscriptions_.push_front(type_url); | ||
| sendDiscoveryRequest(requests_[type_url]); | ||
| } | ||
|
|
||
| return watch; | ||
| } | ||
|
|
||
| void AdsApiImpl::onCreateInitialMetadata(Http::HeaderMap& metadata) { | ||
| UNREFERENCED_PARAMETER(metadata); | ||
| } | ||
|
|
||
| void AdsApiImpl::onReceiveInitialMetadata(Http::HeaderMapPtr&& metadata) { | ||
| UNREFERENCED_PARAMETER(metadata); | ||
| } | ||
|
|
||
| void AdsApiImpl::onReceiveMessage(std::unique_ptr<envoy::api::v2::DiscoveryResponse>&& message) { | ||
| const std::string& type_url = message->type_url(); | ||
| ENVOY_LOG(debug, "Received ADS message for {}", type_url); | ||
| try { | ||
| // To avoid O(n^2) explosion (e.g. when we have 1000s of EDS watches), we | ||
| // build a map here from resource name to resource and then walk watches_. | ||
| // TODO(htuch): Reconsider implementation data structure for watches to make lookups of resource | ||
| // -> watches O(1), to avoid doing this crap. | ||
| std::unordered_map<std::string, ProtobufWkt::Any> resources; | ||
| for (const auto& resource : message->resources()) { | ||
| if (type_url != resource.type_url()) { | ||
| throw EnvoyException(fmt::format("{} does not match {} type URL is DiscoveryResponse {}", | ||
| resource.type_url(), type_url, message->DebugString())); | ||
| } | ||
| resources.emplace(Utility::resourceName(resource), resource); | ||
| } | ||
| for (auto watch : watches_[type_url]) { | ||
| if (watch->resources_.empty()) { | ||
| watch->callbacks_.onConfigUpdate(message->resources()); | ||
| continue; | ||
| } | ||
| Protobuf::RepeatedPtrField<ProtobufWkt::Any> found_resources; | ||
| for (auto watched_resource_name : watch->resources_) { | ||
| auto it = resources.find(watched_resource_name); | ||
| if (it != resources.end()) { | ||
| found_resources.Add()->CopyFrom(it->second); | ||
| } | ||
| } | ||
| if (!found_resources.empty()) { | ||
| watch->callbacks_.onConfigUpdate(found_resources); | ||
| } | ||
| } | ||
| requests_[type_url].set_version_info(message->version_info()); | ||
| } catch (const EnvoyException& e) { | ||
| ENVOY_LOG(warn, "ADS config for {} update rejected: {}", message->type_url(), e.what()); | ||
| for (auto watch : watches_[type_url]) { | ||
| watch->callbacks_.onConfigUpdateFailed(&e); | ||
| } | ||
| } | ||
| sendDiscoveryRequest(requests_[type_url]); | ||
| } | ||
|
|
||
| void AdsApiImpl::onReceiveTrailingMetadata(Http::HeaderMapPtr&& metadata) { | ||
| UNREFERENCED_PARAMETER(metadata); | ||
| } | ||
|
|
||
| void AdsApiImpl::onRemoteClose(Grpc::Status::GrpcStatus status, const std::string& message) { | ||
| ENVOY_LOG(warn, "ADS config stream closed: {}, {}", status, message); | ||
| stream_ = nullptr; | ||
| handleFailure(); | ||
| } | ||
|
|
||
| } // namespace Config | ||
| } // namespace Envoy | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yeah, why not just store internally as map?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We expect that most watches that name resources (as opposed to open CDS/LDS watches, which just wildcard via an empty resource list) will have a single resource, but the current APIs internally allows for more. We need to assemble the subset of resources from those received and pass them to
onConfigUpdate()once per watch. This means we need to do some gather operation. This could be either done by building buckets for each watch and populating them as we walk the watch list (as done today), or building buckets and populating them as we walk the resource list.I'm not a fan of either of these. If we change the API to allow zero (wildcard) or one resource per watch, we can easily use a map. Given that's how it's actually used by the xDS implementations in Envoy today, I think I'm going to modify the ADS internal API to switch to this. Let me know if that sounds good.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
+1 sounds good simpler is better
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
After playing around with this, we still need to do the map construction. This is because we need to deliver empty updates to watches (e.g. EDS) when a resource is not present, so we basically need to walk all the watches and have a O(1) map back to the delivered resource matching the watched name. So, I'll leave it as is.