orca: Add ORCA Load Report processing to Router::Filter. (envoyprox…

…y#35642)

Signed-off-by: Misha Efimov <[email protected]>

api/envoy/config/cluster/v3/cluster.proto

@@ -1154,6 +1154,8 @@ message Cluster {
   // [#not-implemented-hide:]
   // A list of metric names from ORCA load reports to propagate to LRS.
   //
+  // If not specified, then ORCA load reports will not be propagated to LRS.
+  //
   // For map fields in the ORCA proto, the string will be of the form ``<map_field_name>.<map_key>``.
   // For example, the string ``named_metrics.foo`` will mean to look for the key ``foo`` in the ORCA
   // ``named_metrics`` field.

envoy/upstream/upstream.h

@@ -1235,6 +1235,11 @@ class ClusterInfo : public Http::FilterChainFactory {
   virtual OptRef<const envoy::config::cluster::v3::UpstreamConnectionOptions::HappyEyeballsConfig>
   happyEyeballsConfig() const PURE;
 
+  /**
+   * @return Reference to the optional config for LRS endpoint metric reporting.
+   */
+  virtual OptRef<const std::vector<std::string>> lrsReportMetricNames() const PURE;
+
 protected:
   /**
    * Invoked by extensionProtocolOptionsTyped.

source/common/orca/BUILD

@@ -23,3 +23,23 @@ envoy_cc_library(
         "@com_github_cncf_xds//xds/data/orca/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_library(
+    name = "orca_load_metrics_lib",
+    srcs = ["orca_load_metrics.cc"],
+    hdrs = ["orca_load_metrics.h"],
+    external_deps = [
+        "abseil_flat_hash_set",
+        "abseil_status",
+        "abseil_strings",
+        "abseil_statusor",
+        "fmtlib",
+    ],
+    deps = [
+        ":orca_parser",
+        "//envoy/http:header_map_interface",
+        "//source/common/http:header_utility_lib",
+        "//source/common/protobuf:utility_lib_header",
+        "@com_github_cncf_xds//xds/data/orca/v3:pkg_cc_proto",
+    ],
+)

source/common/orca/orca_load_metrics.cc

@@ -0,0 +1,74 @@
+#include "source/common/orca/orca_load_metrics.h"
+
+#include <string>
+
+#include "source/common/orca/orca_parser.h"
+
+#include "absl/strings/match.h"
+#include "absl/strings/str_cat.h"
+#include "absl/strings/string_view.h"
+
+namespace Envoy {
+namespace Orca {
+namespace {
+// The following fields are the names of the metrics tracked in the ORCA load
+// report proto.
+static constexpr absl::string_view kApplicationUtilizationField = "application_utilization";
+static constexpr absl::string_view kCpuUtilizationField = "cpu_utilization";
+static constexpr absl::string_view kMemUtilizationField = "mem_utilization";
+static constexpr absl::string_view kEpsField = "eps";
+static constexpr absl::string_view kRpsFractionalField = "rps_fractional";
+static constexpr absl::string_view kNamedMetricsFieldPrefix = "named_metrics.";
+static constexpr absl::string_view kRequestCostFieldPrefix = "request_cost.";
+static constexpr absl::string_view kUtilizationFieldPrefix = "utilization.";
+} // namespace
+
+void addOrcaNamedMetricToLoadMetricStats(const Protobuf::Map<std::string, double>& metrics_map,
+                                         const absl::string_view metric_name,
+                                         const absl::string_view metric_name_prefix,
+                                         Upstream::LoadMetricStats& stats) {
+  absl::string_view metric_name_without_prefix = absl::StripPrefix(metric_name, metric_name_prefix);
+  // If the metric name is "*", add all metrics from the map.
+  if (metric_name_without_prefix == "*") {
+    for (const auto& [key, value] : metrics_map) {
+      stats.add(absl::StrCat(metric_name_prefix, key), value);
+    }
+  } else {
+    // Add the metric if it exists in the map.
+    const auto metric_it = metrics_map.find(metric_name_without_prefix);
+    if (metric_it != metrics_map.end()) {
+      stats.add(metric_name, metric_it->second);
+    }
+  }
+}
+
+void addOrcaLoadReportToLoadMetricStats(const LrsReportMetricNames& metric_names,
+                                        const xds::data::orca::v3::OrcaLoadReport& report,
+                                        Upstream::LoadMetricStats& stats) {
+  // TODO(efimki): Use InlineMap to speed up this loop.
+  for (const std::string& metric_name : metric_names) {
+    if (metric_name == kCpuUtilizationField) {
+      stats.add(metric_name, report.cpu_utilization());
+    } else if (metric_name == kMemUtilizationField) {
+      stats.add(metric_name, report.mem_utilization());
+    } else if (metric_name == kApplicationUtilizationField) {
+      stats.add(metric_name, report.application_utilization());
+    } else if (metric_name == kEpsField) {
+      stats.add(metric_name, report.eps());
+    } else if (metric_name == kRpsFractionalField) {
+      stats.add(metric_name, report.rps_fractional());
+    } else if (absl::StartsWith(metric_name, kNamedMetricsFieldPrefix)) {
+      addOrcaNamedMetricToLoadMetricStats(report.named_metrics(), metric_name,
+                                          kNamedMetricsFieldPrefix, stats);
+    } else if (absl::StartsWith(metric_name, kUtilizationFieldPrefix)) {
+      addOrcaNamedMetricToLoadMetricStats(report.utilization(), metric_name,
+                                          kUtilizationFieldPrefix, stats);
+    } else if (absl::StartsWith(metric_name, kRequestCostFieldPrefix)) {
+      addOrcaNamedMetricToLoadMetricStats(report.request_cost(), metric_name,
+                                          kRequestCostFieldPrefix, stats);
+    }
+  }
+}
+
+} // namespace Orca
+} // namespace Envoy

source/common/orca/orca_load_metrics.h

@@ -0,0 +1,18 @@
+#pragma once
+
+#include "envoy/upstream/host_description.h"
+
+#include "xds/data/orca/v3/orca_load_report.pb.h"
+
+namespace Envoy {
+namespace Orca {
+
+// List of metric names to report to the LRS.
+typedef std::vector<std::string> LrsReportMetricNames;
+
+void addOrcaLoadReportToLoadMetricStats(const LrsReportMetricNames& metric_names,
+                                        const xds::data::orca::v3::OrcaLoadReport& report,
+                                        Upstream::LoadMetricStats& stats);
+
+} // namespace Orca
+} // namespace Envoy

source/common/router/BUILD

@@ -366,6 +366,8 @@ envoy_cc_library(
         "//source/common/network:socket_option_factory_lib",
         "//source/common/network:transport_socket_options_lib",
         "//source/common/network:upstream_socket_options_filter_state_lib",
+        "//source/common/orca:orca_load_metrics_lib",
+        "//source/common/orca:orca_parser",
         "//source/common/stream_info:stream_info_lib",
         "//source/common/stream_info:uint32_accessor_lib",
         "//source/common/tracing:http_tracer_lib",

source/common/router/router.cc

@@ -35,6 +35,8 @@
 #include "source/common/network/upstream_server_name.h"
 #include "source/common/network/upstream_socket_options_filter_state.h"
 #include "source/common/network/upstream_subject_alt_names.h"
+#include "source/common/orca/orca_load_metrics.h"
+#include "source/common/orca/orca_parser.h"
 #include "source/common/router/config_impl.h"
 #include "source/common/router/debug_config.h"
 #include "source/common/router/retry_state_impl.h"
@@ -1568,6 +1570,8 @@ void Filter::onUpstreamHeaders(uint64_t response_code, Http::ResponseHeaderMapPt
     }
   }
 
+  maybeProcessOrcaLoadReport(*headers, upstream_request);
+
   if (grpc_status.has_value()) {
     upstream_request.upstreamHost()->outlierDetector().putHttpResponseCode(grpc_to_http_status);
   } else {
@@ -1737,6 +1741,8 @@ void Filter::onUpstreamTrailers(Http::ResponseTrailerMapPtr&& trailers,
     }
   }
 
+  maybeProcessOrcaLoadReport(*trailers, upstream_request);
+
   onUpstreamComplete(upstream_request);
 
   callbacks_->encodeTrailers(std::move(trailers));
@@ -2100,6 +2106,38 @@ bool Filter::checkDropOverload(Upstream::ThreadLocalCluster& cluster,
   return false;
 }
 
+void Filter::maybeProcessOrcaLoadReport(const Envoy::Http::HeaderMap& headers_or_trailers,
+                                        UpstreamRequest& upstream_request) {
+  // Process the load report only once, so if response has report in headers,
+  // then don't process it in trailers.
+  if (orca_load_report_received_) {
+    return;
+  }
+  // Check whether we need to send the load report to the LRS or invoke the ORCA
+  // callbacks.
+  auto host = upstream_request.upstreamHost();
+  const bool need_to_send_load_report =
+      (host != nullptr) && cluster_->lrsReportMetricNames().has_value();
+  if (!need_to_send_load_report) {
+    return;
+  }
+
+  absl::StatusOr<xds::data::orca::v3::OrcaLoadReport> orca_load_report =
+      Envoy::Orca::parseOrcaLoadReportHeaders(headers_or_trailers);
+  if (!orca_load_report.ok()) {
+    ENVOY_STREAM_LOG(trace, "Headers don't have orca load report: {}", *callbacks_,
+                     orca_load_report.status().message());
+    return;
+  }
+
+  orca_load_report_received_ = true;
+
+  ENVOY_STREAM_LOG(trace, "Adding ORCA load report {} to load metrics", *callbacks_,
+                   orca_load_report->DebugString());
+  Envoy::Orca::addOrcaLoadReportToLoadMetricStats(
+      cluster_->lrsReportMetricNames().value(), orca_load_report.value(), host->loadMetricStats());
+}
+
 RetryStatePtr
 ProdFilter::createRetryState(const RetryPolicy& policy, Http::RequestHeaderMap& request_headers,
                              const Upstream::ClusterInfo& cluster, const VirtualCluster* vcluster,

source/common/router/router.h

@@ -308,7 +308,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
         downstream_response_started_(false), downstream_end_stream_(false), is_retry_(false),
         request_buffer_overflowed_(false), streaming_shadows_(Runtime::runtimeFeatureEnabled(
                                                "envoy.reloadable_features.streaming_shadow")),
-        upstream_request_started_(false) {}
+        upstream_request_started_(false), orca_load_report_received_(false) {}
 
   ~Filter() override;
 
@@ -559,6 +559,9 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   Http::Context& httpContext() { return config_->http_context_; }
   bool checkDropOverload(Upstream::ThreadLocalCluster& cluster,
                          std::function<void(Http::ResponseHeaderMap&)>& modify_headers);
+  // Process Orca Load Report if necessary (e.g. cluster has lrsReportMetricNames).
+  void maybeProcessOrcaLoadReport(const Envoy::Http::HeaderMap& headers_or_trailers,
+                                  UpstreamRequest& upstream_request);
 
   RetryStatePtr retry_state_;
   const FilterConfigSharedPtr config_;
@@ -611,6 +614,9 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   bool request_buffer_overflowed_ : 1;
   const bool streaming_shadows_ : 1;
   bool upstream_request_started_ : 1;
+  // Indicate that ORCA report is received to process it only once in either response headers or
+  // trailers.
+  bool orca_load_report_received_ : 1;
 };
 
 class ProdFilter : public Filter {

source/common/upstream/BUILD

@@ -457,6 +457,7 @@ envoy_cc_library(
         "//source/common/http/http2:codec_stats_lib",
         "//source/common/http/http3:codec_stats_lib",
         "//source/common/init:manager_lib",
+        "//source/common/orca:orca_load_metrics_lib",
         "//source/common/shared_pool:shared_pool_lib",
         "//source/common/stats:deferred_creation",
         "//source/common/stats:isolated_store_lib",

source/common/upstream/upstream_impl.cc

@@ -1208,6 +1208,11 @@ ClusterInfoImpl::ClusterInfoImpl(
                     envoy::config::cluster::v3::UpstreamConnectionOptions::HappyEyeballsConfig>(
                     config.upstream_connection_options().happy_eyeballs_config())
               : nullptr),
+      lrs_report_metric_names_(!config.lrs_report_endpoint_metrics().empty()
+                                   ? std::make_unique<Envoy::Orca::LrsReportMetricNames>(
+                                         config.lrs_report_endpoint_metrics().begin(),
+                                         config.lrs_report_endpoint_metrics().end())
+                                   : nullptr),
       per_connection_buffer_limit_bytes_(
           PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, per_connection_buffer_limit_bytes, 1024 * 1024)),
       max_response_headers_count_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(

source/common/upstream/upstream_impl.h

@@ -55,6 +55,7 @@
 #include "source/common/http/http3/codec_stats.h"
 #include "source/common/init/manager_impl.h"
 #include "source/common/network/utility.h"
+#include "source/common/orca/orca_load_metrics.h"
 #include "source/common/shared_pool/shared_pool.h"
 #include "source/common/stats/isolated_store_impl.h"
 #include "source/common/upstream/edf_scheduler.h"
@@ -1023,6 +1024,13 @@ class ClusterInfoImpl : public ClusterInfo,
     return *happy_eyeballs_config_;
   }
 
+  OptRef<const Envoy::Orca::LrsReportMetricNames> lrsReportMetricNames() const override {
+    if (lrs_report_metric_names_ == nullptr) {
+      return absl::nullopt;
+    }
+    return *lrs_report_metric_names_;
+  }
+
 protected:
   // Gets the retry budget percent/concurrency from the circuit breaker thresholds. If the retry
   // budget message is specified, defaults will be filled in if either params are unspecified.
@@ -1111,6 +1119,7 @@ class ClusterInfoImpl : public ClusterInfo,
   UpstreamFactoryContextImpl upstream_context_;
   std::unique_ptr<envoy::config::cluster::v3::UpstreamConnectionOptions::HappyEyeballsConfig>
       happy_eyeballs_config_;
+  const std::unique_ptr<Envoy::Orca::LrsReportMetricNames> lrs_report_metric_names_;
 
   // Keep small values like bools and enums at the end of the class to reduce
   // overhead via alignment

test/common/orca/BUILD

@@ -8,6 +8,23 @@ licenses(["notice"])  # Apache 2
 
 envoy_package()
 
+envoy_cc_test(
+    name = "orca_load_metrics_test",
+    srcs = ["orca_load_metrics_test.cc"],
+    external_deps = [
+        "abseil_status",
+        "abseil_strings",
+        "fmtlib",
+    ],
+    deps = [
+        "//source/common/orca:orca_load_metrics_lib",
+        "//source/common/upstream:upstream_lib",
+        "//test/test_common:status_utility_lib",
+        "//test/test_common:utility_lib",
+        "@com_github_cncf_xds//xds/data/orca/v3:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_test(
     name = "orca_parser_test",
     srcs = ["orca_parser_test.cc"],