filter state: add a mark for sharing with the upstream

api/envoy/extensions/transport_sockets/internal_upstream/v3/internal_upstream.proto

@@ -22,20 +22,9 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // extension allows passing additional structured state across the user space
 // socket in addition to the regular byte stream. The purpose is to facilitate
 // communication between filters on the downstream and the upstream internal
-// connections.
-//
-// Because the passthrough state is transferred once per the upstream
-// connection before the bytes payload, every passthrough filter state object
-// is included in the hash key used to select an upstream connection if it
-// implements a hashing interface.
-//
-// .. note::
-//
-//  Using internal upstream transport affects load balancing decisions if the
-//  passthrough state is derived from the downstream connection attributes. As
-//  an example, using the downstream source IP in the passthrough state will
-//  prevent sharing of an upstream internal connection between different source
-//  IPs.
+// connections. All filter state objects that are shared with the upstream
+// connection are also shared with the downstream internal connection using
+// this transport socket.
 message InternalUpstreamTransport {
   // Describes the location of the imported metadata value.
   // If the metadata with the given name is not present at the source location,
@@ -48,26 +37,13 @@ message InternalUpstreamTransport {
     string name = 2 [(validate.rules).string = {min_len: 1}];
   }
 
-  // Describes the location of the imported filter state object from the downstream connection.
-  message FilterStateSource {
-    // Name is the imported filter state object name.
-    string name = 1 [(validate.rules).string = {min_len: 1}];
-  }
-
   // Specifies the metadata namespaces and values to insert into the downstream
   // internal connection dynamic metadata when an internal address is used as a
   // host. If the destination name is repeated across two metadata source
   // locations, and both locations contain the metadata with the given name,
   // then the latter in the list overrides the former.
   repeated MetadataValueSource passthrough_metadata = 1;
 
-  // Specifies the list of the filter state object names to insert into the
-  // server internal connection from the downstream connection when an internal
-  // address is used as a host. The filter state objects must be mutable. These
-  // objects participate in the connection hashing decisions if they implement a
-  // hashing function.
-  repeated FilterStateSource passthrough_filter_state_objects = 2;
-
   // The underlying transport socket being wrapped.
   config.core.v3.TransportSocket transport_socket = 3 [(validate.rules).message = {required: true}];
 }

changelogs/current.yaml

@@ -116,6 +116,10 @@ minor_behavior_changes:
 - area: logging
   change: |
     changed category name for access log filter extensions to ``envoy.access_loggers.extension_filters``.
+- area: filter state
+  change: |
+    revert to respecting the life time of the filter state objects to be bound to the original stream and make sharing
+    filter state objects with the upstream explicit via a flag.
 
 bug_fixes:
 - area: http

docs/root/configuration/other_features/internal_listener.rst

@@ -38,8 +38,6 @@ This extension emits the following statistics:
    :widths: 1, 1, 2
 
    no_metadata, Counter, Metadata key is absent from the import location.
-   no_filter_state, Counter, Filter state object key is absent from the downstream filter state.
-   filter_state_error, Counter, Unable to copy the filter state object (e.g. it is read only).
 
 Example config
 --------------

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,
+      const Network::TransportSocketOptionsConstSharedPtr& transport_options) PURE;
 
   /**
    * @return Filesystem::WatcherPtr a filesystem watcher owned by the caller.

envoy/network/client_connection_factory.h

@@ -23,15 +23,16 @@ class ClientConnectionFactory : public Config::UntypedFactory {
    * @param source_address Optional source address.
    * @param transport_socket The transport socket which supports this client connection.
    * @param options The optional socket options.
+   * @param transport socket options used to create the transport socket.
    * @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,
+      const Network::TransportSocketOptionsConstSharedPtr& transport_options) PURE;
 };
 
 } // namespace Network

envoy/network/transport_socket.h

@@ -235,9 +235,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 connection.
    */
-  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

@@ -27,6 +27,7 @@ using FilterStateSharedPtr = std::shared_ptr<FilterState>;
 class FilterState {
 public:
   enum class StateType { ReadOnly, Mutable };
+
   // 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,8 +46,34 @@ 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 };
 
+  // Objects stored in the filter state can optionally be shared between the
+  // upstrean and downstream filter state.
+  enum class StreamSharing {
+    // None implies the object is exclusive to the stream.
+    None,
+
+    // 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. When
+    // upstream connections are re-used between streams, the downstream objects
+    // are captured for the first, initiating stream. To force distinct
+    // upstream connections and prevent sharing, the shared filter state object
+    // must implement the hashing interface.
+    SharedWithUpstreamConnection,
+  };
+
   class Object {
   public:
     virtual ~Object() = default;
@@ -66,6 +93,16 @@ class FilterState {
     virtual absl::optional<std::string> serializeAsString() const { return absl::nullopt; }
   };
 
+  struct FilterObject {
+    std::shared_ptr<Object> data_;
+    StateType state_type_;
+    StreamSharing stream_sharing_{StreamSharing::None};
+    std::string name_{""};
+  };
+
+  using Objects = std::vector<FilterObject>;
+  using ObjectsPtr = std::unique_ptr<Objects>;
+
   virtual ~FilterState() = default;
 
   /**
@@ -83,7 +120,8 @@ class FilterState {
    * data stored in FilterState.
    */
   virtual void setData(absl::string_view data_name, std::shared_ptr<Object> data,
-                       StateType state_type, LifeSpan life_span = LifeSpan::FilterChain) PURE;
+                       StateType state_type, LifeSpan life_span = LifeSpan::FilterChain,
+                       StreamSharing stream_sharing = StreamSharing::None) PURE;
 
   /**
    * @param data_name the name of the data being looked up (mutable/readonly).
@@ -156,6 +194,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,
+    const Network::TransportSocketOptionsConstSharedPtr& transport_options) {
   ASSERT(isThreadSafe());
 
   auto* factory = Config::Utility::getFactoryByName<Network::ClientConnectionFactory>(
@@ -170,7 +171,7 @@ 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, 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,
+      const Network::TransportSocketOptionsConstSharedPtr& transport_options) override;
   FileEventPtr createFileEvent(os_fd_t fd, FileReadyCb cb, FileTriggerType trigger,
                                uint32_t events) override;
   Filesystem::WatcherPtr createFilesystemWatcher() override;

source/common/network/BUILD

@@ -445,6 +445,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,15 @@ ClientConnectionImpl::ClientConnectionImpl(
       ioHandle().activateFileEvents(Event::FileReadyType::Write);
     }
   }
+
+  if (transport_options) {
+    for (const auto& object : transport_options->filterStateObjects()) {
+      // This does not throw as all objects are distinctly named and the stream info is empty.
+      stream_info_.filterState()->setData(object.name_, object.data_, object.state_type_,
+                                          StreamInfo::FilterState::LifeSpan::Connection,
+                                          object.stream_sharing_);
+    }
+  }
 }
 
 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,
+    const 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,
+      const Network::TransportSocketOptionsConstSharedPtr& transport_options) override;
 };
 
 } // namespace Network

source/common/network/happy_eyeballs_connection_impl.cc

@@ -435,7 +435,8 @@ ClientConnectionPtr HappyEyeballsConnectionImpl::createNextConnection() {
       connection_construction_state_.socket_factory_.createTransportSocket(
           connection_construction_state_.transport_socket_options_,
           connection_construction_state_.host_),
-      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());