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

BUILD

@@ -1783,7 +1783,7 @@ grpc_cc_library(
         "grpc_service_config_impl",
         "grpc_trace",
         "grpcpp_call_metric_recorder",
-        "grpcpp_server_metric_recorder",
+        "grpcpp_backend_metric_recorder",
         "grpcpp_status",
         "iomgr_timer",
         "ref_counted_ptr",
@@ -1853,7 +1853,7 @@ grpc_cc_library(
         "grpc_trace",
         "grpc_unsecure",
         "grpcpp_call_metric_recorder",
-        "grpcpp_server_metric_recorder",
+        "grpcpp_backend_metric_recorder",
         "grpcpp_status",
         "iomgr_timer",
         "ref_counted_ptr",
@@ -1962,9 +1962,12 @@ grpc_cc_library(
 )
 
 grpc_cc_library(
-    name = "grpcpp_server_metric_recorder",
+    name = "grpcpp_backend_metric_recorder",
     srcs = [
-        "src/cpp/server/orca/server_metric_recorder.cc",
+        "src/cpp/server/orca/backend_metric_recorder.cc",
+    ],
+    hdrs = [
+        "src/cpp/server/backend_metric_recorder.h"
     ],
     language = "c++",
     public_hdrs = [
@@ -1973,8 +1976,11 @@ grpc_cc_library(
     visibility = ["@grpc:public"],
     deps = [
         "gpr",
+        "grpcpp_call_metric_recorder",
+        "grpc++_public_hdrs",
         "grpc_trace",
         "//src/core:grpc_backend_metric_data",
+        "//src/core:grpc_backend_metric_provider",
     ],
 )
 
@@ -2000,7 +2006,7 @@ grpc_cc_library(
         "gpr",
         "grpc++",
         "grpc_base",
-        "grpcpp_server_metric_recorder",
+        "grpcpp_backend_metric_recorder",
         "protobuf_duration_upb",
         "ref_counted_ptr",
         "xds_orca_service_upb",

include/grpcpp/ext/orca_service.h

@@ -37,6 +37,8 @@ namespace experimental {
 // utilization metrics to clients.
 class OrcaService : public Service {
  public:
+  OrcaService() = delete;
+
   struct Options {
     // Minimum report interval.  If a client requests an interval lower
     // than this value, this value will be used instead.
@@ -49,16 +51,22 @@ class OrcaService : public Service {
     }
   };
 
-  OrcaService(const ServerMetricRecorder& server_metric_recorder,
+  // ServerMetricRecorder is required.
+  OrcaService(ServerMetricRecorder* const server_metric_recorder,
               Options options);
 
  private:
   class Reactor;
 
-  Slice CreateSerializedResponse();
+  Slice GetOrCreateSerializedResponse();
 
-  const ServerMetricRecorder& server_metric_recorder_;
+  const ServerMetricRecorder* const server_metric_recorder_;
   const absl::Duration min_report_duration_;
+  grpc::internal::Mutex mu_;
+  // Contains the last serialized metrics from server_metric_recorder_.
+  absl::optional<Slice> response_slice_ ABSL_GUARDED_BY(&mu_);
+  // The update sequence number of metrics serialized in response_slice_.
+  uint64_t response_slice_seq_ ABSL_GUARDED_BY(&mu_);
 };
 
 }  // namespace experimental

include/grpcpp/ext/server_metric_recorder.h

@@ -20,12 +20,15 @@
 #define GRPCPP_EXT_SERVER_METRIC_RECORDER_H
 
 #include <atomic>
+#include <map>
 #include <memory>
 #include <string>
 
 #include "absl/strings/string_view.h"
 #include "absl/types/optional.h"
 
+#include <grpcpp/impl/sync.h>
+
 namespace grpc_core {
 struct BackendMetricData;
 }  // namespace grpc_core
@@ -54,26 +57,42 @@ class ServerMetricRecorder {
   /// Values outside of the valid range are rejected.
   /// Overrides the stored value when called again with a valid value.
   void SetQps(double value);
+  /// Records a named resource utilization value in the range [0, 1].
+  /// Values outside of the valid range are rejected. Consumes `name`.
+  /// Overrides the stored value when called again with the same name.
+  void SetNamedUtilization(std::string name, double value);
+  /// Replaces all named resource utilization values. Consumes
+  /// `named_utilization`. No range validation.
+  void SetAllNamedUtilization(std::map<std::string, double> named_utilization);
 
   /// Clears the server CPU utilization if recorded.
   void ClearCpuUtilization();
   /// Clears the server memory utilization if recorded.
   void ClearMemoryUtilization();
   /// Clears number of queries per second to the server if recorded.
   void ClearQps();
+  /// Clears a named utilization value if exists.
+  void ClearNamedUtilization(absl::string_view name);
 
  private:
   // To access GetMetrics().
   friend class grpc::BackendMetricState;
   friend class OrcaService;
 
-  // Only populates fields in `data` that this has recorded metrics.
-  grpc_core::BackendMetricData GetMetrics() const;
+  // Returns a pair of backend metrics recorded and the update sequence number.
+  // Returned metrics are guaranteed to be identical between two calls if the
+  // update sequences are identical.
+  std::pair<grpc_core::BackendMetricData, uint64_t> GetMetrics() const;
+
+  // Incremented every time updated.
+  std::atomic<uint64_t> update_seq_{0};
 
   // 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};
+  mutable grpc::internal::Mutex mu_;
+  std::map<std::string, double> named_utilization_ ABSL_GUARDED_BY(mu_);
 };
 
 }  // namespace experimental

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

@@ -92,17 +92,16 @@ ArenaPromise<ServerMetadataHandle> BackendMetricFilter::MakeCallPromise(
       }));
 }
 
-static bool maybe_prepend_grpc_backend_metric_filter(
-    grpc_core::ChannelStackBuilder* builder) {
-  if (builder->channel_args().Contains(GRPC_ARG_SERVER_CALL_METRIC_RECORDING)) {
-    builder->PrependFilter(&grpc_core::BackendMetricFilter::kFilter);
-  }
-  return true;
-}
-
 void RegisterBackendMetricFilter(CoreConfiguration::Builder* builder) {
   builder->channel_init()->RegisterStage(
-      GRPC_SERVER_CHANNEL, INT_MAX, maybe_prepend_grpc_backend_metric_filter);
+      GRPC_SERVER_CHANNEL, INT_MAX,
+      [](grpc_core::ChannelStackBuilder* builder) {
+        if (builder->channel_args().Contains(
+                GRPC_ARG_SERVER_CALL_METRIC_RECORDING)) {
+          builder->PrependFilter(&grpc_core::BackendMetricFilter::kFilter);
+        }
+        return true;
+      });
 }
 
 }  // namespace grpc_core

src/cpp/server/backend_metric_recorder.h

@@ -0,0 +1,46 @@
+#ifndef LOCAL_GOOGLE_HOME_YSSEUNG_WORK_GRPC_GIT_GRPC_SRC_CPP_SERVER_BACKEND_METRIC_RECORDER_H_
+#define LOCAL_GOOGLE_HOME_YSSEUNG_WORK_GRPC_GIT_GRPC_SRC_CPP_SERVER_BACKEND_METRIC_RECORDER_H_
+
+#include <map>
+
+#include <grpcpp/ext/call_metric_recorder.h>
+
+#include "src/core/ext/filters/backend_metrics/backend_metric_provider.h"
+
+namespace grpc {
+
+class BackendMetricState : public grpc_core::BackendMetricProvider,
+                           public experimental::CallMetricRecorder {
+ public:
+  // `server_metric_recorder` is optional. When set, GetBackendMetricData()
+  // merges metrics from `server_metric_recorder` with metrics recorded to this.
+  explicit BackendMetricState(
+      experimental::ServerMetricRecorder* server_metric_recorder)
+      : server_metric_recorder_(server_metric_recorder) {}
+  ~BackendMetricState() override = default;
+  experimental::CallMetricRecorder& RecordCpuUtilizationMetric(
+      double value) override;
+  experimental::CallMetricRecorder& RecordMemoryUtilizationMetric(
+      double value) override;
+  experimental::CallMetricRecorder& RecordQpsMetric(double value) override;
+  experimental::CallMetricRecorder& RecordUtilizationMetric(
+      string_ref name, double value) override;
+  experimental::CallMetricRecorder& RecordRequestCostMetric(
+      string_ref name, double value) override;
+
+ private:
+  // This clears metrics currently recorded. Don't call twice.
+  grpc_core::BackendMetricData GetBackendMetricData() override;
+
+  experimental::ServerMetricRecorder* server_metric_recorder_;
+  std::atomic<double> cpu_utilization_{-1.0};
+  std::atomic<double> mem_utilization_{-1.0};
+  std::atomic<double> qps_{-1.0};
+  internal::Mutex mu_;
+  std::map<absl::string_view, double> utilization_ ABSL_GUARDED_BY(mu_);
+  std::map<absl::string_view, double> request_cost_ ABSL_GUARDED_BY(mu_);
+};
+
+}  // namespace grpc
+
+#endif  // LOCAL_GOOGLE_HOME_YSSEUNG_WORK_GRPC_GIT_GRPC_SRC_CPP_SERVER_BACKEND_METRIC_RECORDER_H_