Invoke ORCA Load Report callbacks from Router::Filter.

envoy/upstream/BUILD

@@ -86,6 +86,7 @@ envoy_cc_library(
         ":upstream_interface",
         "//envoy/router:router_interface",
         "//envoy/upstream:types_interface",
+        "@com_github_cncf_xds//xds/data/orca/v3:pkg_cc_proto",
     ],
 )
 

envoy/upstream/load_balancer.h

@@ -10,6 +10,8 @@
 #include "envoy/upstream/types.h"
 #include "envoy/upstream/upstream.h"
 
+#include "xds/data/orca/v3/orca_load_report.pb.h"
+
 namespace Envoy {
 namespace Http {
 namespace ConnectionPool {
@@ -103,6 +105,16 @@ class LoadBalancerContext {
    * and return the corresponding host directly.
    */
   virtual absl::optional<OverrideHost> overrideHostToSelect() const PURE;
+
+  // Invoked when a new orca report is received for this LB context.
+  using OrcaLoadReportCb =
+      std::function<absl::Status(const xds::data::orca::v3::OrcaLoadReport& orca_load_report)>;
+
+  /**
+   * Install a callback to be invoked when ORCA Load report is received for this
+   * LB context.
+   */
+  virtual void setOrcaLoadReportCb(OrcaLoadReportCb callback) PURE;
 };
 
 /**

source/common/router/router.cc

@@ -2118,7 +2118,7 @@ void Filter::maybeProcessOrcaLoadReport(const Envoy::Http::HeaderMap& headers_or
   auto host = upstream_request.upstreamHost();
   const bool need_to_send_load_report =
       (host != nullptr) && cluster_->lrsReportMetricNames().has_value();
-  if (!need_to_send_load_report) {
+  if (!need_to_send_load_report && !orca_load_report_cb_) {
     return;
   }
 
@@ -2132,10 +2132,19 @@ void Filter::maybeProcessOrcaLoadReport(const Envoy::Http::HeaderMap& headers_or
 
   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());
+  if (need_to_send_load_report) {
+    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());
+  }
+  if (orca_load_report_cb_) {
+    auto status = orca_load_report_cb_(*orca_load_report);
+    if (!status.ok()) {
+      ENVOY_LOG_MISC(error, "Failed to invoke OrcaLoadReportCb: {}", status.message());
+    }
+  }
 }
 
 RetryStatePtr

source/common/router/router.h

@@ -422,6 +422,10 @@ class Filter : Logger::Loggable<Logger::Id::router>,
     return callbacks_->upstreamOverrideHost();
   }
 
+  void setOrcaLoadReportCb(OrcaLoadReportCb callback) override {
+    orca_load_report_cb_ = std::move(callback);
+  }
+
   /**
    * Set a computed cookie to be sent with the downstream headers.
    * @param key supplies the size of the cookie
@@ -604,6 +608,7 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   Http::Code timeout_response_code_ = Http::Code::GatewayTimeout;
   FilterUtility::HedgingParams hedging_params_;
   Http::StreamFilterSidestreamWatermarkCallbacks watermark_callbacks_;
+  OrcaLoadReportCb orca_load_report_cb_;
   bool grpc_request_ : 1;
   bool exclude_http_code_stats_ : 1;
   bool downstream_response_started_ : 1;

source/common/upstream/load_balancer_context_base.h

@@ -34,6 +34,8 @@ class LoadBalancerContextBase : public LoadBalancerContext {
   }
 
   absl::optional<OverrideHost> overrideHostToSelect() const override { return {}; }
+
+  void setOrcaLoadReportCb(OrcaLoadReportCb) override {}
 };
 
 } // namespace Upstream

source/extensions/load_balancing_policies/subset/subset_lb.h

@@ -200,6 +200,10 @@ class SubsetLoadBalancer : public LoadBalancer, Logger::Loggable<Logger::Id::ups
       return wrapped_->overrideHostToSelect();
     }
 
+    void setOrcaLoadReportCb(OrcaLoadReportCb callback) override {
+      return wrapped_->setOrcaLoadReportCb(std::move(callback));
+    }
+
   private:
     LoadBalancerContext* wrapped_;
     Router::MetadataMatchCriteriaConstPtr metadata_match_;

test/mocks/upstream/load_balancer_context.h

@@ -26,6 +26,7 @@ class MockLoadBalancerContext : public LoadBalancerContext {
   MOCK_METHOD(Network::TransportSocketOptionsConstSharedPtr, upstreamTransportSocketOptions, (),
               (const));
   MOCK_METHOD(absl::optional<OverrideHost>, overrideHostToSelect, (), (const));
+  MOCK_METHOD(void, setOrcaLoadReportCb, (OrcaLoadReportCb));
 
 private:
   HealthyAndDegradedLoad priority_load_;