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 {
@@ -81,6 +82,10 @@ class Subscription {
    * @param resources vector of resource names to fetch.
    */
   virtual void updateResources(const std::vector<std::string>& resources) PURE;
+
+  virtual void updateResourcesViaAliases(const std::vector<std::string>&) {
+    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+  }
 };
 
 /**

include/envoy/http/filter.h

@@ -127,6 +127,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/BUILD

@@ -31,6 +31,16 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "route_config_update_info_interface",
+    hdrs = ["route_config_update_info.h"],
+    external_deps = ["abseil_optional"],
+    deps = [
+        ":router_interface",
+        "//source/common/protobuf",
+    ],
+)
+
 envoy_cc_library(
     name = "router_interface",
     hdrs = ["router.h"],

include/envoy/router/rds.h

@@ -43,6 +43,9 @@ class RouteConfigProvider {
    * @return the last time this RouteConfigProvider was updated. Used for config dumps.
    */
   virtual SystemTime lastUpdated() const PURE;
+
+  virtual void onConfigUpdate() PURE;
+  virtual bool requestConfigUpdate(const std::string for_domain, std::function<void()> cb) PURE;
 };
 
 typedef std::unique_ptr<RouteConfigProvider> RouteConfigProviderPtr;

include/envoy/router/route_config_update_info.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <memory>
+
+#include "envoy/api/v2/rds.pb.h"
+
+namespace Envoy {
+namespace Router {
+
+struct LastConfigInfo {
+  uint64_t last_config_hash_;
+  std::string last_config_version_;
+};
+
+class RouteConfigUpdateInfo {
+public:
+  virtual ~RouteConfigUpdateInfo() = default;
+
+  virtual absl::optional<LastConfigInfo> configInfo() const PURE;
+  virtual envoy::api::v2::RouteConfiguration& routeConfiguration() PURE;
+  virtual SystemTime lastUpdated() const PURE;
+};
+
+} // namespace Router
+} // namespace Envoy

include/envoy/router/router.h

@@ -786,6 +786,8 @@ class Config {
    * @return const std::string the RouteConfiguration name.
    */
   virtual const std::string& name() const PURE;
+
+  virtual bool usesVhds() const PURE;
 };
 
 typedef std::shared_ptr<const Config> ConfigConstSharedPtr;

source/common/config/delta_subscription_impl.h

@@ -215,6 +215,14 @@ class DeltaSubscriptionImpl
     stats_.update_attempt_.inc();
   }
 
+  void updateResourcesViaAliases(const std::vector<std::string>& aliases) override {
+    ResourceNameDiff diff;
+    std::copy(aliases.begin(), aliases.end(), std::inserter(diff.added_, diff.added_.begin()));
+    queueDiscoveryRequest(diff);
+    // sendDiscoveryRequest(diff);
+    stats_.update_attempt_.inc();
+  }
+
 private:
   void disableInitFetchTimeoutTimer() {
     if (init_fetch_timeout_timer_) {

source/common/config/resources.h

@@ -17,6 +17,7 @@ class TypeUrlValues {
   const std::string ClusterLoadAssignment{"type.googleapis.com/envoy.api.v2.ClusterLoadAssignment"};
   const std::string Secret{"type.googleapis.com/envoy.api.v2.auth.Secret"};
   const std::string RouteConfiguration{"type.googleapis.com/envoy.api.v2.RouteConfiguration"};
+  const std::string VirtualHost{"type.googleapis.com/envoy.api.v2.route.VirtualHost"};
 };
 
 typedef ConstSingleton<TypeUrlValues> TypeUrl;

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;
@@ -172,6 +174,7 @@ class AsyncStreamImpl : public AsyncClient::Stream,
     }
 
     const std::string& name() const override { return EMPTY_STRING; }
+    bool usesVhds() const override { return false; }
 
     static const std::list<LowerCaseString> internal_only_headers_;
   };
@@ -370,6 +373,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

@@ -397,6 +397,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(
@@ -1064,6 +1065,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);
@@ -1076,6 +1078,12 @@ void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() {
   }
 }
 
+bool ConnectionManagerImpl::ActiveStream::requestRouteConfigUpdate(std::function<void()> cb) {
+  // TODO check for an empty header?
+  auto host_header = Http::LowerCaseString(request_headers_->Host()->value().c_str()).get();
+  return route_config_provider_.requestConfigUpdate(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,
@@ -1710,13 +1718,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

@@ -121,6 +121,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;
@@ -271,6 +272,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
 
     void responseDataTooLarge();
     void responseDataDrained();
+    bool requestRouteConfigUpdate(std::function<void()>) { return false; }
 
     StreamEncoderFilterSharedPtr handle_;
   };
@@ -351,6 +353,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.
@@ -420,6 +423,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     void onRequestTimeout();
 
     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/BUILD

@@ -66,6 +66,31 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "vhds_lib",
+    srcs = ["vhds.cc"],
+    hdrs = ["vhds.h"],
+    deps = [
+        ":config_lib",
+        "//include/envoy/config:subscription_interface",
+        "//include/envoy/http:codes_interface",
+        "//include/envoy/local_info:local_info_interface",
+        "//include/envoy/router:rds_interface",
+        "//include/envoy/router:route_config_provider_manager_interface",
+        "//include/envoy/router:route_config_update_info_interface",
+        "//include/envoy/singleton:instance_interface",
+        "//include/envoy/thread_local:thread_local_interface",
+        "//source/common/common:assert_lib",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/config:rds_json_lib",
+        "//source/common/config:subscription_factory_lib",
+        "//source/common/config:utility_lib",
+        "//source/common/init:target_lib",
+        "//source/common/protobuf:utility_lib",
+        "@envoy_api//envoy/config/filter/network/http_connection_manager/v2:http_connection_manager_cc",
+    ],
+)
+
 envoy_cc_library(
     name = "rds_lib",
     srcs = ["rds_impl.cc"],
@@ -77,6 +102,7 @@ envoy_cc_library(
         "//include/envoy/local_info:local_info_interface",
         "//include/envoy/router:rds_interface",
         "//include/envoy/router:route_config_provider_manager_interface",
+        "//include/envoy/router:route_config_update_info_interface",
         "//include/envoy/server:admin_interface",
         "//include/envoy/singleton:instance_interface",
         "//include/envoy/thread_local:thread_local_interface",
@@ -87,6 +113,7 @@ envoy_cc_library(
         "//source/common/config:utility_lib",
         "//source/common/init:target_lib",
         "//source/common/protobuf:utility_lib",
+        "//source/common/router:vhds_lib",
         "@envoy_api//envoy/admin/v2alpha:config_dump_cc",
         "@envoy_api//envoy/api/v2:rds_cc",
         "@envoy_api//envoy/config/filter/network/http_connection_manager/v2:http_connection_manager_cc",
@@ -160,6 +187,34 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "on_demand_update_lib",
+    srcs = ["on_demand_update.cc"],
+    hdrs = ["on_demand_update.h"],
+    deps = [
+        "//include/envoy/event:dispatcher_interface",
+        "//include/envoy/event:timer_interface",
+        "//include/envoy/http:filter_interface",
+        "//include/envoy/server:filter_config_interface",
+        "//source/common/access_log:access_log_lib",
+        "//source/common/buffer:watermark_buffer_lib",
+        "//source/common/common:assert_lib",
+        "//source/common/common:empty_string",
+        "//source/common/common:enum_to_int",
+        "//source/common/common:hash_lib",
+        "//source/common/common:hex_lib",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/common:utility_lib",
+        "//source/common/grpc:common_lib",
+        "//source/common/http:codes_lib",
+        "//source/common/http:header_map_lib",
+        "//source/common/http:headers_lib",
+        "//source/common/http:message_lib",
+        "//source/common/http:utility_lib",
+        "@envoy_api//envoy/config/filter/http/router/v2:router_cc",
+    ],
+)
+
 envoy_cc_library(
     name = "router_ratelimit_lib",
     srcs = ["router_ratelimit.cc"],

source/common/router/config_impl.cc

@@ -1086,7 +1086,7 @@ VirtualHostImpl::virtualClusterFromEntries(const Http::HeaderMap& headers) const
 ConfigImpl::ConfigImpl(const envoy::api::v2::RouteConfiguration& config,
                        Server::Configuration::FactoryContext& factory_context,
                        bool validate_clusters_default)
-    : name_(config.name()) {
+    : name_(config.name()), uses_vhds_(config.has_vhds()) {
   route_matcher_ = std::make_unique<RouteMatcher>(
       config, *this, factory_context,
       PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, validate_clusters, validate_clusters_default));

source/common/router/config_impl.h

@@ -768,12 +768,15 @@ class ConfigImpl : public Config {
 
   const std::string& name() const override { return name_; }
 
+  bool usesVhds() const override { return uses_vhds_; }
+
 private:
   std::unique_ptr<RouteMatcher> route_matcher_;
   std::list<Http::LowerCaseString> internal_only_headers_;
   HeaderParserPtr request_headers_parser_;
   HeaderParserPtr response_headers_parser_;
   const std::string name_;
+  const bool uses_vhds_;
 };
 
 /**
@@ -789,6 +792,7 @@ class NullConfigImpl : public Config {
   }
 
   const std::string& name() const override { return name_; }
+  bool usesVhds() const override { return false; }
 
 private:
   std::list<Http::LowerCaseString> internal_only_headers_;

source/common/router/on_demand_update.cc

@@ -0,0 +1,58 @@
+#include "common/router/on_demand_update.h"
+
+#include "common/common/assert.h"
+#include "common/common/enum_to_int.h"
+#include "common/http/codes.h"
+
+#include "extensions/filters/http/well_known_names.h"
+
+namespace Envoy {
+namespace Router {
+
+void OnDemandRouteUpdate::requestRouteConfigUpdate() {
+  if (callbacks_->route() != nullptr) {
+    filter_return_ = FilterReturn::ContinueDecoding;
+  } else {
+    auto configUpdateScheduled =
+        callbacks_->requestRouteConfigUpdate([this]() -> void { onComplete(); });
+    filter_return_ =
+        configUpdateScheduled ? FilterReturn::StopDecoding : FilterReturn::ContinueDecoding;
+  }
+}
+
+Http::FilterHeadersStatus OnDemandRouteUpdate::decodeHeaders(Http::HeaderMap&, bool) {
+  requestRouteConfigUpdate();
+  return filter_return_ == FilterReturn::StopDecoding ? Http::FilterHeadersStatus::StopIteration
+                                                      : Http::FilterHeadersStatus::Continue;
+}
+
+Http::FilterDataStatus OnDemandRouteUpdate::decodeData(Buffer::Instance&, bool) {
+  return filter_return_ == FilterReturn::StopDecoding
+             ? Http::FilterDataStatus::StopIterationAndWatermark
+             : Http::FilterDataStatus::Continue;
+}
+
+Http::FilterTrailersStatus OnDemandRouteUpdate::decodeTrailers(Http::HeaderMap&) {
+  return filter_return_ == FilterReturn::StopDecoding ? Http::FilterTrailersStatus::StopIteration
+                                                      : Http::FilterTrailersStatus::Continue;
+}
+
+void OnDemandRouteUpdate::setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) {
+  callbacks_ = &callbacks;
+}
+
+void OnDemandRouteUpdate::onComplete() {
+  filter_return_ = FilterReturn::ContinueDecoding;
+
+  if (!callbacks_->decodingBuffer() && // Redirects with body not yet supported.
+      callbacks_->recreateStream()) {
+    // cluster_->stats().upstream_internal_redirect_succeeded_total_.inc();
+    return;
+  }
+
+  // recreating stream failed, continue the filter-chain
+  callbacks_->continueDecoding();
+}
+
+} // namespace Router
+} // namespace Envoy