On-demand VHDS implementation

CODEOWNERS

@@ -92,4 +92,6 @@ 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
 /*/extensions/filters/network/local_ratelimit @mattklein123 @junr03

api/BUILD

@@ -51,6 +51,7 @@ proto_library(
         "//envoy/config/filter/http/ip_tagging/v2:pkg",
         "//envoy/config/filter/http/jwt_authn/v2alpha:pkg",
         "//envoy/config/filter/http/lua/v2:pkg",
+        "//envoy/config/filter/http/on_demand/v2:pkg",
         "//envoy/config/filter/http/original_src/v2alpha1:pkg",
         "//envoy/config/filter/http/rate_limit/v2:pkg",
         "//envoy/config/filter/http/rbac/v2:pkg",
@@ -143,6 +144,7 @@ proto_library(
         "//envoy/extensions/filters/http/ip_tagging/v3alpha:pkg",
         "//envoy/extensions/filters/http/jwt_authn/v3alpha:pkg",
         "//envoy/extensions/filters/http/lua/v3alpha:pkg",
+        "//envoy/extensions/filters/http/on_demand/v3alpha:pkg",
         "//envoy/extensions/filters/http/original_src/v3alpha:pkg",
         "//envoy/extensions/filters/http/ratelimit/v3alpha:pkg",
         "//envoy/extensions/filters/http/rbac/v3alpha:pkg",

api/docs/BUILD

@@ -51,6 +51,7 @@ proto_library(
         "//envoy/config/filter/http/ip_tagging/v2:pkg",
         "//envoy/config/filter/http/jwt_authn/v2alpha:pkg",
         "//envoy/config/filter/http/lua/v2:pkg",
+        "//envoy/config/filter/http/on_demand/v2:pkg",
         "//envoy/config/filter/http/original_src/v2alpha1:pkg",
         "//envoy/config/filter/http/rate_limit/v2:pkg",
         "//envoy/config/filter/http/rbac/v2:pkg",

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

@@ -0,0 +1,9 @@
+# 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(
+    deps = ["@com_github_cncf_udpa//udpa/annotations:pkg"],
+)

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

@@ -0,0 +1,19 @@
+syntax = "proto3";
+
+package envoy.config.filter.http.on_demand.v2;
+
+import "udpa/annotations/migrate.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.config.filter.http.on_demand.v2";
+option java_outer_classname = "OnDemandProto";
+option java_multiple_files = true;
+option (udpa.annotations.file_migrate).move_to_package =
+    "envoy.extensions.filters.http.on_demand.v3alpha";
+
+// [#protodoc-title: OnDemand]
+// IP tagging :ref:`configuration overview <config_http_filters_on_demand>`.
+// [#extension: envoy.filters.http.on_demand]
+
+message OnDemand {
+}

api/envoy/extensions/filters/http/on_demand/v3alpha/BUILD

@@ -0,0 +1,12 @@
+# 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(
+    deps = [
+        "//envoy/config/filter/http/on_demand/v2:pkg",
+        "@com_github_cncf_udpa//udpa/annotations:pkg",
+    ],
+)

api/envoy/extensions/filters/http/on_demand/v3alpha/on_demand.proto

@@ -0,0 +1,20 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.http.on_demand.v3alpha;
+
+import "udpa/annotations/versioning.proto";
+
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.http.on_demand.v3alpha";
+option java_outer_classname = "OnDemandProto";
+option java_multiple_files = true;
+
+// [#protodoc-title: OnDemand]
+// IP tagging :ref:`configuration overview <config_http_filters_on_demand>`.
+// [#extension: envoy.filters.http.on_demand]
+
+message OnDemand {
+  option (udpa.annotations.versioning).previous_message_type =
+      "envoy.config.filter.http.on_demand.v2.OnDemand";
+}

docs/root/configuration/http/http_filters/http_filters.rst

@@ -25,6 +25,7 @@ HTTP filters
   ip_tagging_filter
   jwt_authn_filter
   lua_filter
+  on_demand_updates_filter
   original_src_filter
   rate_limit_filter
   rbac_filter

docs/root/configuration/http/http_filters/on_demand_updates_filter.rst

@@ -0,0 +1,16 @@
+.. _config_http_filters_on_demand:
+
+On-demand VHDS Updates
+======================
+
+The on-demand VHDS filter is used to request a :ref:`virtual host <envoy_api_msg_route.VirtualHost>`
+data if it's not already present in the :ref:`Route Configuration <envoy_api_msg_RouteConfiguration>`. The
+contents of the *Host* or *:authority* header is used to create the on-demand request. For an on-demand
+request to be created, :ref:`VHDS <envoy_api_field_RouteConfiguration.vhds>` must be enabled and either *Host*
+or *:authority* header be present.
+
+Configuration
+-------------
+* :ref:`v2 API reference <envoy_api_msg_config.filter.http.on_demand.v2.OnDemand>`
+* This filter should be configured with the name *envoy.on_demand*.
+* The filter should be placed before *envoy.router* filter in the HttpConnectionManager's filter chain.

generated_api_shadow/BUILD

@@ -51,6 +51,7 @@ proto_library(
         "//envoy/config/filter/http/ip_tagging/v2:pkg",
         "//envoy/config/filter/http/jwt_authn/v2alpha:pkg",
         "//envoy/config/filter/http/lua/v2:pkg",
+        "//envoy/config/filter/http/on_demand/v2:pkg",
         "//envoy/config/filter/http/original_src/v2alpha1:pkg",
         "//envoy/config/filter/http/rate_limit/v2:pkg",
         "//envoy/config/filter/http/rbac/v2:pkg",
@@ -143,6 +144,7 @@ proto_library(
         "//envoy/extensions/filters/http/ip_tagging/v3alpha:pkg",
         "//envoy/extensions/filters/http/jwt_authn/v3alpha:pkg",
         "//envoy/extensions/filters/http/lua/v3alpha:pkg",
+        "//envoy/extensions/filters/http/on_demand/v3alpha:pkg",
         "//envoy/extensions/filters/http/original_src/v3alpha:pkg",
         "//envoy/extensions/filters/http/ratelimit/v3alpha:pkg",
         "//envoy/extensions/filters/http/rbac/v3alpha:pkg",

include/envoy/http/filter.h

@@ -193,6 +193,22 @@ class StreamFilterCallbacks {
   virtual const ScopeTrackedObject& scope() PURE;
 };
 
+/**
+ * RouteConfigUpdatedCallback is used to notify an OnDemandRouteUpdate filter about completion of a
+ * RouteConfig update. The filter (and the associated ActiveStream) where the original on-demand
+ * request was originated can be destroyed before a response to an on-demand update request is
+ * received and updates are propagated. To handle this:
+ *
+ * OnDemandRouteUpdate filter instance holds a RouteConfigUpdatedCallbackSharedPtr to a callback.
+ * Envoy::Router::RdsRouteConfigProviderImpl holds a weak pointer to the RouteConfigUpdatedCallback
+ * above in an Envoy::Router::UpdateOnDemandCallback struct
+ *
+ * In RdsRouteConfigProviderImpl::onConfigUpdate(), before invoking the callback, a check is made to
+ * verify if the callback is still available.
+ */
+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 +452,23 @@ 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. Scoped RDS provides are not supported at the moment, as
+   * retrieval of a route configuration in their case requires passing of http request headers
+   * as a parameter.
+   */
+  virtual absl::optional<Router::ConfigConstSharedPtr> routeConfig() PURE;
 };
 
 /**

include/envoy/router/rds.h

@@ -3,6 +3,7 @@
 #include <memory>
 
 #include "envoy/config/route/v3alpha/route.pb.h"
+#include "envoy/http/filter.h"
 #include "envoy/router/router.h"
 
 namespace Envoy {
@@ -54,6 +55,16 @@ class RouteConfigProvider {
    */
   virtual void
   validateConfig(const envoy::config::route::v3alpha::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, Event::Dispatcher& thread_local_dispatcher,
+      std::weak_ptr<Http::RouteConfigUpdatedCallback> route_config_updated_cb) PURE;
 };
 
 using RouteConfigProviderPtr = std::unique_ptr<RouteConfigProvider>;

include/envoy/router/route_config_update_receiver.h

@@ -55,6 +55,14 @@ class RouteConfigUpdateReceiver {
    */
   virtual const std::string& configVersion() PURE;
 
+  /**
+   * @return bool return whether VHDS configuration has been changed in the last RDS update.
+   */
+  // TODO(dmitri-d): Consider splitting RouteConfigUpdateReceiver into a RouteConfig state and a
+  // last update state. The latter could be passed to callbacks as a parameter, which would make the
+  // intent and the lifecycle of the "last update state" less muddled.
+  virtual bool vhdsConfigurationChanged() const PURE;
+
   /**
    * @return uint64_t the hash value of RouteConfiguration.
    */
@@ -76,6 +84,12 @@ class RouteConfigUpdateReceiver {
    * @return SystemTime the time of the last update.
    */
   virtual SystemTime lastUpdated() const PURE;
+
+  /**
+   * @return the union of all resource names and aliases (if any) 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

@@ -29,6 +29,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;
@@ -79,6 +80,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::service::discovery::v3alpha::DeltaDiscoveryResponse& 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) {
+      AddedRemoved 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

@@ -60,6 +60,25 @@ class NewGrpcMuxImpl
                             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);
@@ -94,19 +113,6 @@ class NewGrpcMuxImpl
   // 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_;