Make %UPSTREAM_LOCAL_ADDRESS% access-log format work for HTTP requests.

docs/root/intro/version_history.rst

@@ -5,6 +5,7 @@ Version history
 ================
 * access log: added FILTER_STATE :ref:`access log formatters <config_access_log_format>` and gRPC access logger.
 * access log: added a :ref:`typed JSON logging mode <config_access_log_format_dictionaries>` to output access logs in JSON format with non-string values
+* access log: fixed UPSTREAM_LOCAL_ADDRESS :ref:`access log formatters <config_access_log_format>` to work for http requests
 * api: remove all support for v1
 * buffer: remove old implementation
 * build: official released binary is now built against libc++.

include/envoy/http/BUILD

@@ -29,6 +29,7 @@ envoy_cc_library(
         ":metadata_interface",
         ":protocol_interface",
         "//include/envoy/buffer:buffer_interface",
+        "//include/envoy/network:address_interface",
     ],
 )
 

include/envoy/http/codec.h

@@ -9,6 +9,7 @@
 #include "envoy/http/header_map.h"
 #include "envoy/http/metadata_interface.h"
 #include "envoy/http/protocol.h"
+#include "envoy/network/address.h"
 
 namespace Envoy {
 namespace Http {
@@ -212,6 +213,12 @@ class Stream {
    * not associated with every stream on the connection.
    */
   virtual absl::string_view responseDetails() { return ""; }
+
+  /*
+   * @return const Address::InstanceConstSharedPtr& the local address of the connection associated
+   * with the stream.
+   */
+  virtual const Network::Address::InstanceConstSharedPtr& connectionLocalAddress() PURE;
 };
 
 /**

source/common/http/http1/codec_impl.cc

@@ -269,6 +269,10 @@ void StreamEncoderImpl::readDisable(bool disable) { connection_.readDisable(disa
 
 uint32_t StreamEncoderImpl::bufferLimit() { return connection_.bufferLimit(); }
 
+const Network::Address::InstanceConstSharedPtr& StreamEncoderImpl::connectionLocalAddress() {
+  return connection_.connection().localAddress();
+}
+
 static const char RESPONSE_PREFIX[] = "HTTP/1.1 ";
 static const char HTTP_10_RESPONSE_PREFIX[] = "HTTP/1.0 ";
 

source/common/http/http1/codec_impl.h

@@ -61,6 +61,7 @@ class StreamEncoderImpl : public StreamEncoder,
   void readDisable(bool disable) override;
   uint32_t bufferLimit() override;
   absl::string_view responseDetails() override { return details_; }
+  const Network::Address::InstanceConstSharedPtr& connectionLocalAddress() override;
 
   void isResponseToHeadRequest(bool value) { is_response_to_head_request_ = value; }
   void setDetails(absl::string_view details) { details_ = details; }

source/common/http/http2/codec_impl.h

@@ -173,6 +173,9 @@ class ConnectionImpl : public virtual Connection, protected Logger::Loggable<Log
     void resetStream(StreamResetReason reason) override;
     void readDisable(bool disable) override;
     uint32_t bufferLimit() override { return pending_recv_data_.highWatermark(); }
+    const Network::Address::InstanceConstSharedPtr& connectionLocalAddress() override {
+      return parent_.connection_.localAddress();
+    }
 
     void setWriteBufferWatermarks(uint32_t low_watermark, uint32_t high_watermark) {
       pending_recv_data_.setWatermarks(low_watermark, high_watermark);

source/common/router/router.cc

@@ -1615,10 +1615,13 @@ void Filter::UpstreamRequest::onPoolReady(Http::StreamEncoder& request_encoder,
 
   host->outlierDetector().putResult(Upstream::Outlier::Result::LocalOriginConnectSuccess);
 
-  // TODO(ggreenway): set upstream local address in the StreamInfo.
   onUpstreamHostSelected(host);
   request_encoder.getStream().addCallbacks(*this);
 
+  stream_info_.setUpstreamLocalAddress(request_encoder.getStream().connectionLocalAddress());
+  parent_.callbacks_->streamInfo().setUpstreamLocalAddress(
+      request_encoder.getStream().connectionLocalAddress());
+
   stream_info_.setUpstreamSslConnection(info.downstreamSslConnection());
   parent_.callbacks_->streamInfo().setUpstreamSslConnection(info.downstreamSslConnection());
 

source/extensions/quic_listeners/quiche/envoy_quic_stream.h

@@ -76,6 +76,9 @@ class EnvoyQuicStream : public Http::StreamEncoder,
   }
   void removeCallbacks(Http::StreamCallbacks& callbacks) override { removeCallbacks_(callbacks); }
   uint32_t bufferLimit() override { return send_buffer_simulation_.highWatermark(); }
+  const Network::Address::InstanceConstSharedPtr& connectionLocalAddress() override {
+    return connection()->localAddress();
+  }
 
   // Needs to be called during quic stream creation before the stream receives
   // any headers and data.

test/common/router/router_test.cc

@@ -4374,7 +4374,7 @@ class WatermarkTest : public RouterTest {
     EXPECT_CALL(stream_, addCallbacks(_)).WillOnce(Invoke([&](Http::StreamCallbacks& callbacks) {
       stream_callbacks_ = &callbacks;
     }));
-    EXPECT_CALL(encoder_, getStream()).WillOnce(ReturnRef(stream_));
+    EXPECT_CALL(encoder_, getStream()).WillRepeatedly(ReturnRef(stream_));
     EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
         .WillOnce(Invoke(
             [&](Http::StreamDecoder& decoder,

test/common/router/router_upstream_log_test.cc

@@ -43,7 +43,7 @@ name: envoy.file_access_log
   "@type": type.googleapis.com/envoy.config.accesslog.v2.FileAccessLog
   format: "%REQ(:METHOD)% %REQ(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL% %RESPONSE_CODE%
     %RESPONSE_FLAGS% %BYTES_RECEIVED% %BYTES_SENT% %REQ(:AUTHORITY)% %UPSTREAM_HOST%
-    %RESP(X-UPSTREAM-HEADER)% %TRAILER(X-TRAILER)%\n"
+    %UPSTREAM_LOCAL_ADDRESS% %RESP(X-UPSTREAM-HEADER)% %TRAILER(X-TRAILER)%\n"
   path: "/dev/null"
   )EOF";
 
@@ -133,6 +133,8 @@ class RouterUpstreamLogTest : public testing::Test {
             [&](Http::StreamDecoder& decoder,
                 Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
               response_decoder = &decoder;
+              EXPECT_CALL(encoder.stream_, connectionLocalAddress())
+                  .WillRepeatedly(ReturnRef(upstream_local_address1_));
               callbacks.onPoolReady(encoder, context_.cluster_manager_.conn_pool_.host_,
                                     stream_info_);
               return nullptr;
@@ -161,11 +163,14 @@ class RouterUpstreamLogTest : public testing::Test {
   void runWithRetry() {
     NiceMock<Http::MockStreamEncoder> encoder1;
     Http::StreamDecoder* response_decoder = nullptr;
+
     EXPECT_CALL(context_.cluster_manager_.conn_pool_, newStream(_, _))
         .WillOnce(Invoke(
             [&](Http::StreamDecoder& decoder,
                 Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
               response_decoder = &decoder;
+              EXPECT_CALL(encoder1.stream_, connectionLocalAddress())
+                  .WillRepeatedly(ReturnRef(upstream_local_address1_));
               callbacks.onPoolReady(encoder1, context_.cluster_manager_.conn_pool_.host_,
                                     stream_info_);
               return nullptr;
@@ -193,6 +198,8 @@ class RouterUpstreamLogTest : public testing::Test {
               response_decoder = &decoder;
               EXPECT_CALL(context_.cluster_manager_.conn_pool_.host_->outlier_detector_,
                           putResult(Upstream::Outlier::Result::LocalOriginConnectSuccess, _));
+              EXPECT_CALL(encoder2.stream_, connectionLocalAddress())
+                  .WillRepeatedly(ReturnRef(upstream_local_address2_));
               callbacks.onPoolReady(encoder2, context_.cluster_manager_.conn_pool_.host_,
                                     stream_info_);
               return nullptr;
@@ -215,6 +222,10 @@ class RouterUpstreamLogTest : public testing::Test {
   envoy::api::v2::core::Locality upstream_locality_;
   Network::Address::InstanceConstSharedPtr host_address_{
       Network::Utility::resolveUrl("tcp://10.0.0.5:9211")};
+  Network::Address::InstanceConstSharedPtr upstream_local_address1_{
+      Network::Utility::resolveUrl("tcp://10.0.0.5:10211")};
+  Network::Address::InstanceConstSharedPtr upstream_local_address2_{
+      Network::Utility::resolveUrl("tcp://10.0.0.5:10212")};
   Event::MockTimer* response_timeout_{};
   Event::MockTimer* per_try_timeout_{};
 
@@ -236,24 +247,24 @@ TEST_F(RouterUpstreamLogTest, LogSingleTry) {
   run();
 
   EXPECT_EQ(output_.size(), 1U);
-  EXPECT_EQ(output_.front(), "GET / HTTP/1.0 200 - 0 0 host 10.0.0.5:9211 - -\n");
+  EXPECT_EQ(output_.front(), "GET / HTTP/1.0 200 - 0 0 host 10.0.0.5:9211 10.0.0.5:10211 - -\n");
 }
 
 TEST_F(RouterUpstreamLogTest, LogRetries) {
   init(testUpstreamLog());
   runWithRetry();
 
   EXPECT_EQ(output_.size(), 2U);
-  EXPECT_EQ(output_.front(), "GET / HTTP/1.0 0 UT 0 0 host 10.0.0.5:9211 - -\n");
-  EXPECT_EQ(output_.back(), "GET / HTTP/1.0 200 - 0 0 host 10.0.0.5:9211 - -\n");
+  EXPECT_EQ(output_.front(), "GET / HTTP/1.0 0 UT 0 0 host 10.0.0.5:9211 10.0.0.5:10211 - -\n");
+  EXPECT_EQ(output_.back(), "GET / HTTP/1.0 200 - 0 0 host 10.0.0.5:9211 10.0.0.5:10212 - -\n");
 }
 
 TEST_F(RouterUpstreamLogTest, LogFailure) {
   init(testUpstreamLog());
   run(503, {}, {}, {});
 
   EXPECT_EQ(output_.size(), 1U);
-  EXPECT_EQ(output_.front(), "GET / HTTP/1.0 503 - 0 0 host 10.0.0.5:9211 - -\n");
+  EXPECT_EQ(output_.front(), "GET / HTTP/1.0 503 - 0 0 host 10.0.0.5:9211 10.0.0.5:10211 - -\n");
 }
 
 TEST_F(RouterUpstreamLogTest, LogHeaders) {
@@ -262,7 +273,8 @@ TEST_F(RouterUpstreamLogTest, LogHeaders) {
       {{"x-trailer", "value"}});
 
   EXPECT_EQ(output_.size(), 1U);
-  EXPECT_EQ(output_.front(), "GET /foo HTTP/1.0 200 - 0 0 host 10.0.0.5:9211 abcdef value\n");
+  EXPECT_EQ(output_.front(),
+            "GET /foo HTTP/1.0 200 - 0 0 host 10.0.0.5:9211 10.0.0.5:10211 abcdef value\n");
 }
 
 // Test timestamps and durations are emitted.

test/fuzz/common.proto

@@ -22,4 +22,5 @@ message StreamInfo {
   envoy.api.v2.core.Metadata upstream_metadata = 4;
   string requested_server_name = 5;
   envoy.api.v2.core.Address address = 6;
+  envoy.api.v2.core.Address upstream_local_address = 7;
 }

test/fuzz/utility.h

@@ -138,7 +138,11 @@ inline TestStreamInfo fromStreamInfo(const test::fuzz::StreamInfo& stream_info)
   auto address = stream_info.has_address()
                      ? Envoy::Network::Address::resolveProtoAddress(stream_info.address())
                      : Network::Utility::resolveUrl("tcp://10.0.0.1:443");
-  test_stream_info.upstream_local_address_ = address;
+  auto upstream_local_address =
+      stream_info.has_upstream_local_address()
+          ? Envoy::Network::Address::resolveProtoAddress(stream_info.upstream_local_address())
+          : Network::Utility::resolveUrl("tcp://10.0.0.1:10000");
+  test_stream_info.upstream_local_address_ = upstream_local_address;
   test_stream_info.downstream_local_address_ = address;
   test_stream_info.downstream_direct_remote_address_ = address;
   test_stream_info.downstream_remote_address_ = address;

test/integration/integration_test.cc

@@ -743,6 +743,7 @@ TEST_P(IntegrationTest, TestBind) {
     address_string = "::1";
   }
   config_helper_.setSourceAddress(address_string);
+  useAccessLog("%UPSTREAM_LOCAL_ADDRESS%\n");
   initialize();
 
   codec_client_ = makeHttpConnection(lookupPort("http"));
@@ -763,6 +764,7 @@ TEST_P(IntegrationTest, TestBind) {
   ASSERT_TRUE(upstream_request_->waitForEndStream(*dispatcher_));
 
   cleanupUpstreamAndDownstream();
+  EXPECT_THAT(waitForAccessLog(access_log_name_), HasSubstr(address_string));
 }
 
 TEST_P(IntegrationTest, TestFailedBind) {

test/mocks/http/stream.cc

@@ -2,6 +2,7 @@
 
 using testing::_;
 using testing::Invoke;
+using testing::ReturnRef;
 
 namespace Envoy {
 namespace Http {
@@ -20,6 +21,8 @@ MockStream::MockStream() {
       callbacks->onResetStream(reason, absl::string_view());
     }
   }));
+
+  ON_CALL(*this, connectionLocalAddress()).WillByDefault(ReturnRef(connection_local_address_));
 }
 
 MockStream::~MockStream() = default;

test/mocks/http/stream.h

@@ -19,8 +19,10 @@ class MockStream : public Stream {
   MOCK_METHOD1(readDisable, void(bool disable));
   MOCK_METHOD2(setWriteBufferWatermarks, void(uint32_t, uint32_t));
   MOCK_METHOD0(bufferLimit, uint32_t());
+  MOCK_METHOD0(connectionLocalAddress, const Network::Address::InstanceConstSharedPtr&());
 
   std::list<StreamCallbacks*> callbacks_{};
+  Network::Address::InstanceConstSharedPtr connection_local_address_;
 
   void runHighWatermarkCallbacks() {
     for (auto* callback : callbacks_) {