xDS: add xDS config tracker extension point

api/envoy/config/bootstrap/v3/bootstrap.proto

@@ -41,7 +41,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // <config_overview_bootstrap>` for more detail.
 
 // Bootstrap :ref:`configuration overview <config_overview_bootstrap>`.
-// [#next-free-field: 36]
+// [#next-free-field: 37]
 message Bootstrap {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.bootstrap.v2.Bootstrap";
@@ -342,6 +342,13 @@ message Bootstrap {
   // TODO(abeyad): Add public-facing documentation.
   // [#not-implemented-hide:]
   core.v3.TypedExtensionConfig xds_delegate_extension = 35;
+
+  // Optional XdsConfigTracker configuration, which allows tracking xDS responses in external components,
+  // e.g., external tracer or monitor. It provides the process point when receive, ingest, or fail to
+  // process xDS resources and messages.
+  // If a value is not specified, no XdsConfigTracker will be used.
+  // [#not-implemented-hide:]
+  core.v3.TypedExtensionConfig xds_config_tracker_extension = 36;
 }
 
 // Administration interface :ref:`operations documentation

envoy/config/BUILD

@@ -117,3 +117,16 @@ envoy_cc_library(
         "@envoy_api//envoy/service/discovery/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_library(
+    name = "xds_config_tracker_interface",
+    hdrs = ["xds_config_tracker.h"],
+    deps = [
+        ":subscription_interface",
+        ":typed_config_interface",
+        "//envoy/protobuf:message_validator_interface",
+        "//source/common/config:update_ack_lib",
+        "//source/common/protobuf",
+        "@com_google_googleapis//google/rpc:status_cc_proto",
+    ],
+)

envoy/config/subscription.h

@@ -4,6 +4,7 @@
 #include <vector>
 
 #include "envoy/common/exception.h"
+#include "envoy/common/optref.h"
 #include "envoy/common/pure.h"
 #include "envoy/service/discovery/v3/discovery.pb.h"
 #include "envoy/stats/stats_macros.h"
@@ -59,6 +60,11 @@ class DecodedResource {
    * @return bool does the xDS discovery response have a set resource payload?
    */
   virtual bool hasResource() const PURE;
+
+  /**
+   * @return optional ref<envoy::config::core::v3::Metadata> of a resource.
+   */
+  virtual const OptRef<const envoy::config::core::v3::Metadata> metadata() const PURE;
 };
 
 using DecodedResourcePtr = std::unique_ptr<DecodedResource>;

envoy/config/xds_config_tracker.h

@@ -0,0 +1,109 @@
+#pragma once
+
+#include <string>
+#include <vector>
+
+#include "envoy/common/optref.h"
+#include "envoy/config/subscription.h"
+#include "envoy/config/typed_config.h"
+#include "envoy/protobuf/message_validator.h"
+#include "envoy/stats/scope.h"
+
+#include "source/common/protobuf/protobuf.h"
+
+#include "google/rpc/status.pb.h"
+
+namespace Envoy {
+namespace Config {
+
+/**
+ * An interface for hooking into xDS update events to provide the ablility to use some external
+ * processor in xDS update. This tracker provides the process point when the discovery response
+ * is received, when the resources are successfully processed and applied, and when there is any
+ * failure.
+ *
+ * Instance of this interface get invoked on the main Envoy thread. Thus, it is important
+ * for implementations of this interface to not execute any blocking operations on the same
+ * thread.
+ */
+class XdsConfigTracker {
+public:
+  virtual ~XdsConfigTracker() = default;
+
+  /**
+   * Invoked when SotW xDS configuration updates have been successfully parsed, applied on
+   * the Envoy instance, and are about to be ACK'ed.
+   *
+   * For SotW, the passed resources contain all the received resources except for the heart-beat
+   * ones in the original message. The call of this method means there is a subscriber for this
+   * type_url and the type of resource is same as the message's type_url.
+   *
+   * Note: this method is called when *all* the resouces in a response are accepted.
+   *
+   * @param type_url The type url of xDS message.
+   * @param resources A list of decoded resources to add to the current state.
+   */
+  virtual void onConfigAccepted(const absl::string_view type_url,
+                                const std::vector<DecodedResourcePtr>& resources) PURE;
+
+  /**
+   * Invoked when Delta xDS configuration updates have been successfully accepted, applied on
+   * the Envoy instance, and are about to be ACK'ed.
+   *
+   * For Delta, added_resources contains all the received added resources except for the heart-beat
+   * ones in the original message, and the removed resouces are the same in the xDS message.
+   *
+   * Note: this method is called when *all* the resouces in a response are accepted.
+   *
+   * @param type_url The type url of xDS message.
+   * @param added_resources A list of decoded resources to add to the current state.
+   * @param removed_resources A list of resources to remove from the current state.
+   */
+  virtual void onConfigAccepted(
+      const absl::string_view type_url,
+      const Protobuf::RepeatedPtrField<envoy::service::discovery::v3::Resource>& added_resources,
+      const Protobuf::RepeatedPtrField<std::string>& removed_resources) PURE;
+
+  /**
+   * Invoked when xds configs are rejected during xDS ingestion.
+   *
+   * @param message The SotW discovery response message body.
+   * @param details The process state and error details.
+   */
+  virtual void onConfigRejected(const envoy::service::discovery::v3::DiscoveryResponse& message,
+                                const absl::string_view error_detail) PURE;
+
+  /**
+   * Invoked when xds configs are rejected during xDS ingestion.
+   *
+   * @param message The Delta discovery response message body.
+   * @param details The process state and error details.
+   */
+  virtual void
+  onConfigRejected(const envoy::service::discovery::v3::DeltaDiscoveryResponse& message,
+                   const absl::string_view error_detail) PURE;
+};
+
+using XdsConfigTrackerPtr = std::unique_ptr<XdsConfigTracker>;
+using XdsConfigTrackerOptRef = OptRef<XdsConfigTracker>;
+
+/**
+ * A factory abstract class for creating instances of XdsConfigTracker.
+ */
+class XdsConfigTrackerFactory : public Config::TypedFactory {
+public:
+  ~XdsConfigTrackerFactory() override = default;
+
+  /**
+   * Creates an XdsConfigTracker using the given config.
+   */
+  virtual XdsConfigTrackerPtr
+  createXdsConfigTracker(const ProtobufWkt::Any& config,
+                         ProtobufMessage::ValidationVisitor& validation_visitor,
+                         Event::Dispatcher& dispatcher, Stats::Scope& stats) PURE;
+
+  std::string category() const override { return "envoy.config.xds_tracker"; }
+};
+
+} // namespace Config
+} // namespace Envoy

source/common/config/BUILD

@@ -166,6 +166,7 @@ envoy_cc_library(
         ":xds_source_id_lib",
         "//envoy/config:grpc_mux_interface",
         "//envoy/config:subscription_interface",
+        "//envoy/config:xds_config_tracker_interface",
         "//envoy/config:xds_resources_delegate_interface",
         "//envoy/upstream:cluster_manager_interface",
         "//source/common/common:cleanup_lib",
@@ -203,6 +204,7 @@ envoy_cc_library(
         ":watch_map_lib",
         ":xds_context_params_lib",
         ":xds_resource_lib",
+        "//envoy/config:xds_config_tracker_interface",
         "//envoy/event:dispatcher_interface",
         "//envoy/grpc:async_client_interface",
         "//source/common/memory:utils_lib",
@@ -329,6 +331,7 @@ envoy_cc_library(
         ":xds_resource_lib",
         "//envoy/config:subscription_factory_interface",
         "//envoy/config:subscription_interface",
+        "//envoy/config:xds_config_tracker_interface",
         "//envoy/config:xds_resources_delegate_interface",
         "//envoy/server:instance_interface",
         "//envoy/upstream:cluster_manager_interface",
@@ -434,6 +437,7 @@ envoy_cc_library(
         ":utility_lib",
         ":xds_resource_lib",
         "//envoy/config:subscription_interface",
+        "//envoy/config:xds_config_tracker_interface",
         "//source/common/common:assert_lib",
         "//source/common/common:cleanup_lib",
         "//source/common/common:minimal_logger_lib",

source/common/config/decoded_resource_impl.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "envoy/common/optref.h"
 #include "envoy/config/subscription.h"
 #include "envoy/service/discovery/v3/discovery.pb.h"
 
@@ -40,7 +41,7 @@ class DecodedResourceImpl : public DecodedResource {
 
     return std::unique_ptr<DecodedResourceImpl>(new DecodedResourceImpl(
         resource_decoder, absl::nullopt, Protobuf::RepeatedPtrField<std::string>(), resource, true,
-        version, absl::nullopt));
+        version, absl::nullopt, absl::nullopt));
   }
 
   static DecodedResourceImplPtr
@@ -51,21 +52,22 @@ class DecodedResourceImpl : public DecodedResource {
 
   DecodedResourceImpl(OpaqueResourceDecoder& resource_decoder,
                       const envoy::service::discovery::v3::Resource& resource)
-      : DecodedResourceImpl(resource_decoder, resource.name(), resource.aliases(),
-                            resource.resource(), resource.has_resource(), resource.version(),
-                            resource.has_ttl()
-                                ? absl::make_optional(std::chrono::milliseconds(
-                                      DurationUtil::durationToMilliseconds(resource.ttl())))
-                                : absl::nullopt) {}
+      : DecodedResourceImpl(
+            resource_decoder, resource.name(), resource.aliases(), resource.resource(),
+            resource.has_resource(), resource.version(),
+            resource.has_ttl() ? absl::make_optional(std::chrono::milliseconds(
+                                     DurationUtil::durationToMilliseconds(resource.ttl())))
+                               : absl::nullopt,
+            resource.has_metadata() ? makeOptRef(resource.metadata()) : absl::nullopt) {}
   DecodedResourceImpl(OpaqueResourceDecoder& resource_decoder,
                       const xds::core::v3::CollectionEntry::InlineEntry& inline_entry)
       : DecodedResourceImpl(resource_decoder, inline_entry.name(),
                             Protobuf::RepeatedPtrField<std::string>(), inline_entry.resource(),
-                            true, inline_entry.version(), absl::nullopt) {}
+                            true, inline_entry.version(), absl::nullopt, absl::nullopt) {}
   DecodedResourceImpl(ProtobufTypes::MessagePtr resource, const std::string& name,
                       const std::vector<std::string>& aliases, const std::string& version)
       : resource_(std::move(resource)), has_resource_(true), name_(name), aliases_(aliases),
-        version_(version), ttl_(absl::nullopt) {}
+        version_(version), ttl_(absl::nullopt), metadata_(absl::nullopt) {}
 
   // Config::DecodedResource
   const std::string& name() const override { return name_; }
@@ -74,15 +76,20 @@ class DecodedResourceImpl : public DecodedResource {
   const Protobuf::Message& resource() const override { return *resource_; };
   bool hasResource() const override { return has_resource_; }
   absl::optional<std::chrono::milliseconds> ttl() const override { return ttl_; }
+  const OptRef<const envoy::config::core::v3::Metadata> metadata() const override {
+    return metadata_;
+  }
 
 private:
   DecodedResourceImpl(OpaqueResourceDecoder& resource_decoder, absl::optional<std::string> name,
                       const Protobuf::RepeatedPtrField<std::string>& aliases,
                       const ProtobufWkt::Any& resource, bool has_resource,
-                      const std::string& version, absl::optional<std::chrono::milliseconds> ttl)
+                      const std::string& version, absl::optional<std::chrono::milliseconds> ttl,
+                      const OptRef<const envoy::config::core::v3::Metadata> metadata)
       : resource_(resource_decoder.decodeResource(resource)), has_resource_(has_resource),
         name_(name ? *name : resource_decoder.resourceName(*resource_)),
-        aliases_(repeatedPtrFieldToVector(aliases)), version_(version), ttl_(ttl) {}
+        aliases_(repeatedPtrFieldToVector(aliases)), version_(version), ttl_(ttl),
+        metadata_(metadata) {}
 
   const ProtobufTypes::MessagePtr resource_;
   const bool has_resource_;
@@ -91,6 +98,10 @@ class DecodedResourceImpl : public DecodedResource {
   const std::string version_;
   // Per resource TTL.
   const absl::optional<std::chrono::milliseconds> ttl_;
+
+  // This is the metadata info under the Resource wrapper.
+  // It is intended to be consumed in the xds_config_tracker extension.
+  const OptRef<const envoy::config::core::v3::Metadata> metadata_;
 };
 
 struct DecodedResourcesWrapper {

source/common/config/delta_subscription_state.cc

@@ -9,13 +9,16 @@ namespace {
 DeltaSubscriptionStateVariant getState(std::string type_url,
                                        UntypedConfigUpdateCallbacks& watch_map,
                                        const LocalInfo::LocalInfo& local_info,
-                                       Event::Dispatcher& dispatcher) {
+                                       Event::Dispatcher& dispatcher,
+                                       XdsConfigTrackerOptRef xds_config_tracker) {
   if (Runtime::runtimeFeatureEnabled("envoy.restart_features.explicit_wildcard_resource")) {
     return DeltaSubscriptionStateVariant(absl::in_place_type<NewDeltaSubscriptionState>,
-                                         std::move(type_url), watch_map, local_info, dispatcher);
+                                         std::move(type_url), watch_map, local_info, dispatcher,
+                                         xds_config_tracker);
   } else {
     return DeltaSubscriptionStateVariant(absl::in_place_type<OldDeltaSubscriptionState>,
-                                         std::move(type_url), watch_map, local_info, dispatcher);
+                                         std::move(type_url), watch_map, local_info, dispatcher,
+                                         xds_config_tracker);
   }
 }
 
@@ -24,8 +27,10 @@ DeltaSubscriptionStateVariant getState(std::string type_url,
 DeltaSubscriptionState::DeltaSubscriptionState(std::string type_url,
                                                UntypedConfigUpdateCallbacks& watch_map,
                                                const LocalInfo::LocalInfo& local_info,
-                                               Event::Dispatcher& dispatcher)
-    : state_(getState(std::move(type_url), watch_map, local_info, dispatcher)) {}
+                                               Event::Dispatcher& dispatcher,
+                                               XdsConfigTrackerOptRef xds_config_tracker)
+    : state_(getState(std::move(type_url), watch_map, local_info, dispatcher, xds_config_tracker)) {
+}
 
 void DeltaSubscriptionState::updateSubscriptionInterest(
     const absl::flat_hash_set<std::string>& cur_added,

source/common/config/delta_subscription_state.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "envoy/config/subscription.h"
+#include "envoy/config/xds_config_tracker.h"
 #include "envoy/local_info/local_info.h"
 #include "envoy/service/discovery/v3/discovery.pb.h"
 
@@ -20,7 +21,8 @@ using DeltaSubscriptionStateVariant =
 class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
 public:
   DeltaSubscriptionState(std::string type_url, UntypedConfigUpdateCallbacks& watch_map,
-                         const LocalInfo::LocalInfo& local_info, Event::Dispatcher& dispatcher);
+                         const LocalInfo::LocalInfo& local_info, Event::Dispatcher& dispatcher,
+                         XdsConfigTrackerOptRef xds_config_tracker);
 
   void updateSubscriptionInterest(const absl::flat_hash_set<std::string>& cur_added,
                                   const absl::flat_hash_set<std::string>& cur_removed);
@@ -37,6 +39,7 @@ class DeltaSubscriptionState : public Logger::Loggable<Logger::Id::config> {
 
 private:
   DeltaSubscriptionStateVariant state_;
+  XdsConfigTrackerOptRef xds_config_tracker_;
 };
 
 } // namespace Config

source/common/config/grpc_mux_impl.cc

@@ -39,12 +39,13 @@ GrpcMuxImpl::GrpcMuxImpl(const LocalInfo::LocalInfo& local_info,
                          Random::RandomGenerator& random, Stats::Scope& scope,
                          const RateLimitSettings& rate_limit_settings, bool skip_subsequent_node,
                          CustomConfigValidatorsPtr&& config_validators,
+                         XdsConfigTrackerOptRef xds_config_tracker,
                          XdsResourcesDelegateOptRef xds_resources_delegate,
                          const std::string& target_xds_authority)
     : grpc_stream_(this, std::move(async_client), service_method, random, dispatcher, scope,
                    rate_limit_settings),
       local_info_(local_info), skip_subsequent_node_(skip_subsequent_node),
-      config_validators_(std::move(config_validators)),
+      config_validators_(std::move(config_validators)), xds_config_tracker_(xds_config_tracker),
       xds_resources_delegate_(xds_resources_delegate), target_xds_authority_(target_xds_authority),
       first_stream_request_(true), dispatcher_(dispatcher),
       dynamic_update_callback_handle_(local_info.contextProvider().addDynamicContextUpdateCallback(
@@ -219,6 +220,7 @@ void GrpcMuxImpl::onDiscoveryResponse(
     ControlPlaneStats& control_plane_stats) {
   const std::string type_url = message->type_url();
   ENVOY_LOG(debug, "Received gRPC message for {} at version {}", type_url, message->version_info());
+
   if (api_state_.count(type_url) == 0) {
     // TODO(yuval-k): This should never happen. consider dropping the stream as this is a
     // protocol violation
@@ -286,6 +288,11 @@ void GrpcMuxImpl::onDiscoveryResponse(
 
     processDiscoveryResources(resources, api_state, type_url, message->version_info(),
                               /*call_delegate=*/true);
+
+    // Processing point when resources are successfully ingested.
+    if (xds_config_tracker_.has_value()) {
+      xds_config_tracker_->onConfigAccepted(type_url, resources);
+    }
   }
   END_TRY
   catch (const EnvoyException& e) {
@@ -296,6 +303,11 @@ void GrpcMuxImpl::onDiscoveryResponse(
     ::google::rpc::Status* error_detail = api_state.request_.mutable_error_detail();
     error_detail->set_code(Grpc::Status::WellKnownGrpcStatus::Internal);
     error_detail->set_message(Config::Utility::truncateGrpcStatusMessage(e.what()));
+
+    // Processing point when there is any exception during the parse and ingestion process.
+    if (xds_config_tracker_.has_value()) {
+      xds_config_tracker_->onConfigRejected(*message, error_detail->message());
+    }
   }
   previously_fetched_data_ = true;
   api_state.request_.set_response_nonce(message->nonce());