Init manager checking unready targets with target-aware watchers

include/envoy/runtime/runtime.h

@@ -266,7 +266,7 @@ class Loader {
 public:
   virtual ~Loader() = default;
 
-  using ReadyCallback = std::function<void()>;
+  using ReadyCallback = std::function<void(const std::string)>;
 
   /**
    * Post-construction initialization. Runtime will be generally available after

source/common/config/config_provider_impl.h

@@ -206,7 +206,7 @@ class ConfigSubscriptionCommonBase : protected Logger::Loggable<Logger::Id::conf
         parent_init_target_(fmt::format("ConfigSubscriptionCommonBase init target '{}'", name_),
                             [this]() { local_init_manager_.initialize(local_init_watcher_); }),
         local_init_watcher_(fmt::format("ConfigSubscriptionCommonBase local watcher '{}'", name_),
-                            [this]() { parent_init_target_.ready(); }),
+                            [this](const std::string) { parent_init_target_.ready(); }),
         local_init_manager_(
             fmt::format("ConfigSubscriptionCommonBase local init manager '{}'", name_)),
         manager_identifier_(manager_identifier), config_provider_manager_(config_provider_manager),

source/common/init/manager_impl.cc

@@ -7,7 +7,7 @@ namespace Init {
 
 ManagerImpl::ManagerImpl(absl::string_view name)
     : name_(fmt::format("init manager {}", name)), state_(State::Uninitialized), count_(0),
-      watcher_(name_, [this]() { onTargetReady(); }) {}
+      watcher_(name_, [this](const std::string target_name) { onTargetReady(target_name); }) {}
 
 Manager::State ManagerImpl::state() const { return state_; }
 
@@ -18,13 +18,16 @@ void ManagerImpl::add(const Target& target) {
   case State::Uninitialized:
     // If the manager isn't initialized yet, save the target handle to be initialized later.
     ENVOY_LOG(debug, "added {} to {}", target.name(), name_);
+    target_names_.push_back(std::string(target.name()));
+    ++target_names_count_[std::string(target.name())];
     target_handles_.push_back(std::move(target_handle));
     return;
   case State::Initializing:
     // If the manager is already initializing, initialize the new target immediately. Note that
     // it's important in this case that count_ was incremented above before calling the target,
     // because if the target calls the init manager back immediately, count_ will be decremented
     // here (see the definition of watcher_ above).
+    ++target_names_count_[std::string(target.name())];
     target_handle->initialize(watcher_);
     return;
   case State::Initialized:
@@ -51,18 +54,34 @@ void ManagerImpl::initialize(const Watcher& watcher) {
 
     // Attempt to initialize each target. If a target is unavailable, treat it as though it
     // completed immediately.
+    uint32_t target_index = 0;
     for (const auto& target_handle : target_handles_) {
       if (!target_handle->initialize(watcher_)) {
-        onTargetReady();
+        onTargetReady(target_names_[target_index]);
       }
+      target_index++;
     }
   }
 }
 
-void ManagerImpl::onTargetReady() {
+// For init manager to query unready targets
+void ManagerImpl::checkUnreadyTargets() {
+  ENVOY_LOG(debug, "QUERY: List of unready targets..................");
+  for(const auto &unready_target : target_names_count_) {
+    ENVOY_LOG(debug, "{} x {}", unready_target.first, unready_target.second);
+  }
+  ENVOY_LOG(debug, "QUERY ENDS......................................");
+}
+
+void ManagerImpl::onTargetReady(const std::string target_name) {
   // If there are no remaining targets and one mysteriously calls us back, this manager is haunted.
   ASSERT(count_ != 0, fmt::format("{} called back by target after initialization complete"));
 
+  // Decrease count of a target_name by 1
+  if(--target_names_count_[target_name] == 0) {
+    target_names_count_.erase(target_name);
+  }
+
   // If there are no uninitialized targets remaining when called back by a target, that means it was
   // the last. Signal `ready` to the handle we saved in `initialize`.
   if (--count_ == 0) {

source/common/init/manager_impl.h

@@ -1,6 +1,8 @@
 #pragma once
 
 #include <list>
+#include <vector>
+#include <unordered_map>
 
 #include "envoy/init/manager.h"
 
@@ -35,8 +37,11 @@ class ManagerImpl : public Manager, Logger::Loggable<Logger::Id::init> {
   void add(const Target& target) override;
   void initialize(const Watcher& watcher) override;
 
+  // For init manager to query unready targets
+  void checkUnreadyTargets();
+
 private:
-  void onTargetReady();
+  void onTargetReady(const std::string target_name);
   void ready();
 
   // Human-readable name for logging
@@ -56,6 +61,12 @@ class ManagerImpl : public Manager, Logger::Loggable<Logger::Id::init> {
 
   // All registered targets
   std::list<TargetHandlePtr> target_handles_;
+
+  // Corresponding name of registered targets
+  std::vector<std::string> target_names_;
+
+  // Count of target_name
+  std::unordered_map<std::string, uint32_t> target_names_count_;
 };
 
 } // namespace Init

source/common/init/watcher_impl.cc

@@ -12,7 +12,7 @@ bool WatcherHandleImpl::ready() const {
   if (locked_fn) {
     // If we can "lock" a shared pointer to the watcher's callback function, call it.
     ENVOY_LOG(debug, "{} initialized, notifying {}", handle_name_, name_);
-    (*locked_fn)();
+    (*locked_fn)(handle_name_);
     return true;
   } else {
     // If not, the watcher was already destroyed.

source/common/init/watcher_impl.h

@@ -13,7 +13,7 @@ namespace Init {
  * A watcher is just a glorified callback function, called by a target or a manager when
  * initialization completes.
  */
-using ReadyFn = std::function<void()>;
+using ReadyFn = std::function<void(const std::string)>;
 
 /**
  * A WatcherHandleImpl functions as a weak reference to a Watcher. It is how a TargetImpl safely

source/common/router/rds_impl.cc

@@ -73,7 +73,7 @@ RdsRouteConfigSubscription::RdsRouteConfigSubscription(
       parent_init_target_(fmt::format("RdsRouteConfigSubscription init {}", route_config_name_),
                           [this]() { local_init_manager_.initialize(local_init_watcher_); }),
       local_init_watcher_(fmt::format("RDS local-init-watcher {}", rds.route_config_name()),
-                          [this]() { parent_init_target_.ready(); }),
+                          [this](const std::string) { parent_init_target_.ready(); }),
       local_init_target_(
           fmt::format("RdsRouteConfigSubscription local-init-target {}", route_config_name_),
           [this]() { subscription_->start({route_config_name_}); }),
@@ -169,7 +169,7 @@ void RdsRouteConfigSubscription::maybeCreateInitManager(
       Init::WatcherImpl noop_watcher(
           // Note: we just throw it away.
           fmt::format("VHDS ConfigUpdate watcher {}:{}", route_config_name_, version_info),
-          []() { /*Do nothing.*/ });
+          [](const std::string) { /*Do nothing.*/ });
       init_manager->initialize(noop_watcher);
     });
   }

source/common/router/scoped_rds.cc

@@ -258,7 +258,7 @@ void ScopedRdsConfigSubscription::onConfigUpdate(
           Init::WatcherImpl noop_watcher(
               // Note: we just throw it away.
               fmt::format("SRDS ConfigUpdate watcher {}:{}", name_, version_info),
-              []() { /*Do nothing.*/ });
+              [](const std::string) { /*Do nothing.*/ });
           srds_init_mgr->initialize(noop_watcher);
         });
   }

source/common/runtime/runtime_impl.cc

@@ -475,7 +475,7 @@ LoaderImpl::LoaderImpl(Event::Dispatcher& dispatcher, ThreadLocal::SlotAllocator
                        ProtobufMessage::ValidationVisitor& validation_visitor, Api::Api& api)
     : generator_(generator), stats_(generateStats(store)), tls_(tls.allocateSlot()),
       config_(config), service_cluster_(local_info.clusterName()), api_(api),
-      init_watcher_("RDTS", [this]() { onRdtsReady(); }), store_(store) {
+      init_watcher_("RDTS", [this](const std::string target_name) { onRdtsReady(target_name); }), store_(store) {
   std::unordered_set<std::string> layer_names;
   for (const auto& layer : config_.layers()) {
     auto ret = layer_names.insert(layer.name());
@@ -526,9 +526,9 @@ void LoaderImpl::startRtdsSubscriptions(ReadyCallback on_done) {
   init_manager_.initialize(init_watcher_);
 }
 
-void LoaderImpl::onRdtsReady() {
+void LoaderImpl::onRdtsReady(const std::string target_name) {
   ENVOY_LOG(info, "RTDS has finished initialization");
-  on_rtds_initialized_();
+  on_rtds_initialized_(target_name);
 }
 
 RtdsSubscription::RtdsSubscription(

source/common/runtime/runtime_impl.h

@@ -262,7 +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();
+  void onRdtsReady(const std::string target_name);
 
   RandomGenerator& generator_;
   RuntimeStats stats_;

source/common/upstream/upstream_impl.cc

@@ -875,7 +875,7 @@ ClusterImplBase::ClusterImplBase(
     Server::Configuration::TransportSocketFactoryContextImpl& factory_context,
     Stats::ScopePtr&& stats_scope, bool added_via_api)
     : init_manager_(fmt::format("Cluster {}", cluster.name())),
-      init_watcher_("ClusterImplBase", [this]() { onInitDone(); }), runtime_(runtime),
+      init_watcher_("ClusterImplBase", [this](const std::string) { onInitDone(); }), runtime_(runtime),
       local_cluster_(factory_context.clusterManager().localClusterName().value_or("") ==
                      cluster.name()),
       const_metadata_shared_pool_(Config::Metadata::getConstMetadataSharedPool(

source/server/config_validation/server.h

@@ -177,7 +177,7 @@ class ValidationInstance final : Logger::Loggable<Logger::Id::main>,
   // only after referencing members are gone, since initialization continuation can potentially
   // occur at any point during member lifetime.
   Init::ManagerImpl init_manager_{"Validation server"};
-  Init::WatcherImpl init_watcher_{"(no-op)", []() {}};
+  Init::WatcherImpl init_watcher_{"(no-op)", [](const std::string) {}};
   // secret_manager_ must come before listener_manager_, config_ and dispatcher_, and destructed
   // only after these members can no longer reference it, since:
   // - There may be active filter chains referencing it in listener_manager_.

source/server/listener_impl.cc

@@ -251,7 +251,7 @@ ListenerImpl::ListenerImpl(const envoy::config::listener::v3::Listener& config,
       open_connections_(std::make_shared<BasicResourceLimitImpl>(
           std::numeric_limits<uint64_t>::max(), listener_factory_context_->runtime(),
           cx_limit_runtime_key_)),
-      local_init_watcher_(fmt::format("Listener-local-init-watcher {}", name), [this] {
+      local_init_watcher_(fmt::format("Listener-local-init-watcher {}", name), [this] (const std::string) {
         if (workers_started_) {
           parent_.onListenerWarmed(*this);
         } else {
@@ -321,7 +321,7 @@ ListenerImpl::ListenerImpl(ListenerImpl& origin,
           origin.listener_factory_context_->listener_factory_context_base_, this, *this)),
       filter_chain_manager_(address_, origin.listener_factory_context_->parentFactoryContext(),
                             initManager(), origin.filter_chain_manager_),
-      local_init_watcher_(fmt::format("Listener-local-init-watcher {}", name), [this] {
+      local_init_watcher_(fmt::format("Listener-local-init-watcher {}", name), [this] (const std::string) {
         ASSERT(workers_started_);
         parent_.inPlaceFilterChainUpdate(*this);
       }) {

source/server/server.cc

@@ -506,7 +506,7 @@ void InstanceImpl::initialize(const Options& options,
 
 void InstanceImpl::onClusterManagerPrimaryInitializationComplete() {
   // If RTDS was not configured the `onRuntimeReady` callback is immediately invoked.
-  Runtime::LoaderSingleton::get().startRtdsSubscriptions([this]() { onRuntimeReady(); });
+  Runtime::LoaderSingleton::get().startRtdsSubscriptions([this] (const std::string) { onRuntimeReady(); });
 }
 
 void InstanceImpl::onRuntimeReady() {
@@ -575,7 +575,7 @@ RunHelper::RunHelper(Instance& instance, const Options& options, Event::Dispatch
                      Upstream::ClusterManager& cm, AccessLog::AccessLogManager& access_log_manager,
                      Init::Manager& init_manager, OverloadManager& overload_manager,
                      std::function<void()> post_init_cb)
-    : init_watcher_("RunHelper", [&instance, post_init_cb]() {
+    : init_watcher_("RunHelper", [&instance, post_init_cb](const std::string) {
         if (!instance.isShutdown()) {
           post_init_cb();
         }