xds: refactor ADS related data to xDS-Manager part 1

envoy/config/BUILD

@@ -160,6 +160,7 @@ envoy_cc_library(
     name = "xds_manager_interface",
     hdrs = ["xds_manager.h"],
     deps = [
+        "//envoy/upstream:cluster_manager_interface",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
     ],
 )

envoy/config/xds_manager.h

@@ -2,6 +2,7 @@
 
 #include "envoy/common/pure.h"
 #include "envoy/config/core/v3/config_source.pb.h"
+#include "envoy/upstream/cluster_manager.h"
 
 #include "absl/status/status.h"
 
@@ -22,6 +23,20 @@ class XdsManager {
 public:
   virtual ~XdsManager() = default;
 
+  /**
+   * Initializes the xDS-Manager.
+   * This should be called after the cluster-manager is created.
+   * @param cm - a pointer to a valid cluster manager.
+   * @return Ok if the initialization was successful, or an error otherwise.
+   */
+  virtual absl::Status initialize(Upstream::ClusterManager* cm) PURE;
+
+  /**
+   * Shuts down the xDS-Manager and all the configured connections to the config
+   * servers.
+   */
+  virtual void shutdown() PURE;
+
   /**
    * Set the ADS ConfigSource Envoy should use that will replace the current ADS
    * server.

envoy/server/BUILD

@@ -173,6 +173,7 @@ envoy_cc_library(
         "//envoy/api:api_interface",
         "//envoy/config:typed_config_interface",
         "//envoy/config:typed_metadata_interface",
+        "//envoy/config:xds_manager_interface",
         "//envoy/grpc:context_interface",
         "//envoy/http:codes_interface",
         "//envoy/http:context_interface",

envoy/server/factory_context.h

@@ -9,6 +9,7 @@
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/config/typed_config.h"
 #include "envoy/config/typed_metadata.h"
+#include "envoy/config/xds_manager.h"
 #include "envoy/grpc/context.h"
 #include "envoy/http/codes.h"
 #include "envoy/http/context.h"
@@ -121,6 +122,11 @@ class CommonFactoryContext {
    */
   virtual Upstream::ClusterManager& clusterManager() PURE;
 
+  /**
+   * @return Config::XdsManager& singleton for use by the entire server.
+   */
+  virtual Config::XdsManager& xdsManager() PURE;
+
   /**
    * @return const Http::HttpServerPropertiesCacheManager& instance for use by the entire server.
    */

source/common/config/xds_manager_impl.cc

@@ -7,12 +7,18 @@
 namespace Envoy {
 namespace Config {
 
+absl::Status XdsManagerImpl::initialize(Upstream::ClusterManager* cm) {
+  ASSERT(cm != nullptr);
+  cm_ = cm;
+  return absl::OkStatus();
+}
+
 absl::Status
 XdsManagerImpl::setAdsConfigSource(const envoy::config::core::v3::ApiConfigSource& config_source) {
   ASSERT_IS_MAIN_OR_TEST_THREAD();
   RETURN_IF_NOT_OK(validateAdsConfig(config_source));
 
-  return cm_.replaceAdsMux(config_source);
+  return cm_->replaceAdsMux(config_source);
 }
 
 absl::Status

source/common/config/xds_manager_impl.h

@@ -1,7 +1,6 @@
 #pragma once
 
 #include "envoy/config/xds_manager.h"
-#include "envoy/upstream/cluster_manager.h"
 
 #include "source/common/common/thread.h"
 
@@ -10,20 +9,24 @@ namespace Config {
 
 class XdsManagerImpl : public XdsManager {
 public:
-  XdsManagerImpl(Upstream::ClusterManager& cm,
-                 ProtobufMessage::ValidationContext& validation_context)
-      : cm_(cm), validation_context_(validation_context) {}
+  XdsManagerImpl(ProtobufMessage::ValidationContext& validation_context)
+      : validation_context_(validation_context) {}
 
   // Config::ConfigSourceProvider
+  absl::Status initialize(Upstream::ClusterManager* cm) override;
+  void shutdown() override {}
   absl::Status
   setAdsConfigSource(const envoy::config::core::v3::ApiConfigSource& config_source) override;
 
 private:
   // Validates (syntactically) the config_source by doing the PGV validation.
   absl::Status validateAdsConfig(const envoy::config::core::v3::ApiConfigSource& config_source);
 
-  Upstream::ClusterManager& cm_;
   ProtobufMessage::ValidationContext& validation_context_;
+  // The cm_ will only be valid after the cluster-manager is initialized.
+  // Note that this implies that the xDS-manager must be shut down properly
+  // prior to the cluster-manager deletion.
+  Upstream::ClusterManager* cm_;
 };
 
 } // namespace Config

source/common/upstream/cluster_manager_impl.cc

@@ -325,9 +325,10 @@ 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, absl::Status& creation_status)
+    Router::Context& router_context, Server::Instance& server, Config::XdsManager& xds_manager,
+    absl::Status& creation_status)
     : server_(server), factory_(factory), runtime_(runtime), stats_(stats), tls_(tls),
-      random_(api.randomGenerator()),
+      xds_manager_(xds_manager), 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())
@@ -375,6 +376,12 @@ ClusterManagerImpl::ClusterManagerImpl(
     local_cluster_name_ = cm_config.local_cluster_name();
   }
 
+  // Now that the async-client manager is set, the xDS-Manager can be initialized.
+  absl::Status status = xds_manager_.initialize(this);
+  SET_AND_RETURN_IF_NOT_OK(status, creation_status);
+
+  // TODO(adisuissa): refactor and move the following data members to the
+  // xDS-manager class.
   // Initialize the XdsResourceDelegate extension, if set on the bootstrap config.
   if (bootstrap.has_xds_delegate_extension()) {
     auto& factory = Config::Utility::getAndCheckFactory<Config::XdsResourcesDelegateFactory>(
@@ -2337,7 +2344,7 @@ absl::StatusOr<ClusterManagerPtr> ProdClusterManagerFactory::clusterManagerFromP
       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_, creation_status)};
+      context_.routerContext(), server_, context_.xdsManager(), creation_status)};
   RETURN_IF_NOT_OK(creation_status);
   return cluster_manager_impl;
 }

source/common/upstream/cluster_manager_impl.h

@@ -296,6 +296,7 @@ class ClusterManagerImpl : public ClusterManager,
     // Make sure we destroy all potential outgoing connections before this returns.
     cds_api_.reset();
     ads_mux_.reset();
+    xds_manager_.shutdown();
     active_clusters_.clear();
     warming_clusters_.clear();
     updateClusterCounts();
@@ -381,14 +382,17 @@ 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, absl::Status& creation_status);
+  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,
+                     Config::XdsManager& xds_manager, absl::Status& creation_status);
 
   virtual void postThreadLocalRemoveHosts(const Cluster& cluster, const HostVector& hosts_removed);
 
@@ -913,6 +917,7 @@ class ClusterManagerImpl : public ClusterManager,
   ThreadLocal::TypedSlot<ThreadLocalClusterManagerImpl> tls_;
   // Contains information about ongoing on-demand cluster discoveries.
   ClusterCreationsMap pending_cluster_creations_;
+  Config::XdsManager& xds_manager_;
   Random::RandomGenerator& random_;
   ClusterMap warming_clusters_;
   const bool deferred_cluster_creation_;

source/server/config_validation/cluster_manager.cc

@@ -15,7 +15,7 @@ absl::StatusOr<ClusterManagerPtr> ValidationClusterManagerFactory::clusterManage
       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_, creation_status)};
+      context_.routerContext(), server_, context_.xdsManager(), creation_status)};
   RETURN_IF_NOT_OK(creation_status);
   return cluster_manager;
 }

source/server/config_validation/server.cc

@@ -145,14 +145,15 @@ void ValidationInstance::initialize(const Options& options,
           serverFactoryContext(), messageValidationContext().staticValidationVisitor(),
           thread_local_);
 
+  xds_manager_ = std::make_unique<Config::XdsManagerImpl>(validation_context_);
+
   cluster_manager_factory_ = std::make_unique<Upstream::ValidationClusterManagerFactory>(
       server_contexts_, stats(), threadLocal(), http_context_,
       [this]() -> Network::DnsResolverSharedPtr { return this->dnsResolver(); },
       sslContextManager(), *secret_manager_, quic_stat_names_, *this);
   THROW_IF_NOT_OK(config_.initialize(bootstrap_, *this, *cluster_manager_factory_));
   THROW_IF_NOT_OK(runtime().initialize(clusterManager()));
   clusterManager().setInitializedCb([this]() -> void { init_manager_.initialize(init_watcher_); });
-  xds_manager_ = std::make_unique<Config::XdsManagerImpl>(clusterManager(), validation_context_);
 }
 
 void ValidationInstance::shutdown() {

source/server/config_validation/server.h

@@ -78,6 +78,7 @@ class ValidationInstance final : Logger::Loggable<Logger::Id::main>,
   }
   Api::Api& api() override { return *api_; }
   Upstream::ClusterManager& clusterManager() override { return *config_.clusterManager(); }
+  Config::XdsManager& xdsManager() override { return *xds_manager_; }
   const Upstream::ClusterManager& clusterManager() const override {
     return *config_.clusterManager();
   }
@@ -133,7 +134,6 @@ class ValidationInstance final : Logger::Loggable<Logger::Id::main>,
     http_context_.setDefaultTracingConfig(tracing_config);
   }
   void setSinkPredicates(std::unique_ptr<Stats::SinkPredicates>&&) override {}
-  Config::XdsManager& xdsManager() override { return *xds_manager_; }
 
   // Server::WorkerFactory
   WorkerPtr createWorker(uint32_t, OverloadManager&, OverloadManager&,
@@ -185,8 +185,8 @@ class ValidationInstance final : Logger::Loggable<Logger::Id::main>,
   LocalInfo::LocalInfoPtr local_info_;
   AccessLog::AccessLogManagerImpl access_log_manager_;
   std::unique_ptr<Http::HttpServerPropertiesCacheManager> http_server_properties_cache_manager_;
-  std::unique_ptr<Upstream::ValidationClusterManagerFactory> cluster_manager_factory_;
   Config::XdsManagerPtr xds_manager_;
+  std::unique_ptr<Upstream::ValidationClusterManagerFactory> cluster_manager_factory_;
   std::unique_ptr<ListenerManager> listener_manager_;
   std::unique_ptr<OverloadManager> overload_manager_;
   std::unique_ptr<OverloadManager> null_overload_manager_;

source/server/server.cc

@@ -751,6 +751,10 @@ absl::Status InstanceBase::initializeOrThrow(Network::Address::InstanceConstShar
           serverFactoryContext(), messageValidationContext().staticValidationVisitor(),
           thread_local_);
 
+  // Create the xDS-Manager that will be passed to the cluster manager when it
+  // is initialized below.
+  xds_manager_ = std::make_unique<Config::XdsManagerImpl>(validation_context_);
+
   cluster_manager_factory_ = std::make_unique<Upstream::ProdClusterManagerFactory>(
       serverFactoryContext(), stats_store_, thread_local_, http_context_,
       [this]() -> Network::DnsResolverSharedPtr { return this->getOrCreateDnsResolver(); },
@@ -762,11 +766,6 @@ absl::Status InstanceBase::initializeOrThrow(Network::Address::InstanceConstShar
   // cluster_manager_factory_ is available.
   RETURN_IF_NOT_OK(config_.initialize(bootstrap_, *this, *cluster_manager_factory_));
 
-  // Create the xDS-Manager and pass the cluster manager that was created above.
-  ASSERT(config_.clusterManager());
-  xds_manager_ =
-      std::make_unique<Config::XdsManagerImpl>(*config_.clusterManager(), validation_context_);
-
   // Instruct the listener manager to create the LDS provider if needed. This must be done later
   // because various items do not yet exist when the listener manager is created.
   if (bootstrap_.dynamic_resources().has_lds_config() ||

source/server/server.h

@@ -178,6 +178,7 @@ class ServerFactoryContextImpl : public Configuration::ServerFactoryContext,
 
   // Configuration::ServerFactoryContext
   Upstream::ClusterManager& clusterManager() override { return server_.clusterManager(); }
+  Config::XdsManager& xdsManager() override { return server_.xdsManager(); }
   Http::HttpServerPropertiesCacheManager& httpServerPropertiesCacheManager() override {
     return server_.httpServerPropertiesCacheManager();
   }

test/common/config/xds_manager_impl_test.cc

@@ -17,19 +17,28 @@ using testing::Return;
 
 class XdsManagerImplTest : public testing::Test {
 public:
-  XdsManagerImplTest() : xds_manager_impl_(cm_, validation_context_) {
+  XdsManagerImplTest() : xds_manager_impl_(validation_context_) {
     ON_CALL(validation_context_, staticValidationVisitor())
         .WillByDefault(ReturnRef(validation_visitor_));
   }
 
+  void initialize() { ASSERT_OK(xds_manager_impl_.initialize(&cm_)); }
+
   NiceMock<Upstream::MockClusterManager> cm_;
   NiceMock<ProtobufMessage::MockValidationVisitor> validation_visitor_;
   NiceMock<ProtobufMessage::MockValidationContext> validation_context_;
   XdsManagerImpl xds_manager_impl_;
 };
 
+// Validates that a call to shutdown succeeds.
+TEST_F(XdsManagerImplTest, ShutdownSuccessful) {
+  initialize();
+  xds_manager_impl_.shutdown();
+}
+
 // Validates that setAdsConfigSource invokes the correct method in the cm_.
 TEST_F(XdsManagerImplTest, AdsConfigSourceSetterSuccess) {
+  initialize();
   envoy::config::core::v3::ApiConfigSource config_source;
   config_source.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   EXPECT_CALL(cm_, replaceAdsMux(ProtoEq(config_source))).WillOnce(Return(absl::OkStatus()));
@@ -40,6 +49,7 @@ TEST_F(XdsManagerImplTest, AdsConfigSourceSetterSuccess) {
 // Validates that setAdsConfigSource invokes the correct method in the cm_,
 // and fails if needed.
 TEST_F(XdsManagerImplTest, AdsConfigSourceSetterFailure) {
+  initialize();
   envoy::config::core::v3::ApiConfigSource config_source;
   config_source.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   EXPECT_CALL(cm_, replaceAdsMux(ProtoEq(config_source)))
@@ -51,6 +61,7 @@ TEST_F(XdsManagerImplTest, AdsConfigSourceSetterFailure) {
 
 // Validates that setAdsConfigSource validation failure is detected.
 TEST_F(XdsManagerImplTest, AdsConfigSourceSetterInvalidConfig) {
+  initialize();
   envoy::config::core::v3::ApiConfigSource config_source;
   config_source.set_api_type(envoy::config::core::v3::ApiConfigSource::GRPC);
   // Add an empty gRPC service (without EnvoyGrpc/GoogleGrpc) which should be

test/common/upstream/BUILD

@@ -541,6 +541,7 @@ envoy_cc_test_library(
         "//test/integration/clusters:custom_static_cluster",
         "//test/mocks/access_log:access_log_mocks",
         "//test/mocks/api:api_mocks",
+        "//test/mocks/config:xds_manager_mocks",
         "//test/mocks/http:http_mocks",
         "//test/mocks/local_info:local_info_mocks",
         "//test/mocks/network:network_mocks",