listener: create internal listener

envoy/network/listener.h

@@ -107,6 +107,16 @@ class UdpListenerConfig {
 
 using UdpListenerConfigOptRef = OptRef<UdpListenerConfig>;
 
+/**
+ * Configuration for an internal listener.
+ */
+class InternalListenerConfig {
+public:
+  virtual ~InternalListenerConfig() = default;
+};
+
+using InternalListenerConfigOptRef = OptRef<InternalListenerConfig>;
+
 /**
  * A configuration for an individual listener.
  */
@@ -185,6 +195,11 @@ class ListenerConfig {
    */
   virtual UdpListenerConfigOptRef udpListenerConfig() PURE;
 
+  /**
+   * @return the internal configuration for the listener IFF it is an internal listener.
+   */
+  virtual InternalListenerConfigOptRef internalListenerConfig() PURE;
+
   /**
    * @return traffic direction of the listener.
    */
@@ -428,6 +443,46 @@ class UdpListener : public virtual Listener {
 
 using UdpListenerPtr = std::unique_ptr<UdpListener>;
 
+/**
+ * Internal listener callbacks.
+ */
+class InternalListenerCallbacks {
+public:
+  virtual ~InternalListenerCallbacks() = default;
+
+  /**
+   * Called when a new connection is accepted.
+   * @param socket supplies the socket that is moved into the callee.
+   */
+  virtual void onAccept(ConnectionSocketPtr&& socket) PURE;
+};
+using InternalListenerCallbacksOptRef =
+    absl::optional<std::reference_wrapper<InternalListenerCallbacks>>;
+
+class InternalListener {};
+
+using InternalListenerPtr = std::unique_ptr<InternalListener>;
+using InternalListenerOptRef = absl::optional<std::reference_wrapper<InternalListener>>;
+
+/**
+ * The query interface of the registered internal listener callbacks.
+ */
+class InternalListenerManager {
+public:
+  virtual ~InternalListenerManager() = default;
+
+  /**
+   * Return the internal listener callbacks binding the listener address.
+   *
+   * @param listen_address the internal address of the expected internal listener.
+   */
+  virtual InternalListenerCallbacksOptRef
+  findByAddress(const Address::InstanceConstSharedPtr& listen_address) PURE;
+};
+
+using InternalListenerManagerOptRef =
+    absl::optional<std::reference_wrapper<InternalListenerManager>>;
+
 /**
  * Handles delivering datagrams to the correct worker.
  */

source/common/network/BUILD

@@ -64,6 +64,7 @@ envoy_cc_library(
     hdrs = ["connection_impl_base.h"],
     deps = [
         ":filter_manager_lib",
+        ":listen_socket_lib",
         "//envoy/common:scope_tracker_interface",
         "//envoy/event:dispatcher_interface",
         "//source/common/common:assert_lib",

source/common/network/listen_socket_impl.h

@@ -153,6 +153,25 @@ class UdsListenSocket : public ListenSocketImpl {
   Socket::Type socketType() const override { return Socket::Type::Stream; }
 };
 
+// This socket type adapts the ListenerComponentFactory.
+class InternalListenSocket : public ListenSocketImpl {
+public:
+  InternalListenSocket(const Address::InstanceConstSharedPtr& address)
+      : ListenSocketImpl(/* io_handle= */ nullptr, address) {}
+  Socket::Type socketType() const override { return Socket::Type::Stream; }
+
+  // InternalListenSocket cannot be duplicated.
+  SocketPtr duplicate() override {
+    return std::make_unique<InternalListenSocket>(connectionInfoProvider().localAddress());
+  }
+
+  void close() override { ASSERT(io_handle_ == nullptr); }
+  bool isOpen() const override {
+    ASSERT(io_handle_ == nullptr);
+    return false;
+  }
+};
+
 class ConnectionSocketImpl : public SocketImpl, public ConnectionSocket {
 public:
   ConnectionSocketImpl(IoHandlePtr&& io_handle,

source/common/network/socket_impl.cc

@@ -30,6 +30,11 @@ SocketImpl::SocketImpl(IoHandlePtr&& io_handle,
     return;
   }
 
+  if (connection_info_provider_->remoteAddress() != nullptr) {
+    addr_type_ = connection_info_provider_->remoteAddress()->type();
+    return;
+  }
+
   // Should not happen but some tests inject -1 fds
   if (!io_handle_->isOpen()) {
     return;

source/common/network/utility.cc

@@ -534,6 +534,9 @@ Utility::protobufAddressSocketType(const envoy::config::core::v3::Address& proto
   }
   case envoy::config::core::v3::Address::AddressCase::kPipe:
     return Socket::Type::Stream;
+  case envoy::config::core::v3::Address::AddressCase::kEnvoyInternalAddress:
+    // Currently internal address supports stream operation only.
+    return Socket::Type::Stream;
   default:
     NOT_REACHED_GCOVR_EXCL_LINE;
   }

source/common/runtime/runtime_features.cc

@@ -69,6 +69,7 @@ constexpr const char* runtime_features[] = {
     "envoy.reloadable_features.http_reject_path_with_fragment",
     "envoy.reloadable_features.http_strip_fragment_from_path_unsafe_if_disabled",
     "envoy.reloadable_features.http_transport_failure_reason_in_body",
+    "envoy.reloadable_features.internal_address",
     "envoy.reloadable_features.internal_redirects_with_body",
     "envoy.reloadable_features.listener_reuse_port_default_enabled",
     "envoy.reloadable_features.listener_wildcard_match_ip_family",

source/common/tcp_proxy/tcp_proxy.cc

@@ -531,6 +531,8 @@ Network::FilterStatus Filter::onNewConnection() {
 }
 
 void Filter::onDownstreamEvent(Network::ConnectionEvent event) {
+  ENVOY_CONN_LOG(trace, "on downstream event {}, has upstream = {}", read_callbacks_->connection(),
+                 static_cast<int>(event), upstream_ == nullptr);
   if (upstream_) {
     Tcp::ConnectionPool::ConnectionDataPtr conn_data(upstream_->onDownstreamEvent(event));
     if (conn_data != nullptr &&
@@ -699,6 +701,7 @@ Drainer::Drainer(UpstreamDrainManager& parent, const Config::SharedConfigSharedP
                  const Upstream::HostDescriptionConstSharedPtr& upstream_host)
     : parent_(parent), callbacks_(callbacks), upstream_conn_data_(std::move(conn_data)),
       timer_(std::move(idle_timer)), upstream_host_(upstream_host), config_(config) {
+  ENVOY_CONN_LOG(debug, "draining the upstream connection", upstream_conn_data_->connection());
   config_->stats().upstream_flush_total_.inc();
   config_->stats().upstream_flush_active_.inc();
 }

source/common/tcp_proxy/tcp_proxy.h

@@ -380,7 +380,7 @@ class Filter : public Network::ReadFilter,
 // This class deals with an upstream connection that needs to finish flushing, when the downstream
 // connection has been closed. The TcpProxy is destroyed when the downstream connection is closed,
 // so handling the upstream connection here allows it to finish draining or timeout.
-class Drainer : public Event::DeferredDeletable {
+class Drainer : public Event::DeferredDeletable, protected Logger::Loggable<Logger::Id::filter> {
 public:
   Drainer(UpstreamDrainManager& parent, const Config::SharedConfigSharedPtr& config,
           const std::shared_ptr<Filter::UpstreamCallbacks>& callbacks,

source/common/upstream/upstream_impl.cc

@@ -317,7 +317,10 @@ Network::ClientConnectionPtr HostImpl::createConnection(
   } else {
     connection_options = options;
   }
-  ASSERT(!address->envoyInternalAddress());
+
+  ASSERT(!address->envoyInternalAddress() ||
+         Runtime::runtimeFeatureEnabled("envoy.reloadable_features.internal_address"));
+
   Network::ClientConnectionPtr connection =
       address_list.size() > 1
           ? std::make_unique<Network::HappyEyeballsConnectionImpl>(

source/server/BUILD

@@ -64,6 +64,7 @@ envoy_cc_library(
 envoy_cc_library(
     name = "connection_handler_lib",
     deps = [
+        ":active_internal_listener",
         ":active_tcp_listener",
         ":active_udp_listener",
         ":connection_handler_impl",
@@ -77,6 +78,7 @@ envoy_cc_library(
         "connection_handler_impl.h",
     ],
     deps = [
+        ":active_internal_listener",
         ":active_tcp_listener",
         "//envoy/common:time_interface",
         "//envoy/event:deferred_deletable",
@@ -179,6 +181,36 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "active_internal_listener",
+    srcs = ["active_internal_listener.cc"],
+    hdrs = [
+        "active_internal_listener.h",
+    ],
+    deps = [
+        ":active_listener_base",
+        ":active_tcp_listener_headers",
+        "//envoy/common:time_interface",
+        "//envoy/event:deferred_deletable",
+        "//envoy/event:dispatcher_interface",
+        "//envoy/event:timer_interface",
+        "//envoy/network:connection_handler_interface",
+        "//envoy/network:connection_interface",
+        "//envoy/network:exception_interface",
+        "//envoy/network:filter_interface",
+        "//envoy/network:listen_socket_interface",
+        "//envoy/network:listener_interface",
+        "//envoy/server:listener_manager_interface",
+        "//envoy/stats:timespan_interface",
+        "//source/common/common:linked_object",
+        "//source/common/common:non_copyable",
+        "//source/common/event:deferred_task",
+        "//source/common/network:connection_lib",
+        "//source/common/stats:timespan_lib",
+        "//source/common/stream_info:stream_info_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "active_tcp_socket",
     srcs = ["active_tcp_socket.cc"],

source/server/active_internal_listener.cc

@@ -0,0 +1,79 @@
+#include "source/server/active_internal_listener.h"
+
+#include "envoy/network/filter.h"
+#include "envoy/stats/scope.h"
+
+#include "source/common/network/address_impl.h"
+#include "source/common/stats/timespan_impl.h"
+
+#include "active_stream_listener_base.h"
+
+namespace Envoy {
+namespace Server {
+
+ActiveInternalListener::ActiveInternalListener(Network::ConnectionHandler& conn_handler,
+                                               Event::Dispatcher& dispatcher,
+                                               Network::ListenerConfig& config)
+    : OwnedActiveStreamListenerBase(
+          conn_handler, dispatcher,
+          std::make_unique<ActiveInternalListener::NetworkInternalListener>(), config) {}
+
+ActiveInternalListener::ActiveInternalListener(Network::ConnectionHandler& conn_handler,
+                                               Event::Dispatcher& dispatcher,
+                                               Network::ListenerPtr listener,
+                                               Network::ListenerConfig& config)
+    : OwnedActiveStreamListenerBase(conn_handler, dispatcher, std::move(listener), config) {}
+
+ActiveInternalListener::~ActiveInternalListener() {
+  is_deleting_ = true;
+  // Purge sockets that have not progressed to connections. This should only happen when
+  // a listener filter stops iteration and never resumes.
+  while (!sockets_.empty()) {
+    auto removed = sockets_.front()->removeFromList(sockets_);
+    dispatcher().deferredDelete(std::move(removed));
+  }
+
+  for (auto& [chain, active_connections] : connections_by_context_) {
+    ASSERT(active_connections != nullptr);
+    auto& connections = active_connections->connections_;
+    while (!connections.empty()) {
+      connections.front()->connection_->close(Network::ConnectionCloseType::NoFlush);
+    }
+  }
+  dispatcher().clearDeferredDeleteList();
+}
+
+void ActiveInternalListener::updateListenerConfig(Network::ListenerConfig& config) {
+  ENVOY_LOG(trace, "replacing listener ", config_->listenerTag(), " by ", config.listenerTag());
+  config_ = &config;
+}
+
+void ActiveInternalListener::onAccept(Network::ConnectionSocketPtr&& socket) {
+  // Unlike tcp listener, no rebalancer is applied and won't call pickTargetHandler to account
+  // connections.
+  incNumConnections();
+
+  auto active_socket = std::make_unique<ActiveTcpSocket>(
+      *this, std::move(socket), false /* do not handle off at internal listener */);
+
+  onSocketAccepted(std::move(active_socket));
+}
+
+void ActiveInternalListener::newActiveConnection(
+    const Network::FilterChain& filter_chain, Network::ServerConnectionPtr server_conn_ptr,
+    std::unique_ptr<StreamInfo::StreamInfo> stream_info) {
+  auto& active_connections = getOrCreateActiveConnections(filter_chain);
+  auto active_connection =
+      std::make_unique<ActiveTcpConnection>(active_connections, std::move(server_conn_ptr),
+                                            dispatcher().timeSource(), std::move(stream_info));
+  // If the connection is already closed, we can just let this connection immediately die.
+  if (active_connection->connection_->state() != Network::Connection::State::Closed) {
+    ENVOY_CONN_LOG(
+        debug, "new connection from {}", *active_connection->connection_,
+        active_connection->connection_->connectionInfoProvider().remoteAddress()->asString());
+    active_connection->connection_->addConnectionCallbacks(*active_connection);
+    LinkedList::moveIntoList(std::move(active_connection), active_connections.connections_);
+  }
+}
+} // namespace Server
+} // namespace Envoy