router: inherit upstream connection filter state in upstream request stream info

include/envoy/stream_info/filter_state.h

@@ -30,21 +30,16 @@ class FilterState {
   //
   // - FilterChain has the shortest life span, which is as long as the filter chain lives.
   //
-  // - DownstreamRequest is longer than FilterChain. When internal redirect is enabled, one
-  //   downstream request may create multiple filter chains. DownstreamRequest allows an object to
-  //   survive across filter chains for bookkeeping needs.
+  // - Request is longer than FilterChain. When internal redirect is enabled, one
+  //   downstream request may create multiple filter chains. Request allows an object to
+  //   survive across filter chains for bookkeeping needs. This is not used for the upstream case.
   //
-  // - DownstreamConnection makes an object survive the entire duration of a downstream connection.
+  // - Connection makes an object survive the entire duration of a connection.
   //   Any stream within this connection can see the same object.
   //
   // Note that order matters in this enum because it's assumed that life span grows as enum number
   // grows.
-  enum LifeSpan {
-    FilterChain,
-    DownstreamRequest,
-    DownstreamConnection,
-    TopSpan = DownstreamConnection
-  };
+  enum LifeSpan { FilterChain, Request, Connection, TopSpan = Connection };
 
   class Object {
   public:
@@ -65,7 +60,7 @@ class FilterState {
    * @param data an owning pointer to the data to be stored.
    * @param state_type indicates whether the object is mutable or not.
    * @param life_span indicates the life span of the object: bound to the filter chain, a
-   * downstream request, or a downstream connection.
+   * request, or a connection.
    *
    * Note that it is an error to call setData() twice with the same
    * data_name, if the existing object is immutable. Similarly, it is an

source/common/http/conn_manager_impl.cc

@@ -2383,7 +2383,7 @@ bool ConnectionManagerImpl::ActiveStreamDecoderFilter::recreateStream() {
   // store any objects with a LifeSpan at or above DownstreamRequest. This is to avoid unnecessary
   // heap allocation.
   if (parent_.stream_info_.filter_state_->hasDataAtOrAboveLifeSpan(
-          StreamInfo::FilterState::LifeSpan::DownstreamRequest)) {
+          StreamInfo::FilterState::LifeSpan::Request)) {
     (*parent_.connection_manager_.streams_.begin())->stream_info_.filter_state_ =
         std::make_shared<StreamInfo::FilterStateImpl>(
             parent_.stream_info_.filter_state_->parent(),

source/common/router/router.cc

@@ -83,10 +83,9 @@ bool convertRequestHeadersForInternalRedirect(Http::RequestHeaderMap& downstream
   // Make sure that performing the redirect won't result in exceeding the configured number of
   // redirects allowed for this route.
   if (!filter_state.hasData<StreamInfo::UInt32Accessor>(NumInternalRedirectsFilterStateName)) {
-    filter_state.setData(NumInternalRedirectsFilterStateName,
-                         std::make_shared<StreamInfo::UInt32AccessorImpl>(0),
-                         StreamInfo::FilterState::StateType::Mutable,
-                         StreamInfo::FilterState::LifeSpan::DownstreamRequest);
+    filter_state.setData(
+        NumInternalRedirectsFilterStateName, std::make_shared<StreamInfo::UInt32AccessorImpl>(0),
+        StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Request);
   }
   StreamInfo::UInt32Accessor& num_internal_redirect =
       filter_state.getDataMutable<StreamInfo::UInt32Accessor>(NumInternalRedirectsFilterStateName);

source/common/router/router.h

@@ -447,7 +447,11 @@ class Filter : Logger::Loggable<Logger::Id::router>,
   TimeSource& timeSource() { return config_.timeSource(); }
   Http::Context& httpContext() { return config_.http_context_; }
 
+  // This makes it accessible to the RouterTest class.
+protected:
   FilterConfig& config_;
+
+private:
   Http::StreamDecoderFilterCallbacks* callbacks_{};
   RouteConstSharedPtr route_;
   const RouteEntry* route_entry_{};

source/common/router/upstream_request.cc

@@ -329,6 +329,8 @@ void UpstreamRequest::onPoolReady(
 
   onUpstreamHostSelected(host);
 
+  stream_info_.setUpstreamFilterState(std::make_shared<StreamInfo::FilterStateImpl>(
+      info.filterState().parent()->parent(), StreamInfo::FilterState::LifeSpan::Request));
   stream_info_.setUpstreamLocalAddress(upstream_local_address);
   parent_.callbacks_->streamInfo().setUpstreamLocalAddress(upstream_local_address);
 

source/common/stream_info/stream_info_impl.h

@@ -33,7 +33,7 @@ struct StreamInfoImpl : public StreamInfo {
         start_time_monotonic_(time_source.monotonicTime()), protocol_(protocol),
         filter_state_(std::make_shared<FilterStateImpl>(
             FilterStateImpl::LazyCreateAncestor(parent_filter_state,
-                                                FilterState::LifeSpan::DownstreamConnection),
+                                                FilterState::LifeSpan::Connection),
             FilterState::LifeSpan::FilterChain)) {}
 
   SystemTime startTime() const override { return start_time_; }

source/extensions/access_loggers/grpc/grpc_access_log_utils.cc

@@ -239,9 +239,9 @@ void Utility::extractCommonAccessLogProperties(
   }
 
   for (const auto& key : config.filter_state_objects_to_log()) {
-    if (stream_info.filterState().hasDataWithName(key)) {
+    if (stream_info.filterState()->hasDataWithName(key)) {
       const auto& obj =
-          stream_info.filterState().getDataReadOnly<StreamInfo::FilterState::Object>(key);
+          stream_info.filterState()->getDataReadOnly<StreamInfo::FilterState::Object>(key);
       ProtobufTypes::MessagePtr serialized_proto = obj.serializeAsProto();
       if (serialized_proto != nullptr) {
         auto& filter_state_objects = *common_access_log.mutable_filter_state_objects();

source/extensions/filters/network/sni_cluster/sni_cluster.cc

@@ -21,8 +21,7 @@ Network::FilterStatus SniClusterFilter::onNewConnection() {
     read_callbacks_->connection().streamInfo().filterState()->setData(
         TcpProxy::PerConnectionCluster::key(),
         std::make_unique<TcpProxy::PerConnectionCluster>(sni),
-        StreamInfo::FilterState::StateType::Mutable,
-        StreamInfo::FilterState::LifeSpan::DownstreamConnection);
+        StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Connection);
   }
 
   return Network::FilterStatus::Continue;

test/common/http/conn_manager_impl_test.cc

@@ -5380,11 +5380,11 @@ TEST_F(HttpConnectionManagerImplTest, ConnectionFilterState) {
           decoder_filters_[0]->callbacks_->streamInfo().filterState()->setData(
               "per_downstream_request", std::make_unique<SimpleType>(2),
               StreamInfo::FilterState::StateType::ReadOnly,
-              StreamInfo::FilterState::LifeSpan::DownstreamRequest);
+              StreamInfo::FilterState::LifeSpan::Request);
           decoder_filters_[0]->callbacks_->streamInfo().filterState()->setData(
               "per_downstream_connection", std::make_unique<SimpleType>(3),
               StreamInfo::FilterState::StateType::ReadOnly,
-              StreamInfo::FilterState::LifeSpan::DownstreamConnection);
+              StreamInfo::FilterState::LifeSpan::Connection);
           return FilterHeadersStatus::StopIteration;
         }));
     EXPECT_CALL(*decoder_filters_[1], decodeHeaders(_, true))

test/common/router/router_test.cc

@@ -1,5 +1,6 @@
 #include <chrono>
 #include <cstdint>
+#include <functional>
 #include <string>
 
 #include "envoy/config/core/v3/base.pb.h"
@@ -83,6 +84,8 @@ class RouterTestFilter : public Filter {
     return &downstream_connection_;
   }
 
+  FilterConfig& config() { return config_; }
+
   NiceMock<Network::MockConnection> downstream_connection_;
   MockRetryState* retry_state_{};
   bool reject_all_hosts_ = false;
@@ -304,8 +307,7 @@ class RouterTestBase : public testing::Test {
     callbacks_.streamInfo().filterState()->setData(
         "num_internal_redirects",
         std::make_shared<StreamInfo::UInt32AccessorImpl>(num_previous_redirects),
-        StreamInfo::FilterState::StateType::Mutable,
-        StreamInfo::FilterState::LifeSpan::DownstreamRequest);
+        StreamInfo::FilterState::StateType::Mutable, StreamInfo::FilterState::LifeSpan::Request);
   }
 
   void setIncludeAttemptCountInRequest(bool include) {
@@ -4332,6 +4334,58 @@ TEST_F(RouterTest, DirectResponseWithoutLocation) {
   EXPECT_EQ(1UL, config_.stats_.rq_direct_response_.value());
 }
 
+// Allows verifying the state of the upstream StreamInfo
+class TestAccessLog : public AccessLog::Instance {
+public:
+  explicit TestAccessLog(std::function<void(const StreamInfo::StreamInfo&)> func) : func_(func) {}
+
+  void log(const Http::RequestHeaderMap*, const Http::ResponseHeaderMap*,
+           const Http::ResponseTrailerMap*, const StreamInfo::StreamInfo& info) override {
+    func_(info);
+  }
+
+private:
+  std::function<void(const StreamInfo::StreamInfo&)> func_;
+};
+
+// Verifies that we propagate the upstream connection filter state to the upstream request filter
+// state.
+TEST_F(RouterTest, PropagatesUpstreamFilterState) {
+  NiceMock<Http::MockRequestEncoder> encoder;
+  Http::ResponseDecoder* response_decoder = nullptr;
+
+  // This pattern helps ensure that we're actually invoking the callback.
+  bool filter_state_verified = false;
+  router_.config().upstream_logs_.push_back(
+      std::make_shared<TestAccessLog>([&](const auto& stream_info) {
+        filter_state_verified = stream_info.upstreamFilterState()->hasDataWithName("upstream data");
+      }));
+
+  upstream_stream_info_.filterState()->setData(
+      "upstream data", std::make_unique<StreamInfo::UInt32AccessorImpl>(123),
+      StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::Connection);
+  expectResponseTimerCreate();
+  EXPECT_CALL(cm_.conn_pool_, newStream(_, _))
+      .WillOnce(Invoke(
+          [&](Http::ResponseDecoder& decoder,
+              Http::ConnectionPool::Callbacks& callbacks) -> Http::ConnectionPool::Cancellable* {
+            response_decoder = &decoder;
+            callbacks.onPoolReady(encoder, cm_.conn_pool_.host_, upstream_stream_info_);
+            return nullptr;
+          }));
+
+  Http::TestRequestHeaderMapImpl headers{};
+  HttpTestUtility::addDefaultHeaders(headers);
+  router_.decodeHeaders(headers, true);
+
+  Http::ResponseHeaderMapPtr response_headers(
+      new Http::TestResponseHeaderMapImpl{{":status", "200"}});
+  response_decoder->decodeHeaders(std::move(response_headers), true);
+  EXPECT_TRUE(verifyHostUpstreamStats(1, 0));
+
+  EXPECT_TRUE(filter_state_verified);
+}
+
 TEST_F(RouterTest, UpstreamSSLConnection) {
   NiceMock<Http::MockRequestEncoder> encoder;
   Http::ResponseDecoder* response_decoder = nullptr;

test/common/stream_info/filter_state_impl_test.cc

@@ -249,15 +249,13 @@ TEST_F(FilterStateImplTest, LifeSpanInitFromParent) {
   filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
                          FilterState::LifeSpan::FilterChain);
   filter_state().setData("test_3", std::make_unique<SimpleType>(3),
-                         FilterState::StateType::ReadOnly,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::Request);
   filter_state().setData("test_4", std::make_unique<SimpleType>(4), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::LifeSpan::Request);
   filter_state().setData("test_5", std::make_unique<SimpleType>(5),
-                         FilterState::StateType::ReadOnly,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::Connection);
   filter_state().setData("test_6", std::make_unique<SimpleType>(6), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::LifeSpan::Connection);
 
   FilterStateImpl new_filter_state(filter_state().parent(), FilterState::LifeSpan::FilterChain);
   EXPECT_FALSE(new_filter_state.hasDataWithName("test_1"));
@@ -282,15 +280,13 @@ TEST_F(FilterStateImplTest, LifeSpanInitFromGrandparent) {
   filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
                          FilterState::LifeSpan::FilterChain);
   filter_state().setData("test_3", std::make_unique<SimpleType>(3),
-                         FilterState::StateType::ReadOnly,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::Request);
   filter_state().setData("test_4", std::make_unique<SimpleType>(4), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::LifeSpan::Request);
   filter_state().setData("test_5", std::make_unique<SimpleType>(5),
-                         FilterState::StateType::ReadOnly,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::Connection);
   filter_state().setData("test_6", std::make_unique<SimpleType>(6), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::LifeSpan::Connection);
 
   FilterStateImpl new_filter_state(filter_state().parent()->parent(),
                                    FilterState::LifeSpan::FilterChain);
@@ -312,18 +308,15 @@ TEST_F(FilterStateImplTest, LifeSpanInitFromNonParent) {
   filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
                          FilterState::LifeSpan::FilterChain);
   filter_state().setData("test_3", std::make_unique<SimpleType>(3),
-                         FilterState::StateType::ReadOnly,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::Request);
   filter_state().setData("test_4", std::make_unique<SimpleType>(4), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::LifeSpan::Request);
   filter_state().setData("test_5", std::make_unique<SimpleType>(5),
-                         FilterState::StateType::ReadOnly,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::Connection);
   filter_state().setData("test_6", std::make_unique<SimpleType>(6), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::LifeSpan::Connection);
 
-  FilterStateImpl new_filter_state(filter_state().parent(),
-                                   FilterState::LifeSpan::DownstreamRequest);
+  FilterStateImpl new_filter_state(filter_state().parent(), FilterState::LifeSpan::Request);
   EXPECT_FALSE(new_filter_state.hasDataWithName("test_1"));
   EXPECT_FALSE(new_filter_state.hasDataWithName("test_2"));
   EXPECT_FALSE(new_filter_state.hasDataWithName("test_3"));
@@ -336,29 +329,25 @@ TEST_F(FilterStateImplTest, HasDataAtOrAboveLifeSpan) {
   filter_state().setData("test_1", std::make_unique<SimpleType>(1),
                          FilterState::StateType::ReadOnly, FilterState::LifeSpan::FilterChain);
   EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::FilterChain));
-  EXPECT_FALSE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamRequest));
-  EXPECT_FALSE(
-      filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamConnection));
+  EXPECT_FALSE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::Request));
+  EXPECT_FALSE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::Connection));
 
   filter_state().setData("test_2", std::make_unique<SimpleType>(2),
-                         FilterState::StateType::ReadOnly,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::Request);
   EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::FilterChain));
-  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamRequest));
-  EXPECT_FALSE(
-      filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamConnection));
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::Request));
+  EXPECT_FALSE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::Connection));
 
   filter_state().setData("test_3", std::make_unique<SimpleType>(3),
-                         FilterState::StateType::ReadOnly,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::StateType::ReadOnly, FilterState::LifeSpan::Connection);
   EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::FilterChain));
-  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamRequest));
-  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::DownstreamConnection));
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::Request));
+  EXPECT_TRUE(filter_state().hasDataAtOrAboveLifeSpan(FilterState::LifeSpan::Connection));
 }
 
 TEST_F(FilterStateImplTest, SetSameDataWithDifferentLifeSpan) {
   filter_state().setData("test_1", std::make_unique<SimpleType>(1), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::LifeSpan::Connection);
   // Test reset on smaller LifeSpan
   EXPECT_THROW_WITH_MESSAGE(
       filter_state().setData("test_1", std::make_unique<SimpleType>(2),
@@ -367,18 +356,17 @@ TEST_F(FilterStateImplTest, SetSameDataWithDifferentLifeSpan) {
       "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
   EXPECT_THROW_WITH_MESSAGE(
       filter_state().setData("test_1", std::make_unique<SimpleType>(2),
-                             FilterState::StateType::Mutable,
-                             FilterState::LifeSpan::DownstreamRequest),
+                             FilterState::StateType::Mutable, FilterState::LifeSpan::Request),
       EnvoyException,
       "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
 
   // Still mutable on the correct LifeSpan.
   filter_state().setData("test_1", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamConnection);
+                         FilterState::LifeSpan::Connection);
   EXPECT_EQ(2, filter_state().getDataMutable<SimpleType>("test_1").access());
 
   filter_state().setData("test_2", std::make_unique<SimpleType>(1), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::LifeSpan::Request);
   // Test reset on smaller and greater LifeSpan
   EXPECT_THROW_WITH_MESSAGE(
       filter_state().setData("test_2", std::make_unique<SimpleType>(2),
@@ -387,14 +375,13 @@ TEST_F(FilterStateImplTest, SetSameDataWithDifferentLifeSpan) {
       "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
   EXPECT_THROW_WITH_MESSAGE(
       filter_state().setData("test_2", std::make_unique<SimpleType>(2),
-                             FilterState::StateType::Mutable,
-                             FilterState::LifeSpan::DownstreamConnection),
+                             FilterState::StateType::Mutable, FilterState::LifeSpan::Connection),
       EnvoyException,
       "FilterState::setData<T> called twice with conflicting life_span on the same data_name.");
 
   // Still mutable on the correct LifeSpan.
   filter_state().setData("test_2", std::make_unique<SimpleType>(2), FilterState::StateType::Mutable,
-                         FilterState::LifeSpan::DownstreamRequest);
+                         FilterState::LifeSpan::Request);
   EXPECT_EQ(2, filter_state().getDataMutable<SimpleType>("test_2").access());
 }