orca: make OOB active_sessions gauge composable across managers

changelogs/current/bug_fixes/load_balancing__oob-active-sessions-gauge-delta.rst

@@ -0,0 +1,8 @@
+Fixed the ORCA out-of-band reporting ``lb_orca_oob.active_sessions`` gauge to update by delta
+(``add``/``sub``/``dec``) instead of overwriting it with a single manager's session count.
+Previously, the gauge could be left stale at 0 after a cluster's OOB manager was torn down and
+recreated (for example on a CDS update), since the outgoing manager's ``set(0)`` on destruction
+could run after the incoming manager's ``set(N)``. The overwrite also produced an incorrect value
+whenever more than one load balancing policy in a cluster opened OOB streams against the same
+stats scope, since each manager's ``set()`` clobbered the other's contribution instead of
+composing.

source/extensions/load_balancing_policies/common/orca_oob_manager.cc

@@ -77,12 +77,13 @@ OrcaOobManager::OrcaOobManager(OrcaOobManagerConfig config,
       priority_set_(priority_set), oob_stats_(generateOrcaOobStats(stats_scope)) {}
 
 OrcaOobManager::~OrcaOobManager() {
+  const size_t session_count = oob_sessions_.size();
   for (auto& [host, session] : oob_sessions_) {
     session->disarm();
     dispatcher_.deferredDelete(std::move(session));
   }
   oob_sessions_.clear();
-  oob_stats_.active_sessions_.set(0);
+  oob_stats_.active_sessions_.sub(session_count);
 }
 
 absl::Status OrcaOobManager::initialize() {
@@ -115,8 +116,8 @@ void OrcaOobManager::onHostsAdded(const Upstream::HostVector& hosts) {
     const std::chrono::milliseconds initial_delay(random_.random() % period_ms);
     it->second = std::make_unique<OobSession>(*this, host, initial_delay);
   }
-  if (oob_sessions_.size() != prior_size) {
-    oob_stats_.active_sessions_.set(oob_sessions_.size());
+  if (oob_sessions_.size() > prior_size) {
+    oob_stats_.active_sessions_.add(oob_sessions_.size() - prior_size);
   }
 }
 
@@ -131,8 +132,8 @@ void OrcaOobManager::onHostsRemoved(const Upstream::HostVector& hosts) {
     dispatcher_.deferredDelete(std::move(it->second));
     oob_sessions_.erase(it);
   }
-  if (oob_sessions_.size() != prior_size) {
-    oob_stats_.active_sessions_.set(oob_sessions_.size());
+  if (oob_sessions_.size() < prior_size) {
+    oob_stats_.active_sessions_.sub(prior_size - oob_sessions_.size());
   }
 }
 
@@ -141,7 +142,7 @@ void OrcaOobManager::onSessionTerminated(OobSession* session) {
   ASSERT(it != oob_sessions_.end() && it->second.get() == session);
   dispatcher_.deferredDelete(std::move(it->second));
   oob_sessions_.erase(it);
-  oob_stats_.active_sessions_.set(oob_sessions_.size());
+  oob_stats_.active_sessions_.dec();
 }
 
 OrcaOobManager::OobSession::OobSession(OrcaOobManager& parent, Upstream::HostConstSharedPtr host,

test/extensions/load_balancing_policies/common/orca_oob_manager_test.cc

@@ -192,6 +192,23 @@ TEST_F(OrcaOobManagerLifecycleTest, DestructionDisarmsActiveSessions) {
   manager.reset();
 }
 
+// Two managers sharing one stats scope (e.g. load_aware_locality and its child
+// ClientSideWeightedRoundRobin policy each enabling OOB) must compose deltas on the shared
+// active_sessions gauge rather than clobber each other via absolute set().
+TEST_F(OrcaOobManagerLifecycleTest, ActiveSessionsGaugeComposesAcrossManagers) {
+  auto manager1 = makeManager();
+  ASSERT_OK(manager1->initialize());
+  auto manager2 = makeManager();
+  ASSERT_OK(manager2->initialize());
+
+  priority_set_.runUpdateCallbacks(0, {makeHost(), makeHost()}, {});
+  EXPECT_EQ(activeOobSessions(), 4);
+
+  EXPECT_CALL(dispatcher_, deferredDelete_(_)).Times(AtLeast(1));
+  manager1.reset();
+  EXPECT_EQ(activeOobSessions(), 2);
+}
+
 // Wire fixture: drives end-to-end ORCA OOB decode path through a real CodecClient
 // (CodecClientForTest) layered over Network::MockClientConnection +
 // Http::MockClientConnection. Models the gRPC health checker's wire-test pattern