On-demand VHDS implementation

CODEOWNERS

@@ -92,3 +92,5 @@ extensions/filters/common/original_src @snowp @klarose
 /*/extensions/filters/network/echo @htuch @alyssawilk
 /*/extensions/filters/udp/udp_proxy @mattklein123 @danzh2010
 /*/extensions/clusters/aggregate @yxue @snowp
+# support for on-demand VHDS requests
+/*/extensions/filters/http/on_demand @dmitri-d @htuch @lambdai

api/envoy/config/filter/http/on_demand/v2/BUILD

@@ -0,0 +1,7 @@
+# DO NOT EDIT. This file is generated by tools/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package()

api/envoy/config/filter/http/on_demand/v2/on_demand.proto

@@ -0,0 +1,15 @@
+syntax = "proto3";
+
+package envoy.config.filter.http.on_demand.v2;
+
+option java_outer_classname = "OnDemandProto";
+option java_multiple_files = true;
+option java_package = "io.envoyproxy.envoy.config.filter.http.on_demand.v2";
+
+import "validate/validate.proto";
+
+// [#protodoc-title: OnDemand]
+// [#extension: envoy.filters.http.on_demand]
+
+message OnDemand {
+}

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 {

include/envoy/http/filter.h

@@ -193,6 +193,9 @@ class StreamFilterCallbacks {
   virtual const ScopeTrackedObject& scope() PURE;
 };
 
+using RouteConfigUpdatedCallback = std::function<void(bool)>;
+using RouteConfigUpdatedCallbackSharedPtr = std::shared_ptr<RouteConfigUpdatedCallback>;
+
 /**
  * Stream decoder filter callbacks add additional callbacks that allow a decoding filter to restart
  * decoding if they decide to hold data (e.g. for buffering or rate limiting).
@@ -436,6 +439,21 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
    * @return The socket options to be applied to the upstream request.
    */
   virtual Network::Socket::OptionsSharedPtr getUpstreamSocketOptions() const PURE;
+
+  /**
+   * Schedules a request for a RouteConfiguration update from the management server.
+   * @param route_config_updated_cb callback to be called when the configuration update has been
+   * propagated to the worker thread.
+   */
+  virtual void
+  requestRouteConfigUpdate(RouteConfigUpdatedCallbackSharedPtr route_config_updated_cb) PURE;
+
+  /**
+   *
+   * @return absl::optional<Router::ConfigConstSharedPtr>. Contains a value if a non-scoped RDS
+   * route config provider is used.
+   */
+  virtual absl::optional<Router::ConfigConstSharedPtr> routeConfig() PURE;
 };
 
 /**

include/envoy/router/rds.h

@@ -3,6 +3,7 @@
 #include <memory>
 
 #include "envoy/api/v2/rds.pb.h"
+#include "envoy/http/filter.h"
 #include "envoy/router/router.h"
 
 namespace Envoy {
@@ -53,6 +54,16 @@ class RouteConfigProvider {
    * Validate if the route configuration can be applied to the context of the route config provider.
    */
   virtual void validateConfig(const envoy::api::v2::RouteConfiguration& config) const PURE;
+
+  /**
+   * Callback used to request an update to the route configuration from the management server.
+   * @param for_domain supplies the domain name that virtual hosts must match on
+   * @param route_config_updated_cb callback to be called when the configuration update has been
+   * propagated to worker threads
+   */
+  virtual void
+  requestVirtualHostsUpdate(const std::string& for_domain,
+                            Http::RouteConfigUpdatedCallbackSharedPtr route_config_updated_cb) PURE;
 };
 
 using RouteConfigProviderPtr = std::unique_ptr<RouteConfigProvider>;

include/envoy/router/route_config_update_receiver.h

@@ -54,6 +54,11 @@ class RouteConfigUpdateReceiver {
    */
   virtual const std::string& configVersion() PURE;
 
+  /**
+   * @return bool return whether VHDS configuration has been changed in the last RDS update.
+   */
+  virtual bool vhdsConfigurationChanged() const PURE;
+
   /**
    * @return uint64_t the hash value of RouteConfiguration.
    */
@@ -75,6 +80,11 @@ class RouteConfigUpdateReceiver {
    * @return SystemTime the time of the last update.
    */
   virtual SystemTime lastUpdated() const PURE;
+
+  /**
+   * @return aliases received with the last VHDS update.
+   */
+  virtual const std::set<std::string>& resourceIdsInLastVhdsUpdate() PURE;
 };
 
 using RouteConfigUpdatePtr = std::unique_ptr<RouteConfigUpdateReceiver>;

source/common/config/delta_subscription_impl.h

@@ -43,6 +43,7 @@ class DeltaSubscriptionImpl : public Subscription, public SubscriptionCallbacks
 
   // Config::Subscription
   void start(const std::set<std::string>& resource_names) override;
+
   void updateResourceInterest(const std::set<std::string>& update_to_these_names) override;
 
   // Config::SubscriptionCallbacks (all pass through to callbacks_!)

source/common/config/delta_subscription_state.cc

@@ -75,6 +75,10 @@ void DeltaSubscriptionState::handleGoodResponse(
       throw EnvoyException(
           fmt::format("duplicate name {} found among added/updated resources", resource.name()));
     }
+    // DeltaDiscoveryResponses for unresolved aliases don't contain an actual resource
+    if (!resource.has_resource() && resource.aliases_size() > 0) {
+      continue;
+    }
     if (message.type_url() != resource.resource().type_url()) {
       throw EnvoyException(fmt::format("type URL {} embedded in an individual Any does not match "
                                        "the message-wide type URL {} in DeltaDiscoveryResponse {}",

source/common/config/delta_subscription_state.h

@@ -27,6 +27,7 @@ class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
   // Update which resources we're interested in subscribing to.
   void updateSubscriptionInterest(const std::set<std::string>& cur_added,
                                   const std::set<std::string>& cur_removed);
+  void addAliasesToResolve(const std::set<std::string>& aliases);
 
   // Whether there was a change in our subscription interest we have yet to inform the server of.
   bool subscriptionUpdatePending() const;
@@ -74,6 +75,7 @@ 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);
 
   // 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

source/common/config/new_grpc_mux_impl.cc

@@ -65,6 +65,17 @@ void NewGrpcMuxImpl::onDiscoveryResponse(
               message->system_version_info(), message->type_url());
     return;
   }
+
+  // When an on-demand request is made a Watch is created using an alias, as the resource name isn't
+  // known at that point. When an update containing aliases comes back, we update Watches with
+  // resource names.
+  for (const auto& r : message->resources()) {
+    if (r.aliases_size() > 0) {
+      auto converted = sub->second->watch_map_.convertAliasWatchesToNameWatches(r);
+      sub->second->sub_state_.updateSubscriptionInterest(converted.added_, converted.removed_);
+    }
+  }
+
   kickOffAck(sub->second->sub_state_.handleResponse(*message));
 }
 

source/common/config/new_grpc_mux_impl.h

@@ -56,6 +56,25 @@ class NewGrpcMuxImpl : public GrpcMux,
                             GrpcMuxCallbacks&) override;
   void start() override;
 
+  struct SubscriptionStuff {
+    SubscriptionStuff(const std::string& type_url, std::chrono::milliseconds init_fetch_timeout,
+                      Event::Dispatcher& dispatcher, const LocalInfo::LocalInfo& local_info)
+        : sub_state_(type_url, watch_map_, local_info, init_fetch_timeout, dispatcher),
+          init_fetch_timeout_(init_fetch_timeout) {}
+
+    WatchMap watch_map_;
+    DeltaSubscriptionState sub_state_;
+    const std::chrono::milliseconds init_fetch_timeout_;
+
+    SubscriptionStuff(const SubscriptionStuff&) = delete;
+    SubscriptionStuff& operator=(const SubscriptionStuff&) = delete;
+  };
+
+  // for use in tests only
+  const absl::flat_hash_map<std::string, std::unique_ptr<SubscriptionStuff>>& subscriptions() {
+    return subscriptions_;
+  }
+
 private:
   Watch* addWatch(const std::string& type_url, const std::set<std::string>& resources,
                   SubscriptionCallbacks& callbacks, std::chrono::milliseconds init_fetch_timeout);
@@ -90,19 +109,6 @@ class NewGrpcMuxImpl : public GrpcMux,
   // description of how it interacts with pause() and resume().
   PausableAckQueue pausable_ack_queue_;
 
-  struct SubscriptionStuff {
-    SubscriptionStuff(const std::string& type_url, std::chrono::milliseconds init_fetch_timeout,
-                      Event::Dispatcher& dispatcher, const LocalInfo::LocalInfo& local_info)
-        : sub_state_(type_url, watch_map_, local_info, init_fetch_timeout, dispatcher),
-          init_fetch_timeout_(init_fetch_timeout) {}
-
-    WatchMap watch_map_;
-    DeltaSubscriptionState sub_state_;
-    const std::chrono::milliseconds init_fetch_timeout_;
-
-    SubscriptionStuff(const SubscriptionStuff&) = delete;
-    SubscriptionStuff& operator=(const SubscriptionStuff&) = delete;
-  };
   // Map key is type_url.
   absl::flat_hash_map<std::string, std::unique_ptr<SubscriptionStuff>> subscriptions_;
 

source/common/config/watch_map.cc

@@ -95,6 +95,31 @@ void WatchMap::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>
   }
 }
 
+// For responses to on-demand requests, replace the original watch for an alias
+// with one for the resource's name
+AddedRemoved WatchMap::convertAliasWatchesToNameWatches(const envoy::api::v2::Resource& resource) {
+  absl::flat_hash_set<Watch*> watches_to_update;
+  for (const auto& alias : resource.aliases()) {
+    const auto interested_watches = watch_interest_.find(alias);
+    if (interested_watches != watch_interest_.end()) {
+      for (const auto& interested_watch : interested_watches->second) {
+        watches_to_update.insert(interested_watch);
+      }
+    }
+  }
+
+  auto ret = AddedRemoved({}, {});
+  for (const auto& watch : watches_to_update) {
+    const auto& converted_watches = updateWatchInterest(watch, {resource.name()});
+    std::copy(converted_watches.added_.begin(), converted_watches.added_.end(),
+              std::inserter(ret.added_, ret.added_.end()));
+    std::copy(converted_watches.removed_.begin(), converted_watches.removed_.end(),
+              std::inserter(ret.removed_, ret.removed_.end()));
+  }
+
+  return ret;
+}
+
 void WatchMap::onConfigUpdate(
     const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
     const Protobuf::RepeatedPtrField<std::string>& removed_resources,

source/common/config/watch_map.h

@@ -77,6 +77,9 @@ class WatchMap : public SubscriptionCallbacks, public Logger::Loggable<Logger::I
   // updateWatchInterest().
   void removeWatch(Watch* watch);
 
+  // checks is a watch for an alias exists and replaces it with the resource's name
+  AddedRemoved convertAliasWatchesToNameWatches(const envoy::api::v2::Resource& resource);
+
   // SubscriptionCallbacks
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
                       const std::string& version_info) override;

source/common/http/async_client_impl.h

@@ -83,6 +83,12 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   AsyncStreamImpl(AsyncClientImpl& parent, AsyncClient::StreamCallbacks& callbacks,
                   const AsyncClient::StreamOptions& options);
 
+  // Http::StreamDecoderFilterCallbacks
+  void requestRouteConfigUpdate(Http::RouteConfigUpdatedCallbackSharedPtr) override {
+    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+  }
+  absl::optional<Router::ConfigConstSharedPtr> routeConfig() override { return {}; }
+
   // Http::AsyncClient::Stream
   void sendHeaders(HeaderMap& headers, bool end_stream) override;
   void sendData(Buffer::Instance& data, bool end_stream) override;

source/common/http/conn_manager_impl.cc

@@ -487,6 +487,14 @@ void ConnectionManagerImpl::chargeTracingStats(const Tracing::Reason& tracing_re
   }
 }
 
+void ConnectionManagerImpl::RdsRouteConfigUpdateRequester::requestRouteConfigUpdate(
+    const HeaderString& host, Http::RouteConfigUpdatedCallbackSharedPtr route_config_updated_cb) {
+  ASSERT(!host.empty());
+  const auto& host_header = absl::AsciiStrToLower(host.getStringView());
+  route_config_provider_->requestVirtualHostsUpdate(host_header,
+                                                    std::move(route_config_updated_cb));
+}
+
 ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connection_manager)
     : connection_manager_(connection_manager),
       stream_id_(connection_manager.random_generator_.random()),
@@ -502,7 +510,14 @@ ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connect
                connection_manager.config_.scopedRouteConfigProvider() == nullptr)),
          "Either routeConfigProvider or scopedRouteConfigProvider should be set in "
          "ConnectionManagerImpl.");
-
+  if (connection_manager.config_.routeConfigProvider() != nullptr) {
+    route_config_update_requester_ =
+        std::make_unique<ConnectionManagerImpl::RdsRouteConfigUpdateRequester>(
+            connection_manager.config_.routeConfigProvider());
+  } else if (connection_manager.config_.scopedRouteConfigProvider() != nullptr) {
+    route_config_update_requester_ =
+        std::make_unique<ConnectionManagerImpl::NullRouteConfigUpdateRequester>();
+  }
   ScopeTrackerScopeState scope(this,
                                connection_manager_.read_callbacks_->connection().dispatcher());
 
@@ -1370,6 +1385,20 @@ void ConnectionManagerImpl::ActiveStream::refreshCachedTracingCustomTags() {
   }
 }
 
+void ConnectionManagerImpl::ActiveStream::requestRouteConfigUpdate(
+    Http::RouteConfigUpdatedCallbackSharedPtr route_config_updated_cb) {
+  route_config_update_requester_->requestRouteConfigUpdate(request_headers_->Host()->value(),
+                                                           std::move(route_config_updated_cb));
+}
+
+absl::optional<Router::ConfigConstSharedPtr> ConnectionManagerImpl::ActiveStream::routeConfig() {
+  if (connection_manager_.config_.routeConfigProvider() == nullptr) {
+    return {};
+  }
+  return absl::optional<Router::ConfigConstSharedPtr>(
+      connection_manager_.config_.routeConfigProvider()->config());
+}
+
 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,
@@ -2267,6 +2296,16 @@ bool ConnectionManagerImpl::ActiveStreamDecoderFilter::recreateStream() {
   return true;
 }
 
+void ConnectionManagerImpl::ActiveStreamDecoderFilter::requestRouteConfigUpdate(
+    Http::RouteConfigUpdatedCallbackSharedPtr route_config_updated_cb) {
+  parent_.requestRouteConfigUpdate(std::move(route_config_updated_cb));
+}
+
+absl::optional<Router::ConfigConstSharedPtr>
+ConnectionManagerImpl::ActiveStreamDecoderFilter::routeConfig() {
+  return parent_.routeConfig();
+}
+
 Buffer::WatermarkBufferPtr ConnectionManagerImpl::ActiveStreamEncoderFilter::createBuffer() {
   auto buffer = new Buffer::WatermarkBuffer([this]() -> void { this->responseDataDrained(); },
                                             [this]() -> void { this->responseDataTooLarge(); });