xdstp: LDS glob collection support.

include/envoy/config/BUILD

@@ -29,6 +29,12 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "context_provider_interface",
+    hdrs = ["context_provider.h"],
+    deps = ["@com_github_cncf_udpa//xds/core/v3:pkg_cc_proto"],
+)
+
 envoy_cc_library(
     name = "extension_config_provider_interface",
     hdrs = ["extension_config_provider.h"],

include/envoy/config/context_provider.h

@@ -0,0 +1,26 @@
+#pragma once
+
+#include "envoy/common/pure.h"
+
+#include "xds/core/v3/context_params.pb.h"
+
+namespace Envoy {
+namespace Config {
+
+/**
+ * A provider for xDS context parameters. These are currently derived from the bootstrap, but will
+ * be set dynamically at runtime in the near future as we add support for dynamic context parameter
+ * discovery and updates.
+ */
+class ContextProvider {
+public:
+  virtual ~ContextProvider() = default;
+
+  /**
+   * @return const xds::core::v3::ContextParams& node-level context parameters.
+   */
+  virtual const xds::core::v3::ContextParams& nodeContext() const PURE;
+};
+
+} // namespace Config
+} // namespace Envoy

include/envoy/config/subscription.h

@@ -179,11 +179,8 @@ class Subscription {
    * Start a configuration subscription asynchronously. This should be called once and will continue
    * to fetch throughout the lifetime of the Subscription object.
    * @param resources set of resource names to fetch.
-   * @param use_namespace_matching if the subscription is for a collection of resources. In such a
-   * case a namespace watch will be created.
    */
-  virtual void start(const std::set<std::string>& resource_names,
-                     const bool use_namespace_matching = false) PURE;
+  virtual void start(const std::set<std::string>& resource_names) PURE;
 
   /**
    * Update the resources to fetch.

include/envoy/config/subscription_factory.h

@@ -22,14 +22,14 @@ class SubscriptionFactory {
    * @param callbacks the callbacks needed by all Subscription objects, to deliver config updates.
    *                  The callbacks must not result in the deletion of the Subscription object.
    * @param resource_decoder how incoming opaque resource objects are to be decoded.
+   * @param use_namespace_matching whether to use namespace match semantics on subscription.
    *
    * @return SubscriptionPtr subscription object corresponding for config and type_url.
    */
-  virtual SubscriptionPtr
-  subscriptionFromConfigSource(const envoy::config::core::v3::ConfigSource& config,
-                               absl::string_view type_url, Stats::Scope& scope,
-                               SubscriptionCallbacks& callbacks,
-                               OpaqueResourceDecoder& resource_decoder) PURE;
+  virtual SubscriptionPtr subscriptionFromConfigSource(
+      const envoy::config::core::v3::ConfigSource& config, absl::string_view type_url,
+      Stats::Scope& scope, SubscriptionCallbacks& callbacks,
+      OpaqueResourceDecoder& resource_decoder, bool use_namespace_matching) PURE;
 
   /**
    * Collection subscription factory interface for xDS-TP URLs.

include/envoy/local_info/BUILD

@@ -12,6 +12,7 @@ envoy_cc_library(
     name = "local_info_interface",
     hdrs = ["local_info.h"],
     deps = [
+        "//include/envoy/config:context_provider_interface",
         "//include/envoy/network:address_interface",
         "//include/envoy/stats:symbol_table_interface",
         "@envoy_api//envoy/config/core/v3:pkg_cc_proto",

include/envoy/local_info/local_info.h

@@ -3,6 +3,7 @@
 #include <string>
 
 #include "envoy/common/pure.h"
+#include "envoy/config/context_provider.h"
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/network/address.h"
 #include "envoy/stats/symbol_table.h"
@@ -46,6 +47,11 @@ class LocalInfo {
    * @return the full node identity presented to management servers.
    */
   virtual const envoy::config::core::v3::Node& node() const PURE;
+
+  /**
+   * @return the xDS context provider for the node.
+   */
+  virtual const Config::ContextProvider& contextProvider() const PURE;
 };
 
 using LocalInfoPtr = std::unique_ptr<LocalInfo>;

source/common/config/BUILD

@@ -44,6 +44,15 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "context_provider_lib",
+    hdrs = ["context_provider_impl.h"],
+    deps = [
+        ":xds_context_params_lib",
+        "//include/envoy/config:context_provider_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "datasource_lib",
     srcs = ["datasource.cc"],
@@ -173,6 +182,7 @@ envoy_cc_library(
     hdrs = ["grpc_subscription_impl.h"],
     deps = [
         ":grpc_mux_lib",
+        ":xds_resource_lib",
         "//include/envoy/config:subscription_interface",
         "//include/envoy/event:dispatcher_interface",
         "//include/envoy/grpc:async_client_interface",
@@ -189,6 +199,8 @@ envoy_cc_library(
         ":pausable_ack_queue_lib",
         ":version_converter_lib",
         ":watch_map_lib",
+        ":xds_context_params_lib",
+        ":xds_resource_lib",
         "//include/envoy/event:dispatcher_interface",
         "//include/envoy/grpc:async_client_interface",
         "//source/common/memory:utils_lib",
@@ -426,6 +438,7 @@ envoy_cc_library(
     hdrs = ["watch_map.h"],
     deps = [
         ":decoded_resource_lib",
+        ":xds_resource_lib",
         "//include/envoy/config:subscription_interface",
         "//source/common/common:assert_lib",
         "//source/common/common:cleanup_lib",

source/common/config/context_provider_impl.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include "envoy/config/context_provider.h"
+
+#include "common/config/xds_context_params.h"
+
+namespace Envoy {
+namespace Config {
+
+class ContextProviderImpl : public ContextProvider {
+public:
+  ContextProviderImpl(const envoy::config::core::v3::Node& node,
+                      const Protobuf::RepeatedPtrField<std::string>& node_context_params)
+      : node_context_(XdsContextParams::encodeNodeContext(node, node_context_params)) {}
+
+  const xds::core::v3::ContextParams& nodeContext() const override { return node_context_; }
+
+private:
+  const xds::core::v3::ContextParams node_context_;
+};
+
+} // namespace Config
+} // namespace Envoy

source/common/config/filesystem_subscription_impl.cc

@@ -27,7 +27,7 @@ FilesystemSubscriptionImpl::FilesystemSubscriptionImpl(
 }
 
 // Config::Subscription
-void FilesystemSubscriptionImpl::start(const std::set<std::string>&, const bool) {
+void FilesystemSubscriptionImpl::start(const std::set<std::string>&) {
   started_ = true;
   // Attempt to read in case there is a file there already.
   refresh();

source/common/config/filesystem_subscription_impl.h

@@ -27,7 +27,7 @@ class FilesystemSubscriptionImpl : public Config::Subscription,
   // Config::Subscription
   // We report all discovered resources in the watched file, so the resource names arguments are
   // unused, and updateResourceInterest is a no-op (other than updating a stat).
-  void start(const std::set<std::string>&, const bool use_namespace_matching = false) override;
+  void start(const std::set<std::string>&) override;
   void updateResourceInterest(const std::set<std::string>&) override;
   void requestOnDemandUpdate(const std::set<std::string>&) override {
     NOT_IMPLEMENTED_GCOVR_EXCL_LINE;

source/common/config/grpc_subscription_impl.cc

@@ -3,24 +3,29 @@
 #include "common/common/assert.h"
 #include "common/common/logger.h"
 #include "common/common/utility.h"
+#include "common/config/xds_resource.h"
 #include "common/grpc/common.h"
 #include "common/protobuf/protobuf.h"
+#include "common/protobuf/type_util.h"
 #include "common/protobuf/utility.h"
 
 namespace Envoy {
 namespace Config {
 
-GrpcSubscriptionImpl::GrpcSubscriptionImpl(
-    GrpcMuxSharedPtr grpc_mux, SubscriptionCallbacks& callbacks,
-    OpaqueResourceDecoder& resource_decoder, SubscriptionStats stats, absl::string_view type_url,
-    Event::Dispatcher& dispatcher, std::chrono::milliseconds init_fetch_timeout, bool is_aggregated)
+GrpcSubscriptionImpl::GrpcSubscriptionImpl(GrpcMuxSharedPtr grpc_mux,
+                                           SubscriptionCallbacks& callbacks,
+                                           OpaqueResourceDecoder& resource_decoder,
+                                           SubscriptionStats stats, absl::string_view type_url,
+                                           Event::Dispatcher& dispatcher,
+                                           std::chrono::milliseconds init_fetch_timeout,
+                                           bool is_aggregated, bool use_namespace_matching)
     : grpc_mux_(grpc_mux), callbacks_(callbacks), resource_decoder_(resource_decoder),
       stats_(stats), type_url_(type_url), dispatcher_(dispatcher),
-      init_fetch_timeout_(init_fetch_timeout), is_aggregated_(is_aggregated) {}
+      init_fetch_timeout_(init_fetch_timeout), is_aggregated_(is_aggregated),
+      use_namespace_matching_(use_namespace_matching) {}
 
 // Config::Subscription
-void GrpcSubscriptionImpl::start(const std::set<std::string>& resources,
-                                 const bool use_namespace_matching) {
+void GrpcSubscriptionImpl::start(const std::set<std::string>& resources) {
   if (init_fetch_timeout_.count() > 0) {
     init_fetch_timeout_timer_ = dispatcher_.createTimer([this]() -> void {
       callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::FetchTimedout,
@@ -30,7 +35,7 @@ void GrpcSubscriptionImpl::start(const std::set<std::string>& resources,
   }
 
   watch_ =
-      grpc_mux_->addWatch(type_url_, resources, *this, resource_decoder_, use_namespace_matching);
+      grpc_mux_->addWatch(type_url_, resources, *this, resource_decoder_, use_namespace_matching_);
 
   // The attempt stat here is maintained for the purposes of having consistency between ADS and
   // gRPC/filesystem/REST Subscriptions. Since ADS is push based and muxed, the notion of an
@@ -121,5 +126,21 @@ void GrpcSubscriptionImpl::disableInitFetchTimeoutTimer() {
   }
 }
 
+GrpcCollectionSubscriptionImpl::GrpcCollectionSubscriptionImpl(
+    const xds::core::v3::ResourceLocator& collection_locator, GrpcMuxSharedPtr grpc_mux,
+    SubscriptionCallbacks& callbacks, OpaqueResourceDecoder& resource_decoder,
+    SubscriptionStats stats, Event::Dispatcher& dispatcher,
+    std::chrono::milliseconds init_fetch_timeout, bool is_aggregated)
+    : GrpcSubscriptionImpl(
+          grpc_mux, callbacks, resource_decoder, stats,
+          TypeUtil::descriptorFullNameToTypeUrl(collection_locator.resource_type()), dispatcher,
+          init_fetch_timeout, is_aggregated, false),
+      collection_locator_(collection_locator) {}
+
+void GrpcCollectionSubscriptionImpl::start(const std::set<std::string>& resource_names) {
+  ASSERT(resource_names.empty());
+  GrpcSubscriptionImpl::start({XdsResourceIdentifier::encodeUrl(collection_locator_)});
+}
+
 } // namespace Config
 } // namespace Envoy

source/common/config/grpc_subscription_impl.h

@@ -8,24 +8,26 @@
 
 #include "common/common/logger.h"
 
+#include "xds/core/v3/resource_locator.pb.h"
+
 namespace Envoy {
 namespace Config {
 
 /**
  * Adapter from typed Subscription to untyped GrpcMux. Also handles per-xDS API stats/logging.
  */
 class GrpcSubscriptionImpl : public Subscription,
-                             SubscriptionCallbacks,
+                             protected SubscriptionCallbacks,
                              Logger::Loggable<Logger::Id::config> {
 public:
   GrpcSubscriptionImpl(GrpcMuxSharedPtr grpc_mux, SubscriptionCallbacks& callbacks,
                        OpaqueResourceDecoder& resource_decoder, SubscriptionStats stats,
                        absl::string_view type_url, Event::Dispatcher& dispatcher,
-                       std::chrono::milliseconds init_fetch_timeout, bool is_aggregated);
+                       std::chrono::milliseconds init_fetch_timeout, bool is_aggregated,
+                       bool use_namespace_matching);
 
   // Config::Subscription
-  void start(const std::set<std::string>& resource_names,
-             const bool use_namespace_matching = false) override;
+  void start(const std::set<std::string>& resource_names) override;
   void updateResourceInterest(const std::set<std::string>& update_to_these_names) override;
   void requestOnDemandUpdate(const std::set<std::string>& add_these_names) override;
   // Config::SubscriptionCallbacks (all pass through to callbacks_!)
@@ -55,10 +57,25 @@ class GrpcSubscriptionImpl : public Subscription,
   std::chrono::milliseconds init_fetch_timeout_;
   Event::TimerPtr init_fetch_timeout_timer_;
   const bool is_aggregated_;
+  const bool use_namespace_matching_;
 };
 
 using GrpcSubscriptionImplPtr = std::unique_ptr<GrpcSubscriptionImpl>;
 using GrpcSubscriptionImplSharedPtr = std::shared_ptr<GrpcSubscriptionImpl>;
 
+class GrpcCollectionSubscriptionImpl : public GrpcSubscriptionImpl {
+public:
+  GrpcCollectionSubscriptionImpl(const xds::core::v3::ResourceLocator& collection_locator,
+                                 GrpcMuxSharedPtr grpc_mux, SubscriptionCallbacks& callbacks,
+                                 OpaqueResourceDecoder& resource_decoder, SubscriptionStats stats,
+                                 Event::Dispatcher& dispatcher,
+                                 std::chrono::milliseconds init_fetch_timeout, bool is_aggregated);
+
+  void start(const std::set<std::string>& resource_names) override;
+
+private:
+  xds::core::v3::ResourceLocator collection_locator_;
+};
+
 } // namespace Config
 } // namespace Envoy

source/common/config/http_subscription_impl.cc

@@ -43,7 +43,7 @@ HttpSubscriptionImpl::HttpSubscriptionImpl(
 }
 
 // Config::Subscription
-void HttpSubscriptionImpl::start(const std::set<std::string>& resource_names, const bool) {
+void HttpSubscriptionImpl::start(const std::set<std::string>& resource_names) {
   if (init_fetch_timeout_.count() > 0) {
     init_fetch_timeout_timer_ = dispatcher_.createTimer([this]() -> void {
       handleFailure(Config::ConfigUpdateFailureReason::FetchTimedout, nullptr);

source/common/config/http_subscription_impl.h

@@ -34,8 +34,7 @@ class HttpSubscriptionImpl : public Http::RestApiFetcher,
                        ProtobufMessage::ValidationVisitor& validation_visitor);
 
   // Config::Subscription
-  void start(const std::set<std::string>& resource_names,
-             const bool use_namespace_matching = false) override;
+  void start(const std::set<std::string>& resource_names) override;
   void updateResourceInterest(const std::set<std::string>& update_to_these_names) override;
   void requestOnDemandUpdate(const std::set<std::string>&) override {
     NOT_IMPLEMENTED_GCOVR_EXCL_LINE;