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

BUILD

@@ -509,6 +509,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",
@@ -590,6 +591,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",
@@ -1781,6 +1783,7 @@ grpc_cc_library(
         "grpc_service_config_impl",
         "grpc_trace",
         "grpcpp_call_metric_recorder",
+        "grpcpp_server_metric_recorder",
         "grpcpp_status",
         "iomgr_timer",
         "ref_counted_ptr",
@@ -1795,6 +1798,8 @@ grpc_cc_library(
         "//src/core:error",
         "//src/core:gpr_atm",
         "//src/core:gpr_manual_constructor",
+        "//src/core:grpc_backend_metric_data",
+        "//src/core:grpc_backend_metric_provider",
         "//src/core:grpc_service_config",
         "//src/core:grpc_transport_inproc",
         "//src/core:json",
@@ -1848,6 +1853,7 @@ grpc_cc_library(
         "grpc_trace",
         "grpc_unsecure",
         "grpcpp_call_metric_recorder",
+        "grpcpp_server_metric_recorder",
         "grpcpp_status",
         "iomgr_timer",
         "ref_counted_ptr",
@@ -1858,6 +1864,8 @@ grpc_cc_library(
         "//src/core:error",
         "//src/core:gpr_atm",
         "//src/core:gpr_manual_constructor",
+        "//src/core:grpc_backend_metric_data",
+        "//src/core:grpc_backend_metric_provider",
         "//src/core:grpc_insecure_credentials",
         "//src/core:grpc_service_config",
         "//src/core:grpc_transport_inproc",
@@ -1935,9 +1943,6 @@ grpc_cc_library(
 
 grpc_cc_library(
     name = "grpcpp_call_metric_recorder",
-    srcs = [
-        "src/cpp/server/orca/call_metric_recorder.cc",
-    ],
     external_deps = [
         "absl/strings",
         "absl/types:optional",
@@ -1957,23 +1962,26 @@ grpc_cc_library(
 )
 
 grpc_cc_library(
-    name = "grpcpp_orca_interceptor",
+    name = "grpcpp_server_metric_recorder",
     srcs = [
-        "src/cpp/server/orca/orca_interceptor.cc",
-    ],
-    hdrs = [
-        "src/cpp/server/orca/orca_interceptor.h",
-    ],
-    external_deps = [
-        "absl/strings",
-        "absl/types:optional",
+        "src/cpp/server/orca/server_metric_recorder.cc",
     ],
+#    external_deps = [
+#        "absl/strings",
+#        "absl/types:optional",
+#        "upb_lib",
+#    ],
     language = "c++",
+    public_hdrs = [
+        "include/grpcpp/ext/server_metric_recorder.h",
+    ],
     visibility = ["@grpc:public"],
     deps = [
-        "grpc++",
-        "grpc_base",
-        "grpcpp_call_metric_recorder",
+        "grpc_trace",
+#        "grpc++_public_hdrs",
+#        "xds_orca_upb",
+#        "//src/core:arena",
+        "//src/core:grpc_backend_metric_data",
     ],
 )
 

include/grpc/impl/grpc_types.h

@@ -141,6 +141,9 @@ 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_SERVER_CALL_METRIC_RECORDING \
+  "grpc.server_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/ext/call_metric_recorder.h

@@ -37,7 +37,6 @@ namespace grpc {
 class ServerBuilder;
 
 namespace experimental {
-class OrcaServerInterceptor;
 
 // Registers the per-rpc orca load reporter into the \a ServerBuilder.
 // Once this is done, the server will automatically send the load metrics
@@ -51,45 +50,43 @@ void EnableCallMetricRecording(ServerBuilder*);
 /// method to retrive the recorder for the current call.
 class CallMetricRecorder {
  public:
-  explicit CallMetricRecorder(grpc_core::Arena* arena);
-  ~CallMetricRecorder();
-
-  /// Records a call metric measurement for CPU utilization.
-  /// Multiple calls to this method will override the stored value.
-  CallMetricRecorder& RecordCpuUtilizationMetric(double value);
-
-  /// Records a call metric measurement for memory utilization.
-  /// Multiple calls to this method will override the stored value.
-  CallMetricRecorder& RecordMemoryUtilizationMetric(double value);
-
-  /// Records a call metric measurement for QPS.
-  /// Multiple calls to this method will override the stored value.
-  CallMetricRecorder& RecordQpsMetric(double value);
-
-  /// 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.
-  CallMetricRecorder& RecordUtilizationMetric(string_ref name, double value);
-
-  /// 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.
-  CallMetricRecorder& RecordRequestCostMetric(string_ref name, double value);
-
- private:
-  absl::optional<std::string> CreateSerializedReport();
-
-  internal::Mutex mu_;
-  grpc_core::BackendMetricData* backend_metric_data_ ABSL_GUARDED_BY(&mu_);
-  friend class experimental::OrcaServerInterceptor;
+  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;
 };
 
 }  // namespace experimental

include/grpcpp/ext/server_metric_recorder.h

@@ -0,0 +1,82 @@
+//
+//
+// Copyright 2023 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+//
+
+#ifndef GRPCPP_EXT_SERVER_METRIC_RECORDER_H
+#define GRPCPP_EXT_SERVER_METRIC_RECORDER_H
+
+#include <atomic>
+#include <memory>
+#include <string>
+
+#include "absl/strings/string_view.h"
+#include "absl/types/optional.h"
+
+namespace grpc_core {
+class Arena;
+struct BackendMetricData;
+}  // namespace grpc_core
+
+namespace grpc {
+class BackendMetricState;
+class ServerBuilder;
+
+namespace experimental {
+
+/// Records server wide metrics for the purpose of load balancing.
+/// Server implementation creates an instance and reports server metrics to it,
+/// and then passes it to experimental::EnableCallMetricRecording or
+/// experimental::OrcaService that read metrics to include in the report.
+class ServerMetricRecorder {
+ public:
+  // Records the server CPU utilization in the range [0, 1].
+  // Values outside of the valid range are rejected.
+  // Overrides the stored value when called again with a valid value.
+  void SetCpuUtilization(double value);
+  // Records the server memory utilization in the range [0, 1].
+  // Values outside of the valid range are rejected.
+  // Overrides the stored value when called again with a valid value.
+  void SetMemUtilization(double value);
+  // Records number of queries per second to the server in the range [0, infy).
+  // Values outside of the valid range are rejected.
+  // Overrides the stored value when called again with a valid value.
+  void SetQps(double value);
+
+  // Clears the server CPU utilization if recorded.
+  void ClearCpuUtilization();
+  // Clears the server memory utilization if recorded.
+  void ClearMemUtilization();
+  // Clears number of queries per second to the server if recorded.
+  void ClearQps();
+
+ private:
+  // Only populates fields in `data` that this has recorded metrics.
+  void GetMetrics(grpc_core::BackendMetricData* data);
+
+  // Defaults to -1.0 (unset).
+  std::atomic<double> cpu_utilization_{-1.0};
+  std::atomic<double> mem_utilization_{-1.0};
+  std::atomic<double> qps_{-1.0};
+
+  // To access GetMetrics().
+  friend class grpc::BackendMetricState;
+};
+
+}  // namespace experimental
+}  // namespace grpc
+
+#endif  // GRPCPP_EXT_SERVER_METRIC_RECORDER_H

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>>(),
+         experimental::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_;
+  }
+
+  experimental::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.
+  experimental::ServerMetricRecorder* server_metric_recorder_ = nullptr;
 };
 
 }  // namespace grpc

include/grpcpp/server_builder.h

@@ -283,6 +283,16 @@ class ServerBuilder {
         std::shared_ptr<experimental::AuthorizationPolicyProviderInterface>
             provider);
 
+    // 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.
+    void EnableCallMetricRecording(
+        experimental::ServerMetricRecorder* server_metric_recorder = nullptr);
+
    private:
     ServerBuilder* builder_;
   };
@@ -406,14 +416,12 @@ class ServerBuilder {
   std::vector<
       std::unique_ptr<grpc::experimental::ServerInterceptorFactoryInterface>>
       interceptor_creators_;
-  std::vector<
-      std::unique_ptr<grpc::experimental::ServerInterceptorFactoryInterface>>
-      internal_interceptor_creators_;
   std::vector<std::shared_ptr<grpc::internal::ExternalConnectionAcceptorImpl>>
       acceptors_;
   grpc_server_config_fetcher* server_config_fetcher_ = nullptr;
   std::shared_ptr<experimental::AuthorizationPolicyProviderInterface>
       authorization_provider_;
+  experimental::ServerMetricRecorder* server_metric_recorder_ = nullptr;
 };
 
 }  // namespace grpc