[call v3] add dynamic filter support to client channel (grpc#36877)

Also made some minor improvements to the `ConfigSelector` API. Closes grpc#36877 COPYBARA_INTEGRATE_REVIEW=grpc#36877 from markdroth:client_channel_v3_dynamic_filters 6a539fe PiperOrigin-RevId: 642755276
DerkachAlex · Jun 12, 2024 · f7ce3ee · f7ce3ee
1 parent 8564f72
commit f7ce3ee
Show file tree

Hide file tree

Showing 26 changed files with 180 additions and 105 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
diff --git a/Makefile b/Makefile
diff --git a/Package.swift b/Package.swift
diff --git a/build_autogenerated.yaml b/build_autogenerated.yaml
diff --git a/config.m4 b/config.m4
diff --git a/config.w32 b/config.w32
diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec
diff --git a/grpc.gemspec b/grpc.gemspec
diff --git a/package.xml b/package.xml
diff --git a/src/core/BUILD b/src/core/BUILD
@@ -3295,9 +3295,6 @@ grpc_cc_library(
 
 grpc_cc_library(
     name = "config_selector",
-    srcs = [
-        "client_channel/config_selector.cc",
-    ],
     hdrs = [
         "client_channel/config_selector.h",
     ],
@@ -3313,9 +3310,11 @@ grpc_cc_library(
         "channel_fwd",
         "client_channel_internal_header",
         "grpc_service_config",
+        "interception_chain",
         "metadata_batch",
         "ref_counted",
         "slice",
+        "unique_type_name",
         "useful",
         "//:gpr_public_hdrs",
         "//:grpc_public_hdrs",
@@ -5215,6 +5214,7 @@ grpc_cc_library(
         "grpc_tls_credentials",
         "grpc_transport_chttp2_client_connector",
         "init_internally",
+        "interception_chain",
         "iomgr_fwd",
         "json",
         "json_args",

diff --git a/src/core/client_channel/client_channel.cc b/src/core/client_channel/client_channel.cc
@@ -1195,14 +1195,17 @@ void ClientChannel::UpdateServiceConfigInDataPlaneLocked() {
   }
   CoreConfiguration::Get().channel_init().AddToInterceptionChainBuilder(
       GRPC_CLIENT_CHANNEL, builder);
-  // TODO(roth): add filters returned by config selector
-  // Create call destination.
+  // Add filters returned by the config selector (e.g., xDS HTTP filters).
+  config_selector->AddFilters(builder);
+  // TODO(roth, ctiller): When we implement the retry interceptor, that
+  // needs to be added *after* the filters added by the config selector.
   const bool enable_retries =
       !channel_args_.WantMinimalStack() &&
       channel_args_.GetBool(GRPC_ARG_ENABLE_RETRIES).value_or(true);
   if (enable_retries) {
     Crash("call v3 stack does not yet support retries");
   }
+  // Create call destination.
   auto top_of_stack_call_destination = builder.Build(call_destination_);
   // Send result to data plane.
   if (!top_of_stack_call_destination.ok()) {

diff --git a/src/core/client_channel/config_selector.cc b/src/core/client_channel/config_selector.cc
diff --git a/src/core/client_channel/config_selector.h b/src/core/client_channel/config_selector.h
@@ -35,8 +35,10 @@
 #include "src/core/lib/channel/channel_fwd.h"
 #include "src/core/lib/gprpp/ref_counted.h"
 #include "src/core/lib/gprpp/ref_counted_ptr.h"
+#include "src/core/lib/gprpp/unique_type_name.h"
 #include "src/core/lib/resource_quota/arena.h"
 #include "src/core/lib/slice/slice.h"
+#include "src/core/lib/transport/interception_chain.h"
 #include "src/core/lib/transport/metadata_batch.h"
 #include "src/core/service_config/service_config.h"
 #include "src/core/util/useful.h"
@@ -50,34 +52,32 @@ namespace grpc_core {
 // MethodConfig and provide input to LB policies on a per-call basis.
 class ConfigSelector : public RefCounted<ConfigSelector> {
  public:
-  struct GetCallConfigArgs {
-    grpc_metadata_batch* initial_metadata;
-    Arena* arena;
-    ClientChannelServiceConfigCallData* service_config_call_data;
-  };
-
   ~ConfigSelector() override = default;
 
-  virtual const char* name() const = 0;
+  virtual UniqueTypeName name() const = 0;
 
   static bool Equals(const ConfigSelector* cs1, const ConfigSelector* cs2) {
     if (cs1 == nullptr) return cs2 == nullptr;
     if (cs2 == nullptr) return false;
-    if (strcmp(cs1->name(), cs2->name()) != 0) return false;
+    if (cs1->name() != cs2->name()) return false;
     return cs1->Equals(cs2);
   }
 
   // The channel will call this when the resolver returns a new ConfigSelector
   // to determine what set of dynamic filters will be configured.
+  virtual void AddFilters(InterceptionChainBuilder& /*builder*/) {}
+  // TODO(roth): Remove this once the legacy filter stack goes away.
   virtual std::vector<const grpc_channel_filter*> GetFilters() { return {}; }
 
-  // Returns the call config to use for the call, or a status to fail
-  // the call with.
+  // Gets the configuration for the call and stores it in service config
+  // call data.
+  struct GetCallConfigArgs {
+    grpc_metadata_batch* initial_metadata;
+    Arena* arena;
+    ClientChannelServiceConfigCallData* service_config_call_data;
+  };
   virtual absl::Status GetCallConfig(GetCallConfigArgs args) = 0;
 
-  grpc_arg MakeChannelArg() const;
-  static RefCountedPtr<ConfigSelector> GetFromChannelArgs(
-      const grpc_channel_args& args);
   static absl::string_view ChannelArgName() { return GRPC_ARG_CONFIG_SELECTOR; }
   static int ChannelArgsCompare(const ConfigSelector* a,
                                 const ConfigSelector* b) {
@@ -101,7 +101,10 @@ class DefaultConfigSelector final : public ConfigSelector {
     DCHECK(service_config_ != nullptr);
   }
 
-  const char* name() const override { return "default"; }
+  UniqueTypeName name() const override {
+    static UniqueTypeName::Factory kFactory("default");
+    return kFactory.Create();
+  }
 
   absl::Status GetCallConfig(GetCallConfigArgs args) override {
     Slice* path = args.initial_metadata->get_pointer(HttpPathMetadata());

diff --git a/src/core/resolver/xds/xds_resolver.cc b/src/core/resolver/xds/xds_resolver.cc
@@ -270,7 +270,10 @@ class XdsResolver final : public Resolver {
                       RefCountedPtr<RouteConfigData> route_config_data);
     ~XdsConfigSelector() override;
 
-    const char* name() const override { return "XdsConfigSelector"; }
+    UniqueTypeName name() const override {
+      static UniqueTypeName::Factory kFactory("XdsConfigSelector");
+      return kFactory.Create();
+    }
 
     bool Equals(const ConfigSelector* other) const override {
       const auto* other_xds = static_cast<const XdsConfigSelector*>(other);
@@ -281,14 +284,14 @@ class XdsResolver final : public Resolver {
 
     absl::Status GetCallConfig(GetCallConfigArgs args) override;
 
-    std::vector<const grpc_channel_filter*> GetFilters() override {
-      return filters_;
-    }
+    void AddFilters(InterceptionChainBuilder& builder) override;
+
+    std::vector<const grpc_channel_filter*> GetFilters() override;
 
    private:
     RefCountedPtr<XdsResolver> resolver_;
     RefCountedPtr<RouteConfigData> route_config_data_;
-    std::vector<const grpc_channel_filter*> filters_;
+    std::vector<const XdsHttpFilterImpl*> filters_;
   };
 
   class XdsRouteStateAttributeImpl final : public XdsRouteStateAttribute {
@@ -641,12 +644,9 @@ XdsResolver::XdsConfigSelector::XdsConfigSelector(
         http_filter_registry.GetFilterForType(
             http_filter.config.config_proto_type_name);
     CHECK_NE(filter_impl, nullptr);
-    // Add C-core filter to list.
-    if (filter_impl->channel_filter() != nullptr) {
-      filters_.push_back(filter_impl->channel_filter());
-    }
+    // Add filter to list.
+    filters_.push_back(filter_impl);
   }
-  filters_.push_back(&ClusterSelectionFilter::kFilter);
 }
 
 XdsResolver::XdsConfigSelector::~XdsConfigSelector() {
@@ -799,6 +799,26 @@ absl::Status XdsResolver::XdsConfigSelector::GetCallConfig(
   return absl::OkStatus();
 }
 
+void XdsResolver::XdsConfigSelector::AddFilters(
+    InterceptionChainBuilder& builder) {
+  for (const XdsHttpFilterImpl* filter : filters_) {
+    filter->AddFilter(builder);
+  }
+  builder.Add<ClusterSelectionFilter>();
+}
+
+std::vector<const grpc_channel_filter*>
+XdsResolver::XdsConfigSelector::GetFilters() {
+  std::vector<const grpc_channel_filter*> filters;
+  for (const XdsHttpFilterImpl* filter : filters_) {
+    if (filter->channel_filter() != nullptr) {
+      filters.push_back(filter->channel_filter());
+    }
+  }
+  filters.push_back(&ClusterSelectionFilter::kFilter);
+  return filters;
+}
+
 //
 // XdsResolver::XdsRouteStateAttributeImpl
 //

diff --git a/src/core/xds/grpc/xds_http_fault_filter.cc b/src/core/xds/grpc/xds_http_fault_filter.cc
@@ -214,6 +214,10 @@ XdsHttpFaultFilter::GenerateFilterConfigOverride(
   return GenerateFilterConfig(context, std::move(extension), errors);
 }
 
+void XdsHttpFaultFilter::AddFilter(InterceptionChainBuilder& builder) const {
+  builder.Add<FaultInjectionFilter>();
+}
+
 const grpc_channel_filter* XdsHttpFaultFilter::channel_filter() const {
   return &FaultInjectionFilter::kFilter;
 }

diff --git a/src/core/xds/grpc/xds_http_fault_filter.h b/src/core/xds/grpc/xds_http_fault_filter.h
@@ -44,6 +44,7 @@ class XdsHttpFaultFilter final : public XdsHttpFilterImpl {
   absl::optional<FilterConfig> GenerateFilterConfigOverride(
       const XdsResourceType::DecodeContext& context, XdsExtension extension,
       ValidationErrors* errors) const override;
+  void AddFilter(InterceptionChainBuilder& builder) const override;
   const grpc_channel_filter* channel_filter() const override;
   ChannelArgs ModifyChannelArgs(const ChannelArgs& args) const override;
   absl::StatusOr<ServiceConfigJsonEntry> GenerateServiceConfig(

diff --git a/src/core/xds/grpc/xds_http_filters.h b/src/core/xds/grpc/xds_http_filters.h
@@ -35,6 +35,7 @@
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/channel_fwd.h"
 #include "src/core/lib/gprpp/validation_errors.h"
+#include "src/core/lib/transport/interception_chain.h"
 #include "src/core/util/json/json.h"
 #include "src/core/util/json/json_writer.h"
 #include "src/core/xds/grpc/xds_common_types.h"
@@ -96,6 +97,8 @@ class XdsHttpFilterImpl {
       ValidationErrors* errors) const = 0;
 
   // C-core channel filter implementation.
+  virtual void AddFilter(InterceptionChainBuilder& builder) const = 0;
+  // TODO(roth): Remove this once the legacy filter stack goes away.
   virtual const grpc_channel_filter* channel_filter() const = 0;
 
   // Modifies channel args that may affect service config parsing (not
@@ -135,6 +138,7 @@ class XdsHttpRouterFilter final : public XdsHttpFilterImpl {
   absl::optional<FilterConfig> GenerateFilterConfigOverride(
       const XdsResourceType::DecodeContext& context, XdsExtension extension,
       ValidationErrors* errors) const override;
+  void AddFilter(InterceptionChainBuilder& /*builder*/) const override {}
   const grpc_channel_filter* channel_filter() const override { return nullptr; }
   absl::StatusOr<ServiceConfigJsonEntry> GenerateServiceConfig(
       const FilterConfig& /*hcm_filter_config*/,

diff --git a/src/core/xds/grpc/xds_http_rbac_filter.cc b/src/core/xds/grpc/xds_http_rbac_filter.cc
@@ -564,6 +564,10 @@ XdsHttpRbacFilter::GenerateFilterConfigOverride(
   return FilterConfig{OverrideConfigProtoName(), std::move(rbac_json)};
 }
 
+void XdsHttpRbacFilter::AddFilter(InterceptionChainBuilder& builder) const {
+  builder.Add<RbacFilter>();
+}
+
 const grpc_channel_filter* XdsHttpRbacFilter::channel_filter() const {
   return &RbacFilter::kFilterVtable;
 }