Delta CDS: add on-demand cds support

docs/root/configuration/upstream/cluster_manager/cds.rst

@@ -18,3 +18,40 @@ Statistics
 ----------
 
 CDS has a :ref:`statistics <subscription_statistics>` tree rooted at *cluster_manager.cds.*
+
+On-demand CDS
+-------------
+
+Similar to VHDS on-demand feature in terms of hosts, the on-demand CDS API is an additional API
+that Envoy will call to dynamically fetch upstream clusters which Envoy interested in spontaneously.
+
+By default in CDS, all cluster configurations are sent to every Envoy instance in the mesh. The
+delta CDS provides the ability that the xDS management server can send incremental CDS to the Envoy
+instance, but Envoy instance can not feedback upstream clusters it interested in spontaneously, in
+other words, Envoy instance only can receive the CDS passively.
+
+In order to fix this issue, on-demand CDS uses the delta xDS protocol to allow a cluster configuration
+to be subscribed to and the necessary cluster configuration to be requested as needed. Instead
+of sending all cluster configuration or cluster configuration the Envoy instance aren't interested
+in, using on-demand CDS will allow an Envoy instance to subscribe and unsubscribe from a list of
+cluster configurations stored internally in the xDS management server. The xDS management server
+will monitor the list and use it to filter the configuration sent to an individual Envoy instance
+to only contain the subscribed cluster configurations.
+
+Subscribing to resources
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+On-demand CDS allows resources to be :ref:`subscribed <xds_protocol_delta_subscribe>` to using
+a :ref:`DeltaDiscoveryRequest <envoy_api_msg_DeltaDiscoveryRequest>`
+with the :ref:`type_url <envoy_api_field_DeltaDiscoveryRequest.type_url>` set to
+`type.googleapis.com/envoy.api.v2.Cluster` and
+:ref:`resource_names_subscribe <envoy_api_field_DeltaDiscoveryRequest.resource_names_subscribe>`
+set to a list of cluster resource names for which it would like configuration.
+
+Typical use case
+^^^^^^^^^^^^^^^^
+
+Sometimes, there will be a large amount of broker instances provided for
+`Apache RocketMQ <http://rocketmq.apache.org/>`_ to produce/consume messages. Perhaps the size of
+SToW CDS configurations will be more than 1GB, so it is not practical to deliver the SToW CDS from the
+management server every time, which will cause huge overhead. In this case, on-demand CDS is essential.

include/envoy/config/grpc_mux.h

@@ -125,6 +125,15 @@ class GrpcMux {
                                   const std::set<std::string>& resources,
                                   SubscriptionCallbacks& callbacks,
                                   std::chrono::milliseconds init_fetch_timeout) PURE;
+
+  /**
+   * Adds additional resources to the watch. Unlike addOrUpdateWatch, it never removes resources.
+   */
+  virtual Watch* addToWatch(const std::string& type_url, Watch* watch,
+                            const std::set<std::string>& resources,
+                            SubscriptionCallbacks& callbacks,
+                            std::chrono::milliseconds init_fetch_timeout) PURE;
+
   virtual void removeWatch(const std::string& type_url, Watch* watch) PURE;
 
   /**

include/envoy/config/subscription.h

@@ -90,6 +90,12 @@ class Subscription {
    * be passed to std::set_difference, which must be given sorted collections.
    */
   virtual void updateResourceInterest(const std::set<std::string>& update_to_these_names) PURE;
+
+  /**
+   * Add the resources to fetch.
+   * @param resources vector of resource names to fetch.
+   */
+  virtual void addToResourceInterest(const std::set<std::string>&){};
 };
 
 using SubscriptionPtr = std::unique_ptr<Subscription>;

include/envoy/upstream/cluster_manager.h

@@ -227,6 +227,8 @@ class ClusterManager {
    * @return Config::SubscriptionFactory& the subscription factory.
    */
   virtual Config::SubscriptionFactory& subscriptionFactory() PURE;
+
+  virtual void addToClusterInterest(const std::set<std::string>& add_these_names) PURE;
 };
 
 using ClusterManagerPtr = std::unique_ptr<ClusterManager>;
@@ -253,6 +255,11 @@ class CdsApi {
    * @return std::string last accepted version from fetch.
    */
   virtual const std::string versionInfo() const PURE;
+
+  /**
+   * Add watch set of cluster resources interested.
+   */
+  virtual void addToClusterInterest(const std::set<std::string>& add_these_names) PURE;
 };
 
 using CdsApiPtr = std::unique_ptr<CdsApi>;

source/common/config/delta_subscription_impl.cc

@@ -41,6 +41,11 @@ void DeltaSubscriptionImpl::updateResourceInterest(
   stats_.update_attempt_.inc();
 }
 
+void DeltaSubscriptionImpl::addToResourceInterest(const std::set<std::string>& add_these_names) {
+  watch_ = context_->addToWatch(type_url_, watch_, add_these_names, *this, init_fetch_timeout_);
+  stats_.update_attempt_.inc();
+}
+
 // Config::SubscriptionCallbacks
 void DeltaSubscriptionImpl::onConfigUpdate(
     const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,

source/common/config/delta_subscription_impl.h

@@ -45,6 +45,7 @@ class DeltaSubscriptionImpl : public Subscription, public SubscriptionCallbacks
   void start(const std::set<std::string>& resource_names) override;
 
   void updateResourceInterest(const std::set<std::string>& update_to_these_names) override;
+  void addToResourceInterest(const std::set<std::string>& add_these_names) override;
 
   // Config::SubscriptionCallbacks (all pass through to callbacks_!)
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,

source/common/config/grpc_mux_impl.h

@@ -50,6 +50,11 @@ class GrpcMuxImpl : public GrpcMux,
                           SubscriptionCallbacks&, std::chrono::milliseconds) override {
     NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
   }
+
+  Watch* addToWatch(const std::string&, Watch*, const std::set<std::string>&,
+                    SubscriptionCallbacks&, std::chrono::milliseconds) override {
+    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
+  }
   void removeWatch(const std::string&, Watch*) override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
 
   void handleDiscoveryResponse(
@@ -156,6 +161,10 @@ class NullGrpcMuxImpl : public GrpcMux,
                           SubscriptionCallbacks&, std::chrono::milliseconds) override {
     throw EnvoyException("ADS must be configured to support an ADS config source");
   }
+  Watch* addToWatch(const std::string&, Watch*, const std::set<std::string>&,
+                    SubscriptionCallbacks&, std::chrono::milliseconds) override {
+    throw EnvoyException("ADS must be configured to support an ADS config source");
+  }
   void removeWatch(const std::string&, Watch*) override {
     throw EnvoyException("ADS must be configured to support an ADS config source");
   }

source/common/config/new_grpc_mux_impl.cc

@@ -30,11 +30,24 @@ Watch* NewGrpcMuxImpl::addOrUpdateWatch(const std::string& type_url, Watch* watc
                                         SubscriptionCallbacks& callbacks,
                                         std::chrono::milliseconds init_fetch_timeout) {
   if (watch == nullptr) {
-    return addWatch(type_url, resources, callbacks, init_fetch_timeout);
-  } else {
-    updateWatch(type_url, watch, resources);
-    return watch;
+    watch = addWatch(type_url, callbacks, init_fetch_timeout);
   }
+  // updateWatch() queues a discovery request if any of 'resources' are not yet subscribed.
+  updateWatch(type_url, watch, resources);
+  return watch;
+}
+
+Watch* NewGrpcMuxImpl::addToWatch(const std::string& type_url, Watch* watch,
+                                  const std::set<std::string>& resources,
+                                  SubscriptionCallbacks& callbacks,
+                                  std::chrono::milliseconds init_fetch_timeout) {
+  if (watch == nullptr) {
+    watch = addWatch(type_url, callbacks, init_fetch_timeout);
+  }
+  // addToWatch() queues a discovery request for any of *extra* 'resources' we are not yet
+  // subscribed.
+  addToWatch(type_url, watch, resources);
+  return watch;
 }
 
 void NewGrpcMuxImpl::removeWatch(const std::string& type_url, Watch* watch) {
@@ -124,22 +137,32 @@ GrpcMuxWatchPtr NewGrpcMuxImpl::subscribe(const std::string&, const std::set<std
 
 void NewGrpcMuxImpl::start() { grpc_stream_.establishNewStream(); }
 
-Watch* NewGrpcMuxImpl::addWatch(const std::string& type_url, const std::set<std::string>& resources,
-                                SubscriptionCallbacks& callbacks,
+Watch* NewGrpcMuxImpl::addWatch(const std::string& type_url, SubscriptionCallbacks& callbacks,
                                 std::chrono::milliseconds init_fetch_timeout) {
   auto entry = subscriptions_.find(type_url);
   if (entry == subscriptions_.end()) {
     // We don't yet have a subscription for type_url! Make one!
     addSubscription(type_url, init_fetch_timeout);
-    return addWatch(type_url, resources, callbacks, init_fetch_timeout);
+    return addWatch(type_url, callbacks, init_fetch_timeout);
   }
-
   Watch* watch = entry->second->watch_map_.addWatch(callbacks);
-  // updateWatch() queues a discovery request if any of 'resources' are not yet subscribed.
-  updateWatch(type_url, watch, resources);
   return watch;
 }
 
+void NewGrpcMuxImpl::addToWatch(const std::string& type_url, Watch* watch,
+                                const std::set<std::string>& resources) {
+  ASSERT(watch != nullptr);
+  auto sub = subscriptions_.find(type_url);
+  RELEASE_ASSERT(sub != subscriptions_.end(),
+                 fmt::format("Watch of {} has no subscription to update.", type_url));
+  auto added_removed = sub->second->watch_map_.addToWatchInterest(watch, resources);
+  sub->second->sub_state_.updateSubscriptionInterest(added_removed.added_, added_removed.removed_);
+  // Tell the server about our change in interest, if any.
+  if (sub->second->sub_state_.subscriptionUpdatePending()) {
+    trySendDiscoveryRequests();
+  }
+}
+
 // Updates the list of resource names watched by the given watch. If an added name is new across
 // the whole subscription, or if a removed name has no other watch interested in it, then the
 // subscription will enqueue and attempt to send an appropriate discovery request.

source/common/config/new_grpc_mux_impl.h

@@ -37,6 +37,11 @@ class NewGrpcMuxImpl
   Watch* addOrUpdateWatch(const std::string& type_url, Watch* watch,
                           const std::set<std::string>& resources, SubscriptionCallbacks& callbacks,
                           std::chrono::milliseconds init_fetch_timeout) override;
+
+  Watch* addToWatch(const std::string& type_url, Watch* watch,
+                    const std::set<std::string>& resources, SubscriptionCallbacks& callbacks,
+                    std::chrono::milliseconds init_fetch_timeout) override;
+
   void removeWatch(const std::string& type_url, Watch* watch) override;
 
   // TODO(fredlas) PR #8478 will remove this.
@@ -81,8 +86,11 @@ class NewGrpcMuxImpl
   }
 
 private:
-  Watch* addWatch(const std::string& type_url, const std::set<std::string>& resources,
-                  SubscriptionCallbacks& callbacks, std::chrono::milliseconds init_fetch_timeout);
+  Watch* addWatch(const std::string& type_url, SubscriptionCallbacks& callbacks,
+                  std::chrono::milliseconds init_fetch_timeout);
+
+  void addToWatch(const std::string& type_url, Watch* watch,
+                  const std::set<std::string>& resources);
 
   // Updates the list of resource names watched by the given watch. If an added name is new across
   // the whole subscription, or if a removed name has no other watch interested in it, then the

source/common/config/watch_map.cc

@@ -42,6 +42,15 @@ AddedRemoved WatchMap::updateWatchInterest(Watch* watch,
                       findRemovals(newly_removed_from_watch, watch));
 }
 
+AddedRemoved WatchMap::addToWatchInterest(Watch* watch,
+                                          const std::set<std::string>& add_these_names) {
+  std::vector<std::string> newly_added_to_watch(add_these_names.begin(), add_these_names.end());
+  watch->resource_names_.insert(add_these_names.begin(), add_these_names.end());
+  std::set<std::string> removals;
+
+  return AddedRemoved(findAdditions(newly_added_to_watch, watch), std::move(removals));
+}
+
 absl::flat_hash_set<Watch*> WatchMap::watchesInterestedIn(const std::string& resource_name) {
   absl::flat_hash_set<Watch*> ret = wildcard_watches_;
   const auto watches_interested = watch_interest_.find(resource_name);

source/common/config/watch_map.h

@@ -73,6 +73,9 @@ class WatchMap : public SubscriptionCallbacks, public Logger::Loggable<Logger::I
   AddedRemoved updateWatchInterest(Watch* watch,
                                    const std::set<std::string>& update_to_these_names);
 
+  // Adds the extra set of resource names that the given watch should watch.
+  AddedRemoved addToWatchInterest(Watch* watch, const std::set<std::string>& add_these_names);
+
   // Expects that the watch to be removed has already had all of its resource names removed via
   // updateWatchInterest().
   void removeWatch(Watch* watch);

source/common/upstream/cds_api_impl.h

@@ -35,6 +35,10 @@ class CdsApiImpl : public CdsApi,
   }
   const std::string versionInfo() const override { return system_version_info_; }
 
+  void addToClusterInterest(const std::set<std::string>& add_these_names) override {
+    subscription_->addToResourceInterest(add_these_names);
+  }
+
 private:
   // Config::SubscriptionCallbacks
   void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,

source/common/upstream/cluster_manager_impl.h

@@ -243,6 +243,10 @@ class ClusterManagerImpl : public ClusterManager, Logger::Loggable<Logger::Id::u
 
   Config::SubscriptionFactory& subscriptionFactory() override { return subscription_factory_; }
 
+  void addToClusterInterest(const std::set<std::string>& add_these_names) override {
+    cds_api_->addToClusterInterest(add_these_names);
+  }
+
 protected:
   virtual void postThreadLocalDrainConnections(const Cluster& cluster,
                                                const HostVector& hosts_removed);

test/common/config/delta_subscription_impl_test.cc

@@ -56,6 +56,12 @@ TEST_F(DeltaSubscriptionImplTest, PauseHoldsRequest) {
   subscription_->resume();
 }
 
+TEST_F(DeltaSubscriptionImplTest, AddResourceCauseRequest) {
+  startSubscription({});
+  expectSendMessage({"name1", "name2"}, {}, Grpc::Status::WellKnownGrpcStatus::Ok, "", {});
+  subscription_->addToResourceInterest({"name1", "name2"});
+}
+
 TEST_F(DeltaSubscriptionImplTest, ResponseCausesAck) {
   startSubscription({"name1"});
   deliverConfigUpdate({"name1"}, "someversion", true);

test/mocks/config/mocks.h

@@ -99,6 +99,9 @@ class MockGrpcMux : public GrpcMux {
   MOCK_METHOD(Watch*, addOrUpdateWatch,
               (const std::string& type_url, Watch* watch, const std::set<std::string>& resources,
                SubscriptionCallbacks& callbacks, std::chrono::milliseconds init_fetch_timeout));
+  MOCK_METHOD(Watch*, addToWatch,
+              (const std::string& type_url, Watch* watch, const std::set<std::string>& resources,
+               SubscriptionCallbacks& callbacks, std::chrono::milliseconds init_fetch_timeout));
   MOCK_METHOD(void, removeWatch, (const std::string& type_url, Watch* watch));
 };
 

test/mocks/upstream/mocks.h

@@ -330,6 +330,7 @@ class MockClusterManager : public ClusterManager {
   MOCK_METHOD(ClusterUpdateCallbacksHandle*, addThreadLocalClusterUpdateCallbacks_,
               (ClusterUpdateCallbacks & callbacks));
   MOCK_METHOD(Config::SubscriptionFactory&, subscriptionFactory, ());
+  MOCK_METHOD(void, addToClusterInterest, (const std::set<std::string>&));
 
   NiceMock<Http::ConnectionPool::MockInstance> conn_pool_;
   NiceMock<Http::MockAsyncClient> async_client_;
@@ -385,6 +386,7 @@ class MockCdsApi : public CdsApi {
   MOCK_METHOD(void, initialize, ());
   MOCK_METHOD(void, setInitializedCb, (std::function<void()> callback));
   MOCK_METHOD(const std::string, versionInfo, (), (const));
+  MOCK_METHOD(void, addToClusterInterest, (const std::set<std::string>&));
 
   std::function<void()> initialized_callback_;
 };