stream info: add upstream connection id

envoy/stream_info/stream_info.h

@@ -242,6 +242,10 @@ class StreamInfo {
 public:
   virtual ~StreamInfo() = default;
 
+  virtual void setUpstreamConnectionId(uint64_t id) PURE;
+
+  virtual absl::optional<uint64_t> upstreamConnectionId() const PURE;
+
   /**
    * @param response_flag the response flag. Each filter can set independent response flags. The
    * flags are accumulated.

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/router.cc

@@ -1231,7 +1231,7 @@ void Filter::resetOtherUpstreams(UpstreamRequest& upstream_request) {
 
 void Filter::onUpstreamHeaders(uint64_t response_code, Http::ResponseHeaderMapPtr&& headers,
                                UpstreamRequest& upstream_request, bool end_stream) {
-  ENVOY_STREAM_LOG(debug, "upstream headers complete: end_stream={}", *callbacks_, end_stream);
+  ENVOY_STREAM_LOG(debug, "upstream headers complete: end_stream={} upstream_conn={}", *callbacks_, end_stream, callbacks_->streamInfo().upstreamConnectionId().value_or(0));
 
   modify_headers_(*headers);
   // When grpc-status appears in response headers, convert grpc-status to HTTP status code

source/common/router/upstream_request.cc

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

source/common/stream_info/stream_info_impl.h

@@ -68,6 +68,14 @@ 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 +309,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() {