transport sockets: support passthrough state for internal connections

CODEOWNERS

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

api/envoy/extensions/transport_sockets/internal/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/v3/internal_upstream.proto

@@ -0,0 +1,39 @@
+syntax = "proto3";
+
+package envoy.extensions.transport_sockets.internal.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.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/v3;internalv3";
+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 is a
+// loopback user space socket residing in the same proxy instance.
+// [#not-implemented-hide:]
+message InternalUpstreamTransport {
+  // Describes the location of the metadata value in the upstream cluster context.
+  message MetadataValueSource {
+    // Specifies what kind of metadata.
+    type.metadata.v3.MetadataKind kind = 1;
+
+    // Name is the filter namespace used in the dynamic metadata.
+    string name = 2 [(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.
+  map<string, MetadataValueSource> passthrough_metadata = 1;
+
+  // The underlying transport socket being wrapped.
+  config.core.v3.TransportSocket transport_socket = 2 [(validate.rules).message = {required: true}];
+}

api/versioning/BUILD

@@ -178,6 +178,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/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",

envoy/network/transport_socket.h

@@ -12,6 +12,11 @@
 #include "absl/types/optional.h"
 
 namespace Envoy {
+
+namespace Upstream {
+class HostDescription;
+}
+
 namespace Network {
 
 class TransportSocketFactory;
@@ -216,6 +221,11 @@ class TransportSocketOptions {
    */
   virtual void hashKey(std::vector<uint8_t>& key,
                        const Network::TransportSocketFactory& factory) const PURE;
+
+  /**
+   * @return optional upstream host description.
+   */
+  virtual std::shared_ptr<const Upstream::HostDescription> host() const PURE;
 };
 
 using TransportSocketOptionsConstSharedPtr = std::shared_ptr<const TransportSocketOptions>;

source/common/network/transport_socket_options_impl.h

@@ -7,12 +7,11 @@
 namespace Envoy {
 namespace Network {
 
-// A wrapper around another TransportSocketOptions that overrides the ALPN fallback.
-class AlpnDecoratingTransportSocketOptions : public TransportSocketOptions {
+// Base wrapper for transport socket options.
+class BaseWrapperTransportSocketOptions : public TransportSocketOptions {
 public:
-  AlpnDecoratingTransportSocketOptions(std::vector<std::string>&& alpn,
-                                       TransportSocketOptionsConstSharedPtr inner_options)
-      : alpn_fallback_(std::move(alpn)), inner_options_(std::move(inner_options)) {}
+  BaseWrapperTransportSocketOptions(TransportSocketOptionsConstSharedPtr inner_options)
+      : inner_options_(std::move(inner_options)) {}
   // Network::TransportSocketOptions
   const absl::optional<std::string>& serverNameOverride() const override {
     return inner_options_->serverNameOverride();
@@ -24,17 +23,50 @@ class AlpnDecoratingTransportSocketOptions : public TransportSocketOptions {
     return inner_options_->applicationProtocolListOverride();
   }
   const std::vector<std::string>& applicationProtocolFallback() const override {
-    return alpn_fallback_;
+    return inner_options_->applicationProtocolFallback();
   }
   absl::optional<Network::ProxyProtocolData> proxyProtocolOptions() const override {
     return inner_options_->proxyProtocolOptions();
   }
+  void hashKey(std::vector<uint8_t>& key,
+               const Network::TransportSocketFactory& factory) const override {
+    return inner_options_->hashKey(key, factory);
+  }
+  std::shared_ptr<const Upstream::HostDescription> host() const override {
+    return inner_options_->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)) {}
+  const std::vector<std::string>& applicationProtocolFallback() const override {
+    return alpn_fallback_;
+  }
   void hashKey(std::vector<uint8_t>& key,
                const Network::TransportSocketFactory& factory) const override;
 
 private:
   const std::vector<std::string> alpn_fallback_;
-  const TransportSocketOptionsConstSharedPtr inner_options_;
+};
+
+// 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) {}
+  std::shared_ptr<const Upstream::HostDescription> host() const override { return host_; }
+
+private:
+  const std::shared_ptr<const Upstream::HostDescription> host_;
 };
 
 class TransportSocketOptionsImpl : public TransportSocketOptions {
@@ -70,6 +102,8 @@ class TransportSocketOptionsImpl : public TransportSocketOptions {
   void hashKey(std::vector<uint8_t>& key,
                const Network::TransportSocketFactory& factory) const override;
 
+  std::shared_ptr<const Upstream::HostDescription> host() const override { return nullptr; }
+
 private:
   const absl::optional<std::string> override_server_name_;
   const std::vector<std::string> override_verify_san_list_;

source/common/upstream/logical_host.cc

@@ -6,15 +6,14 @@ namespace Upstream {
 Upstream::Host::CreateConnectionData LogicalHost::createConnection(
     Event::Dispatcher& dispatcher, const Network::ConnectionSocket::OptionsSharedPtr& options,
     Network::TransportSocketOptionsConstSharedPtr transport_socket_options) const {
-  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,
+  HostDescriptionConstSharedPtr host =
+      std::make_shared<RealHostDescription>(address(), shared_from_this());
+  return {HostImpl::createConnection(dispatcher, host, transportSocketFactory(), options,
                                      override_transport_socket_options_ != nullptr
                                          ? override_transport_socket_options_
-                                         : transport_socket_options),
-          std::make_shared<RealHostDescription>(current_address, shared_from_this())};
+                                         : transport_socket_options,
+                                     false),
+          host};
 }
 
 } // namespace Upstream

source/common/upstream/upstream_impl.cc

@@ -277,9 +277,10 @@ 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()};
+  HostDescriptionConstSharedPtr host = shared_from_this();
+  return {createConnection(dispatcher, host, transportSocketFactory(), options,
+                           transport_socket_options, false),
+          host};
 }
 
 void HostImpl::setEdsHealthFlag(envoy::config::core::v3::HealthStatus health_status) {
@@ -308,19 +309,19 @@ 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()};
-}
-
-Network::ClientConnectionPtr 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) {
+  HostDescriptionConstSharedPtr host = shared_from_this();
+  return {createConnection(dispatcher, host, factory, nullptr, transport_socket_options, true),
+          host};
+}
+
+Network::ClientConnectionPtr
+HostImpl::createConnection(Event::Dispatcher& dispatcher, HostDescriptionConstSharedPtr& host,
+                           Network::TransportSocketFactory& socket_factory,
+                           const Network::ConnectionSocket::OptionsSharedPtr& options,
+                           Network::TransportSocketOptionsConstSharedPtr transport_socket_options,
+                           bool health_check) {
   Network::ConnectionSocket::OptionsSharedPtr connection_options;
+  const auto& cluster = host->cluster();
   if (cluster.clusterSocketOptions() != nullptr) {
     if (options) {
       connection_options = std::make_shared<Network::ConnectionSocket::Options>();
@@ -334,17 +335,24 @@ Network::ClientConnectionPtr HostImpl::createConnection(
     connection_options = options;
   }
 
+  const auto& address = health_check ? host->healthCheckAddress() : host->address();
+  const auto& address_list =
+      health_check ? std::vector<Network::Address::InstanceConstSharedPtr>() : host->addressList();
+
   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)
+                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->setBufferLimits(cluster.perConnectionBufferLimitBytes());

source/common/upstream/upstream_impl.h

@@ -282,12 +282,11 @@ class HostImpl : public HostDescriptionImpl,
 
 protected:
   static Network::ClientConnectionPtr
-  createConnection(Event::Dispatcher& dispatcher, const ClusterInfo& cluster,
-                   const Network::Address::InstanceConstSharedPtr& address,
-                   const std::vector<Network::Address::InstanceConstSharedPtr>& address_list,
+  createConnection(Event::Dispatcher& dispatcher, HostDescriptionConstSharedPtr& host,
                    Network::TransportSocketFactory& socket_factory,
                    const Network::ConnectionSocket::OptionsSharedPtr& options,
-                   Network::TransportSocketOptionsConstSharedPtr transport_socket_options);
+                   Network::TransportSocketOptionsConstSharedPtr transport_socket_options,
+                   bool health_check);
 
 private:
   void setEdsHealthFlag(envoy::config::core::v3::HealthStatus health_status);

source/extensions/extensions_build_config.bzl

@@ -200,6 +200,7 @@ EXTENSIONS = {
     "envoy.transport_sockets.tap":                      "//source/extensions/transport_sockets/tap:config",
     "envoy.transport_sockets.starttls":                 "//source/extensions/transport_sockets/starttls:config",
     "envoy.transport_sockets.tcp_stats":                "//source/extensions/transport_sockets/tcp_stats:config",
+    "envoy.transport_sockets.internal_upstream":        "//source/extensions/transport_sockets/internal:config",
 
     #
     # Retry host predicates

source/extensions/io_socket/user_space/BUILD

@@ -24,6 +24,7 @@ envoy_cc_library(
         "//source/common/buffer:buffer_lib",
         "//source/common/buffer:watermark_buffer_lib",
         "//source/common/common:empty_string",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
     ],
 )
 

source/extensions/io_socket/user_space/io_handle.h

@@ -2,6 +2,7 @@
 
 #include "envoy/buffer/buffer.h"
 #include "envoy/common/pure.h"
+#include "envoy/config/core/v3/base.pb.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -60,6 +61,16 @@ class IoHandle {
    * @return true if the pending receive buffer is not empty or read_end is set.
    */
   virtual bool isReadable() const PURE;
+
+  /**
+   * @return user space socket metadata.
+   */
+  virtual const envoy::config::core::v3::Metadata& metadata() const PURE;
+
+  /**
+   * Set user space socket metadata by merging the protobuf.
+   */
+  virtual void setMetadata(const envoy::config::core::v3::Metadata& metadata) PURE;
 };
 } // namespace UserSpace
 } // namespace IoSocket

source/extensions/io_socket/user_space/io_handle_impl.h

@@ -147,6 +147,10 @@ class IoHandleImpl final : public Network::IoHandle,
     ENVOY_LOG(trace, "io handle {} set peer handle to {}.", static_cast<void*>(this),
               static_cast<void*>(writable_peer));
   }
+  const envoy::config::core::v3::Metadata& metadata() const override { return metadata_; }
+  virtual void setMetadata(const envoy::config::core::v3::Metadata& metadata) override {
+    metadata_.MergeFrom(metadata);
+  }
 
 private:
   friend class IoHandleFactory;
@@ -174,6 +178,9 @@ class IoHandleImpl final : public Network::IoHandle,
 
   // The flag whether the peer is valid. Any write attempt must check this flag.
   bool write_shutdown_{false};
+
+  // User space metadata passed through the user space socket pair.
+  envoy::config::core::v3::Metadata metadata_;
 };
 
 using IoHandleImplPtr = std::unique_ptr<IoHandleImpl>;

source/extensions/transport_sockets/internal/BUILD

@@ -0,0 +1,40 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_library(
+    name = "internal_lib",
+    srcs = ["internal.cc"],
+    hdrs = ["internal.h"],
+    deps = [
+        "//envoy/network:transport_socket_interface",
+        "//envoy/upstream:host_description_interface",
+        "//envoy/upstream:upstream_interface",
+        "//source/common/common:assert_lib",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/protobuf:utility_lib",
+        "//source/extensions/io_socket/user_space:io_handle_lib",
+        "//source/extensions/transport_sockets/common:passthrough_lib",
+        "@envoy_api//envoy/extensions/transport_sockets/internal/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":internal_lib",
+        "//envoy/registry",
+        "//envoy/server:transport_socket_config_interface",
+        "//source/common/config:utility_lib",
+        "@envoy_api//envoy/extensions/transport_sockets/internal/v3:pkg_cc_proto",
+    ],
+)