WIP: VHDS: on-demand updates

include/envoy/config/subscription.h

@@ -8,6 +8,7 @@
 #include "envoy/common/pure.h"
 #include "envoy/stats/stats_macros.h"
 
+#include "common/common/assert.h"
 #include "common/protobuf/protobuf.h"
 
 namespace Envoy {
@@ -79,6 +80,10 @@ class Subscription {
    * be passed to std::set_difference, which must be given sorted collections.
    */
   virtual void updateResources(const std::set<std::string>& update_to_these_names) PURE;
+
+  virtual void updateResourcesViaAliases(const std::set<std::string>&) {
+    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+  }
 };
 
 /**

include/envoy/http/filter.h

@@ -150,6 +150,8 @@ class StreamFilterCallbacks {
    */
   virtual Router::RouteConstSharedPtr route() PURE;
 
+  virtual bool requestRouteConfigUpdate(std::function<void()> cb) PURE;
+
   /**
    * Returns the clusterInfo for the cached route.
    * This method is to avoid multiple look ups in the filter chain, it also provides a consistent

include/envoy/router/rds.h

@@ -48,6 +48,17 @@ class RouteConfigProvider {
    * Callback used to notify RouteConfigProvider about configuration changes.
    */
   virtual void onConfigUpdate() PURE;
+
+  /**
+   * Callback used to request an update to the route configuration.
+   * @param for_domain supplies the domain name that virtual hosts contained in the VHDS response
+   * must match on
+   * @param cb callback to be called when the configuration update has been propagated to worker
+   * threads
+   * @return whether a request for a configuration update has been successfully scheduled
+   */
+  virtual bool requestVirtualHostsUpdate(const std::string& for_domain,
+                                         std::function<void()> cb) PURE;
 };
 
 typedef std::unique_ptr<RouteConfigProvider> RouteConfigProviderPtr;

source/common/config/delta_subscription_impl.cc

@@ -42,6 +42,13 @@ void DeltaSubscriptionImpl::updateResources(const std::set<std::string>& update_
   trySendDiscoveryRequests();
 }
 
+void DeltaSubscriptionImpl::updateResourcesViaAliases(
+    const std::set<std::string>& updates_to_these_aliases) {
+  state_->updateResourceInterestViaAliases(updates_to_these_aliases);
+  // Tell the server about our new interests, if there are any.
+  trySendDiscoveryRequests();
+}
+
 // Config::GrpcStreamCallbacks
 void DeltaSubscriptionImpl::onStreamEstablished() {
   state_->markStreamFresh();

source/common/config/delta_subscription_impl.h

@@ -38,6 +38,7 @@ class DeltaSubscriptionImpl : public Subscription,
   // Config::Subscription
   void start(const std::set<std::string>& resources, SubscriptionCallbacks& callbacks) override;
   void updateResources(const std::set<std::string>& update_to_these_names) override;
+  void updateResourcesViaAliases(const std::set<std::string>& updates_to_these_aliases) override;
 
   // Config::GrpcStreamCallbacks
   void onStreamEstablished() override;

source/common/config/delta_subscription_state.cc

@@ -76,10 +76,15 @@ void DeltaSubscriptionState::updateResourceInterest(
   }
 }
 
+void DeltaSubscriptionState::updateResourceInterestViaAliases(
+    const std::set<std::string>& updates_to_these_aliases) {
+  aliases_added_.insert(updates_to_these_aliases.begin(), updates_to_these_aliases.end());
+}
+
 // Not having sent any requests yet counts as an "update pending" since you're supposed to resend
 // the entirety of your interest at the start of a stream, even if nothing has changed.
 bool DeltaSubscriptionState::subscriptionUpdatePending() const {
-  return !names_added_.empty() || !names_removed_.empty() ||
+  return !aliases_added_.empty() || !names_added_.empty() || !names_removed_.empty() ||
          !any_request_sent_yet_in_current_stream_;
 }
 
@@ -157,6 +162,28 @@ void DeltaSubscriptionState::handleEstablishmentFailure() {
 
 envoy::api::v2::DeltaDiscoveryRequest DeltaSubscriptionState::getNextRequest() {
   envoy::api::v2::DeltaDiscoveryRequest request;
+  if (!any_request_sent_yet_in_current_stream_) {
+    populateDiscoveryRequest(request);
+  } else if (!aliases_added_.empty()) {
+    populateDiscoveryRequestWithAliases(request);
+  } else {
+    populateDiscoveryRequest(request);
+  }
+
+  request.set_type_url(type_url_);
+  request.mutable_node()->MergeFrom(local_info_.node());
+  return request;
+}
+
+void DeltaSubscriptionState::populateDiscoveryRequestWithAliases(
+    envoy::api::v2::DeltaDiscoveryRequest& request) {
+  std::copy(aliases_added_.begin(), aliases_added_.end(),
+            Protobuf::RepeatedFieldBackInserter(request.mutable_resource_names_subscribe()));
+  aliases_added_.clear();
+}
+
+void DeltaSubscriptionState::populateDiscoveryRequest(
+    envoy::api::v2::DeltaDiscoveryRequest& request) {
   if (!any_request_sent_yet_in_current_stream_) {
     any_request_sent_yet_in_current_stream_ = true;
     // initial_resource_versions "must be populated for first request in a stream".
@@ -181,10 +208,6 @@ envoy::api::v2::DeltaDiscoveryRequest DeltaSubscriptionState::getNextRequest() {
             Protobuf::RepeatedFieldBackInserter(request.mutable_resource_names_unsubscribe()));
   names_added_.clear();
   names_removed_.clear();
-
-  request.set_type_url(type_url_);
-  request.mutable_node()->MergeFrom(local_info_.node());
-  return request;
 }
 
 void DeltaSubscriptionState::disableInitFetchTimeoutTimer() {

source/common/config/delta_subscription_state.h

@@ -35,6 +35,7 @@ class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
 
   // Update which resources we're interested in subscribing to.
   void updateResourceInterest(const std::set<std::string>& update_to_these_names);
+  void updateResourceInterestViaAliases(const std::set<std::string>& updates_to_these_aliases);
 
   // Whether there was a change in our subscription interest we have yet to inform the server of.
   bool subscriptionUpdatePending() const;
@@ -75,6 +76,8 @@ class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
   void setResourceVersion(const std::string& resource_name, const std::string& resource_version);
   void setResourceWaitingForServer(const std::string& resource_name);
   void setLostInterestInResource(const std::string& resource_name);
+  void populateDiscoveryRequest(envoy::api::v2::DeltaDiscoveryRequest& request);
+  void populateDiscoveryRequestWithAliases(envoy::api::v2::DeltaDiscoveryRequest& request);
 
   // A map from resource name to per-resource version. The keys of this map are exactly the resource
   // names we are currently interested in. Those in the waitingForServer state currently don't have
@@ -99,6 +102,7 @@ class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
   // Feel free to change to unordered if you can figure out how to make it work.
   std::set<std::string> names_added_;
   std::set<std::string> names_removed_;
+  std::set<std::string> aliases_added_;
 
   SubscriptionStats& stats_;
 };

source/common/http/async_client_impl.h

@@ -78,6 +78,8 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   AsyncStreamImpl(AsyncClientImpl& parent, AsyncClient::StreamCallbacks& callbacks,
                   const AsyncClient::StreamOptions& options);
 
+  bool requestRouteConfigUpdate(std::function<void()>) override { return false; }
+
   // Http::AsyncClient::Stream
   void sendHeaders(HeaderMap& headers, bool end_stream) override;
   void sendData(Buffer::Instance& data, bool end_stream) override;
@@ -367,6 +369,8 @@ class AsyncRequestImpl final : public AsyncClient::Request,
   AsyncRequestImpl(MessagePtr&& request, AsyncClientImpl& parent, AsyncClient::Callbacks& callbacks,
                    const AsyncClient::RequestOptions& options);
 
+  bool requestRouteConfigUpdate(std::function<void()>) override { return false; }
+
   // AsyncClient::Request
   virtual void cancel() override;
 

source/common/http/conn_manager_impl.cc

@@ -399,6 +399,7 @@ void ConnectionManagerImpl::chargeTracingStats(const Tracing::Reason& tracing_re
 
 ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connection_manager)
     : connection_manager_(connection_manager),
+      route_config_provider_(connection_manager.config_.routeConfigProvider()),
       snapped_route_config_(connection_manager.config_.routeConfigProvider().config()),
       stream_id_(connection_manager.random_generator_.random()),
       request_response_timespan_(new Stats::Timespan(
@@ -1125,6 +1126,7 @@ void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() {
   Router::RouteConstSharedPtr route;
   if (request_headers_ != nullptr) {
     route = snapped_route_config_->route(*request_headers_, stream_id_);
+    // route = route_config_provider_.config()->route(*request_headers_, stream_id_);
   }
   stream_info_.route_entry_ = route ? route->routeEntry() : nullptr;
   cached_route_ = std::move(route);
@@ -1137,6 +1139,13 @@ void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() {
   }
 }
 
+bool ConnectionManagerImpl::ActiveStream::requestRouteConfigUpdate(std::function<void()> cb) {
+  ASSERT(!request_headers_->Host()->value().empty());
+  auto host_header =
+      Http::LowerCaseString(std::string(request_headers_->Host()->value().getStringView())).get();
+  return route_config_provider_.requestVirtualHostsUpdate(host_header, cb);
+}
+
 void ConnectionManagerImpl::ActiveStream::sendLocalReply(
     bool is_grpc_request, Code code, absl::string_view body,
     const std::function<void(HeaderMap& headers)>& modify_headers, bool is_head_request,
@@ -1819,13 +1828,18 @@ Upstream::ClusterInfoConstSharedPtr ConnectionManagerImpl::ActiveStreamFilterBas
 }
 
 Router::RouteConstSharedPtr ConnectionManagerImpl::ActiveStreamFilterBase::route() {
-  if (!parent_.cached_route_.has_value()) {
+  if (!parent_.cached_route_.has_value() || parent_.cached_route_.value() == nullptr) {
     parent_.refreshCachedRoute();
   }
 
   return parent_.cached_route_.value();
 }
 
+bool ConnectionManagerImpl::ActiveStreamFilterBase::requestRouteConfigUpdate(
+    std::function<void()> cb) {
+  return parent_.requestRouteConfigUpdate(cb);
+}
+
 void ConnectionManagerImpl::ActiveStreamFilterBase::clearRouteCache() {
   parent_.cached_route_ = absl::optional<Router::RouteConstSharedPtr>();
   parent_.cached_cluster_info_ = absl::optional<Upstream::ClusterInfoConstSharedPtr>();

source/common/http/conn_manager_impl.h

@@ -131,6 +131,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     Event::Dispatcher& dispatcher() override;
     void resetStream() override;
     Router::RouteConstSharedPtr route() override;
+    bool requestRouteConfigUpdate(std::function<void()> cb) override;
     Upstream::ClusterInfoConstSharedPtr clusterInfo() override;
     void clearRouteCache() override;
     uint64_t streamId() override;
@@ -318,6 +319,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
     void responseDataTooLarge();
     void responseDataDrained();
+    bool requestRouteConfigUpdate(std::function<void()>) override { return false; }
 
     StreamEncoderFilterSharedPtr handle_;
   };
@@ -419,6 +421,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     void traceRequest();
 
     void refreshCachedRoute();
+    bool requestRouteConfigUpdate(std::function<void()> cb);
 
     // Pass on watermark callbacks to watermark subscribers. This boils down to passing watermark
     // events for this stream and the downstream connection to the router filter.
@@ -490,6 +493,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     bool hasCachedRoute() { return cached_route_.has_value() && cached_route_.value(); }
 
     ConnectionManagerImpl& connection_manager_;
+    Router::RouteConfigProvider& route_config_provider_;
     Router::ConfigConstSharedPtr snapped_route_config_;
     Tracing::SpanPtr active_span_;
     const uint64_t stream_id_;

source/common/router/rds_impl.cc

@@ -153,6 +153,13 @@ void RdsRouteConfigSubscription::onConfigUpdateFailed(const EnvoyException*) {
   init_target_.ready();
 }
 
+void RdsRouteConfigSubscription::ondemandUpdate(const std::set<std::string>& aliases) {
+  if (vhds_subscription_.get() == nullptr) {
+    return;
+  }
+  vhds_subscription_->ondemandUpdate(aliases);
+}
+
 bool RdsRouteConfigSubscription::validateUpdateSize(int num_resources) {
   if (num_resources == 0) {
     ENVOY_LOG(debug, "Missing RouteConfiguration for {} in onConfigUpdate()", route_config_name_);
@@ -172,7 +179,8 @@ RdsRouteConfigProviderImpl::RdsRouteConfigProviderImpl(
     Server::Configuration::FactoryContext& factory_context)
     : subscription_(std::move(subscription)),
       config_update_info_(subscription_->routeConfigUpdate()), factory_context_(factory_context),
-      tls_(factory_context.threadLocal().allocateSlot()) {
+      tls_(factory_context.threadLocal().allocateSlot()),
+      config_update_callbacks_(factory_context.threadLocal().allocateSlot()) {
   ConfigConstSharedPtr initial_config;
   if (config_update_info_->configInfo().has_value()) {
     initial_config = std::make_shared<ConfigImpl>(config_update_info_->routeConfiguration(),
@@ -183,6 +191,9 @@ RdsRouteConfigProviderImpl::RdsRouteConfigProviderImpl(
   tls_->set([initial_config](Event::Dispatcher&) -> ThreadLocal::ThreadLocalObjectSharedPtr {
     return std::make_shared<ThreadLocalConfig>(initial_config);
   });
+  config_update_callbacks_->set([](Event::Dispatcher&) -> ThreadLocal::ThreadLocalObjectSharedPtr {
+    return std::make_shared<ThreadLocalCallbacks>();
+  });
   subscription_->routeConfigProviders().insert(this);
 }
 
@@ -194,11 +205,30 @@ Router::ConfigConstSharedPtr RdsRouteConfigProviderImpl::config() {
   return tls_->getTyped<ThreadLocalConfig>().config_;
 }
 
+bool RdsRouteConfigProviderImpl::requestVirtualHostsUpdate(const std::string& for_domain,
+                                                           std::function<void()> cb) {
+  if (!config()->usesVhds()) {
+    return false;
+  }
+  factory_context_.dispatcher().post(
+      [this, for_domain]() -> void { subscription_->ondemandUpdate({for_domain}); });
+  config_update_callbacks_->getTyped<ThreadLocalCallbacks>().callbacks_.push(cb);
+
+  return true;
+}
+
 void RdsRouteConfigProviderImpl::onConfigUpdate() {
   ConfigConstSharedPtr new_config(
       new ConfigImpl(config_update_info_->routeConfiguration(), factory_context_, false));
-  tls_->runOnAllThreads(
-      [this, new_config]() -> void { tls_->getTyped<ThreadLocalConfig>().config_ = new_config; });
+  tls_->runOnAllThreads([this, new_config]() -> void {
+    tls_->getTyped<ThreadLocalConfig>().config_ = new_config;
+    auto callbacks = config_update_callbacks_->getTyped<ThreadLocalCallbacks>().callbacks_;
+    if (!callbacks.empty()) {
+      auto cb = callbacks.front();
+      callbacks.pop();
+      cb();
+    }
+  });
 }
 
 RouteConfigProviderManagerImpl::RouteConfigProviderManagerImpl(Server::Admin& admin) {

source/common/router/rds_impl.h

@@ -2,6 +2,7 @@
 
 #include <cstdint>
 #include <functional>
+#include <queue>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
@@ -67,6 +68,9 @@ class StaticRouteConfigProviderImpl : public RouteConfigProvider {
   }
   SystemTime lastUpdated() const override { return last_updated_; }
   void onConfigUpdate() override {}
+  bool requestVirtualHostsUpdate(const std::string&, std::function<void()>) override {
+    return false;
+  }
 
 private:
   ConfigConstSharedPtr config_;
@@ -108,6 +112,7 @@ class RdsRouteConfigSubscription : Envoy::Config::SubscriptionCallbacks,
   }
   RouteConfigUpdatePtr& routeConfigUpdate() { return config_update_info_; }
 
+  void ondemandUpdate(const std::set<std::string>& aliases);
   // Config::SubscriptionCallbacks
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                       const std::string& version_info) override;
@@ -146,6 +151,10 @@ class RdsRouteConfigSubscription : Envoy::Config::SubscriptionCallbacks,
 
 using RdsRouteConfigSubscriptionSharedPtr = std::shared_ptr<RdsRouteConfigSubscription>;
 
+struct ThreadLocalCallbacks : public ThreadLocal::ThreadLocalObject {
+  std::queue<std::function<void()>> callbacks_;
+};
+
 /**
  * Implementation of RouteConfigProvider that fetches the route configuration dynamically using
  * the subscription.
@@ -157,6 +166,7 @@ class RdsRouteConfigProviderImpl : public RouteConfigProvider,
 
   RdsRouteConfigSubscription& subscription() { return *subscription_; }
   void onConfigUpdate() override;
+  bool requestVirtualHostsUpdate(const std::string& for_domain, std::function<void()> cb) override;
 
   // Router::RouteConfigProvider
   Router::ConfigConstSharedPtr config() override;
@@ -174,10 +184,13 @@ class RdsRouteConfigProviderImpl : public RouteConfigProvider,
   RdsRouteConfigProviderImpl(RdsRouteConfigSubscriptionSharedPtr&& subscription,
                              Server::Configuration::FactoryContext& factory_context);
 
+  void addConfigUpdateCallback(std::function<void()> cb);
+
   RdsRouteConfigSubscriptionSharedPtr subscription_;
   RouteConfigUpdatePtr& config_update_info_;
   Server::Configuration::FactoryContext& factory_context_;
   ThreadLocal::SlotPtr tls_;
+  ThreadLocal::SlotPtr config_update_callbacks_;
 
   friend class RouteConfigProviderManagerImpl;
 };

source/common/router/router.cc

@@ -306,6 +306,20 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::HeaderMap& headers, bool e
   // Determine if there is a route entry or a direct response for the request.
   route_ = callbacks_->route();
   if (!route_) {
+    /*
+        attempting_internal_redirect_with_complete_stream_ = true;
+               // upstream_request_->upstream_timing_.last_upstream_rx_byte_received_ &&
+       downstream_end_stream_;
+
+        if (//downstream_end_stream_ &&
+            !callbacks_->decodingBuffer() && // Redirects with body not yet supported.
+            callbacks_->recreateStream()) {
+          //cluster_->stats().upstream_internal_redirect_succeeded_total_.inc();
+          return Http::FilterHeadersStatus::StopIteration;
+        }
+
+        attempting_internal_redirect_with_complete_stream_ = false;
+    */
     config_.stats_.no_route_.inc();
     ENVOY_STREAM_LOG(debug, "no cluster match for URL '{}'", *callbacks_,
                      headers.Path()->value().getStringView());