filter state: add a mark for sharing with the upstream

envoy/event/dispatcher.h

@@ -210,13 +210,15 @@ class Dispatcher : public DispatcherBase, public ScopeTracker {
    * @param transport_socket supplies a transport socket to be used by the connection.
    * @param options the socket options to be set on the underlying socket before anything is sent
    *        on the socket.
+   * @param transport socket options used to create the transport socket.
    * @return Network::ClientConnectionPtr a client connection that is owned by the caller.
    */
-  virtual Network::ClientConnectionPtr
-  createClientConnection(Network::Address::InstanceConstSharedPtr address,
-                         Network::Address::InstanceConstSharedPtr source_address,
-                         Network::TransportSocketPtr&& transport_socket,
-                         const Network::ConnectionSocket::OptionsSharedPtr& options) PURE;
+  virtual Network::ClientConnectionPtr createClientConnection(
+      Network::Address::InstanceConstSharedPtr address,
+      Network::Address::InstanceConstSharedPtr source_address,
+      Network::TransportSocketPtr&& transport_socket,
+      const Network::ConnectionSocket::OptionsSharedPtr& options,
+      Network::TransportSocketOptionsConstSharedPtr transport_options = nullptr) PURE;
 
   /**
    * @return Filesystem::WatcherPtr a filesystem watcher owned by the caller.

envoy/network/client_connection_factory.h

@@ -26,12 +26,12 @@ class ClientConnectionFactory : public Config::UntypedFactory {
    * @return Network::ClientConnectionPtr The created connection. It's never nullptr but the return
    * connection may be closed upon return.
    */
-  virtual Network::ClientConnectionPtr
-  createClientConnection(Event::Dispatcher& dispatcher,
-                         Network::Address::InstanceConstSharedPtr address,
-                         Network::Address::InstanceConstSharedPtr source_address,
-                         Network::TransportSocketPtr&& transport_socket,
-                         const Network::ConnectionSocket::OptionsSharedPtr& options) PURE;
+  virtual Network::ClientConnectionPtr createClientConnection(
+      Event::Dispatcher& dispatcher, Network::Address::InstanceConstSharedPtr address,
+      Network::Address::InstanceConstSharedPtr source_address,
+      Network::TransportSocketPtr&& transport_socket,
+      const Network::ConnectionSocket::OptionsSharedPtr& options,
+      Network::TransportSocketOptionsConstSharedPtr transport_options = nullptr) PURE;
 };
 
 } // namespace Network

envoy/network/transport_socket.h

@@ -231,9 +231,10 @@ class TransportSocketOptions {
   virtual absl::optional<Network::ProxyProtocolData> proxyProtocolOptions() const PURE;
 
   /**
-   * @return filter state from the downstream request or connection.
+   * @return filter state objects from the downstream request or connection
+   * that are marked as shared with the upstream.
    */
-  virtual const StreamInfo::FilterStateSharedPtr& filterState() const PURE;
+  virtual const StreamInfo::FilterState::Objects& filterStateObjects() const PURE;
 };
 
 using TransportSocketOptionsConstSharedPtr = std::shared_ptr<const TransportSocketOptions>;

envoy/stream_info/filter_state.h

@@ -26,7 +26,24 @@ using FilterStateSharedPtr = std::shared_ptr<FilterState>;
  */
 class FilterState {
 public:
-  enum class StateType { ReadOnly, Mutable };
+  enum StateType {
+    // Readonly objects must be set once.
+    ReadOnly = 0x1,
+
+    // Mutable objects can be overwritten with the same state type, life span, and name.
+    Mutable = 0,
+
+    // Mark a filter state object as shared with the upstream connection.
+    // Shared filter state objects are copied by reference from the downstream
+    // requests and connections to the upstream connection filter state, and
+    // also participate in the hashing decisions in the transport socket
+    // options when connections are re-used between downstream requests.
+    SharedWithUpstreamConnection = 0x2,
+
+    // ATTENTION: MAKE SURE THIS REMAINS EQUAL TO THE LAST FLAG.
+    LastFlag = SharedWithUpstreamConnection,
+  };
+
   // Objects stored in the FilterState may have different life span. Life span is what controls
   // how long an object stored in FilterState lives. Implementation of this interface actually
   // stores objects in a (reverse) tree manner - multiple FilterStateImpl with shorter life span may
@@ -45,6 +62,16 @@ class FilterState {
   //
   // Note that order matters in this enum because it's assumed that life span grows as enum number
   // grows.
+  //
+  // Note that for more accurate book-keeping it is recommended to subscribe to
+  // the stream callbacks instead of relying on the destruction of the filter
+  // state.
+  //
+  // As a special case, objects that are marked as shared with the upstream
+  // become bound to the upstream connection life span, regardless of the
+  // original life span. That means, for example, that a shared request span
+  // object may outlive the original request when it is shared, because it may
+  // be captured by an upstream connection for the original downstream request.
   enum LifeSpan { FilterChain, Request, Connection, TopSpan = Connection };
 
   class Object {
@@ -66,6 +93,15 @@ class FilterState {
     virtual absl::optional<std::string> serializeAsString() const { return absl::nullopt; }
   };
 
+  struct FilterObject {
+    std::shared_ptr<Object> data_;
+    FilterState::StateType state_type_;
+    std::string name_{""};
+  };
+
+  using Objects = std::vector<FilterObject>;
+  using ObjectsPtr = std::unique_ptr<Objects>;
+
   virtual ~FilterState() = default;
 
   /**
@@ -156,6 +192,11 @@ class FilterState {
    * either the top LifeSpan or the parent is not yet created.
    */
   virtual FilterStateSharedPtr parent() const PURE;
+
+  /**
+   * @return filter objects that are shared with the upstream connection.
+   **/
+  virtual ObjectsPtr objectsSharedWithUpstreamConnection() const PURE;
 };
 
 } // namespace StreamInfo

source/common/event/dispatcher_impl.cc

@@ -155,11 +155,12 @@ DispatcherImpl::createServerConnection(Network::ConnectionSocketPtr&& socket,
       *this, std::move(socket), std::move(transport_socket), stream_info, true);
 }
 
-Network::ClientConnectionPtr
-DispatcherImpl::createClientConnection(Network::Address::InstanceConstSharedPtr address,
-                                       Network::Address::InstanceConstSharedPtr source_address,
-                                       Network::TransportSocketPtr&& transport_socket,
-                                       const Network::ConnectionSocket::OptionsSharedPtr& options) {
+Network::ClientConnectionPtr DispatcherImpl::createClientConnection(
+    Network::Address::InstanceConstSharedPtr address,
+    Network::Address::InstanceConstSharedPtr source_address,
+    Network::TransportSocketPtr&& transport_socket,
+    const Network::ConnectionSocket::OptionsSharedPtr& options,
+    Network::TransportSocketOptionsConstSharedPtr transport_options) {
   ASSERT(isThreadSafe());
 
   auto* factory = Config::Utility::getFactoryByName<Network::ClientConnectionFactory>(
@@ -170,7 +171,8 @@ DispatcherImpl::createClientConnection(Network::Address::InstanceConstSharedPtr
   // expects a non-null connection as of today so we cannot gracefully handle unsupported address
   // type.
   return factory->createClientConnection(*this, address, source_address,
-                                         std::move(transport_socket), options);
+                                         std::move(transport_socket), options,
+                                         std::move(transport_options));
 }
 
 FileEventPtr DispatcherImpl::createFileEvent(os_fd_t fd, FileReadyCb cb, FileTriggerType trigger,

source/common/event/dispatcher_impl.h

@@ -66,11 +66,12 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
   createServerConnection(Network::ConnectionSocketPtr&& socket,
                          Network::TransportSocketPtr&& transport_socket,
                          StreamInfo::StreamInfo& stream_info) override;
-  Network::ClientConnectionPtr
-  createClientConnection(Network::Address::InstanceConstSharedPtr address,
-                         Network::Address::InstanceConstSharedPtr source_address,
-                         Network::TransportSocketPtr&& transport_socket,
-                         const Network::ConnectionSocket::OptionsSharedPtr& options) override;
+  Network::ClientConnectionPtr createClientConnection(
+      Network::Address::InstanceConstSharedPtr address,
+      Network::Address::InstanceConstSharedPtr source_address,
+      Network::TransportSocketPtr&& transport_socket,
+      const Network::ConnectionSocket::OptionsSharedPtr& options,
+      Network::TransportSocketOptionsConstSharedPtr transport_options = nullptr) override;
   FileEventPtr createFileEvent(os_fd_t fd, FileReadyCb cb, FileTriggerType trigger,
                                uint32_t events) override;
   Filesystem::WatcherPtr createFilesystemWatcher() override;

source/common/network/BUILD

@@ -442,6 +442,7 @@ envoy_cc_library(
         ":proxy_protocol_filter_state_lib",
         ":upstream_server_name_lib",
         ":upstream_subject_alt_names_lib",
+        "//envoy/common:hashable_interface",
         "//envoy/network:proxy_protocol_options_lib",
         "//envoy/network:transport_socket_interface",
         "//envoy/stream_info:filter_state_interface",

source/common/network/connection_impl.cc

@@ -855,15 +855,18 @@ ClientConnectionImpl::ClientConnectionImpl(
     Event::Dispatcher& dispatcher, const Address::InstanceConstSharedPtr& remote_address,
     const Network::Address::InstanceConstSharedPtr& source_address,
     Network::TransportSocketPtr&& transport_socket,
-    const Network::ConnectionSocket::OptionsSharedPtr& options)
+    const Network::ConnectionSocket::OptionsSharedPtr& options,
+    const Network::TransportSocketOptionsConstSharedPtr& transport_options)
     : ClientConnectionImpl(dispatcher, std::make_unique<ClientSocketImpl>(remote_address, options),
-                           source_address, std::move(transport_socket), options) {}
+                           source_address, std::move(transport_socket), options,
+                           transport_options) {}
 
 ClientConnectionImpl::ClientConnectionImpl(
     Event::Dispatcher& dispatcher, std::unique_ptr<ConnectionSocket> socket,
     const Address::InstanceConstSharedPtr& source_address,
     Network::TransportSocketPtr&& transport_socket,
-    const Network::ConnectionSocket::OptionsSharedPtr& options)
+    const Network::ConnectionSocket::OptionsSharedPtr& options,
+    const Network::TransportSocketOptionsConstSharedPtr& transport_options)
     : ConnectionImpl(dispatcher, std::move(socket), std::move(transport_socket), stream_info_,
                      false),
       stream_info_(dispatcher_.timeSource(), socket_->connectionInfoProviderSharedPtr()) {
@@ -905,6 +908,14 @@ ClientConnectionImpl::ClientConnectionImpl(
       ioHandle().activateFileEvents(Event::FileReadyType::Write);
     }
   }
+
+  if (transport_options) {
+    for (const auto& object : transport_options->filterStateObjects()) {
+      // TODO: handle exception
+      stream_info_.filterState()->setData(object.name_, object.data_, object.state_type_,
+                                          StreamInfo::FilterState::LifeSpan::Connection);
+    }
+  }
 }
 
 void ClientConnectionImpl::connect() {

source/common/network/connection_impl.h

@@ -262,11 +262,14 @@ class ClientConnectionImpl : public ConnectionImpl, virtual public ClientConnect
                        const Address::InstanceConstSharedPtr& remote_address,
                        const Address::InstanceConstSharedPtr& source_address,
                        Network::TransportSocketPtr&& transport_socket,
-                       const Network::ConnectionSocket::OptionsSharedPtr& options);
+                       const Network::ConnectionSocket::OptionsSharedPtr& options,
+                       const Network::TransportSocketOptionsConstSharedPtr& transport_options);
+
   ClientConnectionImpl(Event::Dispatcher& dispatcher, std::unique_ptr<ConnectionSocket> socket,
                        const Address::InstanceConstSharedPtr& source_address,
                        Network::TransportSocketPtr&& transport_socket,
-                       const Network::ConnectionSocket::OptionsSharedPtr& options);
+                       const Network::ConnectionSocket::OptionsSharedPtr& options,
+                       const Network::TransportSocketOptionsConstSharedPtr& transport_options);
 
   // Network::ClientConnection
   void connect() override;

source/common/network/default_client_connection_factory.cc

@@ -13,10 +13,11 @@ Network::ClientConnectionPtr DefaultClientConnectionFactory::createClientConnect
     Event::Dispatcher& dispatcher, Network::Address::InstanceConstSharedPtr address,
     Network::Address::InstanceConstSharedPtr source_address,
     Network::TransportSocketPtr&& transport_socket,
-    const Network::ConnectionSocket::OptionsSharedPtr& options) {
+    const Network::ConnectionSocket::OptionsSharedPtr& options,
+    Network::TransportSocketOptionsConstSharedPtr transport_options) {
   ASSERT(address->ip() || address->pipe());
-  return std::make_unique<Network::ClientConnectionImpl>(dispatcher, address, source_address,
-                                                         std::move(transport_socket), options);
+  return std::make_unique<Network::ClientConnectionImpl>(
+      dispatcher, address, source_address, std::move(transport_socket), options, transport_options);
 }
 REGISTER_FACTORY(DefaultClientConnectionFactory, Network::ClientConnectionFactory);
 

source/common/network/default_client_connection_factory.h

@@ -20,12 +20,12 @@ class DefaultClientConnectionFactory : public ClientConnectionFactory {
   std::string name() const override { return "default"; }
 
   // Network::ClientConnectionFactory
-  Network::ClientConnectionPtr
-  createClientConnection(Event::Dispatcher& dispatcher,
-                         Network::Address::InstanceConstSharedPtr address,
-                         Network::Address::InstanceConstSharedPtr source_address,
-                         Network::TransportSocketPtr&& transport_socket,
-                         const Network::ConnectionSocket::OptionsSharedPtr& options) override;
+  Network::ClientConnectionPtr createClientConnection(
+      Event::Dispatcher& dispatcher, Network::Address::InstanceConstSharedPtr address,
+      Network::Address::InstanceConstSharedPtr source_address,
+      Network::TransportSocketPtr&& transport_socket,
+      const Network::ConnectionSocket::OptionsSharedPtr& options,
+      Network::TransportSocketOptionsConstSharedPtr transport_options = nullptr) override;
 };
 
 } // namespace Network

source/common/network/happy_eyeballs_connection_impl.cc

@@ -433,7 +433,8 @@ ClientConnectionPtr HappyEyeballsConnectionImpl::createNextConnection() {
       address_list_[next_address_++], connection_construction_state_.source_address_,
       connection_construction_state_.socket_factory_.createTransportSocket(
           connection_construction_state_.transport_socket_options_),
-      connection_construction_state_.options_);
+      connection_construction_state_.options_,
+      connection_construction_state_.transport_socket_options_);
   ENVOY_LOG_EVENT(debug, "happy_eyeballs_cx_attempt", "C[{}] address={}", id_, next_address_);
   callbacks_wrappers_.push_back(std::make_unique<ConnectionCallbacksWrapper>(*this, *connection));
   connection->addConnectionCallbacks(*callbacks_wrappers_.back());

source/common/network/transport_socket_options_impl.cc

@@ -6,6 +6,8 @@
 #include <utility>
 #include <vector>
 
+#include "envoy/common/hashable.h"
+
 #include "source/common/common/scalar_to_byte_vector.h"
 #include "source/common/common/utility.h"
 #include "source/common/network/application_protocol.h"
@@ -40,6 +42,14 @@ void CommonTransportSocketFactory::hashKey(std::vector<uint8_t>& key,
   for (const auto& protocol : alpn_fallback) {
     pushScalarToByteVector(StringUtil::CaseInsensitiveHash()(protocol), key);
   }
+
+  for (const auto& object : options->filterStateObjects()) {
+    if (auto hashable = dynamic_cast<const Hashable*>(object.data_.get()); hashable != nullptr) {
+      if (auto hash = hashable->hash(); hash) {
+        pushScalarToByteVector(hash.value(), key);
+      }
+    }
+  }
 }
 
 TransportSocketOptionsConstSharedPtr TransportSocketOptionsUtility::fromFilterState(
@@ -79,7 +89,8 @@ TransportSocketOptionsConstSharedPtr TransportSocketOptionsUtility::fromFilterSt
 
   return std::make_shared<Network::TransportSocketOptionsImpl>(
       server_name, std::move(subject_alt_names), std::move(application_protocols),
-      std::move(alpn_fallback), proxy_protocol_options, filter_state);
+      std::move(alpn_fallback), proxy_protocol_options,
+      filter_state->objectsSharedWithUpstreamConnection());
 }
 
 } // namespace Network

source/common/network/transport_socket_options_impl.h

@@ -29,8 +29,8 @@ class AlpnDecoratingTransportSocketOptions : public TransportSocketOptions {
   absl::optional<Network::ProxyProtocolData> proxyProtocolOptions() const override {
     return inner_options_->proxyProtocolOptions();
   }
-  const StreamInfo::FilterStateSharedPtr& filterState() const override {
-    return inner_options_->filterState();
+  const StreamInfo::FilterState::Objects& filterStateObjects() const override {
+    return inner_options_->filterStateObjects();
   }
 
 private:
@@ -45,13 +45,15 @@ class TransportSocketOptionsImpl : public TransportSocketOptions {
       std::vector<std::string>&& override_verify_san_list = {},
       std::vector<std::string>&& override_alpn = {}, std::vector<std::string>&& fallback_alpn = {},
       absl::optional<Network::ProxyProtocolData> proxy_proto_options = absl::nullopt,
-      const StreamInfo::FilterStateSharedPtr filter_state = nullptr)
+      StreamInfo::FilterState::ObjectsPtr filter_state_objects =
+          std::make_unique<StreamInfo::FilterState::Objects>())
       : override_server_name_(override_server_name.empty()
                                   ? absl::nullopt
                                   : absl::optional<std::string>(override_server_name)),
         override_verify_san_list_{std::move(override_verify_san_list)},
         override_alpn_list_{std::move(override_alpn)}, alpn_fallback_{std::move(fallback_alpn)},
-        proxy_protocol_options_(proxy_proto_options), filter_state_(filter_state) {}
+        proxy_protocol_options_(proxy_proto_options),
+        filter_state_objects_(std::move(filter_state_objects)) {}
 
   // Network::TransportSocketOptions
   const absl::optional<std::string>& serverNameOverride() const override {
@@ -69,15 +71,17 @@ class TransportSocketOptionsImpl : public TransportSocketOptions {
   absl::optional<Network::ProxyProtocolData> proxyProtocolOptions() const override {
     return proxy_protocol_options_;
   }
-  const StreamInfo::FilterStateSharedPtr& filterState() const override { return filter_state_; }
+  const StreamInfo::FilterState::Objects& filterStateObjects() const override {
+    return *filter_state_objects_;
+  }
 
 private:
   const absl::optional<std::string> override_server_name_;
   const std::vector<std::string> override_verify_san_list_;
   const std::vector<std::string> override_alpn_list_;
   const std::vector<std::string> alpn_fallback_;
   const absl::optional<Network::ProxyProtocolData> proxy_protocol_options_;
-  const StreamInfo::FilterStateSharedPtr filter_state_;
+  const StreamInfo::FilterState::ObjectsPtr filter_state_objects_;
 };
 
 class TransportSocketOptionsUtility {