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 @lizan
 # sni_cluster extension
 /*/extensions/filters/network/sni_cluster @rshriram @lizan
 # sni_dynamic_forward_proxy extension

api/BUILD

@@ -249,6 +249,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

@@ -190,6 +190,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

@@ -151,6 +151,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.
 
 deprecated:
 - area: dubbo_proxy

docs/root/configuration/other_features/internal_listener.rst

@@ -21,6 +21,16 @@ 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.
 
+Passing state to the internal listener
+--------------------------------------
+
+:ref:`Internal upstream transport
+<envoy_v3_api_msg_extensions.transport_sockets.internal_upstream.v3.InternalUpstreamTransport>`
+extension enables passing additional state from the downstream listener to the
+internal listener through an internal address besides the raw byte stream. This
+additional state can be in the form of metadata or :ref:`filter state objects
+<arch_overview_data_sharing_between_filters>`.
+
 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;
@@ -234,6 +239,11 @@ class TransportSocketOptions {
    * @return filter state from the downstream request or connection.
    */
   virtual const StreamInfo::FilterStateSharedPtr& filterState() const PURE;
+
+  /**
+   * @return optional upstream host description.
+   */
+  virtual const std::shared_ptr<const Upstream::HostDescription>& host() const PURE;
 };
 
 using TransportSocketOptionsConstSharedPtr = std::shared_ptr<const TransportSocketOptions>;

source/common/network/BUILD

@@ -447,6 +447,7 @@ envoy_cc_library(
         "//envoy/stream_info:filter_state_interface",
         "//source/common/common:scalar_to_byte_vector_lib",
         "//source/common/common:utility_lib",
+        "//source/common/singleton:const_singleton",
     ],
 )
 

source/common/network/transport_socket_options_impl.h

@@ -4,40 +4,11 @@
 #include "envoy/network/transport_socket.h"
 #include "envoy/stream_info/filter_state.h"
 
+#include "source/common/singleton/const_singleton.h"
+
 namespace Envoy {
 namespace Network {
 
-// A wrapper around another TransportSocketOptions that overrides the ALPN fallback.
-class AlpnDecoratingTransportSocketOptions : public TransportSocketOptions {
-public:
-  AlpnDecoratingTransportSocketOptions(std::vector<std::string>&& alpn,
-                                       TransportSocketOptionsConstSharedPtr inner_options)
-      : alpn_fallback_(std::move(alpn)), inner_options_(std::move(inner_options)) {}
-  // Network::TransportSocketOptions
-  const absl::optional<std::string>& serverNameOverride() const override {
-    return inner_options_->serverNameOverride();
-  }
-  const std::vector<std::string>& verifySubjectAltNameListOverride() const override {
-    return inner_options_->verifySubjectAltNameListOverride();
-  }
-  const std::vector<std::string>& applicationProtocolListOverride() const override {
-    return inner_options_->applicationProtocolListOverride();
-  }
-  const std::vector<std::string>& applicationProtocolFallback() const override {
-    return alpn_fallback_;
-  }
-  absl::optional<Network::ProxyProtocolData> proxyProtocolOptions() const override {
-    return inner_options_->proxyProtocolOptions();
-  }
-  const StreamInfo::FilterStateSharedPtr& filterState() const override {
-    return inner_options_->filterState();
-  }
-
-private:
-  const std::vector<std::string> alpn_fallback_;
-  const TransportSocketOptionsConstSharedPtr inner_options_;
-};
-
 class TransportSocketOptionsImpl : public TransportSocketOptions {
 public:
   TransportSocketOptionsImpl(
@@ -71,13 +42,16 @@ class TransportSocketOptionsImpl : public TransportSocketOptions {
   }
   const StreamInfo::FilterStateSharedPtr& filterState() const override { return filter_state_; }
 
+  const std::shared_ptr<const Upstream::HostDescription>& host() const override { return host_; }
+
 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 std::shared_ptr<const Upstream::HostDescription> host_;
 };
 
 class TransportSocketOptionsUtility {
@@ -101,5 +75,73 @@ class CommonTransportSocketFactory : public TransportSocketFactory {
                TransportSocketOptionsConstSharedPtr options) const override;
 };
 
+using EmptyTransportSocketOptions = ConstSingleton<TransportSocketOptionsImpl>;
+
+// Base wrapper for transport socket options.
+class BaseWrapperTransportSocketOptions : public TransportSocketOptions {
+public:
+  BaseWrapperTransportSocketOptions(TransportSocketOptionsConstSharedPtr inner_options)
+      : inner_options_(std::move(inner_options)) {}
+  // Network::TransportSocketOptions
+  const absl::optional<std::string>& serverNameOverride() const override {
+    return inner_options_ ? inner_options_->serverNameOverride()
+                          : EmptyTransportSocketOptions::get().serverNameOverride();
+  }
+  const std::vector<std::string>& verifySubjectAltNameListOverride() const override {
+    return inner_options_ ? inner_options_->verifySubjectAltNameListOverride()
+                          : EmptyTransportSocketOptions::get().verifySubjectAltNameListOverride();
+  }
+  const std::vector<std::string>& applicationProtocolListOverride() const override {
+    return inner_options_ ? inner_options_->applicationProtocolListOverride()
+                          : EmptyTransportSocketOptions::get().applicationProtocolListOverride();
+  }
+  const std::vector<std::string>& applicationProtocolFallback() const override {
+    return inner_options_ ? inner_options_->applicationProtocolFallback()
+                          : EmptyTransportSocketOptions::get().applicationProtocolFallback();
+  }
+  absl::optional<Network::ProxyProtocolData> proxyProtocolOptions() const override {
+    return inner_options_ ? inner_options_->proxyProtocolOptions() : absl::nullopt;
+  }
+  const StreamInfo::FilterStateSharedPtr& filterState() const override {
+    return inner_options_ ? inner_options_->filterState()
+                          : EmptyTransportSocketOptions::get().filterState();
+  }
+  const std::shared_ptr<const Upstream::HostDescription>& host() const override {
+    return inner_options_ ? inner_options_->host() : EmptyTransportSocketOptions::get().host();
+  }
+
+private:
+  const TransportSocketOptionsConstSharedPtr inner_options_;
+};
+
+// A wrapper around another TransportSocketOptions that overrides the ALPN fallback.
+class AlpnDecoratingTransportSocketOptions : public BaseWrapperTransportSocketOptions {
+public:
+  AlpnDecoratingTransportSocketOptions(std::vector<std::string>&& alpn,
+                                       TransportSocketOptionsConstSharedPtr inner_options)
+      : BaseWrapperTransportSocketOptions(std::move(inner_options)),
+        alpn_fallback_(std::move(alpn)) {}
+  // Network::TransportSocketOptions
+  const std::vector<std::string>& applicationProtocolFallback() const override {
+    return alpn_fallback_;
+  }
+
+private:
+  const std::vector<std::string> alpn_fallback_;
+};
+
+// A wrapper around another TransportSocketOptions that overrides the upstream host.
+class HostDecoratingTransportSocketOptions : public BaseWrapperTransportSocketOptions {
+public:
+  HostDecoratingTransportSocketOptions(std::shared_ptr<const Upstream::HostDescription>& host,
+                                       TransportSocketOptionsConstSharedPtr inner_options)
+      : BaseWrapperTransportSocketOptions(std::move(inner_options)), host_(host) {}
+  // Network::TransportSocketOptions
+  const std::shared_ptr<const Upstream::HostDescription>& host() const override { return host_; }
+
+private:
+  const std::shared_ptr<const Upstream::HostDescription> host_;
+};
+
 } // namespace Network
 } // namespace Envoy

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) {
@@ -339,21 +338,25 @@ Network::ClientConnectionPtr HostImpl::createConnection(
   ASSERT(!address->envoyInternalAddress() ||
          Runtime::runtimeFeatureEnabled("envoy.reloadable_features.internal_address"));
 
+  auto host_transport_socket_options =
+      std::make_shared<Network::HostDecoratingTransportSocketOptions>(host,
+                                                                      transport_socket_options);
+
   Network::ClientConnectionPtr connection =
       address_list.size() > 1
           ? std::make_unique<Network::HappyEyeballsConnectionImpl>(
                 dispatcher, address_list, cluster.sourceAddress(), socket_factory,
-                transport_socket_options, connection_options)
+                std::move(host_transport_socket_options), connection_options)
           : dispatcher.createClientConnection(
                 address, cluster.sourceAddress(),
-                socket_factory.createTransportSocket(std::move(transport_socket_options)),
+                socket_factory.createTransportSocket(std::move(host_transport_socket_options)),
                 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); }