init: order dynamic resource initialization to make RTDS always be fi…

…rst (#10362)

The new order of initialization:
1. Initialize all primary clusters
2. Initialize RTDS
3. Initialize secondary clusters
4. Initialize the rest of dynamic resources

Signed-off-by: Yan Avlasov <yavlasov@google.com>

include/envoy/runtime/runtime.h

@@ -259,6 +259,8 @@ class Loader {
 public:
   virtual ~Loader() = default;
 
+  using ReadyCallback = std::function<void()>;
+
   /**
    * Post-construction initialization. Runtime will be generally available after
    * the constructor is finished, with the exception of dynamic RTDS layers,
@@ -286,6 +288,12 @@ class Loader {
    * @param values the values to merge
    */
   virtual void mergeValues(const std::unordered_map<std::string, std::string>& values) PURE;
+
+  /**
+   * Initiate all RTDS subscriptions. The `on_done` callback is invoked when all RTDS requests
+   * have either received and applied their responses or timed out.
+   */
+  virtual void startRtdsSubscriptions(ReadyCallback on_done) PURE;
 };
 
 using LoaderPtr = std::unique_ptr<Loader>;

include/envoy/upstream/cluster_manager.h

@@ -73,6 +73,13 @@ class ClusterManagerFactory;
 /**
  * Manages connection pools and load balancing for upstream clusters. The cluster manager is
  * persistent and shared among multiple ongoing requests/connections.
+ * Cluster manager is initialized in two phases. In the first phase which begins at the construction
+ * all primary clusters (i.e. with endpoint assignments provisioned statically in bootstrap,
+ * discovered through DNS or file based CDS) are initialized.
+ * After the first phase has completed the server instance initializes services (i.e. RTDS) needed
+ * to successfully deploy the rest of dynamic configuration.
+ * In the second phase all secondary clusters (with endpoint assignments provisioned by xDS servers)
+ * are initialized and then the rest of the configuration provisioned through xDS.
  */
 class ClusterManager {
 public:
@@ -96,6 +103,14 @@ class ClusterManager {
    */
   virtual void setInitializedCb(std::function<void()> callback) PURE;
 
+  /**
+   * Start initialization of secondary clusters and then dynamically configured clusters.
+   * The "initialized callback" set in the method above is invoked when secondary and
+   * dynamically provisioned clusters have finished initializing.
+   */
+  virtual void
+  initializeSecondaryClusters(const envoy::config::bootstrap::v3::Bootstrap& bootstrap) PURE;
+
   using ClusterInfoMap = std::unordered_map<std::string, std::reference_wrapper<const Cluster>>;
 
   /**

source/common/runtime/BUILD

@@ -67,7 +67,9 @@ envoy_cc_library(
         "//source/common/config:subscription_base_interface",
         "//source/common/filesystem:directory_lib",
         "//source/common/grpc:common_lib",
+        "//source/common/init:manager_lib",
         "//source/common/init:target_lib",
+        "//source/common/init:watcher_lib",
         "//source/common/protobuf:message_validator_lib",
         "//source/common/protobuf:utility_lib",
         "@envoy_api//envoy/config/bootstrap/v3:pkg_cc_proto",

source/common/runtime/runtime_impl.cc

@@ -465,11 +465,12 @@ void ProtoLayer::walkProtoValue(const ProtobufWkt::Value& v, const std::string&
 
 LoaderImpl::LoaderImpl(Event::Dispatcher& dispatcher, ThreadLocal::SlotAllocator& tls,
                        const envoy::config::bootstrap::v3::LayeredRuntime& config,
-                       const LocalInfo::LocalInfo& local_info, Init::Manager& init_manager,
-                       Stats::Store& store, RandomGenerator& generator,
+                       const LocalInfo::LocalInfo& local_info, Stats::Store& store,
+                       RandomGenerator& generator,
                        ProtobufMessage::ValidationVisitor& validation_visitor, Api::Api& api)
     : generator_(generator), stats_(generateStats(store)), tls_(tls.allocateSlot()),
-      config_(config), service_cluster_(local_info.clusterName()), api_(api) {
+      config_(config), service_cluster_(local_info.clusterName()), api_(api),
+      init_watcher_("RDTS", [this]() { onRdtsReady(); }) {
   std::unordered_set<std::string> layer_names;
   for (const auto& layer : config_.layers()) {
     auto ret = layer_names.insert(layer.name());
@@ -497,7 +498,7 @@ LoaderImpl::LoaderImpl(Event::Dispatcher& dispatcher, ThreadLocal::SlotAllocator
     case envoy::config::bootstrap::v3::RuntimeLayer::LayerSpecifierCase::kRtdsLayer:
       subscriptions_.emplace_back(
           std::make_unique<RtdsSubscription>(*this, layer.rtds_layer(), store, validation_visitor));
-      init_manager.add(subscriptions_.back()->init_target_);
+      init_manager_.add(subscriptions_.back()->init_target_);
       break;
     default:
       NOT_REACHED_GCOVR_EXCL_LINE;
@@ -509,6 +510,16 @@ LoaderImpl::LoaderImpl(Event::Dispatcher& dispatcher, ThreadLocal::SlotAllocator
 
 void LoaderImpl::initialize(Upstream::ClusterManager& cm) { cm_ = &cm; }
 
+void LoaderImpl::startRtdsSubscriptions(ReadyCallback on_done) {
+  on_rtds_initialized_ = on_done;
+  init_manager_.initialize(init_watcher_);
+}
+
+void LoaderImpl::onRdtsReady() {
+  ENVOY_LOG(info, "RTDS has finished initialization");
+  on_rtds_initialized_();
+}
+
 RtdsSubscription::RtdsSubscription(
     LoaderImpl& parent, const envoy::config::bootstrap::v3::RuntimeLayer::RtdsLayer& rtds_layer,
     Stats::Store& store, ProtobufMessage::ValidationVisitor& validation_visitor)

source/common/runtime/runtime_impl.h

@@ -24,6 +24,7 @@
 #include "common/common/logger.h"
 #include "common/common/thread.h"
 #include "common/config/subscription_base.h"
+#include "common/init/manager_impl.h"
 #include "common/init/target_impl.h"
 #include "common/singleton/threadsafe_singleton.h"
 
@@ -242,15 +243,16 @@ class LoaderImpl : public Loader, Logger::Loggable<Logger::Id::runtime> {
 public:
   LoaderImpl(Event::Dispatcher& dispatcher, ThreadLocal::SlotAllocator& tls,
              const envoy::config::bootstrap::v3::LayeredRuntime& config,
-             const LocalInfo::LocalInfo& local_info, Init::Manager& init_manager,
-             Stats::Store& store, RandomGenerator& generator,
-             ProtobufMessage::ValidationVisitor& validation_visitor, Api::Api& api);
+             const LocalInfo::LocalInfo& local_info, Stats::Store& store,
+             RandomGenerator& generator, ProtobufMessage::ValidationVisitor& validation_visitor,
+             Api::Api& api);
 
   // Runtime::Loader
   void initialize(Upstream::ClusterManager& cm) override;
   const Snapshot& snapshot() override;
   std::shared_ptr<const Snapshot> threadsafeSnapshot() override;
   void mergeValues(const std::unordered_map<std::string, std::string>& values) override;
+  void startRtdsSubscriptions(ReadyCallback on_done) override;
 
 private:
   friend RtdsSubscription;
@@ -260,6 +262,7 @@ class LoaderImpl : public Loader, Logger::Loggable<Logger::Id::runtime> {
   // Load a new Snapshot into TLS
   void loadNewSnapshot();
   RuntimeStats generateStats(Stats::Store& store);
+  void onRdtsReady();
 
   RandomGenerator& generator_;
   RuntimeStats stats_;
@@ -269,6 +272,9 @@ class LoaderImpl : public Loader, Logger::Loggable<Logger::Id::runtime> {
   const std::string service_cluster_;
   Filesystem::WatcherPtr watcher_;
   Api::Api& api_;
+  ReadyCallback on_rtds_initialized_;
+  Init::WatcherImpl init_watcher_;
+  Init::ManagerImpl init_manager_{"RTDS"};
   std::vector<RtdsSubscriptionPtr> subscriptions_;
   Upstream::ClusterManager* cm_{};
 

source/common/upstream/cluster_manager_impl.cc

@@ -124,12 +124,12 @@ void ClusterManagerInitHelper::maybeFinishInitialize() {
   // Do not do anything if we are still doing the initial static load or if we are waiting for
   // CDS initialize.
   ENVOY_LOG(debug, "maybe finish initialize state: {}", enumToInt(state_));
-  if (state_ == State::Loading || state_ == State::WaitingForCdsInitialize) {
+  if (state_ == State::Loading || state_ == State::WaitingToStartCdsInitialization) {
     return;
   }
 
   // If we are still waiting for primary clusters to initialize, do nothing.
-  ASSERT(state_ == State::WaitingForStaticInitialize || state_ == State::CdsInitialized);
+  ASSERT(state_ == State::WaitingToStartSecondaryInitialization || state_ == State::CdsInitialized);
   ENVOY_LOG(debug, "maybe finish initialize primary init clusters empty: {}",
             primary_init_clusters_.empty());
   if (!primary_init_clusters_.empty()) {
@@ -163,9 +163,9 @@ void ClusterManagerInitHelper::maybeFinishInitialize() {
   // directly to initialized.
   started_secondary_initialize_ = false;
   ENVOY_LOG(debug, "maybe finish initialize cds api ready: {}", cds_ != nullptr);
-  if (state_ == State::WaitingForStaticInitialize && cds_) {
+  if (state_ == State::WaitingToStartSecondaryInitialization && cds_) {
     ENVOY_LOG(info, "cm init: initializing cds");
-    state_ = State::WaitingForCdsInitialize;
+    state_ = State::WaitingToStartCdsInitialization;
     cds_->initialize();
   } else {
     ENVOY_LOG(info, "cm init: all clusters initialized");
@@ -178,7 +178,14 @@ void ClusterManagerInitHelper::maybeFinishInitialize() {
 
 void ClusterManagerInitHelper::onStaticLoadComplete() {
   ASSERT(state_ == State::Loading);
-  state_ = State::WaitingForStaticInitialize;
+  // After initialization of primary clusters has completed, transition to
+  // waiting for signal to initialize secondary clusters and then CDS.
+  state_ = State::WaitingToStartSecondaryInitialization;
+}
+
+void ClusterManagerInitHelper::startInitializingSecondaryClusters() {
+  ASSERT(state_ == State::WaitingToStartSecondaryInitialization);
+  ENVOY_LOG(debug, "continue initializing secondary clusters");
   maybeFinishInitialize();
 }
 
@@ -187,7 +194,7 @@ void ClusterManagerInitHelper::setCds(CdsApi* cds) {
   cds_ = cds;
   if (cds_) {
     cds_->setInitializedCb([this]() -> void {
-      ASSERT(state_ == State::WaitingForCdsInitialize);
+      ASSERT(state_ == State::WaitingToStartCdsInitialization);
       state_ = State::CdsInitialized;
       maybeFinishInitialize();
     });
@@ -347,15 +354,22 @@ ClusterManagerImpl::ClusterManagerImpl(
   init_helper_.onStaticLoadComplete();
 
   ads_mux_->start();
+}
 
+void ClusterManagerImpl::initializeSecondaryClusters(
+    const envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+  init_helper_.startInitializingSecondaryClusters();
+
+  const auto& cm_config = bootstrap.cluster_manager();
   if (cm_config.has_load_stats_config()) {
     const auto& load_stats_config = cm_config.load_stats_config();
+
     load_stats_reporter_ = std::make_unique<LoadStatsReporter>(
-        local_info, *this, stats,
+        local_info_, *this, stats_,
         Config::Utility::factoryForGrpcApiConfigSource(*async_client_manager_, load_stats_config,
-                                                       stats, false)
+                                                       stats_, false)
             ->create(),
-        load_stats_config.transport_api_version(), main_thread_dispatcher);
+        load_stats_config.transport_api_version(), dispatcher_);
   }
 }
 

source/common/upstream/cluster_manager_impl.h

@@ -110,15 +110,18 @@ class ClusterManagerInitHelper : Logger::Loggable<Logger::Id::upstream> {
       : cm_(cm), per_cluster_init_callback_(per_cluster_init_callback) {}
 
   enum class State {
-    // Initial state. During this state all static clusters are loaded. Any phase 1 clusters
+    // Initial state. During this state all static clusters are loaded. Any primary clusters
     // are immediately initialized.
     Loading,
-    // During this state we wait for all static clusters to fully initialize. This requires
-    // completing phase 1 clusters, initializing phase 2 clusters, and then waiting for them.
-    WaitingForStaticInitialize,
-    // If CDS is configured, this state tracks waiting for the first CDS response to populate
-    // clusters.
-    WaitingForCdsInitialize,
+    // During this state cluster manager waits to start initializing secondary clusters. In this
+    // state all
+    // primary clusters have completed initialization. Initialization of the secondary clusters
+    // is started by the `initializeSecondaryClusters` method.
+    WaitingToStartSecondaryInitialization,
+    // In this state cluster manager waits for all secondary clusters (if configured) to finish
+    // initialization. Then, if CDS is configured, this state tracks waiting for the first CDS
+    // response to populate dynamically configured clusters.
+    WaitingToStartCdsInitialization,
     // During this state, all CDS populated clusters are undergoing either phase 1 or phase 2
     // initialization.
     CdsInitialized,
@@ -133,6 +136,8 @@ class ClusterManagerInitHelper : Logger::Loggable<Logger::Id::upstream> {
   void setInitializedCb(std::function<void()> callback);
   State state() const { return state_; }
 
+  void startInitializingSecondaryClusters();
+
 private:
   // To enable invariant assertions on the cluster lists.
   friend ClusterManagerImpl;
@@ -242,6 +247,9 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
 
   Config::SubscriptionFactory& subscriptionFactory() override { return subscription_factory_; }
 
+  void
+  initializeSecondaryClusters(const envoy::config::bootstrap::v3::Bootstrap& bootstrap) override;
+
 protected:
   virtual void postThreadLocalDrainConnections(const Cluster& cluster,
                                                const HostVector& hosts_removed);

source/server/server.cc

@@ -379,6 +379,21 @@ void InstanceImpl::initialize(const Options& options,
     dispatcher_->initializeStats(stats_store_, "server.");
   }
 
+  // The broad order of initialization from this point on is the following:
+  // 1. Statically provisioned configuration (bootstrap) are loaded.
+  // 2. Cluster manager is created and all primary clusters (i.e. with endpoint assignments
+  //    provisioned statically in bootstrap, discovered through DNS or file based CDS) are
+  //    initialized.
+  // 3. Various services are initialized and configured using the bootstrap config.
+  // 4. RTDS is initialized using primary clusters. This  allows runtime overrides to be fully
+  //    configured before the rest of xDS configuration is provisioned.
+  // 5. Secondary clusters (with endpoint assignments provisioned by xDS servers) are initialized.
+  // 6. The rest of the dynamic configuration is provisioned.
+  //
+  // Please note: this order requires that RTDS is provisioned using a primary cluster. If RTDS is
+  // provisioned through ADS then ADS must use primary cluster as well. This invariant is enforced
+  // during RTDS initialization and invalid configuration will be rejected.
+
   // Runtime gets initialized before the main configuration since during main configuration
   // load things may grab a reference to the loader for later use.
   runtime_singleton_ = std::make_unique<Runtime::ScopedLoaderSingleton>(
@@ -413,6 +428,27 @@ void InstanceImpl::initialize(const Options& options,
   // instantiated (which in turn relies on runtime...).
   Runtime::LoaderSingleton::get().initialize(clusterManager());
 
+  // If RTDS was not configured the `onRuntimeReady` callback is immediately invoked.
+  Runtime::LoaderSingleton::get().startRtdsSubscriptions([this]() { onRuntimeReady(); });
+
+  for (Stats::SinkPtr& sink : config_.statsSinks()) {
+    stats_store_.addSink(*sink);
+  }
+
+  // Some of the stat sinks may need dispatcher support so don't flush until the main loop starts.
+  // Just setup the timer.
+  stat_flush_timer_ = dispatcher_->createTimer([this]() -> void { flushStats(); });
+  stat_flush_timer_->enableTimer(config_.statsFlushInterval());
+
+  // GuardDog (deadlock detection) object and thread setup before workers are
+  // started and before our own run() loop runs.
+  guard_dog_ = std::make_unique<Server::GuardDogImpl>(stats_store_, config_, *api_);
+}
+
+void InstanceImpl::onRuntimeReady() {
+  // Begin initializing secondary clusters after RTDS configuration has been applied.
+  clusterManager().initializeSecondaryClusters(bootstrap_);
+
   if (bootstrap_.has_hds_config()) {
     const auto& hds_config = bootstrap_.hds_config();
     async_client_manager_ = std::make_unique<Grpc::AsyncClientManagerImpl>(
@@ -427,19 +463,6 @@ void InstanceImpl::initialize(const Options& options,
         *config_.clusterManager(), *local_info_, *admin_, *singleton_manager_, thread_local_,
         messageValidationContext().dynamicValidationVisitor(), *api_);
   }
-
-  for (Stats::SinkPtr& sink : config_.statsSinks()) {
-    stats_store_.addSink(*sink);
-  }
-
-  // Some of the stat sinks may need dispatcher support so don't flush until the main loop starts.
-  // Just setup the timer.
-  stat_flush_timer_ = dispatcher_->createTimer([this]() -> void { flushStats(); });
-  stat_flush_timer_->enableTimer(config_.statsFlushInterval());
-
-  // GuardDog (deadlock detection) object and thread setup before workers are
-  // started and before our own run() loop runs.
-  guard_dog_ = std::make_unique<Server::GuardDogImpl>(stats_store_, config_, *api_);
 }
 
 void InstanceImpl::startWorkers() {
@@ -459,8 +482,8 @@ Runtime::LoaderPtr InstanceUtil::createRuntime(Instance& server,
   ENVOY_LOG(info, "runtime: {}", MessageUtil::getYamlStringFromMessage(config.runtime()));
   return std::make_unique<Runtime::LoaderImpl>(
       server.dispatcher(), server.threadLocal(), config.runtime(), server.localInfo(),
-      server.initManager(), server.stats(), server.random(),
-      server.messageValidationContext().dynamicValidationVisitor(), server.api());
+      server.stats(), server.random(), server.messageValidationContext().dynamicValidationVisitor(),
+      server.api());
 }
 
 void InstanceImpl::loadServerFlags(const absl::optional<std::string>& flags_path) {
@@ -679,4 +702,4 @@ ProtobufTypes::MessagePtr InstanceImpl::dumpBootstrapConfig() {
 }
 
 } // namespace Server
-} // namespace Envoy
+} // namespace Envoy

source/server/server.h

@@ -285,6 +285,7 @@ class InstanceImpl final : Logger::Loggable<Logger::Id::main>,
   void terminate();
   void notifyCallbacksForStage(
       Stage stage, Event::PostCb completion_cb = [] {});
+  void onRuntimeReady();
 
   using LifecycleNotifierCallbacks = std::list<StageCallback>;
   using LifecycleNotifierCompletionCallbacks = std::list<StageCallbackWithCompletion>;
@@ -305,6 +306,9 @@ class InstanceImpl final : Logger::Loggable<Logger::Id::main>,
   const Options& options_;
   ProtobufMessage::ProdValidationContextImpl validation_context_;
   TimeSource& time_source_;
+  // Delete local_info_ as late as possible as some members below may reference it during their
+  // destruction.
+  LocalInfo::LocalInfoPtr local_info_;
   HotRestart& restarter_;
   const time_t start_time_;
   time_t original_start_time_;
@@ -328,7 +332,6 @@ class InstanceImpl final : Logger::Loggable<Logger::Id::main>,
   Configuration::MainImpl config_;
   Network::DnsResolverSharedPtr dns_resolver_;
   Event::TimerPtr stat_flush_timer_;
-  LocalInfo::LocalInfoPtr local_info_;
   DrainManagerPtr drain_manager_;
   AccessLog::AccessLogManagerImpl access_log_manager_;
   std::unique_ptr<Upstream::ClusterManagerFactory> cluster_manager_factory_;