stream info: add upstream connection id

envoy/stream_info/stream_info.h

@@ -593,6 +593,16 @@ class StreamInfo {
    * @return Network filter chain name of the downstream connection.
    */
   virtual const std::string& filterChainName() const PURE;
+
+  /**
+   * @param connection ID of the upstream connection.
+   */
+  virtual void setUpstreamConnectionId(uint64_t id) PURE;
+
+  /**
+   * @return the ID of the upstream connection, nullopt if not available.
+   */
+  virtual absl::optional<uint64_t> upstreamConnectionId() const PURE;
 };
 
 } // namespace StreamInfo

source/common/network/connection_impl.cc

@@ -824,6 +824,8 @@ ClientConnectionImpl::ClientConnectionImpl(
     : ConnectionImpl(dispatcher, std::make_unique<ClientSocketImpl>(remote_address, options),
                      std::move(transport_socket), stream_info_, false),
       stream_info_(dispatcher.timeSource(), socket_->addressProviderSharedPtr()) {
+
+  stream_info_.setUpstreamConnectionId(id());
   // There are no meaningful socket options or source address semantics for
   // non-IP sockets, so skip.
   if (remote_address->ip() == nullptr) {

source/common/router/upstream_request.cc

@@ -420,6 +420,11 @@ void UpstreamRequest::onPoolReady(
   stream_info_.setUpstreamSslConnection(info.downstreamSslConnection());
   parent_.callbacks()->streamInfo().setUpstreamSslConnection(info.downstreamSslConnection());
 
+  if (info.upstreamConnectionId().has_value()) {
+    stream_info_.setUpstreamConnectionId(info.upstreamConnectionId().value());
+    parent_.callbacks()->streamInfo().setUpstreamConnectionId(info.upstreamConnectionId().value());
+  }
+
   if (parent_.downstreamEndStream()) {
     setupPerTryTimeout();
   } else {

source/common/stream_info/stream_info_impl.h

@@ -68,6 +68,10 @@ struct StreamInfoImpl : public StreamInfo {
                                                                 start_time_monotonic_);
   }
 
+  void setUpstreamConnectionId(uint64_t id) override { upstream_connection_id_ = id; }
+
+  absl::optional<uint64_t> upstreamConnectionId() const override { return upstream_connection_id_; }
+
   absl::optional<std::chrono::nanoseconds> lastDownstreamRxByteReceived() const override {
     return duration(last_downstream_rx_byte_received);
   }
@@ -301,6 +305,7 @@ struct StreamInfoImpl : public StreamInfo {
   FilterStateSharedPtr filter_state_;
   FilterStateSharedPtr upstream_filter_state_;
   std::string route_name_;
+  absl::optional<uint64_t> upstream_connection_id_;
 
 private:
   static Network::SocketAddressProviderSharedPtr emptyDownstreamAddressProvider() {

test/common/router/router_test.cc

@@ -4870,6 +4870,7 @@ TEST_F(RouterTest, UpstreamSSLConnection) {
 
   ASSERT_NE(nullptr, callbacks_.streamInfo().upstreamSslConnection());
   EXPECT_EQ(session_id, callbacks_.streamInfo().upstreamSslConnection()->sessionId());
+  EXPECT_FALSE(callbacks_.streamInfo().upstreamConnectionId().has_value());
 }
 
 // Verify that upstream timing information is set into the StreamInfo after the upstream

test/common/stream_info/stream_info_impl_test.cc

@@ -192,6 +192,11 @@ TEST_F(StreamInfoImplTest, MiscSettersAndGetters) {
     EXPECT_CALL(*ssl_info, sessionId()).WillRepeatedly(testing::ReturnRef(session_id));
     stream_info.setUpstreamSslConnection(ssl_info);
     EXPECT_EQ(session_id, stream_info.upstreamSslConnection()->sessionId());
+
+    EXPECT_FALSE(stream_info.upstreamConnectionId().has_value());
+    stream_info.setUpstreamConnectionId(12345);
+    ASSERT_TRUE(stream_info.upstreamConnectionId().has_value());
+    EXPECT_EQ(12345, stream_info.upstreamConnectionId().value());
   }
 }
 

test/common/stream_info/test_util.h

@@ -219,6 +219,10 @@ class TestStreamInfo : public StreamInfo::StreamInfo {
 
   const std::string& filterChainName() const override { return filter_chain_name_; }
 
+  void setUpstreamConnectionId(uint64_t id) override { upstream_connection_id_ = id; }
+
+  absl::optional<uint64_t> upstreamConnectionId() const override { return upstream_connection_id_; }
+
   Random::RandomGeneratorImpl random_;
   SystemTime start_time_;
   MonotonicTime start_time_monotonic_;
@@ -258,9 +262,9 @@ class TestStreamInfo : public StreamInfo::StreamInfo {
   Envoy::Event::SimulatedTimeSystem test_time_;
   absl::optional<Upstream::ClusterInfoConstSharedPtr> upstream_cluster_info_{};
   Http::RequestIdStreamInfoProviderSharedPtr request_id_provider_;
-  absl::optional<uint64_t> connection_id_;
   std::string filter_chain_name_;
   Tracing::Reason trace_reason_{Tracing::Reason::NotTraceable};
+  absl::optional<uint64_t> upstream_connection_id_;
 };
 
 } // namespace Envoy

test/mocks/stream_info/mocks.cc

@@ -126,6 +126,12 @@ MockStreamInfo::MockStreamInfo()
         filter_chain_name_ = std::string(filter_chain_name);
       }));
   ON_CALL(*this, filterChainName()).WillByDefault(ReturnRef(filter_chain_name_));
+  ON_CALL(*this, setUpstreamConnectionId(_)).WillByDefault(Invoke([this](uint64_t id) {
+    upstream_connection_id_ = id;
+  }));
+  ON_CALL(*this, upstreamConnectionId()).WillByDefault(Invoke([this]() {
+    return upstream_connection_id_;
+  }));
 }
 
 MockStreamInfo::~MockStreamInfo() = default;

test/mocks/stream_info/mocks.h

@@ -96,6 +96,8 @@ class MockStreamInfo : public StreamInfo {
   MOCK_METHOD(void, setConnectionID, (uint64_t));
   MOCK_METHOD(void, setFilterChainName, (const absl::string_view));
   MOCK_METHOD(const std::string&, filterChainName, (), (const));
+  MOCK_METHOD(void, setUpstreamConnectionId, (uint64_t));
+  MOCK_METHOD(absl::optional<uint64_t>, upstreamConnectionId, (), (const));
 
   std::shared_ptr<testing::NiceMock<Upstream::MockHostDescription>> host_{
       new testing::NiceMock<Upstream::MockHostDescription>()};
@@ -128,6 +130,7 @@ class MockStreamInfo : public StreamInfo {
   std::string route_name_;
   std::string upstream_transport_failure_reason_;
   std::string filter_chain_name_;
+  absl::optional<uint64_t> upstream_connection_id_;
 };
 
 } // namespace StreamInfo