[router] Add SRDS configUpdate impl

source/common/config/config_provider_impl.h

@@ -1,7 +1,5 @@
 #pragma once
 
-#include <memory>
-
 #include "envoy/config/config_provider.h"
 #include "envoy/config/config_provider_manager.h"
 #include "envoy/init/manager.h"
@@ -146,9 +144,7 @@ class MutableConfigProviderCommonBase;
  * shared ownership of the underlying subscription.
  *
  */
-class ConfigSubscriptionCommonBase
-    : protected Logger::Loggable<Logger::Id::config>,
-      public std::enable_shared_from_this<ConfigSubscriptionCommonBase> {
+class ConfigSubscriptionCommonBase : protected Logger::Loggable<Logger::Id::config> {
 public:
   // Callback for updating a Config implementation held in each worker thread, the callback is
   // called in applyConfigUpdate() with the current version Config, and is expected to return the
@@ -224,21 +220,14 @@ class ConfigSubscriptionCommonBase
    */
   void applyConfigUpdate(
       const ConfigUpdateCb& update_fn, const Event::PostCb& complete_cb = []() {}) {
-    // It is safe to call shared_from_this here as this is in main thread, and destruction of a
-    // ConfigSubscriptionCommonBase owner (i.e., a provider) happens in main thread as well.
-    auto shared_this = shared_from_this();
     tls_->runOnAllThreads(
         [this, update_fn]() {
-          tls_->getTyped<ThreadLocalConfig>().config_ = update_fn(this->getConfig());
+          // NOTE: there is a known race condition between *this* subscription be teared down in
+          // main thread and the posted callback been executed before the destruction. See more
+          // details in https://github.com/envoyproxy/envoy/issues/7902
+          tls_->getTyped<ThreadLocalConfig>().config_ = update_fn(getConfig());
         },
-        // During the update propagation, a subscription may get teared down in main thread due to
-        // all owners/providers destructed in a xDS update (e.g. LDS demolishes a
-        // RouteConfigProvider and its subscription).
-        // If such a race condition happens, holding a reference to the "*this" subscription
-        // instance in this cb will ensure the shared "*this" gets posted back to main thread, after
-        // all the workers finish calling the update_fn, at which point it's safe to destruct
-        // "*this" instance.
-        [shared_this, complete_cb]() { complete_cb(); });
+        complete_cb);
   }
 
   void setLastUpdated() { last_updated_ = time_source_.systemTime(); }
@@ -287,8 +276,8 @@ class ConfigSubscriptionInstance : public ConfigSubscriptionCommonBase {
 
   /**
    * Must be called by the derived class' constructor.
-   * @param initial_config supplies an initial Envoy::Config::ConfigProvider::Config associated with
-   *                       the underlying subscription, shared across all providers and workers.
+   * @param initial_config supplies an initial Envoy::Config::ConfigProvider::Config associated
+   * with the underlying subscription, shared across all providers and workers.
    */
   void initialize(const ConfigProvider::ConfigConstSharedPtr& initial_config) {
     tls_->set([initial_config](Event::Dispatcher&) -> ThreadLocal::ThreadLocalObjectSharedPtr {
@@ -302,7 +291,8 @@ class ConfigSubscriptionInstance : public ConfigSubscriptionCommonBase {
    * @param config_proto supplies the newly received config proto.
    * @param config_name supplies the name associated with the config.
    * @param version_info supplies the version associated with the config.
-   * @return bool false when the config proto has no delta from the previous config, true otherwise.
+   * @return bool false when the config proto has no delta from the previous config, true
+   * otherwise.
    */
   bool checkAndApplyConfigUpdate(const Protobuf::Message& config_proto,
                                  const std::string& config_name, const std::string& version_info);
@@ -369,8 +359,8 @@ class MutableConfigProviderCommonBase : public ConfigProvider {
 };
 
 /**
- * Provides generic functionality required by all config provider managers, such as managing shared
- * lifetime of subscriptions and dynamic config providers, along with determining which
+ * Provides generic functionality required by all config provider managers, such as managing
+ * shared lifetime of subscriptions and dynamic config providers, along with determining which
  * subscriptions should be associated with newly instantiated providers.
  *
  * The implementation of this class is not thread safe. Note that ImmutableConfigProviderBase
@@ -379,9 +369,9 @@ class MutableConfigProviderCommonBase : public ConfigProvider {
  *
  * All config processing is done on the main thread, so instantiation of *ConfigProvider* objects
  * via createStaticConfigProvider() and createXdsConfigProvider() is naturally thread safe. Care
- * must be taken with regards to destruction of these objects, since it must also happen on the main
- * thread _prior_ to destruction of the ConfigProviderManagerImplBase object from which they were
- * created.
+ * must be taken with regards to destruction of these objects, since it must also happen on the
+ * main thread _prior_ to destruction of the ConfigProviderManagerImplBase object from which they
+ * were created.
  *
  * This class can not be instantiated directly; instead, it provides the foundation for
  * dynamic config provider implementations which derive from it.
@@ -415,8 +405,8 @@ class ConfigProviderManagerImplBase : public ConfigProviderManager, public Singl
   const ConfigProviderSet& immutableConfigProviders(ConfigProviderInstanceType type) const;
 
   /**
-   * Returns the subscription associated with the config_source_proto; if none exists, a new one is
-   * allocated according to the subscription_factory_fn.
+   * Returns the subscription associated with the config_source_proto; if none exists, a new one
+   * is allocated according to the subscription_factory_fn.
    * @param config_source_proto supplies the proto specifying the config subscription parameters.
    * @param init_manager supplies the init manager.
    * @param subscription_factory_fn supplies a function to be called when a new subscription needs

source/common/router/BUILD

@@ -132,6 +132,7 @@ envoy_cc_library(
         "//include/envoy/singleton:instance_interface",
         "//include/envoy/thread_local:thread_local_interface",
         "//source/common/common:assert_lib",
+        "//source/common/common:callback_impl_lib",
         "//source/common/common:minimal_logger_lib",
         "//source/common/config:rds_json_lib",
         "//source/common/config:subscription_factory_lib",
@@ -168,12 +169,14 @@ envoy_cc_library(
     srcs = ["scoped_rds.cc"],
     hdrs = ["scoped_rds.h"],
     deps = [
+        ":rds_lib",
         ":scoped_config_lib",
         "//include/envoy/config:config_provider_interface",
         "//include/envoy/config:subscription_interface",
         "//include/envoy/router:route_config_provider_manager_interface",
         "//include/envoy/stats:stats_interface",
         "//source/common/common:assert_lib",
+        "//source/common/common:cleanup_lib",
         "//source/common/common:minimal_logger_lib",
         "//source/common/config:config_provider_lib",
         "//source/common/init:manager_lib",

source/common/router/rds_impl.cc

@@ -120,6 +120,7 @@ void RdsRouteConfigSubscription::onConfigUpdate(
       }
       vhds_subscription_.release();
     }
+    update_callback_manager_.runCallbacks();
   }
 
   init_target_.ready();

source/common/router/rds_impl.h

@@ -22,6 +22,7 @@
 #include "envoy/stats/scope.h"
 #include "envoy/thread_local/thread_local.h"
 
+#include "common/common/callback_impl.h"
 #include "common/common/logger.h"
 #include "common/init/target_impl.h"
 #include "common/protobuf/utility.h"
@@ -31,6 +32,9 @@
 namespace Envoy {
 namespace Router {
 
+// For friend class declaration in RdsRouteConfigSubscription.
+class ScopedRdsConfigSubscription;
+
 /**
  * Route configuration provider utilities.
  */
@@ -122,6 +126,10 @@ class RdsRouteConfigSubscription : Envoy::Config::SubscriptionCallbacks,
         .name();
   }
 
+  Common::CallbackHandle* addUpdateCallback(std::function<void()> callback) {
+    return update_callback_manager_.add(callback);
+  }
+
   RdsRouteConfigSubscription(
       const envoy::config::filter::network::http_connection_manager::v2::Rds& rds,
       const uint64_t manager_identifier, Server::Configuration::FactoryContext& factory_context,
@@ -143,8 +151,11 @@ class RdsRouteConfigSubscription : Envoy::Config::SubscriptionCallbacks,
   VhdsSubscriptionPtr vhds_subscription_;
   RouteConfigUpdatePtr config_update_info_;
   ProtobufMessage::ValidationVisitor& validation_visitor_;
+  Common::CallbackManager<> update_callback_manager_;
 
   friend class RouteConfigProviderManagerImpl;
+  // Access to addUpdateCallback
+  friend class ScopedRdsConfigSubscription;
 };
 
 using RdsRouteConfigSubscriptionSharedPtr = std::shared_ptr<RdsRouteConfigSubscription>;

source/common/router/scoped_config_impl.cc

@@ -70,6 +70,22 @@ HeaderValueExtractorImpl::computeFragment(const Http::HeaderMap& headers) const
   return nullptr;
 }
 
+ScopedRouteInfo::ScopedRouteInfo(envoy::api::v2::ScopedRouteConfiguration&& config_proto,
+                                 ConfigConstSharedPtr&& route_config)
+    : config_proto_(std::move(config_proto)), route_config_(std::move(route_config)) {
+  // TODO(stevenzzzz): Maybe worth a KeyBuilder abstraction when there are more than one type of
+  // Fragment.
+  for (const auto& fragment : config_proto_.key().fragments()) {
+    switch (fragment.type_case()) {
+    case envoy::api::v2::ScopedRouteConfiguration::Key::Fragment::kStringKey:
+      scope_key_.addFragment(std::make_unique<StringKeyFragment>(fragment.string_key()));
+      break;
+    default:
+      NOT_REACHED_GCOVR_EXCL_LINE;
+    }
+  }
+}
+
 ScopeKeyBuilderImpl::ScopeKeyBuilderImpl(ScopedRoutes::ScopeKeyBuilder&& config)
     : ScopeKeyBuilderBase(std::move(config)) {
   for (const auto& fragment_builder : config_.fragments()) {
@@ -102,6 +118,7 @@ void ScopedConfigImpl::addOrUpdateRoutingScope(
     const ScopedRouteInfoConstSharedPtr& scoped_route_info) {
   const auto iter = scoped_route_info_by_name_.find(scoped_route_info->scopeName());
   if (iter != scoped_route_info_by_name_.end()) {
+    ASSERT(scoped_route_info_by_key_.contains(iter->second->scopeKey().hash()));
     scoped_route_info_by_key_.erase(iter->second->scopeKey().hash());
   }
   scoped_route_info_by_name_[scoped_route_info->scopeName()] = scoped_route_info;
@@ -111,7 +128,7 @@ void ScopedConfigImpl::addOrUpdateRoutingScope(
 void ScopedConfigImpl::removeRoutingScope(const std::string& scope_name) {
   const auto iter = scoped_route_info_by_name_.find(scope_name);
   if (iter != scoped_route_info_by_name_.end()) {
-    ASSERT(scoped_route_info_by_key_.count(iter->second->scopeKey().hash()) == 1);
+    ASSERT(scoped_route_info_by_key_.contains(iter->second->scopeKey().hash()));
     scoped_route_info_by_key_.erase(iter->second->scopeKey().hash());
     scoped_route_info_by_name_.erase(iter);
   }

source/common/router/scoped_config_impl.h

@@ -66,8 +66,11 @@ class ScopeKey {
 private:
   // Update the key's hash with the new fragment hash.
   void updateHash(const ScopeKeyFragmentBase& fragment) {
-    absl::uint128 buffer = absl::MakeUint128(hash_, fragment.hash());
-    hash_ = HashUtil::xxHash64(absl::string_view(reinterpret_cast<char*>(&buffer), 16));
+    std::stringbuf buffer;
+    buffer.sputn(reinterpret_cast<const char*>(&hash_), sizeof(hash_));
+    const auto& fragment_hash = fragment.hash();
+    buffer.sputn(reinterpret_cast<const char*>(&fragment_hash), sizeof(fragment_hash));
+    hash_ = HashUtil::xxHash64(buffer.str());
   }
 
   uint64_t hash_{0};
@@ -147,36 +150,17 @@ class ScopeKeyBuilderImpl : public ScopeKeyBuilderBase {
 class ScopedRouteInfo {
 public:
   ScopedRouteInfo(envoy::api::v2::ScopedRouteConfiguration&& config_proto,
-                  std::unique_ptr<RouteConfigProvider>&& route_provider)
-      : config_proto_(std::move(config_proto)), route_provider_(std::move(route_provider)) {
-    ASSERT(route_provider_ != nullptr, "ScopedRouteInfo expects a valid RouteConfigProvider.");
-    ASSERT(
-        !route_provider_->configInfo().has_value() ||
-            route_provider_->config()->name() == config_proto_.route_configuration_name(),
-        fmt::format("RouteConfigProvider's name '{}' doesn't match route_configuration_name '{}'.",
-                    route_provider_->config()->name(), config_proto_.route_configuration_name()));
-    // TODO(stevenzzzz): Maybe worth a KeyBuilder abstraction when there are more than one type of
-    // Fragment.
-    for (const auto& fragment : config_proto_.key().fragments()) {
-      switch (fragment.type_case()) {
-      case envoy::api::v2::ScopedRouteConfiguration::Key::Fragment::kStringKey:
-        scope_key_.addFragment(std::make_unique<StringKeyFragment>(fragment.string_key()));
-        break;
-      default:
-        NOT_REACHED_GCOVR_EXCL_LINE;
-      }
-    }
-  }
+                  ConfigConstSharedPtr&& route_config);
 
-  Router::ConfigConstSharedPtr routeConfig() const { return route_provider_->config(); }
+  ConfigConstSharedPtr routeConfig() const { return route_config_; }
   const ScopeKey& scopeKey() const { return scope_key_; }
   const envoy::api::v2::ScopedRouteConfiguration& configProto() const { return config_proto_; }
   const std::string& scopeName() const { return config_proto_.name(); }
 
 private:
-  const envoy::api::v2::ScopedRouteConfiguration config_proto_;
+  envoy::api::v2::ScopedRouteConfiguration config_proto_;
   ScopeKey scope_key_;
-  std::unique_ptr<RouteConfigProvider> route_provider_;
+  ConfigConstSharedPtr route_config_;
 };
 using ScopedRouteInfoConstSharedPtr = std::shared_ptr<const ScopedRouteInfo>;
 // Ordered map for consistent config dumping.