http: fixing UPSTREAM_REMOTE_ADDRESS for happy eyeballs, proxied connections

changelogs/current.yaml

@@ -59,6 +59,8 @@ minor_behavior_changes:
 - area: http
   change: |
     changed the filter callback interfaces to make sure that downstream-only functionality is explicit.
+  change: |
+    the upstream remote address is now available to downstream filters via the ``upstreamRemoteAddress`` function.
 - area: stats
   change: |
     Default tag extraction rules were changed for ``worker_id`` extraction. Previously, ``worker_`` was removed from the original name during the extraction. This

envoy/http/codec.h

@@ -341,10 +341,10 @@ class Stream : public StreamResetHandler {
   virtual absl::string_view responseDetails() { return ""; }
 
   /**
-   * @return const Address::InstanceConstSharedPtr& the local address of the connection associated
-   * with the stream.
+   * @return const Network::ConnectionInfoProvider& the adderess provider  of the connection
+   * associated with the stream.
    */
-  virtual const Network::Address::InstanceConstSharedPtr& connectionLocalAddress() PURE;
+  virtual const Network::ConnectionInfoProvider& connectionInfoProvider() PURE;
 
   /**
    * Set the flush timeout for the stream. At the codec level this is used to bound the amount of

envoy/router/router.h

@@ -1400,13 +1400,13 @@ class GenericConnectionPoolCallbacks {
    * @param upstream supplies the generic upstream for the stream.
    * @param host supplies the description of the host that will carry the request. For logical
    *             connection pools the description may be different each time this is called.
-   * @param upstream_local_address supplies the local address of the upstream connection.
+   * @param connection_info_provider, supplies the address provider of the upstream connection.
    * @param info supplies the stream info object associated with the upstream connection.
    * @param protocol supplies the protocol associated with the upstream connection.
    */
   virtual void onPoolReady(std::unique_ptr<GenericUpstream>&& upstream,
                            Upstream::HostDescriptionConstSharedPtr host,
-                           const Network::Address::InstanceConstSharedPtr& upstream_local_address,
+                           const Network::ConnectionInfoProvider& connection_info_provider,
                            StreamInfo::StreamInfo& info,
                            absl::optional<Http::Protocol> protocol) PURE;
 

envoy/stream_info/stream_info.h

@@ -402,6 +402,17 @@ class UpstreamInfo {
    */
   virtual const Network::Address::InstanceConstSharedPtr& upstreamLocalAddress() const PURE;
 
+  /**
+   * @param upstream_remote_address sets the remote address of the upstream connection.
+   */
+  virtual void setUpstreamRemoteAddress(
+      const Network::Address::InstanceConstSharedPtr& upstream_remote_address) PURE;
+
+  /**
+   * @return the upstream remote address.
+   */
+  virtual const Network::Address::InstanceConstSharedPtr& upstreamRemoteAddress() const PURE;
+
   /**
    * @param failure_reason the upstream transport failure reason.
    */

envoy/tcp/upstream.h

@@ -67,14 +67,14 @@ class GenericConnectionPoolCallbacks {
    * @param info supplies the stream info object associated with the upstream connection.
    * @param upstream supplies the generic upstream for the stream.
    * @param host supplies the description of the host that will carry the request.
-   * @param upstream_local_address supplies the local address of the upstream connection.
+   * @param address_provider supplies the address provider of the upstream connection.
    * @param ssl_info supplies the ssl information of the upstream connection.
    */
-  virtual void
-  onGenericPoolReady(StreamInfo::StreamInfo* info, std::unique_ptr<GenericUpstream>&& upstream,
-                     Upstream::HostDescriptionConstSharedPtr& host,
-                     const Network::Address::InstanceConstSharedPtr& upstream_local_address,
-                     Ssl::ConnectionInfoConstSharedPtr ssl_info) PURE;
+  virtual void onGenericPoolReady(StreamInfo::StreamInfo* info,
+                                  std::unique_ptr<GenericUpstream>&& upstream,
+                                  Upstream::HostDescriptionConstSharedPtr& host,
+                                  const Network::ConnectionInfoProvider& address_provider,
+                                  Ssl::ConnectionInfoConstSharedPtr ssl_info) PURE;
 
   /**
    * Called to indicate a failure for GenericConnPool::newStream to establish a stream.

source/common/http/http1/codec_impl.cc

@@ -375,8 +375,8 @@ void StreamEncoderImpl::readDisable(bool disable) {
 
 uint32_t StreamEncoderImpl::bufferLimit() const { return connection_.bufferLimit(); }
 
-const Network::Address::InstanceConstSharedPtr& StreamEncoderImpl::connectionLocalAddress() {
-  return connection_.connection().connectionInfoProvider().localAddress();
+const Network::ConnectionInfoProvider& StreamEncoderImpl::connectionInfoProvider() {
+  return connection_.connection().connectionInfoProvider();
 }
 
 static constexpr absl::string_view RESPONSE_PREFIX = "HTTP/1.1 ";

source/common/http/http1/codec_impl.h

@@ -62,7 +62,7 @@ class StreamEncoderImpl : public virtual StreamEncoder,
   void readDisable(bool disable) override;
   uint32_t bufferLimit() const override;
   absl::string_view responseDetails() override { return details_; }
-  const Network::Address::InstanceConstSharedPtr& connectionLocalAddress() override;
+  const Network::ConnectionInfoProvider& connectionInfoProvider() override;
   void setFlushTimeout(std::chrono::milliseconds) override {
     // HTTP/1 has one stream per connection, thus any data encoded is immediately written to the
     // connection, invoking any watermarks as necessary. There is no internal buffering that would

source/common/http/http2/codec_impl.h

@@ -266,8 +266,8 @@ class ConnectionImpl : public virtual Connection,
     void resetStream(StreamResetReason reason) override;
     void readDisable(bool disable) override;
     uint32_t bufferLimit() const override { return pending_recv_data_->highWatermark(); }
-    const Network::Address::InstanceConstSharedPtr& connectionLocalAddress() override {
-      return parent_.connection_.connectionInfoProvider().localAddress();
+    const Network::ConnectionInfoProvider& connectionInfoProvider() override {
+      return parent_.connection_.connectionInfoProvider();
     }
     absl::string_view responseDetails() override { return details_; }
     void setAccount(Buffer::BufferMemoryAccountSharedPtr account) override;

source/common/quic/envoy_quic_stream.h

@@ -78,8 +78,8 @@ class EnvoyQuicStream : public virtual Http::StreamEncoder,
     removeCallbacksHelper(callbacks);
   }
   uint32_t bufferLimit() const override { return send_buffer_simulation_.highWatermark(); }
-  const Network::Address::InstanceConstSharedPtr& connectionLocalAddress() override {
-    return connection()->connectionInfoProvider().localAddress();
+  const Network::ConnectionInfoProvider& connectionInfoProvider() override {
+    return connection()->connectionInfoProvider();
   }
 
   void setAccount(Buffer::BufferMemoryAccountSharedPtr account) override {

source/common/router/upstream_request.cc

@@ -710,10 +710,11 @@ void UpstreamRequest::onPoolFailure(ConnectionPool::PoolFailureReason reason,
   onResetStream(reset_reason, transport_failure_reason);
 }
 
-void UpstreamRequest::onPoolReady(
-    std::unique_ptr<GenericUpstream>&& upstream, Upstream::HostDescriptionConstSharedPtr host,
-    const Network::Address::InstanceConstSharedPtr& upstream_local_address,
-    StreamInfo::StreamInfo& info, absl::optional<Http::Protocol> protocol) {
+void UpstreamRequest::onPoolReady(std::unique_ptr<GenericUpstream>&& upstream,
+                                  Upstream::HostDescriptionConstSharedPtr host,
+                                  const Network::ConnectionInfoProvider& address_provider,
+                                  StreamInfo::StreamInfo& info,
+                                  absl::optional<Http::Protocol> protocol) {
   // This may be called under an existing ScopeTrackerScopeState but it will unwind correctly.
   ScopeTrackerScopeState scope(&parent_.callbacks()->scope(), parent_.callbacks()->dispatcher());
   ENVOY_STREAM_LOG(debug, "pool ready", *parent_.callbacks());
@@ -763,7 +764,8 @@ void UpstreamRequest::onPoolReady(
   } else {
     upstream_info.setUpstreamFilterState(filter_state);
   }
-  upstream_info.setUpstreamLocalAddress(upstream_local_address);
+  upstream_info.setUpstreamLocalAddress(address_provider.localAddress());
+  upstream_info.setUpstreamRemoteAddress(address_provider.remoteAddress());
   upstream_info.setUpstreamSslConnection(info.downstreamAddressProvider().sslConnection());
 
   if (info.downstreamAddressProvider().connectionID().has_value()) {

source/common/router/upstream_request.h

@@ -137,7 +137,7 @@ class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,
                      Upstream::HostDescriptionConstSharedPtr host) override;
   void onPoolReady(std::unique_ptr<GenericUpstream>&& upstream,
                    Upstream::HostDescriptionConstSharedPtr host,
-                   const Network::Address::InstanceConstSharedPtr& upstream_local_address,
+                   const Network::ConnectionInfoProvider& address_provider,
                    StreamInfo::StreamInfo& info, absl::optional<Http::Protocol> protocol) override;
   UpstreamToDownstream& upstreamToDownstream() override;
 

source/common/stream_info/stream_info_impl.h

@@ -49,10 +49,17 @@ struct UpstreamInfoImpl : public UpstreamInfo {
   const Network::Address::InstanceConstSharedPtr& upstreamLocalAddress() const override {
     return upstream_local_address_;
   }
+  const Network::Address::InstanceConstSharedPtr& upstreamRemoteAddress() const override {
+    return upstream_remote_address_;
+  }
   void setUpstreamLocalAddress(
       const Network::Address::InstanceConstSharedPtr& upstream_local_address) override {
     upstream_local_address_ = upstream_local_address;
   }
+  void setUpstreamRemoteAddress(
+      const Network::Address::InstanceConstSharedPtr& upstream_remote_address) override {
+    upstream_remote_address_ = upstream_remote_address;
+  }
   void setUpstreamTransportFailureReason(absl::string_view failure_reason) override {
     upstream_transport_failure_reason_ = std::string(failure_reason);
   }
@@ -84,6 +91,7 @@ struct UpstreamInfoImpl : public UpstreamInfo {
 
   Upstream::HostDescriptionConstSharedPtr upstream_host_{};
   Network::Address::InstanceConstSharedPtr upstream_local_address_;
+  Network::Address::InstanceConstSharedPtr upstream_remote_address_;
   UpstreamTiming upstream_timing_;
   Ssl::ConnectionInfoConstSharedPtr upstream_ssl_info_;
   absl::optional<uint64_t> upstream_connection_id_;

source/common/tcp_proxy/tcp_proxy.cc

@@ -506,14 +506,15 @@ void Filter::onGenericPoolFailure(ConnectionPool::PoolFailureReason reason,
 void Filter::onGenericPoolReady(StreamInfo::StreamInfo* info,
                                 std::unique_ptr<GenericUpstream>&& upstream,
                                 Upstream::HostDescriptionConstSharedPtr& host,
-                                const Network::Address::InstanceConstSharedPtr& local_address,
+                                const Network::ConnectionInfoProvider& address_provider,
                                 Ssl::ConnectionInfoConstSharedPtr ssl_info) {
   upstream_ = std::move(upstream);
   generic_conn_pool_.reset();
   read_callbacks_->upstreamHost(host);
   StreamInfo::UpstreamInfo& upstream_info = *getStreamInfo().upstreamInfo();
   upstream_info.setUpstreamHost(host);
-  upstream_info.setUpstreamLocalAddress(local_address);
+  upstream_info.setUpstreamLocalAddress(address_provider.localAddress());
+  upstream_info.setUpstreamRemoteAddress(address_provider.remoteAddress());
   upstream_info.setUpstreamSslConnection(ssl_info);
   onUpstreamConnection();
   read_callbacks_->continueReading();

source/common/tcp_proxy/tcp_proxy.h

@@ -341,7 +341,7 @@ class Filter : public Network::ReadFilter,
   // GenericConnectionPoolCallbacks
   void onGenericPoolReady(StreamInfo::StreamInfo* info, std::unique_ptr<GenericUpstream>&& upstream,
                           Upstream::HostDescriptionConstSharedPtr& host,
-                          const Network::Address::InstanceConstSharedPtr& local_address,
+                          const Network::ConnectionInfoProvider& address_provider,
                           Ssl::ConnectionInfoConstSharedPtr ssl_info) override;
   void onGenericPoolFailure(ConnectionPool::PoolFailureReason reason,
                             absl::string_view failure_reason,

source/common/tcp_proxy/upstream.cc

@@ -189,7 +189,7 @@ void TcpConnPool::onPoolReady(Tcp::ConnectionPool::ConnectionDataPtr&& conn_data
   auto upstream = std::make_unique<TcpUpstream>(std::move(conn_data), upstream_callbacks_);
   callbacks_->onGenericPoolReady(
       &connection.streamInfo(), std::move(upstream), host,
-      latched_data->connection().connectionInfoProvider().localAddress(),
+      latched_data->connection().connectionInfoProvider(),
       latched_data->connection().streamInfo().downstreamAddressProvider().sslConnection());
 }
 
@@ -252,9 +252,9 @@ void HttpConnPool::onPoolReady(Http::RequestEncoder& request_encoder,
 }
 
 void HttpConnPool::onGenericPoolReady(Upstream::HostDescriptionConstSharedPtr& host,
-                                      const Network::Address::InstanceConstSharedPtr& local_address,
+                                      const Network::ConnectionInfoProvider& address_provider,
                                       Ssl::ConnectionInfoConstSharedPtr ssl_info) {
-  callbacks_->onGenericPoolReady(nullptr, std::move(upstream_), host, local_address, ssl_info);
+  callbacks_->onGenericPoolReady(nullptr, std::move(upstream_), host, address_provider, ssl_info);
 }
 
 Http2Upstream::Http2Upstream(Tcp::ConnectionPool::UpstreamCallbacks& callbacks,

source/common/tcp_proxy/upstream.h

@@ -71,7 +71,7 @@ class HttpConnPool : public GenericConnPool, public Http::ConnectionPool::Callba
     virtual ~Callbacks() = default;
     virtual void onSuccess(Http::RequestEncoder& request_encoder) {
       ASSERT(conn_pool_ != nullptr);
-      conn_pool_->onGenericPoolReady(host_, request_encoder.getStream().connectionLocalAddress(),
+      conn_pool_->onGenericPoolReady(host_, request_encoder.getStream().connectionInfoProvider(),
                                      ssl_info_);
     }
     virtual void onFailure() {
@@ -91,7 +91,7 @@ class HttpConnPool : public GenericConnPool, public Http::ConnectionPool::Callba
 
 private:
   void onGenericPoolReady(Upstream::HostDescriptionConstSharedPtr& host,
-                          const Network::Address::InstanceConstSharedPtr& local_address,
+                          const Network::ConnectionInfoProvider& address_provider,
                           Ssl::ConnectionInfoConstSharedPtr ssl_info);
   const TunnelingConfigHelper& config_;
   Http::CodecType type_;

source/extensions/upstreams/http/http/upstream_request.cc

@@ -63,7 +63,7 @@ void HttpConnPool::onPoolReady(Envoy::Http::RequestEncoder& request_encoder,
   auto upstream =
       std::make_unique<HttpUpstream>(callbacks_->upstreamToDownstream(), &request_encoder);
   callbacks_->onPoolReady(std::move(upstream), host,
-                          request_encoder.getStream().connectionLocalAddress(), info, protocol);
+                          request_encoder.getStream().connectionInfoProvider(), info, protocol);
 }
 
 } // namespace Http

source/extensions/upstreams/http/tcp/upstream_request.cc

@@ -27,8 +27,7 @@ void TcpConnPool::onPoolReady(Envoy::Tcp::ConnectionPool::ConnectionDataPtr&& co
   Network::Connection& latched_conn = conn_data->connection();
   auto upstream =
       std::make_unique<TcpUpstream>(&callbacks_->upstreamToDownstream(), std::move(conn_data));
-  callbacks_->onPoolReady(std::move(upstream), host,
-                          latched_conn.connectionInfoProvider().localAddress(),
+  callbacks_->onPoolReady(std::move(upstream), host, latched_conn.connectionInfoProvider(),
                           latched_conn.streamInfo(), {});
 }
 

test/common/router/router_upstream_log_test.cc

@@ -156,8 +156,8 @@ class RouterUpstreamLogTest : public testing::Test {
                              const Http::ConnectionPool::Instance::StreamOptions&)
                              -> Http::ConnectionPool::Cancellable* {
           response_decoder = &decoder;
-          EXPECT_CALL(encoder.stream_, connectionLocalAddress())
-              .WillRepeatedly(ReturnRef(upstream_local_address1_));
+          EXPECT_CALL(encoder.stream_, connectionInfoProvider())
+              .WillRepeatedly(ReturnRef(connection_info1_));
           callbacks.onPoolReady(encoder,
                                 context_.cluster_manager_.thread_local_cluster_.conn_pool_.host_,
                                 stream_info_, Http::Protocol::Http10);
@@ -198,8 +198,8 @@ class RouterUpstreamLogTest : public testing::Test {
                              const Http::ConnectionPool::Instance::StreamOptions&)
                              -> Http::ConnectionPool::Cancellable* {
           response_decoder = &decoder;
-          EXPECT_CALL(encoder1.stream_, connectionLocalAddress())
-              .WillRepeatedly(ReturnRef(upstream_local_address1_));
+          EXPECT_CALL(encoder1.stream_, connectionInfoProvider())
+              .WillRepeatedly(ReturnRef(connection_info1_));
           callbacks.onPoolReady(encoder1,
                                 context_.cluster_manager_.thread_local_cluster_.conn_pool_.host_,
                                 stream_info_, Http::Protocol::Http10);
@@ -230,8 +230,8 @@ class RouterUpstreamLogTest : public testing::Test {
           EXPECT_CALL(
               context_.cluster_manager_.thread_local_cluster_.conn_pool_.host_->outlier_detector_,
               putResult(Upstream::Outlier::Result::LocalOriginConnectSuccess, _));
-          EXPECT_CALL(encoder2.stream_, connectionLocalAddress())
-              .WillRepeatedly(ReturnRef(upstream_local_address2_));
+          EXPECT_CALL(encoder2.stream_, connectionInfoProvider())
+              .WillRepeatedly(ReturnRef(connection_info2_));
           callbacks.onPoolReady(encoder2,
                                 context_.cluster_manager_.thread_local_cluster_.conn_pool_.host_,
                                 stream_info_, Http::Protocol::Http10);
@@ -263,6 +263,10 @@ class RouterUpstreamLogTest : public testing::Test {
       Network::Utility::resolveUrl("tcp://10.0.0.5:10211")};
   Network::Address::InstanceConstSharedPtr upstream_local_address2_{
       Network::Utility::resolveUrl("tcp://10.0.0.5:10212")};
+  Network::ConnectionInfoSetterImpl connection_info1_{*upstream_local_address2_,
+                                                      *upstream_local_address1_};
+  Network::ConnectionInfoSetterImpl connection_info2_{*upstream_local_address2_,
+                                                      *upstream_local_address2_};
   Event::MockTimer* response_timeout_{};
   Event::MockTimer* per_try_timeout_{};
 

test/integration/filters/stream_info_to_headers_filter.cc

@@ -50,6 +50,11 @@ class StreamInfoToHeadersFilter : public Http::PassThroughFilter {
             Http::LowerCaseString("alpn"),
             decoder_callbacks_->streamInfo().upstreamInfo()->upstreamSslConnection()->alpn());
       }
+      if (decoder_callbacks_->streamInfo().upstreamInfo()->upstreamRemoteAddress()) {
+        headers.addCopy(
+            Http::LowerCaseString("remote_address"),
+            decoder_callbacks_->streamInfo().upstreamInfo()->upstreamRemoteAddress()->asString());
+      }
 
       headers.addCopy(Http::LowerCaseString("num_streams"),
                       decoder_callbacks_->streamInfo().upstreamInfo()->upstreamNumStreams());

test/integration/multiplexed_upstream_integration_test.cc

@@ -161,6 +161,9 @@ void MultiplexedUpstreamIntegrationTest::bidirectionalStreaming(uint32_t bytes)
   EXPECT_EQ(
       "1",
       response->headers().get(Http::LowerCaseString("num_streams"))[0]->value().getStringView());
+  EXPECT_EQ(
+      fake_upstreams_[0]->localAddress()->ip()->addressAsString(),
+      response->headers().get(Http::LowerCaseString("remote_address"))[0]->value().getStringView());
 }
 
 TEST_P(MultiplexedUpstreamIntegrationTest, BidirectionalStreaming) { bidirectionalStreaming(1024); }

test/integration/upstreams/per_host_upstream_config.h

@@ -83,7 +83,7 @@ class PerHostHttpConnPool : public Extensions::Upstreams::Http::Http::HttpConnPo
     auto upstream = std::make_unique<PerHostHttpUpstream>(callbacks_->upstreamToDownstream(),
                                                           &callbacks_encoder, host);
     callbacks_->onPoolReady(std::move(upstream), host,
-                            callbacks_encoder.getStream().connectionLocalAddress(), info, protocol);
+                            callbacks_encoder.getStream().connectionInfoProvider(), info, protocol);
   }
 };
 

test/mocks/http/BUILD

@@ -85,6 +85,7 @@ envoy_cc_mock(
     hdrs = ["stream.h"],
     deps = [
         "//envoy/http:codec_interface",
+        "//source/common/network:socket_lib",
     ],
 )