Share filterstate between upstream and downstream filters

include/envoy/stream_info/stream_info.h

@@ -482,6 +482,14 @@ class StreamInfo {
   virtual const FilterStateSharedPtr& filterState() PURE;
   virtual const FilterState& filterState() const PURE;
 
+  /**
+   * Filter State object to be shared between upstream and downstream filters.
+   * @param pointer to upstream connections filter state.
+   * @return pointer to filter state to be used by upstream connections.
+   */
+  virtual const FilterStateSharedPtr& upstreamFilterState() const PURE;
+  virtual void setUpstreamFilterState(const FilterStateSharedPtr& filter_state) PURE;
+
   /**
    * @param SNI value requested.
    */

source/common/stream_info/stream_info_impl.h

@@ -219,6 +219,13 @@ struct StreamInfoImpl : public StreamInfo {
   const FilterStateSharedPtr& filterState() override { return filter_state_; }
   const FilterState& filterState() const override { return *filter_state_; }
 
+  const FilterStateSharedPtr& upstreamFilterState() const override {
+    return upstream_filter_state_;
+  }
+  void setUpstreamFilterState(const FilterStateSharedPtr& filter_state) override {
+    upstream_filter_state_ = filter_state;
+  }
+
   void setRequestedServerName(absl::string_view requested_server_name) override {
     requested_server_name_ = std::string(requested_server_name);
   }
@@ -262,6 +269,7 @@ struct StreamInfoImpl : public StreamInfo {
   const Router::RouteEntry* route_entry_{};
   envoy::config::core::v3::Metadata metadata_{};
   FilterStateSharedPtr filter_state_;
+  FilterStateSharedPtr upstream_filter_state_;
   std::string route_name_;
 
 private:

source/common/tcp_proxy/tcp_proxy.cc

@@ -467,6 +467,8 @@ void Filter::onPoolReady(Tcp::ConnectionPool::ConnectionDataPtr&& conn_data,
   getStreamInfo().onUpstreamHostSelected(host);
   getStreamInfo().setUpstreamLocalAddress(connection.localAddress());
   getStreamInfo().setUpstreamSslConnection(connection.streamInfo().downstreamSslConnection());
+  read_callbacks_->connection().streamInfo().setUpstreamFilterState(
+      connection.streamInfo().filterState());
 
   // Simulate the event that onPoolReady represents.
   upstream_callbacks_->onEvent(Network::ConnectionEvent::Connected);

test/common/stream_info/stream_info_impl_test.cc

@@ -167,6 +167,10 @@ TEST_F(StreamInfoImplTest, MiscSettersAndGetters) {
                                        FilterState::LifeSpan::FilterChain);
     EXPECT_EQ(1, stream_info.filterState()->getDataReadOnly<TestIntAccessor>("test").access());
 
+    stream_info.setUpstreamFilterState(stream_info.filterState());
+    EXPECT_EQ(1,
+              stream_info.upstreamFilterState()->getDataReadOnly<TestIntAccessor>("test").access());
+
     EXPECT_EQ("", stream_info.requestedServerName());
     absl::string_view sni_name = "stubserver.org";
     stream_info.setRequestedServerName(sni_name);

test/common/stream_info/test_util.h

@@ -179,6 +179,14 @@ class TestStreamInfo : public StreamInfo::StreamInfo {
   const Envoy::StreamInfo::FilterStateSharedPtr& filterState() override { return filter_state_; }
   const Envoy::StreamInfo::FilterState& filterState() const override { return *filter_state_; }
 
+  const Envoy::StreamInfo::FilterStateSharedPtr& upstreamFilterState() const override {
+    return upstream_filter_state_;
+  }
+  void
+  setUpstreamFilterState(const Envoy::StreamInfo::FilterStateSharedPtr& filter_state) override {
+    upstream_filter_state_ = filter_state;
+  }
+
   void setRequestedServerName(const absl::string_view requested_server_name) override {
     requested_server_name_ = std::string(requested_server_name);
   }
@@ -229,6 +237,7 @@ class TestStreamInfo : public StreamInfo::StreamInfo {
   Envoy::StreamInfo::FilterStateSharedPtr filter_state_{
       std::make_shared<Envoy::StreamInfo::FilterStateImpl>(
           Envoy::StreamInfo::FilterState::LifeSpan::FilterChain)};
+  Envoy::StreamInfo::FilterStateSharedPtr upstream_filter_state_;
   Envoy::StreamInfo::UpstreamTiming upstream_timing_;
   std::string requested_server_name_;
   std::string upstream_transport_failure_reason_;

test/common/tcp_proxy/tcp_proxy_test.cc

@@ -1694,6 +1694,22 @@ TEST_F(TcpProxyTest, DEPRECATED_FEATURE_TEST(UpstreamFlushReceiveUpstreamData))
   upstream_callbacks_->onUpstreamData(buffer, false);
 }
 
+// Tests that downstream connection can access upstream connections filter state.
+TEST_F(TcpProxyTest, ShareFilterState) {
+  setup(1);
+
+  upstream_connections_.at(0)->streamInfo().filterState()->setData(
+      "envoy.tcp_proxy.cluster", std::make_unique<PerConnectionCluster>("filter_state_cluster"),
+      StreamInfo::FilterState::StateType::Mutable,
+      StreamInfo::FilterState::LifeSpan::DownstreamConnection);
+  raiseEventUpstreamConnected(0);
+  EXPECT_EQ("filter_state_cluster",
+            filter_callbacks_.connection_.streamInfo()
+                .upstreamFilterState()
+                ->getDataReadOnly<PerConnectionCluster>("envoy.tcp_proxy.cluster")
+                .value());
+}
+
 class TcpProxyRoutingTest : public testing::Test {
 public:
   TcpProxyRoutingTest() = default;

test/mocks/stream_info/mocks.cc

@@ -89,6 +89,11 @@ MockStreamInfo::MockStreamInfo()
   ON_CALL(Const(*this), dynamicMetadata()).WillByDefault(ReturnRef(metadata_));
   ON_CALL(*this, filterState()).WillByDefault(ReturnRef(filter_state_));
   ON_CALL(Const(*this), filterState()).WillByDefault(ReturnRef(*filter_state_));
+  ON_CALL(*this, upstreamFilterState()).WillByDefault(ReturnRef(upstream_filter_state_));
+  ON_CALL(*this, setUpstreamFilterState(_))
+      .WillByDefault(Invoke([this](const FilterStateSharedPtr& filter_state) {
+        upstream_filter_state_ = filter_state;
+      }));
   ON_CALL(*this, setRequestedServerName(_))
       .WillByDefault(Invoke([this](const absl::string_view requested_server_name) {
         requested_server_name_ = std::string(requested_server_name);

test/mocks/stream_info/mocks.h

@@ -80,6 +80,8 @@ class MockStreamInfo : public StreamInfo {
               (const std::string&, const std::string&, const std::string&));
   MOCK_METHOD(const FilterStateSharedPtr&, filterState, ());
   MOCK_METHOD(const FilterState&, filterState, (), (const));
+  MOCK_METHOD(const FilterStateSharedPtr&, upstreamFilterState, (), (const));
+  MOCK_METHOD(void, setUpstreamFilterState, (const FilterStateSharedPtr&));
   MOCK_METHOD(void, setRequestedServerName, (const absl::string_view));
   MOCK_METHOD(const std::string&, requestedServerName, (), (const));
   MOCK_METHOD(void, setUpstreamTransportFailureReason, (absl::string_view));
@@ -103,6 +105,7 @@ class MockStreamInfo : public StreamInfo {
   absl::optional<uint32_t> response_code_;
   absl::optional<std::string> response_code_details_;
   envoy::config::core::v3::Metadata metadata_;
+  FilterStateSharedPtr upstream_filter_state_;
   FilterStateSharedPtr filter_state_;
   uint64_t bytes_received_{};
   uint64_t bytes_sent_{};