server: introduce ServerMetricRecorder API and move per-call reporting from a C++ interceptor to a C-core filter

BUILD

@@ -504,6 +504,7 @@ grpc_cc_library(
         "grpc_trace",
         "http_connect_handshaker",
         "iomgr_timer",
+        "//src/core:backend_metric_filter",
         "//src/core:channel_args",
         "//src/core:channel_init",
         "//src/core:channel_stack_type",
@@ -585,6 +586,7 @@ grpc_cc_library(
         "sockaddr_utils",
         "tsi_base",
         "uri_parser",
+        "//src/core:backend_metric_filter",
         "//src/core:channel_args",
         "//src/core:channel_init",
         "//src/core:channel_stack_type",
@@ -1775,7 +1777,6 @@ grpc_cc_library(
         "grpc_security_base",
         "grpc_service_config_impl",
         "grpc_trace",
-        "grpcpp_call_metric_recorder",
         "grpcpp_status",
         "iomgr_timer",
         "ref_counted_ptr",
@@ -1790,6 +1791,7 @@ grpc_cc_library(
         "//src/core:error",
         "//src/core:gpr_atm",
         "//src/core:gpr_manual_constructor",
+        "//src/core:grpc_backend_metric_data",
         "//src/core:grpc_service_config",
         "//src/core:grpc_transport_inproc",
         "//src/core:json",
@@ -1842,7 +1844,6 @@ grpc_cc_library(
         "grpc_service_config_impl",
         "grpc_trace",
         "grpc_unsecure",
-        "grpcpp_call_metric_recorder",
         "grpcpp_status",
         "iomgr_timer",
         "ref_counted_ptr",
@@ -1853,6 +1854,7 @@ grpc_cc_library(
         "//src/core:error",
         "//src/core:gpr_atm",
         "//src/core:gpr_manual_constructor",
+        "//src/core:grpc_backend_metric_data",
         "//src/core:grpc_insecure_credentials",
         "//src/core:grpc_service_config",
         "//src/core:grpc_transport_inproc",
@@ -1928,50 +1930,6 @@ grpc_cc_library(
     alwayslink = 1,
 )
 
-grpc_cc_library(
-    name = "grpcpp_call_metric_recorder",
-    srcs = [
-        "src/cpp/server/orca/call_metric_recorder.cc",
-    ],
-    external_deps = [
-        "absl/strings",
-        "absl/types:optional",
-        "upb_lib",
-    ],
-    language = "c++",
-    public_hdrs = [
-        "include/grpcpp/ext/call_metric_recorder.h",
-    ],
-    visibility = ["@grpc:public"],
-    deps = [
-        "grpc++_public_hdrs",
-        "xds_orca_upb",
-        "//src/core:arena",
-        "//src/core:grpc_backend_metric_data",
-    ],
-)
-
-grpc_cc_library(
-    name = "grpcpp_orca_interceptor",
-    srcs = [
-        "src/cpp/server/orca/orca_interceptor.cc",
-    ],
-    hdrs = [
-        "src/cpp/server/orca/orca_interceptor.h",
-    ],
-    external_deps = [
-        "absl/strings",
-        "absl/types:optional",
-    ],
-    language = "c++",
-    visibility = ["@grpc:public"],
-    deps = [
-        "grpc++",
-        "grpc_base",
-        "grpcpp_call_metric_recorder",
-    ],
-)
-
 grpc_cc_library(
     name = "grpcpp_orca_service",
     srcs = [

include/grpc/impl/grpc_types.h

@@ -141,6 +141,8 @@ typedef struct {
 #define GRPC_ARG_ENABLE_CENSUS "grpc.census"
 /** If non-zero, enable load reporting. */
 #define GRPC_ARG_ENABLE_LOAD_REPORTING "grpc.loadreporting"
+/** If non-zero, call metric recording is enabled. */
+#define GRPC_ARG_CALL_METRIC_RECORDING "grpc.call_metric_recording"
 /** Request that optional features default to off (regardless of what they
     usually default to) - to enable tight control over what gets enabled */
 #define GRPC_ARG_MINIMAL_STACK "grpc.minimal_stack"

include/grpcpp/server.h

@@ -183,7 +183,8 @@ class Server : public ServerInterface, private internal::GrpcLibrary {
          std::vector<
              std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
              interceptor_creators = std::vector<std::unique_ptr<
-                 experimental::ServerInterceptorFactoryInterface>>());
+                 experimental::ServerInterceptorFactoryInterface>>(),
+         grpc_core::ServerMetricRecorder* server_metric_recorder = nullptr);
 
   /// Start the server.
   ///
@@ -255,6 +256,14 @@ class Server : public ServerInterface, private internal::GrpcLibrary {
     return max_receive_message_size_;
   }
 
+  bool call_metric_recording_enabled() const override {
+    return call_metric_recording_enabled_;
+  }
+
+  grpc_core::ServerMetricRecorder* server_metric_recorder() const override {
+    return server_metric_recorder_;
+  }
+
   CompletionQueue* CallbackCQ() ABSL_LOCKS_EXCLUDED(mu_) override;
 
   ServerInitializer* initializer();
@@ -338,6 +347,12 @@ class Server : public ServerInterface, private internal::GrpcLibrary {
   // Shutdown.  Even though this is only used with NDEBUG, instantiate it in all
   // cases since otherwise the size will be inconsistent.
   std::vector<CompletionQueue*> cq_list_;
+
+  // Whetner per-call load reporting is enabled.
+  bool call_metric_recording_enabled_ = false;
+
+  // Interface to read or update server-wide metrics. Optional.
+  grpc_core::ServerMetricRecorder* server_metric_recorder_ = nullptr;
 };
 
 }  // namespace grpc

include/grpcpp/server_builder.h

@@ -251,6 +251,16 @@ class ServerBuilder {
   /// doc/workarounds.md.
   ServerBuilder& EnableWorkaround(grpc_workaround_list id);
 
+  // Enables per-call load reporting. The server will automatically send the
+  // load metrics after each RPC. The caller can report load metrics for the
+  // current call to what \a ServerContext::GetCallMetricRecorder() returns.
+  // The server merges metrics from the optional \a server_metric_recorder when
+  // provided where the call metric recorder take a higher precedence.
+  // The caller owns and must ensure the server metric recorder outlives the
+  // server.
+  ServerBuilder& EnableCallMetricRecording(
+      grpc_core::ServerMetricRecorder* server_metric_recorder = nullptr);
+
   /// NOTE: class experimental_type is not part of the public API of this class.
   /// TODO(yashykt): Integrate into public API when this is no longer
   /// experimental.
@@ -414,6 +424,7 @@ class ServerBuilder {
   grpc_server_config_fetcher* server_config_fetcher_ = nullptr;
   std::shared_ptr<experimental::AuthorizationPolicyProviderInterface>
       authorization_provider_;
+  grpc_core::ServerMetricRecorder* server_metric_recorder_ = nullptr;
 };
 
 }  // namespace grpc

include/grpcpp/server_context.h

@@ -51,6 +51,11 @@ struct grpc_metadata;
 struct grpc_call;
 struct census_context;
 
+namespace grpc_core {
+struct BackendMetricData;
+class ServerMetricRecorder;
+}  // namespace grpc_core
+
 namespace grpc {
 template <class W, class R>
 class ServerAsyncReader;
@@ -115,10 +120,46 @@ class ServerContextTestSpouse;
 class DefaultReactorTestPeer;
 }  // namespace testing
 
-namespace experimental {
-class OrcaServerInterceptor;
-class CallMetricRecorder;
-}  // namespace experimental
+class CallMetricRecorder {
+ public:
+  virtual ~CallMetricRecorder() = default;
+
+  // Records a call metric measurement for CPU utilization.
+  // Multiple calls to this method will override the stored value.
+  // Values outside of the valid range [0, 1] are ignored.
+  virtual CallMetricRecorder& RecordCpuUtilizationMetric(double value) = 0;
+
+  // Records a call metric measurement for memory utilization.
+  // Multiple calls to this method will override the stored value.
+  // Values outside of the valid range [0, 1] are ignored.
+  virtual CallMetricRecorder& RecordMemoryUtilizationMetric(double value) = 0;
+
+  // Records a call metric measurement for queries per second.
+  // Multiple calls to this method will override the stored value.
+  // Values outside of the valid range [0, infy) are ignored.
+  virtual CallMetricRecorder& RecordQpsMetric(double value) = 0;
+
+  // Records a call metric measurement for utilization.
+  // Multiple calls to this method with the same name will
+  // override the corresponding stored value. The lifetime of the
+  // name string needs to be longer than the lifetime of the RPC
+  // itself, since it's going to be sent as trailers after the RPC
+  // finishes. It is assumed the strings are common names that
+  // are global constants.
+  // Values outside of the valid range [0, 1] are ignored.
+  virtual CallMetricRecorder& RecordUtilizationMetric(string_ref name,
+                                                      double value) = 0;
+
+  // Records a call metric measurement for request cost.
+  // Multiple calls to this method with the same name will
+  // override the corresponding stored value. The lifetime of the
+  // name string needs to be longer than the lifetime of the RPC
+  // itself, since it's going to be sent as trailers after the RPC
+  // finishes. It is assumed the strings are common names that
+  // are global constants.
+  virtual CallMetricRecorder& RecordRequestCostMetric(string_ref name,
+                                                      double value) = 0;
+};
 
 /// Base class of ServerContext.
 class ServerContextBase {
@@ -291,9 +332,7 @@ class ServerContextBase {
   /// client in order to make load balancing decisions. This will
   /// return nullptr if the feature hasn't been enabled using
   /// \a EnableCallMetricRecording.
-  experimental::CallMetricRecorder* ExperimentalGetCallMetricRecorder() {
-    return call_metric_recorder_;
-  }
+  CallMetricRecorder* GetCallMetricRecorder() { return call_metric_recorder_; }
 
   /// EXPERIMENTAL API
   /// Returns the call's authority.
@@ -404,7 +443,6 @@ class ServerContextBase {
   friend class grpc::ClientContext;
   friend class grpc::GenericServerContext;
   friend class grpc::GenericCallbackServerContext;
-  friend class grpc::experimental::OrcaServerInterceptor;
 
   /// Prevent copying.
   ServerContextBase(const ServerContextBase&);
@@ -445,7 +483,14 @@ class ServerContextBase {
     }
   }
 
-  void CreateCallMetricRecorder();
+  // Arena allocates a new BackendMetricState for the current call.
+  // BackendMetricState serves as both the CallMetricRecorder and
+  // BackendMetricProvider interfaces. Sets the provided ServerMetricRecorder to
+  // the BackendMetricProvider when not null.
+  // This should be called only once and only when call metric recording is
+  // enabled.
+  void CreateCallMetricRecorder(
+      grpc_core::ServerMetricRecorder* server_metric_recorder = nullptr);
 
   struct CallWrapper {
     ~CallWrapper();
@@ -484,7 +529,7 @@ class ServerContextBase {
   grpc::experimental::ServerRpcInfo* rpc_info_ = nullptr;
   RpcAllocatorState* message_allocator_state_ = nullptr;
   ContextAllocator* context_allocator_ = nullptr;
-  experimental::CallMetricRecorder* call_metric_recorder_ = nullptr;
+  CallMetricRecorder* call_metric_recorder_ = nullptr;
 
   class Reactor : public grpc::ServerUnaryReactor {
    public:

include/grpcpp/server_interface.h

@@ -350,6 +350,12 @@ class ServerInterface : public internal::CallHook {
     return nullptr;
   }
 
+  // Whether per-call load reporting is enabled.
+  virtual bool call_metric_recording_enabled() const = 0;
+
+  // Interface to read or update server-wide metrics. Returns null when not set.
+  virtual grpc_core::ServerMetricRecorder* server_metric_recorder() const = 0;
+
   // A method to get the callbackable completion queue associated with this
   // server. If the return value is nullptr, this server doesn't support
   // callback operations.

src/core/BUILD

@@ -3589,12 +3589,19 @@ grpc_cc_library(
 
 grpc_cc_library(
     name = "grpc_backend_metric_data",
+    srcs = [
+        "ext/filters/load_reporting/backend_metric_data.cc",
+    ],
     hdrs = [
-        "ext/filters/client_channel/lb_policy/backend_metric_data.h",
+        "ext/filters/load_reporting/backend_metric_data.h",
     ],
     external_deps = ["absl/strings"],
     language = "c++",
-    deps = ["//:gpr_platform"],
+    deps = [
+        "//:gpr",
+        "//:gpr_platform",
+        "//:grpc_trace",
+    ],
 )
 
 grpc_cc_library(
@@ -4577,6 +4584,36 @@ grpc_cc_library(
     alwayslink = 1,
 )
 
+grpc_cc_library(
+    name = "backend_metric_filter",
+    srcs = [
+        "ext/filters/load_reporting/backend_metric_filter.cc",
+    ],
+    hdrs = [
+        "ext/filters/load_reporting/backend_metric_filter.h",
+    ],
+    external_deps = [
+        "absl/status:statusor",
+        "absl/types:optional",
+        "upb_lib",
+    ],
+    language = "c++",
+    deps = [
+        "arena",
+        "arena_promise",
+        "channel_args",
+        "grpc_backend_metric_data",
+        "map",
+        "xds_orca_service_upb",
+        "xds_orca_upb",
+        "//:gpr_platform",
+        "//:grpc_base",
+        "//:grpc_trace",
+        "//:promise",
+    ],
+    alwayslink = 1,
+)
+
 grpc_cc_library(
     name = "polling_resolver",
     srcs = [

src/core/ext/filters/client_channel/backend_metric.h

@@ -23,7 +23,7 @@
 
 #include "absl/strings/string_view.h"
 
-#include "src/core/ext/filters/client_channel/lb_policy/backend_metric_data.h"
+#include "src/core/ext/filters/load_reporting/backend_metric_data.h"
 
 namespace grpc_core {
 

src/core/ext/filters/client_channel/client_channel.h

@@ -41,9 +41,9 @@
 #include "src/core/ext/filters/client_channel/config_selector.h"
 #include "src/core/ext/filters/client_channel/dynamic_filters.h"
 #include "src/core/ext/filters/client_channel/lb_call_state_internal.h"
-#include "src/core/ext/filters/client_channel/lb_policy/backend_metric_data.h"
 #include "src/core/ext/filters/client_channel/subchannel.h"
 #include "src/core/ext/filters/client_channel/subchannel_pool_interface.h"
+#include "src/core/ext/filters/load_reporting/backend_metric_data.h"
 #include "src/core/lib/channel/call_tracer.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/channel/channel_fwd.h"

src/core/ext/filters/client_channel/lb_policy/oob_backend_metric.h

@@ -21,7 +21,7 @@
 
 #include <memory>
 
-#include "src/core/ext/filters/client_channel/lb_policy/backend_metric_data.h"
+#include "src/core/ext/filters/load_reporting/backend_metric_data.h"
 #include "src/core/lib/gprpp/time.h"
 #include "src/core/lib/load_balancing/subchannel_interface.h"