listener: add envoy internal listener

CODEOWNERS

@@ -152,3 +152,5 @@ extensions/filters/http/oauth2 @rgs1 @derekargueta @snowp
 # HTTP Local Rate Limit
 /*/extensions/filters/http/local_ratelimit @rgs1 @mattklein123
 /*/extensions/filters/common/local_ratelimit @mattklein123 @rgs1
+# user space socket pair and event
+/*/extensions/io_socket/buffered_io_socket @lambdai @antoniovicente

api/envoy/config/listener/v3/internal_listener.proto

@@ -0,0 +1,20 @@
+syntax = "proto3";
+
+package envoy.config.listener.v3;
+
+import "udpa/annotations/status.proto";
+import "udpa/annotations/versioning.proto";
+
+option java_package = "io.envoyproxy.envoy.config.listener.v3";
+option java_outer_classname = "InternalListenerProto";
+option java_multiple_files = true;
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#not-implemented-hide:]
+// [#protodoc-title: internal listener]
+// Describes a type of internal listener which expects to serve the cluster in
+// the same envoy process.
+message InternalListener {
+  option (udpa.annotations.versioning).previous_message_type =
+      "envoy.config.listener.v2.InternalListener";
+}

api/envoy/config/listener/v3/listener.proto

@@ -8,6 +8,7 @@ import "envoy/config/core/v3/base.proto";
 import "envoy/config/core/v3/extension.proto";
 import "envoy/config/core/v3/socket_option.proto";
 import "envoy/config/listener/v3/api_listener.proto";
+import "envoy/config/listener/v3/internal_listener.proto";
 import "envoy/config/listener/v3/listener_components.proto";
 import "envoy/config/listener/v3/udp_listener_config.proto";
 
@@ -36,7 +37,7 @@ message ListenerCollection {
   repeated udpa.core.v1.CollectionEntry entries = 1;
 }
 
-// [#next-free-field: 26]
+// [#next-free-field: 27]
 message Listener {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.Listener";
 
@@ -267,4 +268,10 @@ message Listener {
   // The maximum length a tcp listener's pending connections queue can grow to. If no value is
   // provided net.core.somaxconn will be used on Linux and 128 otherwise.
   google.protobuf.UInt32Value tcp_backlog_size = 24;
+
+  // Used to represent an internal listener, which accepts connection from the cluster in the same envoy process.
+  // When this field is set, the address field must be :ref:`envoy internal address
+  // <envoy_api_field_config.core.v3.EnvoyInternalAddress>`.
+  // [#not-implemented-hide:]
+  InternalListener internal_listener = 26;
 }

include/envoy/event/dispatcher.h

@@ -123,6 +123,18 @@ class Dispatcher {
                          Network::TransportSocketPtr&& transport_socket,
                          const Network::ConnectionSocket::OptionsSharedPtr& options) PURE;
 
+  /**
+   * Creates an client internal connection. Does NOT initiate the connection;
+   * the caller must then call connect() on the returned Network::ClientConnection.
+   * @param internal_address supplies the internal address to connect to.
+   * @param local_address supplies an address to bind to or nullptr. If nullptr is provided, an
+   * internal local_address is automatically generated. This address is accessible by the server
+   * connection as source address.
+   * @return Network::ClientConnectionPtr a client connection that is owned by the caller.
+   */
+  virtual Network::ClientConnectionPtr
+  createInternalConnection(Network::Address::InstanceConstSharedPtr internal_address,
+                           Network::Address::InstanceConstSharedPtr local_address) PURE;
   /**
    * Creates an async DNS resolver. The resolver should only be used on the thread that runs this
    * dispatcher.

include/envoy/network/address.h

@@ -134,6 +134,10 @@ class EnvoyInternalAddress {
 
 enum class Type { Ip, Pipe, EnvoyInternal };
 
+constexpr absl::string_view IpName = "Ip";
+constexpr absl::string_view PipeName = "Pipe";
+constexpr absl::string_view EnvoyInternalName = "EnvoyInternal";
+
 /**
  * Interface for all network addresses.
  */

include/envoy/network/listener.h

@@ -146,6 +146,11 @@ class ListenerConfig {
    */
   virtual UdpPacketWriterFactoryOptRef udpPacketWriterFactory() PURE;
 
+  /**
+   * @return true if this listener is internal listener.
+   */
+  virtual bool isInternalListener() PURE;
+
   /**
    * @return the ``UdpListenerWorkerRouter`` for this listener. This will
    * be non-empty iff this is a UDP listener.
@@ -207,6 +212,21 @@ class TcpListenerCallbacks {
   virtual void onReject(RejectCause cause) PURE;
 };
 
+/**
+ * Callbacks invoked by a internal listener.
+ */
+class InternalListenerCallbacks {
+public:
+  virtual ~InternalListenerCallbacks() = default;
+
+  /**
+   * Called when a new internal connection is created.
+   * @param socket supplies the socket. Note that the underlying IoHandle has no file descriptor
+   * although the interface provides fd.
+   */
+  virtual void onNewSocket(Network::ConnectionSocketPtr socket) PURE;
+};
+
 /**
  * Utility struct that encapsulates the information from a udp socket's recvmmsg call.
  */

source/common/event/BUILD

@@ -30,8 +30,9 @@ envoy_cc_library(
         "//source/common/common:assert_lib",
         "//source/common/common:thread_lib",
         "//source/common/filesystem:watcher_lib",
-        "//source/common/network:dns_lib",
+        "//source/extensions/io_socket/buffered_io_socket:buffered_io_socket_handle_lib",
         "//source/common/network:connection_lib",
+        "//source/common/network:dns_lib",
         "//source/common/network:listener_lib",
         "//source/common/runtime:runtime_features_lib",
     ] + select({

source/common/event/dispatcher_impl.cc

@@ -3,6 +3,7 @@
 #include <chrono>
 #include <cstdint>
 #include <functional>
+#include <memory>
 #include <string>
 #include <vector>
 
@@ -18,8 +19,11 @@
 #include "common/event/signal_impl.h"
 #include "common/event/timer_impl.h"
 #include "common/filesystem/watcher_impl.h"
+#include "common/network/address_impl.h"
+#include "extensions/io_socket/buffered_io_socket/buffered_io_socket_handle_impl.h"
 #include "common/network/connection_impl.h"
 #include "common/network/dns_impl.h"
+#include "common/network/raw_buffer_socket.h"
 #include "common/network/tcp_listener_impl.h"
 #include "common/network/udp_listener_impl.h"
 #include "common/runtime/runtime_features.h"
@@ -129,6 +133,72 @@ DispatcherImpl::createClientConnection(Network::Address::InstanceConstSharedPtr
                                                          std::move(transport_socket), options);
 }
 
+namespace {
+Network::Address::InstanceConstSharedPtr
+nextClientAddress(const Network::Address::InstanceConstSharedPtr& server_address) {
+  static uint64_t id = 0;
+  return std::make_shared<Network::Address::EnvoyInternalInstance>(
+      absl::StrCat(server_address->asStringView(), "_", ++id));
+}
+} // namespace
+
+Network::ClientConnectionPtr
+DispatcherImpl::createInternalConnection(Network::Address::InstanceConstSharedPtr internal_address,
+                                         Network::Address::InstanceConstSharedPtr local_address) {
+  ASSERT(isThreadSafe());
+  if (internal_address == nullptr) {
+    return nullptr;
+  }
+  if (local_address == nullptr) {
+    local_address = nextClientAddress(internal_address);
+  }
+  // Find the internal listener callback. The listener will set up the server connection.
+  auto iter = internal_listeners_.find(internal_address->asString());
+  for (const auto& [name, _] : internal_listeners_) {
+    ENVOY_LOG_MISC(debug, "lambdai: p listener {}", name);
+  }
+  if (iter == internal_listeners_.end()) {
+    ENVOY_LOG_MISC(debug, "lambdai: no valid listener registered for envoy internal address {}",
+                   internal_address->asString());
+    return std::make_unique<Network::ClosingClientConnectionImpl>(*this, internal_address,
+                                                                  local_address);
+  }
+
+  auto client_io_handle_ = std::make_unique<Extensions::IoSocket::BufferedIoSocket::BufferedIoSocketHandleImpl>();
+  auto server_io_handle_ = std::make_unique<Extensions::IoSocket::BufferedIoSocket::BufferedIoSocketHandleImpl>();
+  client_io_handle_->setWritablePeer(server_io_handle_.get());
+  server_io_handle_->setWritablePeer(client_io_handle_.get());
+
+  Network::RawBufferSocketFactory client_transport_socket_factory;
+  // ConnectionSocket conn_socket
+  auto client_conn_socket = std::make_unique<Network::InternalConnectionSocketImpl>(
+      std::move(client_io_handle_), local_address, internal_address);
+  auto server_conn_socket = std::make_unique<Network::InternalConnectionSocketImpl>(
+      std::move(server_io_handle_), internal_address, local_address);
+  ENVOY_LOG_MISC(debug, "lambdai: internal address {}", internal_address->asString());
+  ENVOY_LOG_MISC(debug, "lambdai: client address {}", local_address->asString());
+
+  auto client_conn = std::make_unique<Network::ClientConnectionImpl>(
+      *this, internal_address, local_address,
+      client_transport_socket_factory.createTransportSocket(nullptr), nullptr,
+      std::move(client_conn_socket));
+
+  (iter->second)(internal_address, std::move(server_conn_socket));
+  return client_conn;
+}
+
+void DispatcherImpl::registerInternalListener(
+    const std::string& internal_listener_id,
+    DispatcherImpl::InternalConnectionCallback internal_conn_callback) {
+  if (internal_conn_callback == nullptr) {
+    ENVOY_LOG_MISC(debug, "lambdai: unregister pipe factory on address {}", internal_listener_id);
+    internal_listeners_.erase(internal_listener_id);
+  } else {
+    ENVOY_LOG_MISC(debug, "lambdai: register pipe factory on address {}", internal_listener_id);
+    internal_listeners_[internal_listener_id] = internal_conn_callback;
+  }
+}
+
 Network::DnsResolverSharedPtr DispatcherImpl::createDnsResolver(
     const std::vector<Network::Address::InstanceConstSharedPtr>& resolvers,
     const bool use_tcp_for_dns_lookups) {