cluster: removing exceptions (envoyproxy#37500)

Risk Level: low
Testing: updated tests
Docs Changes: n/a
Release Notes: n/a
envoyproxy/envoy-mobile#176

Signed-off-by: Alyssa Wilk <[email protected]>

Author: alyssawilk

envoy/upstream/cluster_manager.h

@@ -274,11 +274,12 @@ class ClusterManager {
    *                       update. It can be overridden by setting `remove_ignored` to true while
    *                       calling removeCluster(). This is useful for clusters whose lifecycle
    *                       is managed with custom implementation, e.g., DFP clusters.
-   * @return true if the action results in an add/update of a cluster.
+   * @return true if the action results in an add/update of a cluster, an error
+   * status if the config is invalid.
    */
-  virtual bool addOrUpdateCluster(const envoy::config::cluster::v3::Cluster& cluster,
-                                  const std::string& version_info,
-                                  const bool avoid_cds_removal = false) PURE;
+  virtual absl::StatusOr<bool>
+  addOrUpdateCluster(const envoy::config::cluster::v3::Cluster& cluster,
+                     const std::string& version_info, const bool avoid_cds_removal = false) PURE;
 
   /**
    * Set a callback that will be invoked when all primary clusters have been initialized.
@@ -572,7 +573,7 @@ class ClusterManagerFactory {
    * The cluster manager initialize() method needs to be called right after this method.
    * Please check https://github.com/envoyproxy/envoy/issues/33218 for details.
    */
-  virtual ClusterManagerPtr
+  virtual absl::StatusOr<ClusterManagerPtr>
   clusterManagerFromProto(const envoy::config::bootstrap::v3::Bootstrap& bootstrap) PURE;
 
   /**

envoy/upstream/upstream.h

@@ -1307,7 +1307,7 @@ class Cluster {
    *        time initialization. E.g., for a dynamic DNS cluster the initialize callback will be
    *        called when initial DNS resolution is complete.
    */
-  virtual void initialize(std::function<void()> callback) PURE;
+  virtual void initialize(std::function<absl::Status()> callback) PURE;
 
   /**
    * @return the phase in which the cluster is initialized at boot. This mechanism is used such that

source/common/upstream/cds_api_helper.cc

@@ -46,7 +46,13 @@ CdsApiHelper::onConfigUpdate(const std::vector<Config::DecodedResourceRef>& adde
             fmt::format("{}: duplicate cluster {} found", cluster.name(), cluster.name()));
         continue;
       }
-      if (cm_.addOrUpdateCluster(cluster, resource.get().version())) {
+      auto update_or_error = cm_.addOrUpdateCluster(cluster, resource.get().version());
+      if (!update_or_error.status().ok()) {
+        exception_msgs.push_back(
+            fmt::format("{}: {}", cluster.name(), update_or_error.status().message()));
+        continue;
+      }
+      if (*update_or_error) {
         any_applied = true;
         ENVOY_LOG(debug, "{}: add/update cluster '{}'", name_, cluster.name());
         ++added_or_updated;

source/common/upstream/cluster_manager_impl.cc

@@ -117,8 +117,9 @@ void ClusterManagerInitHelper::addCluster(ClusterManagerCluster& cm_cluster) {
   ASSERT(state_ != State::AllClustersInitialized);
 
   const auto initialize_cb = [&cm_cluster, this] {
-    onClusterInit(cm_cluster);
+    RETURN_IF_NOT_OK(onClusterInit(cm_cluster));
     cm_cluster.cluster().info()->configUpdateStats().warming_state_.set(0);
+    return absl::OkStatus();
   };
   Cluster& cluster = cm_cluster.cluster();
 
@@ -142,10 +143,11 @@ void ClusterManagerInitHelper::addCluster(ClusterManagerCluster& cm_cluster) {
             primary_init_clusters_.size(), secondary_init_clusters_.size());
 }
 
-void ClusterManagerInitHelper::onClusterInit(ClusterManagerCluster& cluster) {
+absl::Status ClusterManagerInitHelper::onClusterInit(ClusterManagerCluster& cluster) {
   ASSERT(state_ != State::AllClustersInitialized);
-  per_cluster_init_callback_(cluster);
+  RETURN_IF_NOT_OK(per_cluster_init_callback_(cluster));
   removeCluster(cluster);
+  return absl::OkStatus();
 }
 
 void ClusterManagerInitHelper::removeCluster(ClusterManagerCluster& cluster) {
@@ -186,7 +188,7 @@ void ClusterManagerInitHelper::initializeSecondaryClusters() {
     ClusterManagerCluster* cluster = iter->second;
     ENVOY_LOG(debug, "initializing secondary cluster {}", iter->first);
     ++iter;
-    cluster->cluster().initialize([cluster, this] { onClusterInit(*cluster); });
+    cluster->cluster().initialize([cluster, this] { return onClusterInit(*cluster); });
   }
 }
 
@@ -309,15 +311,16 @@ ClusterManagerImpl::ClusterManagerImpl(
     AccessLog::AccessLogManager& log_manager, Event::Dispatcher& main_thread_dispatcher,
     OptRef<Server::Admin> admin, ProtobufMessage::ValidationContext& validation_context,
     Api::Api& api, Http::Context& http_context, Grpc::Context& grpc_context,
-    Router::Context& router_context, Server::Instance& server)
+    Router::Context& router_context, Server::Instance& server, absl::Status& creation_status)
     : server_(server), factory_(factory), runtime_(runtime), stats_(stats), tls_(tls),
       random_(api.randomGenerator()),
       deferred_cluster_creation_(bootstrap.cluster_manager().enable_deferred_cluster_creation()),
       bind_config_(bootstrap.cluster_manager().has_upstream_bind_config()
                        ? absl::make_optional(bootstrap.cluster_manager().upstream_bind_config())
                        : absl::nullopt),
       local_info_(local_info), cm_stats_(generateStats(*stats.rootScope())),
-      init_helper_(*this, [this](ClusterManagerCluster& cluster) { onClusterInit(cluster); }),
+      init_helper_(*this,
+                   [this](ClusterManagerCluster& cluster) { return onClusterInit(cluster); }),
       time_source_(main_thread_dispatcher.timeSource()), dispatcher_(main_thread_dispatcher),
       http_context_(http_context), validation_context_(validation_context),
       router_context_(router_context), cluster_stat_names_(stats.symbolTable()),
@@ -345,9 +348,10 @@ ClusterManagerImpl::ClusterManagerImpl(
   if (cm_config.has_outlier_detection()) {
     const std::string event_log_file_path = cm_config.outlier_detection().event_log_path();
     if (!event_log_file_path.empty()) {
-      outlier_event_logger_ = THROW_OR_RETURN_VALUE(
-          Outlier::EventLoggerImpl::create(log_manager, event_log_file_path, time_source_),
-          std::unique_ptr<Outlier::EventLoggerImpl>);
+      auto outlier_or_error =
+          Outlier::EventLoggerImpl::create(log_manager, event_log_file_path, time_source_);
+      SET_AND_RETURN_IF_NOT_OK(outlier_or_error.status(), creation_status);
+      outlier_event_logger_ = std::move(*outlier_or_error);
     }
   }
 
@@ -693,7 +697,7 @@ ClusterManagerImpl::ThreadLocalClusterManagerImpl::generateStats(Stats::Scope& s
   return {ALL_THREAD_LOCAL_CLUSTER_MANAGER_STATS(POOL_GAUGE_PREFIX(scope, final_prefix))};
 }
 
-void ClusterManagerImpl::onClusterInit(ClusterManagerCluster& cm_cluster) {
+absl::Status ClusterManagerImpl::onClusterInit(ClusterManagerCluster& cm_cluster) {
   // This routine is called when a cluster has finished initializing. The cluster has not yet
   // been setup for cross-thread updates to avoid needless updates during initialization. The order
   // of operations here is important. We start by initializing the thread aware load balancer if
@@ -710,7 +714,7 @@ void ClusterManagerImpl::onClusterInit(ClusterManagerCluster& cm_cluster) {
   cluster_data = active_clusters_.find(cluster.info()->name());
 
   if (cluster_data->second->thread_aware_lb_ != nullptr) {
-    THROW_IF_NOT_OK(cluster_data->second->thread_aware_lb_->initialize());
+    RETURN_IF_NOT_OK(cluster_data->second->thread_aware_lb_->initialize());
   }
 
   // Now setup for cross-thread updates.
@@ -793,6 +797,7 @@ void ClusterManagerImpl::onClusterInit(ClusterManagerCluster& cm_cluster) {
   // clusters being added/updated. We could gate the below update on hosts being available on
   // the cluster or the cluster not already existing, but the special logic is not worth it.
   postThreadLocalClusterUpdate(cm_cluster, std::move(params));
+  return absl::OkStatus();
 }
 
 bool ClusterManagerImpl::scheduleUpdate(ClusterManagerCluster& cluster, uint32_t priority,
@@ -870,9 +875,10 @@ void ClusterManagerImpl::applyUpdates(ClusterManagerCluster& cluster, uint32_t p
   updates.last_updated_ = time_source_.monotonicTime();
 }
 
-bool ClusterManagerImpl::addOrUpdateCluster(const envoy::config::cluster::v3::Cluster& cluster,
-                                            const std::string& version_info,
-                                            const bool avoid_cds_removal) {
+absl::StatusOr<bool>
+ClusterManagerImpl::addOrUpdateCluster(const envoy::config::cluster::v3::Cluster& cluster,
+                                       const std::string& version_info,
+                                       const bool avoid_cds_removal) {
   // First we need to see if this new config is new or an update to an existing dynamic cluster.
   // We don't allow updates to statically configured clusters in the main configuration. We check
   // both the warming clusters and the active clusters to see if we need an update or the update
@@ -924,7 +930,7 @@ bool ClusterManagerImpl::addOrUpdateCluster(const envoy::config::cluster::v3::Cl
   auto status_or_cluster =
       loadCluster(cluster, new_hash, version_info, /*added_via_api=*/true,
                   /*required_for_ads=*/false, warming_clusters_, avoid_cds_removal);
-  THROW_IF_NOT_OK_REF(status_or_cluster.status());
+  RETURN_IF_NOT_OK_REF(status_or_cluster.status());
   const ClusterDataPtr previous_cluster = std::move(status_or_cluster.value());
   auto& cluster_entry = warming_clusters_.at(cluster_name);
   cluster_entry->cluster_->info()->configUpdateStats().warming_state_.set(1);
@@ -938,7 +944,7 @@ bool ClusterManagerImpl::addOrUpdateCluster(const envoy::config::cluster::v3::Cl
       auto state_changed_cluster_entry = warming_clusters_.find(cluster_name);
       state_changed_cluster_entry->second->cluster_->info()->configUpdateStats().warming_state_.set(
           0);
-      onClusterInit(*state_changed_cluster_entry->second);
+      return onClusterInit(*state_changed_cluster_entry->second);
     });
   }
 
@@ -2296,13 +2302,15 @@ void ClusterManagerImpl::ThreadLocalClusterManagerImpl::tcpConnPoolIsIdle(
   }
 }
 
-ClusterManagerPtr ProdClusterManagerFactory::clusterManagerFromProto(
+absl::StatusOr<ClusterManagerPtr> ProdClusterManagerFactory::clusterManagerFromProto(
     const envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+  absl::Status creation_status = absl::OkStatus();
   auto cluster_manager_impl = std::unique_ptr<ClusterManagerImpl>{new ClusterManagerImpl(
       bootstrap, *this, context_, stats_, tls_, context_.runtime(), context_.localInfo(),
       context_.accessLogManager(), context_.mainThreadDispatcher(), context_.admin(),
       context_.messageValidationContext(), context_.api(), http_context_, context_.grpcContext(),
-      context_.routerContext(), server_)};
+      context_.routerContext(), server_, creation_status)};
+  RETURN_IF_NOT_OK(creation_status);
   return cluster_manager_impl;
 }
 

source/common/upstream/cluster_manager_impl.h

@@ -65,7 +65,7 @@ class ProdClusterManagerFactory : public ClusterManagerFactory {
         server_(server) {}
 
   // Upstream::ClusterManagerFactory
-  ClusterManagerPtr
+  absl::StatusOr<ClusterManagerPtr>
   clusterManagerFromProto(const envoy::config::bootstrap::v3::Bootstrap& bootstrap) override;
   Http::ConnectionPool::InstancePtr allocateConnPool(
       Event::Dispatcher& dispatcher, HostConstSharedPtr host, ResourcePriority priority,
@@ -147,7 +147,7 @@ class ClusterManagerInitHelper : Logger::Loggable<Logger::Id::upstream> {
    */
   ClusterManagerInitHelper(
       ClusterManager& cm,
-      const std::function<void(ClusterManagerCluster&)>& per_cluster_init_callback)
+      const std::function<absl::Status(ClusterManagerCluster&)>& per_cluster_init_callback)
       : cm_(cm), per_cluster_init_callback_(per_cluster_init_callback) {}
 
   enum class State {
@@ -190,10 +190,10 @@ class ClusterManagerInitHelper : Logger::Loggable<Logger::Id::upstream> {
 
   void initializeSecondaryClusters();
   void maybeFinishInitialize();
-  void onClusterInit(ClusterManagerCluster& cluster);
+  absl::Status onClusterInit(ClusterManagerCluster& cluster);
 
   ClusterManager& cm_;
-  std::function<void(ClusterManagerCluster& cluster)> per_cluster_init_callback_;
+  std::function<absl::Status(ClusterManagerCluster& cluster)> per_cluster_init_callback_;
   CdsApi* cds_{};
   ClusterManager::PrimaryClustersReadyCallback primary_clusters_initialized_callback_;
   ClusterManager::InitializationCompleteCallback initialized_callback_;
@@ -251,9 +251,9 @@ class ClusterManagerImpl : public ClusterManager,
   std::size_t warmingClusterCount() const { return warming_clusters_.size(); }
 
   // Upstream::ClusterManager
-  bool addOrUpdateCluster(const envoy::config::cluster::v3::Cluster& cluster,
-                          const std::string& version_info,
-                          const bool avoid_cds_removal = false) override;
+  absl::StatusOr<bool> addOrUpdateCluster(const envoy::config::cluster::v3::Cluster& cluster,
+                                          const std::string& version_info,
+                                          const bool avoid_cds_removal = false) override;
 
   void setPrimaryClustersInitializedCb(PrimaryClustersReadyCallback callback) override {
     init_helper_.setPrimaryClustersInitializedCb(callback);
@@ -381,16 +381,14 @@ class ClusterManagerImpl : public ClusterManager,
 protected:
   // ClusterManagerImpl's constructor should not be invoked directly; create instances from the
   // clusterManagerFromProto() static method. The init() method must be called after construction.
-  ClusterManagerImpl(const envoy::config::bootstrap::v3::Bootstrap& bootstrap,
-                     ClusterManagerFactory& factory,
-                     Server::Configuration::CommonFactoryContext& context, Stats::Store& stats,
-                     ThreadLocal::Instance& tls, Runtime::Loader& runtime,
-                     const LocalInfo::LocalInfo& local_info,
-                     AccessLog::AccessLogManager& log_manager,
-                     Event::Dispatcher& main_thread_dispatcher, OptRef<Server::Admin> admin,
-                     ProtobufMessage::ValidationContext& validation_context, Api::Api& api,
-                     Http::Context& http_context, Grpc::Context& grpc_context,
-                     Router::Context& router_context, Server::Instance& server);
+  ClusterManagerImpl(
+      const envoy::config::bootstrap::v3::Bootstrap& bootstrap, ClusterManagerFactory& factory,
+      Server::Configuration::CommonFactoryContext& context, Stats::Store& stats,
+      ThreadLocal::Instance& tls, Runtime::Loader& runtime, const LocalInfo::LocalInfo& local_info,
+      AccessLog::AccessLogManager& log_manager, Event::Dispatcher& main_thread_dispatcher,
+      OptRef<Server::Admin> admin, ProtobufMessage::ValidationContext& validation_context,
+      Api::Api& api, Http::Context& http_context, Grpc::Context& grpc_context,
+      Router::Context& router_context, Server::Instance& server, absl::Status& creation_status);
 
   virtual void postThreadLocalRemoveHosts(const Cluster& cluster, const HostVector& hosts_removed);
 
@@ -872,7 +870,7 @@ class ClusterManagerImpl : public ClusterManager,
                                              const std::string& version_info, bool added_via_api,
                                              bool required_for_ads, ClusterMap& cluster_map,
                                              bool avoid_cds_removal = false);
-  void onClusterInit(ClusterManagerCluster& cluster);
+  absl::Status onClusterInit(ClusterManagerCluster& cluster);
   void postThreadLocalHealthFailure(const HostSharedPtr& host);
   void updateClusterCounts();
   void clusterWarmingToActive(const std::string& cluster_name);

source/common/upstream/health_discovery_service.cc

@@ -211,7 +211,7 @@ HdsDelegate::createHdsCluster(const envoy::config::cluster::v3::Cluster& cluster
                                    store_stats_, ssl_context_manager_, false, *info_factory_, tls_);
 
   // Begin HCs in the background.
-  new_cluster->initialize([] {});
+  new_cluster->initialize([] { return absl::OkStatus(); });
   new_cluster->initHealthchecks();
 
   return new_cluster;
@@ -547,7 +547,7 @@ void HdsCluster::initHealthchecks() {
   }
 }
 
-void HdsCluster::initialize(std::function<void()> callback) {
+void HdsCluster::initialize(std::function<absl::Status()> callback) {
   initialization_complete_callback_ = callback;
 
   // If this function gets called again we do not want to touch the priority set again with the

source/common/upstream/health_discovery_service.h

@@ -59,7 +59,7 @@ class HdsCluster : public Cluster, Logger::Loggable<Logger::Id::upstream> {
   ClusterInfoConstSharedPtr info() const override { return info_; }
   Outlier::Detector* outlierDetector() override { return outlier_detector_.get(); }
   const Outlier::Detector* outlierDetector() const override { return outlier_detector_.get(); }
-  void initialize(std::function<void()> callback) override;
+  void initialize(std::function<absl::Status()> callback) override;
   // Compare changes in the cluster proto, and update parts of the cluster as needed.
   absl::Status update(envoy::config::cluster::v3::Cluster cluster,
                       const envoy::config::core::v3::BindConfig& bind_config,
@@ -80,7 +80,7 @@ class HdsCluster : public Cluster, Logger::Loggable<Logger::Id::upstream> {
   Outlier::DetectorSharedPtr outlier_detector_;
 
 private:
-  std::function<void()> initialization_complete_callback_;
+  std::function<absl::Status()> initialization_complete_callback_;
 
   Server::Configuration::ServerFactoryContext& server_context_;
   envoy::config::cluster::v3::Cluster cluster_;

source/common/upstream/upstream_impl.cc

@@ -1736,7 +1736,7 @@ ResourceManager& ClusterInfoImpl::resourceManager(ResourcePriority priority) con
   return *resource_managers_.managers_[enumToInt(priority)];
 }
 
-void ClusterImplBase::initialize(std::function<void()> callback) {
+void ClusterImplBase::initialize(std::function<absl::Status()> callback) {
   ASSERT(!initialization_started_);
   ASSERT(initialization_complete_callback_ == nullptr);
   initialization_complete_callback_ = callback;
@@ -1798,7 +1798,7 @@ void ClusterImplBase::finishInitialization() {
   }
 
   if (snapped_callback != nullptr) {
-    snapped_callback();
+    THROW_IF_NOT_OK(snapped_callback());
   }
 }
 

source/common/upstream/upstream_impl.h

@@ -1232,7 +1232,7 @@ class ClusterImplBase : public Cluster, protected Logger::Loggable<Logger::Id::u
   ClusterInfoConstSharedPtr info() const override { return info_; }
   Outlier::Detector* outlierDetector() override { return outlier_detector_.get(); }
   const Outlier::Detector* outlierDetector() const override { return outlier_detector_.get(); }
-  void initialize(std::function<void()> callback) override;
+  void initialize(std::function<absl::Status()> callback) override;
   UnitFloat dropOverload() const override { return drop_overload_; }
   const std::string& dropCategory() const override { return drop_category_; }
   void setDropOverload(UnitFloat drop_overload) override { drop_overload_ = drop_overload; }
@@ -1301,7 +1301,7 @@ class ClusterImplBase : public Cluster, protected Logger::Loggable<Logger::Id::u
   void reloadHealthyHosts(const HostSharedPtr& host);
 
   bool initialization_started_{};
-  std::function<void()> initialization_complete_callback_;
+  std::function<absl::Status()> initialization_complete_callback_;
   uint64_t pending_initialize_health_checks_{};
   const bool local_cluster_;
   Config::ConstMetadataSharedPoolSharedPtr const_metadata_shared_pool_;

source/extensions/common/aws/credentials_provider_impl.cc

@@ -191,7 +191,7 @@ void MetadataCredentialsProviderBase::createCluster(bool new_timer) {
                    cluster_type_str, cluster_name_, host_port);
   }
 
-  context_->clusterManager().addOrUpdateCluster(cluster, "");
+  THROW_IF_NOT_OK(context_->clusterManager().addOrUpdateCluster(cluster, "").status());
 }
 
 // A thread local callback that occurs on every worker thread during cluster initialization.