envoyproxy · yanavlasov · Apr 2, 2022 · Sep 2, 2021 · Sep 9, 2021 · Sep 10, 2021
diff --git a/CODEOWNERS b/CODEOWNERS
@@ -170,8 +170,9 @@ extensions/filters/http/oauth2 @rgs1 @derekargueta @snowp
 /*/extensions/matching/input_matchers/consistent_hashing @snowp @donyu
 # environment generic input
 /*/extensions/matching/common_inputs/environment @snowp @donyu
-# user space socket pair and event
+# user space socket pair, event, connection and listener
 /*/extensions/io_socket/user_space @lambdai @antoniovicente
+/*/extensions/bootstrap/internal_listener @lambdai @adisuissa
 # Default UUID4 request ID extension
 /*/extensions/request_id/uuid @mattklein123 @alyssawilk
 # HTTP header formatters

diff --git a/api/BUILD b/api/BUILD
@@ -120,6 +120,7 @@ proto_library(
         "//envoy/extensions/access_loggers/open_telemetry/v3:pkg",
         "//envoy/extensions/access_loggers/stream/v3:pkg",
         "//envoy/extensions/access_loggers/wasm/v3:pkg",
+        "//envoy/extensions/bootstrap/internal_listener/v3:pkg",
         "//envoy/extensions/cache/simple_http_cache/v3:pkg",
         "//envoy/extensions/clusters/aggregate/v3:pkg",
         "//envoy/extensions/clusters/dynamic_forward_proxy/v3:pkg",

diff --git a/api/envoy/config/core/v3/address.proto b/api/envoy/config/core/v3/address.proto
@@ -31,9 +31,9 @@ message Pipe {
   uint32 mode = 2 [(validate.rules).uint32 = {lte: 511}];
 }
 
-// [#not-implemented-hide:] The address represents an envoy internal listener.
-// TODO(lambdai): Make this address available for listener and endpoint.
-// TODO(asraa): When address available, remove workaround from test/server/server_fuzz_test.cc:30.
+// The address represents an envoy internal listener.
+// [#comment: TODO(lambdai): Make this address available for listener and endpoint.
+// TODO(asraa): When address available, remove workaround from test/server/server_fuzz_test.cc:30.]
 message EnvoyInternalAddress {
   oneof address_name_specifier {
     option (validate.required) = true;

diff --git a/api/envoy/extensions/bootstrap/internal_listener/v3/BUILD b/api/envoy/extensions/bootstrap/internal_listener/v3/BUILD
@@ -0,0 +1,12 @@
+# 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 = [
+        "@com_github_cncf_udpa//udpa/annotations:pkg",
+        "@com_github_cncf_udpa//xds/annotations/v3:pkg",
+    ],
+)
diff --git a/api/envoy/extensions/bootstrap/internal_listener/v3/internal_listener.proto b/api/envoy/extensions/bootstrap/internal_listener/v3/internal_listener.proto
@@ -0,0 +1,22 @@
+syntax = "proto3";
+
+package envoy.extensions.bootstrap.internal_listener.v3;
+
+import "xds/annotations/v3/status.proto";
+
+import "udpa/annotations/status.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.bootstrap.internal_listener.v3";
+option java_outer_classname = "InternalListenerProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/bootstrap/internal_listener/v3;internal_listenerv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+option (xds.annotations.v3.file_status).work_in_progress = true;
+
+// [#protodoc-title: Internal Listener]
+// Internal Listener :ref:`overview <config_internal_listener>`.
+// [#extension: envoy.bootstrap.internal_listener]
+
+// Configuration for internal listener.
+message InternalListener {
+}
diff --git a/api/versioning/BUILD b/api/versioning/BUILD
@@ -57,6 +57,7 @@ proto_library(
         "//envoy/extensions/access_loggers/open_telemetry/v3:pkg",
         "//envoy/extensions/access_loggers/stream/v3:pkg",
         "//envoy/extensions/access_loggers/wasm/v3:pkg",
+        "//envoy/extensions/bootstrap/internal_listener/v3:pkg",
         "//envoy/extensions/cache/simple_http_cache/v3:pkg",
         "//envoy/extensions/clusters/aggregate/v3:pkg",
         "//envoy/extensions/clusters/dynamic_forward_proxy/v3:pkg",

diff --git a/docs/root/api-v3/bootstrap/bootstrap.rst b/docs/root/api-v3/bootstrap/bootstrap.rst
@@ -10,5 +10,6 @@ Bootstrap
   ../config/metrics/v3/metrics_service.proto
   ../config/overload/v3/overload.proto
   ../config/ratelimit/v3/rls.proto
+  ../extensions/bootstrap/internal_listener/v3/internal_listener.proto
   ../extensions/vcl/v3alpha/vcl_socket_interface.proto
   ../extensions/wasm/v3/wasm.proto
diff --git a/docs/root/configuration/other_features/internal_listener.rst b/docs/root/configuration/other_features/internal_listener.rst
@@ -0,0 +1,110 @@
+.. _config_internal_listener:
+
+Internal Listener
+=================
+
+How it works
+------------
+
+This extension contains 2 major components to add a listener with
+an :ref:`Envoy internal address <envoy_v3_api_msg_config.core.v3.EnvoyInternalAddress>`
+and to create a client connection to that :ref:`listener <envoy_v3_api_msg_config.listener.v3.Listener>`
+
+envoy.bootstrap.internal_listener
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+This bootstrap extension is required to support looking up the target listener via an
+:ref:`envoy internal address <envoy_v3_api_msg_config.core.v3.EnvoyInternalAddress>` on each worker threads.
+
+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.
+
+Example config
+--------------
+Below is a smallest static config that redirect TCP proxy on port 19000 to the TCP proxy binding to the internal address.
+
+.. code-block:: yaml
+
+  static_resources:
+    listeners:
+    - name: outbound_tcp_svc_19000
+      address:
+        socket_address:
+          address: 0.0.0.0
+          port_value: 19000
+      filter_chains:
+      - filters:
+        - name: tcp_proxy
+          typed_config:
+            "@type": type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy
+            cluster: bridge_internal_listener
+            stat_prefix: svc_tcp_proxy
+    - name: singleton_internal_encap
+      address:
+        envoy_internal_address:
+          server_listener_name: singleton_internal_encap
+      filter_chains:
+      - filters:
+        - name: tcp_proxy
+          typed_config:
+            "@type": type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy
+            cluster: singleton_internal_encap
+            stat_prefix: encap_tcp_proxy
+    clusters:
+    - name: bridge_internal_listener
+      connect_timeout: 3600s
+      type: STATIC
+      load_assignment:
+        cluster_name: "bridge_internal_listener"
+        endpoints:
+        - lb_endpoints:
+          - endpoint:
+              address:
+                envoy_internal_address:
+                  server_listener_name: singleton_internal_encap
+      transport_socket:
+        name: envoy.transport_sockets.raw_buffer
+        typed_config:
+          "@type": type.googleapis.com/envoy.extensions.transport_sockets.raw_buffer.v3.RawBuffer
+    - name: singleton_internal_encap
+      connect_timeout: 3600s
+      type: STATIC
+      load_assignment:
+        cluster_name: "singleton_internal_encap"
+        endpoints:
+        - lb_endpoints:
+          - endpoint:
+              address:
+                socket_address:
+                  address: 0.0.0.0
+                  port_value: 19001
+  bootstrap_extensions:
+  - name: envoy.bootstrap.internal_listener
+    typed_config:
+      "@type": "type.googleapis.com/envoy.extensions.bootstrap.internal_listener.v3.InternalListener"
+  layered_runtime:
+    layers:
+    - name: enable_internal_address
+      static_layer:
+        envoy.reloadable_features.internal_address: true
+
+Real world use cases
+--------------------
+
+Encap HTTP GET requests in a HTTP CONNECT request
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Currently Envoy :ref:`HTTP connection manager <config_http_conn_man>`
+cannot proxy a GET request in an upstream HTTP CONNECT request. This requirement
+can be acomplished by setting up the upstream endpoint of HTTP connection manager to the internal listener address.
+Meanwhile, another internal listener binding to the above listener address includes a TCP proxy with :ref:`tunneling config <envoy_v3_api_field_extensions.filters.network.tcp_proxy.v3.TcpProxy.tunneling_config>`.
+
+Decap the CONNECT requests
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+There are some complicated GET-in-CONNECT requests across services or edges.
+In order to proxy the GET request within Envoy, two layer of :ref:`HTTP connection manager <config_http_conn_man>`
+is demanded. The first HHTTP connection manager layer extract the TCP stream from a CONNECT request and redirect the TCP stream to the second
+HTTP connection manager layer to parse the common GET requests.
diff --git a/docs/root/configuration/other_features/other_features.rst b/docs/root/configuration/other_features/other_features.rst
@@ -4,6 +4,7 @@ Other features
 .. toctree::
   :maxdepth: 2
 
+  internal_listener
   rate_limit
   vcl
   wasm

diff --git a/envoy/network/BUILD b/envoy/network/BUILD
@@ -28,6 +28,18 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "client_connection_factory",
+    hdrs = ["client_connection_factory.h"],
+    deps = [
+        ":address_interface",
+        ":connection_interface",
+        ":listen_socket_interface",
+        ":transport_socket_interface",
+        "//envoy/config:typed_config_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "connection_handler_interface",
     hdrs = ["connection_handler.h"],

diff --git a/envoy/network/address.h b/envoy/network/address.h
@@ -210,6 +210,12 @@ class Instance {
    */
   virtual Type type() const PURE;
 
+  /**
+   * Return the address type in string_view. The returned type name is used to find the
+   * ClientConnectionFactory.
+   */
+  virtual absl::string_view addressType() const PURE;
+
   /**
    * @return SocketInterface to be used with the address.
    */

diff --git a/envoy/network/client_connection_factory.h b/envoy/network/client_connection_factory.h
@@ -0,0 +1,38 @@
+#pragma once
+
+#include "envoy/config/typed_config.h"
+#include "envoy/network/address.h"
+#include "envoy/network/connection.h"
+#include "envoy/network/listen_socket.h"
+#include "envoy/network/transport_socket.h"
+
+namespace Envoy {
+namespace Network {
+
+// The factory to create a client connection. This factory hides the details of various remote
+// address type and transport socket.
+class ClientConnectionFactory : public Config::UntypedFactory {
+public:
+  ~ClientConnectionFactory() override = default;
+
+  // Config::UntypedFactory
+  std::string category() const override { return "network.connection.client"; }
+
+  /**
+   * @param address The target remote address.
+   * @param source_address Optional source address.
+   * @param transport_socket The transport socket which supports this client connection.
+   * @param options The optional socket options.
+   * @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;
+};
+
+} // namespace Network
+} // namespace Envoy
diff --git a/envoy/network/listener.h b/envoy/network/listener.h
@@ -107,12 +107,21 @@ class UdpListenerConfig {
 
 using UdpListenerConfigOptRef = OptRef<UdpListenerConfig>;
 
+// Forward declare.
+class InternalListenerRegistry;
+
 /**
  * Configuration for an internal listener.
  */
 class InternalListenerConfig {
 public:
   virtual ~InternalListenerConfig() = default;
+
+  /**
+   * @return InternalListenerRegistry& The internal registry of this internal listener config. The
+   *         registry outlives this listener config.
+   */
+  virtual InternalListenerRegistry& internalListenerRegistry() PURE;
 };
 
 using InternalListenerConfigOptRef = OptRef<InternalListenerConfig>;
@@ -472,7 +481,7 @@ class InternalListenerManager {
   virtual ~InternalListenerManager() = default;
 
   /**
-   * Return the internal listener callbacks binding the listener address.
+   * Return the internal listener binding the listener address.
    *
    * @param listen_address the internal address of the expected internal listener.
    */
@@ -483,6 +492,30 @@ class InternalListenerManager {
 using InternalListenerManagerOptRef =
     absl::optional<std::reference_wrapper<InternalListenerManager>>;
 
+// The thread local registry.
+class LocalInternalListenerRegistry {
+public:
+  virtual ~LocalInternalListenerRegistry() = default;
+
+  // Set the internal listener manager which maintains life of internal listeners. Called by
+  // connection handler.
+  virtual void setInternalListenerManager(InternalListenerManager& internal_listener_manager) PURE;
+
+  // Get the internal listener manager to obtain a listener. Called by client connection factory.
+  virtual Network::InternalListenerManagerOptRef getInternalListenerManager() PURE;
+};
+
+// The central internal listener registry interface providing the thread local accessor.
+class InternalListenerRegistry {
+public:
+  virtual ~InternalListenerRegistry() = default;
+
+  /**
+   * @return The thread local registry.
+   */
+  virtual LocalInternalListenerRegistry* getLocalRegistry() PURE;
+};
+
 /**
  * Handles delivering datagrams to the correct worker.
  */

diff --git a/source/common/event/BUILD b/source/common/event/BUILD
@@ -39,12 +39,15 @@ envoy_cc_library(
         "//envoy/common:scope_tracker_interface",
         "//envoy/common:time_interface",
         "//envoy/event:signal_interface",
+        "//envoy/network:client_connection_factory",
         "//envoy/network:listen_socket_interface",
         "//envoy/network:listener_interface",
         "//source/common/common:assert_lib",
         "//source/common/common:thread_lib",
+        "//source/common/config:utility_lib",
         "//source/common/filesystem:watcher_lib",
-        "//source/common/network:connection_lib",
+        "//source/common/network:address_lib",
+        "//source/common/network:default_client_connection_factory",
         "//source/common/network:listener_lib",
         "@envoy_api//envoy/config/overload/v3:pkg_cc_proto",
     ],

diff --git a/source/common/event/dispatcher_impl.cc b/source/common/event/dispatcher_impl.cc
@@ -9,19 +9,22 @@
 #include "envoy/api/api.h"
 #include "envoy/common/scope_tracker.h"
 #include "envoy/config/overload/v3/overload.pb.h"
+#include "envoy/network/client_connection_factory.h"
 #include "envoy/network/listen_socket.h"
 #include "envoy/network/listener.h"
 
 #include "source/common/buffer/buffer_impl.h"
 #include "source/common/common/assert.h"
 #include "source/common/common/lock_guard.h"
 #include "source/common/common/thread.h"
+#include "source/common/config/utility.h"
 #include "source/common/event/file_event_impl.h"
 #include "source/common/event/libevent_scheduler.h"
 #include "source/common/event/scaled_range_timer_manager_impl.h"
 #include "source/common/event/signal_impl.h"
 #include "source/common/event/timer_impl.h"
 #include "source/common/filesystem/watcher_impl.h"
+#include "source/common/network/address_impl.h"
 #include "source/common/network/connection_impl.h"
 #include "source/common/network/tcp_listener_impl.h"
 #include "source/common/network/udp_listener_impl.h"
@@ -158,8 +161,16 @@ DispatcherImpl::createClientConnection(Network::Address::InstanceConstSharedPtr
                                        Network::TransportSocketPtr&& transport_socket,
                                        const Network::ConnectionSocket::OptionsSharedPtr& options) {
   ASSERT(isThreadSafe());
-  return std::make_unique<Network::ClientConnectionImpl>(*this, address, source_address,
-                                                         std::move(transport_socket), options);
+
+  auto* factory = Config::Utility::getFactoryByName<Network::ClientConnectionFactory>(
+      std::string(address->addressType()));
+  // The target address is usually offered by EDS and the EDS api should reject the unsupported
+  // address.
+  // TODO(lambdai): Return a closed connection if the factory is not found. Note that the caller
+  // 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);
 }
 
 FileEventPtr DispatcherImpl::createFileEvent(os_fd_t fd, FileReadyCb cb, FileTriggerType trigger,