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 XdsConfigTracer configuration, which allows tracking xDS responses in external components,
+  // e.g., external logger, tracer, or monitor. It provides the log point when receive, ingest, or fail to
+  // process xDS resources and messages.
+  // If a value is not specified, no XdsConfigTracer will be used.
+  // [#not-implemented-hide:]
+  core.v3.TypedExtensionConfig xds_config_tracer_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_tracer_interface",
+    hdrs = ["xds_config_tracer.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

@@ -59,6 +59,11 @@ class DecodedResource {
    * @return bool does the xDS discovery response have a set resource payload?
    */
   virtual bool hasResource() const PURE;
+
+  /**
+   * @return absl::optional<envoy::config::core::v3::Metadata> of a resource.
+   */
+  virtual absl::optional<envoy::config::core::v3::Metadata> metadata() const PURE;
 };
 
 using DecodedResourcePtr = std::unique_ptr<DecodedResource>;

envoy/config/xds_config_tracer.h

@@ -0,0 +1,96 @@
+#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 "source/common/protobuf/protobuf.h"
+
+#include "google/rpc/status.pb.h"
+
+namespace Envoy {
+namespace Config {
+
+// The status of processing Delta/DiscoveryResponse.
+enum TraceState {
+  // Successfully got the resources or message.
+  RECEIVE = 0,
+  // Successfully ingested the resouces.
+  INGESTED = 1,
+  // Failed to apply the resources.
+  FAILED = 2,
+};
+
+struct TraceDetails {
+  TraceDetails(const TraceState state) : state_(state){};
+  TraceDetails(const TraceState state, const ::google::rpc::Status error_detail)
+      : state_(state), error_detail_(error_detail){};
+  const TraceState state_;
+  ::google::rpc::Status error_detail_;
+};
+
+/**
+ * An interface for hooking into xDS update events to provide the ablility to use some
+ * external logger or processor in xDS update.
+ *
+ * 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 XdsConfigTracer {
+public:
+  virtual ~XdsConfigTracer() = default;
+
+  /**
+   * Log the decoded SotW xDS resources that are about to be ingested.
+   * @param type_url The type url of xDS message.
+   * @param resources List of decoded resources that reflect the latest state.
+   * @param details The log point state and details.
+   */
+  virtual void log(const absl::string_view type_url,
+                   const std::vector<DecodedResourcePtr>& resources,
+                   const TraceDetails& details) PURE;
+
+  /**
+   * Log point for SotW xDS discovery response.
+   * @param message The SotW discovery response message body.
+   * @param details The log point state and details.
+   */
+  virtual void log(const envoy::service::discovery::v3::DiscoveryResponse& message,
+                   const TraceDetails& details) PURE;
+
+  /**
+   * Log point for Delta xDS discovery response message when received, ingested, and failed.
+   * @param message The Delta discovery response message body.
+   * @param details The log point state and details.
+   */
+  virtual void log(const envoy::service::discovery::v3::DeltaDiscoveryResponse& message,
+                   const TraceDetails& details) PURE;
+};
+
+using XdsConfigTracerPtr = std::unique_ptr<XdsConfigTracer>;
+using XdsConfigTracerOptRef = OptRef<XdsConfigTracer>;
+
+/**
+ * A factory abstract class for creating instances of XdsConfigTracer.
+ */
+class XdsConfigTracerFactory : public Config::TypedFactory {
+public:
+  ~XdsConfigTracerFactory() override = default;
+
+  /**
+   * Creates an XdsConfigTracer using the given config.
+   */
+  virtual XdsConfigTracerPtr
+  createXdsConfigTracer(const ProtobufWkt::Any& config,
+                        ProtobufMessage::ValidationVisitor& validation_visitor) PURE;
+
+  std::string category() const override { return "envoy.config.xds_tracers"; }
+};
+
+} // 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_tracer_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_tracer_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_tracer_interface",
         "//envoy/config:xds_resources_delegate_interface",
         "//envoy/server:instance_interface",
         "//envoy/upstream:cluster_manager_interface",

source/common/config/decoded_resource_impl.h

@@ -40,7 +40,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 +51,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() ? absl::make_optional(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 +75,18 @@ 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_; }
+  absl::optional<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 absl::optional<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 +95,7 @@ class DecodedResourceImpl : public DecodedResource {
   const std::string version_;
   // Per resource TTL.
   const absl::optional<std::chrono::milliseconds> ttl_;
+  const absl::optional<envoy::config::core::v3::Metadata> metadata_;
 };
 
 struct DecodedResourcesWrapper {

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,
+                         XdsConfigTracerOptRef xds_config_tracer,
                          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_tracer_(xds_config_tracer),
       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(
@@ -284,8 +285,16 @@ void GrpcMuxImpl::onDiscoveryResponse(
       }
     }
 
+    if (xds_config_tracer_.has_value()) {
+      xds_config_tracer_->log(type_url, resources, TraceDetails(TraceState::RECEIVE));
+    }
+
     processDiscoveryResources(resources, api_state, type_url, message->version_info(),
                               /*call_delegate=*/true);
+
+    if (xds_config_tracer_.has_value()) {
+      xds_config_tracer_->log(type_url, resources, TraceDetails(TraceState::INGESTED));
+    }
   }
   END_TRY
   catch (const EnvoyException& e) {
@@ -296,6 +305,9 @@ 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()));
+    if (xds_config_tracer_.has_value()) {
+      xds_config_tracer_->log(*message, TraceDetails(TraceState::FAILED, *error_detail));
+    }
   }
   previously_fetched_data_ = true;
   api_state.request_.set_response_nonce(message->nonce());

source/common/config/grpc_mux_impl.h

@@ -8,6 +8,7 @@
 #include "envoy/common/time.h"
 #include "envoy/config/grpc_mux.h"
 #include "envoy/config/subscription.h"
+#include "envoy/config/xds_config_tracer.h"
 #include "envoy/config/xds_resources_delegate.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/grpc/status.h"
@@ -39,6 +40,7 @@ class GrpcMuxImpl : public GrpcMux,
               Random::RandomGenerator& random, Stats::Scope& scope,
               const RateLimitSettings& rate_limit_settings, bool skip_subsequent_node,
               CustomConfigValidatorsPtr&& config_validators,
+              XdsConfigTracerOptRef xds_config_tracer,
               XdsResourcesDelegateOptRef xds_resources_delegate,
               const std::string& target_xds_authority);
 
@@ -182,6 +184,7 @@ class GrpcMuxImpl : public GrpcMux,
   const LocalInfo::LocalInfo& local_info_;
   const bool skip_subsequent_node_;
   CustomConfigValidatorsPtr config_validators_;
+  XdsConfigTracerOptRef xds_config_tracer_;
   XdsResourcesDelegateOptRef xds_resources_delegate_;
   const std::string target_xds_authority_;
   bool first_stream_request_;

source/common/config/new_grpc_mux_impl.cc

@@ -40,15 +40,16 @@ NewGrpcMuxImpl::NewGrpcMuxImpl(Grpc::RawAsyncClientPtr&& async_client,
                                Random::RandomGenerator& random, Stats::Scope& scope,
                                const RateLimitSettings& rate_limit_settings,
                                const LocalInfo::LocalInfo& local_info,
-                               CustomConfigValidatorsPtr&& config_validators)
+                               CustomConfigValidatorsPtr&& config_validators,
+                               XdsConfigTracerOptRef xds_config_tracer)
     : grpc_stream_(this, std::move(async_client), service_method, random, dispatcher, scope,
                    rate_limit_settings),
       local_info_(local_info), config_validators_(std::move(config_validators)),
       dynamic_update_callback_handle_(local_info.contextProvider().addDynamicContextUpdateCallback(
           [this](absl::string_view resource_type_url) {
             onDynamicContextUpdate(resource_type_url);
           })),
-      dispatcher_(dispatcher) {
+      dispatcher_(dispatcher), xds_config_tracer_(xds_config_tracer) {
   AllMuxes::get().insert(this);
 }
 
@@ -89,6 +90,11 @@ void NewGrpcMuxImpl::onDiscoveryResponse(
     ControlPlaneStats& control_plane_stats) {
   ENVOY_LOG(debug, "Received DeltaDiscoveryResponse for {} at version {}", message->type_url(),
             message->system_version_info());
+
+  if (xds_config_tracer_.has_value()) {
+    xds_config_tracer_->log(*message, TraceDetails(TraceState::RECEIVE));
+  }
+
   auto sub = subscriptions_.find(message->type_url());
   if (sub == subscriptions_.end()) {
     ENVOY_LOG(warn,
@@ -108,7 +114,15 @@ void NewGrpcMuxImpl::onDiscoveryResponse(
     }
   }
 
-  kickOffAck(sub->second->sub_state_.handleResponse(*message));
+  auto ack = sub->second->sub_state_.handleResponse(*message);
+  if (xds_config_tracer_.has_value()) {
+    xds_config_tracer_->log(
+        *message, TraceDetails(ack.error_detail_.code() == Grpc::Status::WellKnownGrpcStatus::Ok
+                                   ? TraceState::INGESTED
+                                   : TraceState::FAILED,
+                               ack.error_detail_));
+  }
+  kickOffAck(ack);
   Memory::Utils::tryShrinkHeap();
 }
 

source/common/config/new_grpc_mux_impl.h

@@ -6,6 +6,7 @@
 #include "envoy/common/token_bucket.h"
 #include "envoy/config/grpc_mux.h"
 #include "envoy/config/subscription.h"
+#include "envoy/config/xds_config_tracer.h"
 #include "envoy/service/discovery/v3/discovery.pb.h"
 
 #include "source/common/common/logger.h"
@@ -34,7 +35,8 @@ class NewGrpcMuxImpl
                  const Protobuf::MethodDescriptor& service_method, Random::RandomGenerator& random,
                  Stats::Scope& scope, const RateLimitSettings& rate_limit_settings,
                  const LocalInfo::LocalInfo& local_info,
-                 CustomConfigValidatorsPtr&& config_validators);
+                 CustomConfigValidatorsPtr&& config_validators,
+                 XdsConfigTracerOptRef xds_config_tracer);
 
   ~NewGrpcMuxImpl() override;
 
@@ -178,6 +180,7 @@ class NewGrpcMuxImpl
   CustomConfigValidatorsPtr config_validators_;
   Common::CallbackHandlePtr dynamic_update_callback_handle_;
   Event::Dispatcher& dispatcher_;
+  XdsConfigTracerOptRef xds_config_tracer_;
 
   // True iff Envoy is shutting down; no messages should be sent on the `grpc_stream_` when this is
   // true because it may contain dangling pointers.