[call-v3] Move BackendMetricFilter to the v3 interface (#35416)

Closes #35416

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/35416 from ctiller:megawrt 9884323db9
PiperOrigin-RevId: 595151552

bazel/experiments.bzl

@@ -50,6 +50,7 @@ EXPERIMENT_ENABLES = {
     "tcp_rcv_lowat": "tcp_rcv_lowat",
     "trace_record_callops": "trace_record_callops",
     "unconstrained_max_quota_buffer_size": "unconstrained_max_quota_buffer_size",
+    "v3_backend_metric_filter": "v3_backend_metric_filter",
     "v3_channel_idle_filters": "v3_channel_idle_filters",
     "v3_compression_filter": "v3_compression_filter",
     "v3_server_auth_filter": "v3_server_auth_filter",

src/core/BUILD

@@ -5488,6 +5488,7 @@ grpc_cc_library(
         "channel_fwd",
         "channel_stack_type",
         "context",
+        "experiments",
         "grpc_backend_metric_data",
         "grpc_backend_metric_provider",
         "map",

src/core/ext/filters/backend_metrics/backend_metric_filter.cc

@@ -37,6 +37,7 @@
 #include "src/core/lib/channel/context.h"
 #include "src/core/lib/config/core_configuration.h"
 #include "src/core/lib/debug/trace.h"
+#include "src/core/lib/experiments/experiments.h"
 #include "src/core/lib/promise/context.h"
 #include "src/core/lib/promise/map.h"
 #include "src/core/lib/slice/slice.h"
@@ -47,8 +48,15 @@ namespace grpc_core {
 
 TraceFlag grpc_backend_metric_filter_trace(false, "backend_metric_filter");
 
-absl::optional<std::string> BackendMetricFilter::MaybeSerializeBackendMetrics(
-    BackendMetricProvider* provider) const {
+const NoInterceptor BackendMetricFilter::Call::OnClientInitialMetadata;
+const NoInterceptor BackendMetricFilter::Call::OnServerInitialMetadata;
+const NoInterceptor BackendMetricFilter::Call::OnClientToServerMessage;
+const NoInterceptor BackendMetricFilter::Call::OnServerToClientMessage;
+const NoInterceptor BackendMetricFilter::Call::OnFinalize;
+
+namespace {
+absl::optional<std::string> MaybeSerializeBackendMetrics(
+    BackendMetricProvider* provider) {
   if (provider == nullptr) return absl::nullopt;
   BackendMetricData data = provider->GetBackendMetricData();
   upb::Arena arena;
@@ -107,17 +115,27 @@ absl::optional<std::string> BackendMetricFilter::MaybeSerializeBackendMetrics(
       xds_data_orca_v3_OrcaLoadReport_serialize(response, arena.ptr(), &len);
   return std::string(buf, len);
 }
+}  // namespace
+
+const grpc_channel_filter LegacyBackendMetricFilter::kFilter =
+    MakePromiseBasedFilter<LegacyBackendMetricFilter, FilterEndpoint::kServer>(
+        "backend_metric");
 
 const grpc_channel_filter BackendMetricFilter::kFilter =
     MakePromiseBasedFilter<BackendMetricFilter, FilterEndpoint::kServer>(
         "backend_metric");
 
+absl::StatusOr<LegacyBackendMetricFilter> LegacyBackendMetricFilter::Create(
+    const ChannelArgs&, ChannelFilter::Args) {
+  return LegacyBackendMetricFilter();
+}
+
 absl::StatusOr<BackendMetricFilter> BackendMetricFilter::Create(
     const ChannelArgs&, ChannelFilter::Args) {
   return BackendMetricFilter();
 }
 
-ArenaPromise<ServerMetadataHandle> BackendMetricFilter::MakeCallPromise(
+ArenaPromise<ServerMetadataHandle> LegacyBackendMetricFilter::MakeCallPromise(
     CallArgs call_args, NextPromiseFactory next_promise_factory) {
   return ArenaPromise<ServerMetadataHandle>(Map(
       next_promise_factory(std::move(call_args)),
@@ -148,9 +166,35 @@ ArenaPromise<ServerMetadataHandle> BackendMetricFilter::MakeCallPromise(
       }));
 }
 
+void BackendMetricFilter::Call::OnServerTrailingMetadata(ServerMetadata& md) {
+  auto* ctx = &GetContext<
+      grpc_call_context_element>()[GRPC_CONTEXT_BACKEND_METRIC_PROVIDER];
+  if (ctx == nullptr) {
+    if (GRPC_TRACE_FLAG_ENABLED(grpc_backend_metric_filter_trace)) {
+      gpr_log(GPR_INFO, "[%p] No BackendMetricProvider.", this);
+    }
+    return;
+  }
+  absl::optional<std::string> serialized = MaybeSerializeBackendMetrics(
+      reinterpret_cast<BackendMetricProvider*>(ctx->value));
+  if (serialized.has_value() && !serialized->empty()) {
+    if (GRPC_TRACE_FLAG_ENABLED(grpc_backend_metric_filter_trace)) {
+      gpr_log(GPR_INFO, "[%p] Backend metrics serialized. size: %" PRIuPTR,
+              this, serialized->size());
+    }
+    md.Set(EndpointLoadMetricsBinMetadata(),
+           Slice::FromCopiedString(std::move(*serialized)));
+  } else if (GRPC_TRACE_FLAG_ENABLED(grpc_backend_metric_filter_trace)) {
+    gpr_log(GPR_INFO, "[%p] No backend metrics.", this);
+  }
+}
+
 void RegisterBackendMetricFilter(CoreConfiguration::Builder* builder) {
   builder->channel_init()
-      ->RegisterFilter(GRPC_SERVER_CHANNEL, &BackendMetricFilter::kFilter)
+      ->RegisterFilter(GRPC_SERVER_CHANNEL,
+                       IsV3BackendMetricFilterEnabled()
+                           ? &BackendMetricFilter::kFilter
+                           : &LegacyBackendMetricFilter::kFilter)
       .IfHasChannelArg(GRPC_ARG_SERVER_CALL_METRIC_RECORDING);
 }
 

src/core/ext/filters/backend_metrics/backend_metric_filter.h

@@ -31,20 +31,34 @@
 
 namespace grpc_core {
 
-class BackendMetricFilter : public ChannelFilter {
+class LegacyBackendMetricFilter : public ChannelFilter {
  public:
   static const grpc_channel_filter kFilter;
 
-  static absl::StatusOr<BackendMetricFilter> Create(const ChannelArgs& args,
-                                                    ChannelFilter::Args);
+  static absl::StatusOr<LegacyBackendMetricFilter> Create(
+      const ChannelArgs& args, ChannelFilter::Args);
 
   // Construct a promise for one call.
   ArenaPromise<ServerMetadataHandle> MakeCallPromise(
       CallArgs call_args, NextPromiseFactory next_promise_factory) override;
+};
+
+class BackendMetricFilter : public ImplementChannelFilter<BackendMetricFilter> {
+ public:
+  static const grpc_channel_filter kFilter;
+
+  static absl::StatusOr<BackendMetricFilter> Create(const ChannelArgs& args,
+                                                    ChannelFilter::Args);
 
- private:
-  absl::optional<std::string> MaybeSerializeBackendMetrics(
-      BackendMetricProvider* provider) const;
+  class Call {
+   public:
+    static const NoInterceptor OnClientInitialMetadata;
+    static const NoInterceptor OnServerInitialMetadata;
+    void OnServerTrailingMetadata(ServerMetadata& md);
+    static const NoInterceptor OnClientToServerMessage;
+    static const NoInterceptor OnServerToClientMessage;
+    static const NoInterceptor OnFinalize;
+  };
 };
 
 }  // namespace grpc_core

src/core/lib/experiments/experiments.cc

@@ -152,6 +152,9 @@ const char* const description_unconstrained_max_quota_buffer_size =
     "Discard the cap on the max free pool size for one memory allocator";
 const char* const additional_constraints_unconstrained_max_quota_buffer_size =
     "{}";
+const char* const description_v3_backend_metric_filter =
+    "Use the backend metric filter utilizing the v3 filter api";
+const char* const additional_constraints_v3_backend_metric_filter = "{}";
 const char* const description_v3_channel_idle_filters =
     "Use the v3 filter API version of the idle filters.";
 const char* const additional_constraints_v3_channel_idle_filters = "{}";
@@ -273,6 +276,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_unconstrained_max_quota_buffer_size,
      additional_constraints_unconstrained_max_quota_buffer_size, nullptr, 0,
      false, true},
+    {"v3_backend_metric_filter", description_v3_backend_metric_filter,
+     additional_constraints_v3_backend_metric_filter, nullptr, 0, false, true},
     {"v3_channel_idle_filters", description_v3_channel_idle_filters,
      additional_constraints_v3_channel_idle_filters, nullptr, 0, false, true},
     {"v3_compression_filter", description_v3_compression_filter,
@@ -426,6 +431,9 @@ const char* const description_unconstrained_max_quota_buffer_size =
     "Discard the cap on the max free pool size for one memory allocator";
 const char* const additional_constraints_unconstrained_max_quota_buffer_size =
     "{}";
+const char* const description_v3_backend_metric_filter =
+    "Use the backend metric filter utilizing the v3 filter api";
+const char* const additional_constraints_v3_backend_metric_filter = "{}";
 const char* const description_v3_channel_idle_filters =
     "Use the v3 filter API version of the idle filters.";
 const char* const additional_constraints_v3_channel_idle_filters = "{}";
@@ -547,6 +555,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_unconstrained_max_quota_buffer_size,
      additional_constraints_unconstrained_max_quota_buffer_size, nullptr, 0,
      false, true},
+    {"v3_backend_metric_filter", description_v3_backend_metric_filter,
+     additional_constraints_v3_backend_metric_filter, nullptr, 0, false, true},
     {"v3_channel_idle_filters", description_v3_channel_idle_filters,
      additional_constraints_v3_channel_idle_filters, nullptr, 0, false, true},
     {"v3_compression_filter", description_v3_compression_filter,
@@ -700,6 +710,9 @@ const char* const description_unconstrained_max_quota_buffer_size =
     "Discard the cap on the max free pool size for one memory allocator";
 const char* const additional_constraints_unconstrained_max_quota_buffer_size =
     "{}";
+const char* const description_v3_backend_metric_filter =
+    "Use the backend metric filter utilizing the v3 filter api";
+const char* const additional_constraints_v3_backend_metric_filter = "{}";
 const char* const description_v3_channel_idle_filters =
     "Use the v3 filter API version of the idle filters.";
 const char* const additional_constraints_v3_channel_idle_filters = "{}";
@@ -821,6 +834,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      description_unconstrained_max_quota_buffer_size,
      additional_constraints_unconstrained_max_quota_buffer_size, nullptr, 0,
      false, true},
+    {"v3_backend_metric_filter", description_v3_backend_metric_filter,
+     additional_constraints_v3_backend_metric_filter, nullptr, 0, false, true},
     {"v3_channel_idle_filters", description_v3_channel_idle_filters,
      additional_constraints_v3_channel_idle_filters, nullptr, 0, false, true},
     {"v3_compression_filter", description_v3_compression_filter,

src/core/lib/experiments/experiments.h

@@ -107,6 +107,7 @@ inline bool IsTcpFrameSizeTuningEnabled() { return false; }
 inline bool IsTcpRcvLowatEnabled() { return false; }
 inline bool IsTraceRecordCallopsEnabled() { return false; }
 inline bool IsUnconstrainedMaxQuotaBufferSizeEnabled() { return false; }
+inline bool IsV3BackendMetricFilterEnabled() { return false; }
 inline bool IsV3ChannelIdleFiltersEnabled() { return false; }
 inline bool IsV3CompressionFilterEnabled() { return false; }
 inline bool IsV3ServerAuthFilterEnabled() { return false; }
@@ -172,6 +173,7 @@ inline bool IsTcpFrameSizeTuningEnabled() { return false; }
 inline bool IsTcpRcvLowatEnabled() { return false; }
 inline bool IsTraceRecordCallopsEnabled() { return false; }
 inline bool IsUnconstrainedMaxQuotaBufferSizeEnabled() { return false; }
+inline bool IsV3BackendMetricFilterEnabled() { return false; }
 inline bool IsV3ChannelIdleFiltersEnabled() { return false; }
 inline bool IsV3CompressionFilterEnabled() { return false; }
 inline bool IsV3ServerAuthFilterEnabled() { return false; }
@@ -237,6 +239,7 @@ inline bool IsTcpFrameSizeTuningEnabled() { return false; }
 inline bool IsTcpRcvLowatEnabled() { return false; }
 inline bool IsTraceRecordCallopsEnabled() { return false; }
 inline bool IsUnconstrainedMaxQuotaBufferSizeEnabled() { return false; }
+inline bool IsV3BackendMetricFilterEnabled() { return false; }
 inline bool IsV3ChannelIdleFiltersEnabled() { return false; }
 inline bool IsV3CompressionFilterEnabled() { return false; }
 inline bool IsV3ServerAuthFilterEnabled() { return false; }
@@ -286,6 +289,7 @@ enum ExperimentIds {
   kExperimentIdTcpRcvLowat,
   kExperimentIdTraceRecordCallops,
   kExperimentIdUnconstrainedMaxQuotaBufferSize,
+  kExperimentIdV3BackendMetricFilter,
   kExperimentIdV3ChannelIdleFilters,
   kExperimentIdV3CompressionFilter,
   kExperimentIdV3ServerAuthFilter,
@@ -428,6 +432,10 @@ inline bool IsTraceRecordCallopsEnabled() {
 inline bool IsUnconstrainedMaxQuotaBufferSizeEnabled() {
   return IsExperimentEnabled(kExperimentIdUnconstrainedMaxQuotaBufferSize);
 }
+#define GRPC_EXPERIMENT_IS_INCLUDED_V3_BACKEND_METRIC_FILTER
+inline bool IsV3BackendMetricFilterEnabled() {
+  return IsExperimentEnabled(kExperimentIdV3BackendMetricFilter);
+}
 #define GRPC_EXPERIMENT_IS_INCLUDED_V3_CHANNEL_IDLE_FILTERS
 inline bool IsV3ChannelIdleFiltersEnabled() {
   return IsExperimentEnabled(kExperimentIdV3ChannelIdleFilters);

src/core/lib/experiments/experiments.yaml

@@ -253,6 +253,12 @@
   expiry: 2024/02/01
   owner: ctiller@google.com
   test_tags: [resource_quota_test]
+- name: v3_backend_metric_filter
+  description:
+    Use the backend metric filter utilizing the v3 filter api
+  expiry: 2024/05/05
+  owner: ctiller@google.com
+  test_tags: []
 - name: v3_channel_idle_filters
   description:
     Use the v3 filter API version of the idle filters.