transport sockets: support passthrough state for internal connections

CODEOWNERS

@@ -47,6 +47,8 @@ extensions/filters/common/original_src @snowp @klarose
 /*/extensions/transport_sockets/common @alyssawilk @wez470
 # starttls transport socket
 /*/extensions/transport_sockets/starttls @cpakulski @lizan
+# internal upstream transport socket
+/*/extensions/transport_sockets/internal_upstream @kyessenov @lambdai
 # sni_cluster extension
 /*/extensions/filters/network/sni_cluster @rshriram @lizan
 # sni_dynamic_forward_proxy extension

api/BUILD

@@ -251,6 +251,7 @@ proto_library(
         "//envoy/extensions/stat_sinks/graphite_statsd/v3:pkg",
         "//envoy/extensions/stat_sinks/wasm/v3:pkg",
         "//envoy/extensions/transport_sockets/alts/v3:pkg",
+        "//envoy/extensions/transport_sockets/internal_upstream/v3:pkg",
         "//envoy/extensions/transport_sockets/proxy_protocol/v3:pkg",
         "//envoy/extensions/transport_sockets/quic/v3:pkg",
         "//envoy/extensions/transport_sockets/raw_buffer/v3:pkg",

api/envoy/extensions/transport_sockets/internal_upstream/v3/BUILD

@@ -0,0 +1,13 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = [
+        "//envoy/config/core/v3:pkg",
+        "//envoy/type/metadata/v3:pkg",
+        "@com_github_cncf_udpa//udpa/annotations:pkg",
+    ],
+)

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

@@ -0,0 +1,73 @@
+syntax = "proto3";
+
+package envoy.extensions.transport_sockets.internal_upstream.v3;
+
+import "envoy/config/core/v3/base.proto";
+import "envoy/type/metadata/v3/metadata.proto";
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.transport_sockets.internal_upstream.v3";
+option java_outer_classname = "InternalUpstreamProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/internal_upstream/v3;internal_upstreamv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Internal Upstream]
+// [#extension: envoy.transport_sockets.internal_upstream]
+
+// Configuration for the internal upstream address. An internal address defines
+// a loopback user space socket residing in the same proxy instance. This
+// 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.
+message InternalUpstreamTransport {
+  // Describes the location of the imported metadata value.
+  // If the metadata with the given name is not present at the source location,
+  // then no metadata is passed through for this particular instance.
+  message MetadataValueSource {
+    // Specifies what kind of metadata.
+    type.metadata.v3.MetadataKind kind = 1 [(validate.rules).message = {required: true}];
+
+    // Name is the filter namespace used in the dynamic metadata.
+    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}];
+}

api/versioning/BUILD

@@ -192,6 +192,7 @@ proto_library(
         "//envoy/extensions/stat_sinks/graphite_statsd/v3:pkg",
         "//envoy/extensions/stat_sinks/wasm/v3:pkg",
         "//envoy/extensions/transport_sockets/alts/v3:pkg",
+        "//envoy/extensions/transport_sockets/internal_upstream/v3:pkg",
         "//envoy/extensions/transport_sockets/proxy_protocol/v3:pkg",
         "//envoy/extensions/transport_sockets/quic/v3:pkg",
         "//envoy/extensions/transport_sockets/raw_buffer/v3:pkg",

changelogs/current.yaml

@@ -180,6 +180,9 @@ new_features:
 - area: http
   change: |
     added :ref:`cluster_header <envoy_v3_api_field_config.route.v3.RouteAction.RequestMirrorPolicy.cluster_header>` in :ref:`request_mirror_policies <envoy_v3_api_field_config.route.v3.RouteAction.request_mirror_policies>` to allow routing shadow request to the cluster specified in the request_header.
+- area: upstream
+  change: |
+    added :ref:`internal upstream transport <envoy_v3_api_msg_extensions.transport_sockets.internal_upstream.v3.InternalUpstreamTransport>` for passing metadata and filter state across the user space sockets and the internal listeners.
 - area: router
   change:
     added :ref:`keep_empty_value <envoy_v3_api_field_config.core.v3.HeaderValueOption.keep_empty_value>` to allow keeping empty values in custom headers.

docs/root/configuration/other_features/internal_listener.rst

@@ -21,6 +21,26 @@ network.connection.client.envoy_internal
 It is a client connection factory. The factory is implicitly instantiated by the dispatcher to establish a client connection to an
 internal listener address.  This client connection factory is installed automatically when ``envoy.bootstrap.internal_listener`` is specified.
 
+Internal upstream transport
+---------------------------
+
+:ref:`Internal upstream transport
+<envoy_v3_api_msg_extensions.transport_sockets.internal_upstream.v3.InternalUpstreamTransport>`
+extension enables exchange of the filter state from the downstream listener to
+the internal listener through a user space socket. This additional state can be
+in the form of the resource metadata obtained from the upstream host or
+:ref:`the filter state objects <arch_overview_data_sharing_between_filters>`.
+
+This extension emits the following statistics:
+
+.. csv-table::
+   :header: Name, Type, Description
+   :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
 --------------
 Below is a smallest static config that redirect TCP proxy on port 19000 to the TCP proxy binding to the internal address.

envoy/network/transport_socket.h

@@ -14,6 +14,11 @@
 #include "absl/types/optional.h"
 
 namespace Envoy {
+
+namespace Upstream {
+class HostDescription;
+}
+
 namespace Network {
 
 class TransportSocketFactory;
@@ -252,10 +257,12 @@ class TransportSocketFactory {
 
   /**
    * @param options for creating the transport socket
+   * @param host description for the destination upstream host
    * @return Network::TransportSocketPtr a transport socket to be passed to connection.
    */
   virtual TransportSocketPtr
-  createTransportSocket(TransportSocketOptionsConstSharedPtr options) const PURE;
+  createTransportSocket(TransportSocketOptionsConstSharedPtr options,
+                        std::shared_ptr<const Upstream::HostDescription> host) const PURE;
 
   /**
    * Returns true if the transport socket created by this factory supports some form of ALPN

source/common/network/happy_eyeballs_connection_impl.cc

@@ -9,11 +9,12 @@ HappyEyeballsConnectionImpl::HappyEyeballsConnectionImpl(
     Event::Dispatcher& dispatcher, const std::vector<Address::InstanceConstSharedPtr>& address_list,
     Address::InstanceConstSharedPtr source_address, TransportSocketFactory& socket_factory,
     TransportSocketOptionsConstSharedPtr transport_socket_options,
+    const Upstream::HostDescriptionConstSharedPtr& host,
     const ConnectionSocket::OptionsSharedPtr options)
     : id_(ConnectionImpl::next_global_id_++), dispatcher_(dispatcher),
       address_list_(sortAddresses(address_list)),
       connection_construction_state_(
-          {source_address, socket_factory, transport_socket_options, options}),
+          {source_address, socket_factory, transport_socket_options, host, options}),
       next_attempt_timer_(dispatcher_.createTimer([this]() -> void { tryAnotherConnection(); })) {
   ENVOY_LOG_EVENT(debug, "happy_eyeballs_new_cx", "[C{}] addresses={}", id_, address_list_.size());
   connections_.push_back(createNextConnection());
@@ -432,7 +433,8 @@ ClientConnectionPtr HappyEyeballsConnectionImpl::createNextConnection() {
   auto connection = dispatcher_.createClientConnection(
       address_list_[next_address_++], connection_construction_state_.source_address_,
       connection_construction_state_.socket_factory_.createTransportSocket(
-          connection_construction_state_.transport_socket_options_),
+          connection_construction_state_.transport_socket_options_,
+          connection_construction_state_.host_),
       connection_construction_state_.options_);
   ENVOY_LOG_EVENT(debug, "happy_eyeballs_cx_attempt", "C[{}] address={}", id_, next_address_);
   callbacks_wrappers_.push_back(std::make_unique<ConnectionCallbacksWrapper>(*this, *connection));

source/common/network/happy_eyeballs_connection_impl.h

@@ -40,6 +40,7 @@ class HappyEyeballsConnectionImpl : public ClientConnection,
                               Address::InstanceConstSharedPtr source_address,
                               TransportSocketFactory& socket_factory,
                               TransportSocketOptionsConstSharedPtr transport_socket_options,
+                              const Upstream::HostDescriptionConstSharedPtr& host,
                               const ConnectionSocket::OptionsSharedPtr options);
 
   ~HappyEyeballsConnectionImpl() override;
@@ -199,6 +200,7 @@ class HappyEyeballsConnectionImpl : public ClientConnection,
     Address::InstanceConstSharedPtr source_address_;
     TransportSocketFactory& socket_factory_;
     TransportSocketOptionsConstSharedPtr transport_socket_options_;
+    const Upstream::HostDescriptionConstSharedPtr host_;
     const ConnectionSocket::OptionsSharedPtr options_;
   };
 

source/common/network/raw_buffer_socket.cc

@@ -87,7 +87,8 @@ absl::string_view RawBufferSocket::failureReason() const { return EMPTY_STRING;
 void RawBufferSocket::onConnected() { callbacks_->raiseEvent(ConnectionEvent::Connected); }
 
 TransportSocketPtr
-RawBufferSocketFactory::createTransportSocket(TransportSocketOptionsConstSharedPtr) const {
+RawBufferSocketFactory::createTransportSocket(TransportSocketOptionsConstSharedPtr,
+                                              Upstream::HostDescriptionConstSharedPtr) const {
   return std::make_unique<RawBufferSocket>();
 }
 

source/common/network/raw_buffer_socket.h

@@ -36,8 +36,8 @@ class RawBufferSocket : public TransportSocket, protected Logger::Loggable<Logge
 class RawBufferSocketFactory : public CommonTransportSocketFactory {
 public:
   // Network::TransportSocketFactory
-  TransportSocketPtr
-  createTransportSocket(TransportSocketOptionsConstSharedPtr options) const override;
+  TransportSocketPtr createTransportSocket(TransportSocketOptionsConstSharedPtr,
+                                           Upstream::HostDescriptionConstSharedPtr) const override;
   bool implementsSecureTransport() const override;
   absl::string_view defaultServerNameIndication() const override { return ""; }
 };

source/common/quic/quic_transport_socket_factory.h

@@ -48,7 +48,8 @@ class QuicTransportSocketFactoryBase : public Network::CommonTransportSocketFact
 
   // Network::TransportSocketFactory
   Network::TransportSocketPtr
-  createTransportSocket(Network::TransportSocketOptionsConstSharedPtr /*options*/) const override {
+  createTransportSocket(Network::TransportSocketOptionsConstSharedPtr /*options*/,
+                        Upstream::HostDescriptionConstSharedPtr /* host */) const override {
     PANIC("not implemented");
   }
   bool implementsSecureTransport() const override { return true; }
@@ -115,8 +116,9 @@ class QuicClientTransportSocketFactory : public QuicTransportSocketFactoryBase {
   // is needed. In this case the QuicClientTransportSocketFactory falls over to
   // using the fallback factory.
   Network::TransportSocketPtr
-  createTransportSocket(Network::TransportSocketOptionsConstSharedPtr options) const override {
-    return fallback_factory_->createTransportSocket(options);
+  createTransportSocket(Network::TransportSocketOptionsConstSharedPtr options,
+                        Upstream::HostDescriptionConstSharedPtr host) const override {
+    return fallback_factory_->createTransportSocket(options, host);
   }
 
   virtual Envoy::Ssl::ClientContextSharedPtr sslCtx() { return fallback_factory_->sslCtx(); }

source/common/upstream/logical_host.cc

@@ -9,12 +9,11 @@ Upstream::Host::CreateConnectionData LogicalHost::createConnection(
   const auto current_address = address();
   const std::vector<Network::Address::InstanceConstSharedPtr>& address_list = addressList();
 
-  return {HostImpl::createConnection(dispatcher, cluster(), current_address, address_list,
-                                     transportSocketFactory(), options,
-                                     override_transport_socket_options_ != nullptr
-                                         ? override_transport_socket_options_
-                                         : transport_socket_options),
-          std::make_shared<RealHostDescription>(current_address, shared_from_this())};
+  return HostImpl::createConnection(
+      dispatcher, cluster(), current_address, address_list, transportSocketFactory(), options,
+      override_transport_socket_options_ != nullptr ? override_transport_socket_options_
+                                                    : transport_socket_options,
+      std::make_shared<RealHostDescription>(current_address, shared_from_this()));
 }
 
 } // namespace Upstream

source/common/upstream/upstream_impl.cc

@@ -279,9 +279,8 @@ Network::TransportSocketFactory& HostDescriptionImpl::resolveTransportSocketFact
 Host::CreateConnectionData HostImpl::createConnection(
     Event::Dispatcher& dispatcher, const Network::ConnectionSocket::OptionsSharedPtr& options,
     Network::TransportSocketOptionsConstSharedPtr transport_socket_options) const {
-  return {createConnection(dispatcher, cluster(), address(), addressList(),
-                           transportSocketFactory(), options, transport_socket_options),
-          shared_from_this()};
+  return createConnection(dispatcher, cluster(), address(), addressList(), transportSocketFactory(),
+                          options, transport_socket_options, shared_from_this());
 }
 
 void HostImpl::setEdsHealthFlag(envoy::config::core::v3::HealthStatus health_status) {
@@ -310,18 +309,18 @@ Host::CreateConnectionData HostImpl::createHealthCheckConnection(
   Network::TransportSocketFactory& factory =
       (metadata != nullptr) ? resolveTransportSocketFactory(healthCheckAddress(), metadata)
                             : transportSocketFactory();
-  return {createConnection(dispatcher, cluster(), healthCheckAddress(), {}, factory, nullptr,
-                           transport_socket_options),
-          shared_from_this()};
+  return createConnection(dispatcher, cluster(), healthCheckAddress(), {}, factory, nullptr,
+                          transport_socket_options, shared_from_this());
 }
 
-Network::ClientConnectionPtr HostImpl::createConnection(
+Host::CreateConnectionData HostImpl::createConnection(
     Event::Dispatcher& dispatcher, const ClusterInfo& cluster,
     const Network::Address::InstanceConstSharedPtr& address,
     const std::vector<Network::Address::InstanceConstSharedPtr>& address_list,
     Network::TransportSocketFactory& socket_factory,
     const Network::ConnectionSocket::OptionsSharedPtr& options,
-    Network::TransportSocketOptionsConstSharedPtr transport_socket_options) {
+    Network::TransportSocketOptionsConstSharedPtr transport_socket_options,
+    HostDescriptionConstSharedPtr host) {
   Network::ConnectionSocket::OptionsSharedPtr connection_options;
   if (cluster.clusterSocketOptions() != nullptr) {
     if (options) {
@@ -343,17 +342,17 @@ Network::ClientConnectionPtr HostImpl::createConnection(
       address_list.size() > 1
           ? std::make_unique<Network::HappyEyeballsConnectionImpl>(
                 dispatcher, address_list, cluster.sourceAddress(), socket_factory,
-                transport_socket_options, connection_options)
+                transport_socket_options, host, connection_options)
           : dispatcher.createClientConnection(
                 address, cluster.sourceAddress(),
-                socket_factory.createTransportSocket(std::move(transport_socket_options)),
+                socket_factory.createTransportSocket(transport_socket_options, host),
                 connection_options);
 
   connection->connectionInfoSetter().enableSettingInterfaceName(
       cluster.setLocalInterfaceNameOnUpstreamConnections());
   connection->setBufferLimits(cluster.perConnectionBufferLimitBytes());
   cluster.createNetworkFilterChain(*connection);
-  return connection;
+  return {std::move(connection), std::move(host)};
 }
 
 void HostImpl::weight(uint32_t new_weight) { weight_ = std::max(1U, new_weight); }

source/common/upstream/upstream_impl.h

@@ -289,13 +289,14 @@ class HostImpl : public HostDescriptionImpl,
   void used(bool new_used) override { used_ = new_used; }
 
 protected:
-  static Network::ClientConnectionPtr
+  static CreateConnectionData
   createConnection(Event::Dispatcher& dispatcher, const ClusterInfo& cluster,
                    const Network::Address::InstanceConstSharedPtr& address,
                    const std::vector<Network::Address::InstanceConstSharedPtr>& address_list,
                    Network::TransportSocketFactory& socket_factory,
                    const Network::ConnectionSocket::OptionsSharedPtr& options,
-                   Network::TransportSocketOptionsConstSharedPtr transport_socket_options);
+                   Network::TransportSocketOptionsConstSharedPtr transport_socket_options,
+                   HostDescriptionConstSharedPtr host);
 
 private:
   void setEdsHealthFlag(envoy::config::core::v3::HealthStatus health_status);

source/extensions/bootstrap/internal_listener/BUILD

@@ -96,6 +96,7 @@ envoy_cc_library(
         "//source/common/network:listener_filter_buffer_lib",
         "//source/common/stats:timespan_lib",
         "//source/common/stream_info:stream_info_lib",
+        "//source/extensions/io_socket/user_space:config",
         "//source/server:active_stream_listener_base",
         "//source/server:active_tcp_listener",
     ],

source/extensions/bootstrap/internal_listener/active_internal_listener.cc

@@ -5,6 +5,7 @@
 
 #include "source/common/network/address_impl.h"
 #include "source/common/stats/timespan_impl.h"
+#include "source/extensions/io_socket/user_space/io_handle.h"
 #include "source/server/active_stream_listener_base.h"
 
 namespace Envoy {
@@ -55,8 +56,14 @@ void ActiveInternalListener::onAccept(Network::ConnectionSocketPtr&& socket) {
   // connections.
   incNumConnections();
 
+  auto* io_handle = dynamic_cast<Extensions::IoSocket::UserSpace::IoHandle*>(&socket->ioHandle());
   auto active_socket = std::make_unique<Server::ActiveTcpSocket>(
       *this, std::move(socket), false /* do not hand off at internal listener */);
+  // Transfer internal passthrough state to the active socket from downstream.
+  if (io_handle != nullptr && io_handle->passthroughState()) {
+    io_handle->passthroughState()->mergeInto(active_socket->dynamicMetadata(),
+                                             active_socket->filterState());
+  }
 
   onSocketAccepted(std::move(active_socket));
 }