Envoy core changes for reverse connections

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

@@ -12,6 +12,7 @@ import "envoy/config/listener/v3/listener_components.proto";
 import "envoy/config/listener/v3/udp_listener_config.proto";
 
 import "google/protobuf/duration.proto";
+import "google/protobuf/any.proto";
 import "google/protobuf/wrappers.proto";
 
 import "xds/annotations/v3/status.proto";
@@ -53,7 +54,7 @@ message ListenerCollection {
   repeated xds.core.v3.CollectionEntry entries = 1;
 }
 
-// [#next-free-field: 36]
+// [#next-free-field: 37]
 message Listener {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.Listener";
 
@@ -378,6 +379,11 @@ message Listener {
     // * :ref:`freebind <envoy_v3_api_field_config.listener.v3.Listener.freebind>`
     // * :ref:`transparent <envoy_v3_api_field_config.listener.v3.Listener.transparent>`
     InternalListenerConfig internal_listener = 27;
+
+    // Used to represent a reverse connection listener which, instead of binding to a port and listening,
+    // initiates reverse connections to a remote endpoint. The used sockets are cached on the remote
+    // endpoint and can be used to send request to local services.
+    google.protobuf.Any reverse_connection_listener_config = 36;
   }
 
   // Enable MPTCP (multi-path TCP) on this listener. Clients will be allowed to establish

envoy/network/connection.h

@@ -318,6 +318,27 @@ class Connection : public Event::DeferredDeletable,
    */
   virtual bool aboveHighWatermark() const PURE;
 
+  /**
+   * @return ConnectionSocketPtr& To get socket from current connection.
+   */
+  virtual const ConnectionSocketPtr& getSocket() const PURE;
+
+  /**
+   * Set the flag connection_reused_ to value. The flag connection_reused_
+   * indicates whether the client connection is reused.
+   */
+  virtual void setConnectionReused(bool value) PURE;
+
+  /**
+   *  Set flag to convey active connection (listener) is reused.
+   */
+  virtual void setActiveConnectionReused(bool value) PURE;
+
+  /**
+   *  return boolean telling if active connection (listener) is reused.
+   */
+  virtual bool isActiveConnectionReused() PURE;
+
   /**
    * Get the socket options set on this connection.
    */

envoy/network/connection_handler.h

@@ -19,6 +19,8 @@
 namespace Envoy {
 namespace Network {
 
+class LocalRevConnRegistry;
+
 // This interface allows for a listener to perform an alternative behavior when a
 // packet can't be routed correctly during draining; for example QUIC packets that
 // are not for an existing connection.
@@ -127,6 +129,20 @@ class ConnectionHandler {
    */
   virtual const std::string& statPrefix() const PURE;
 
+  /**
+   * Pass the reverse connection socket to the listener that initiated it.
+   * @param upstream_socket the socket to be passed.
+   * @param listener_tag the tag of the listener that initiated the reverse
+   * connection.
+   */
+  virtual void saveUpstreamConnection(Network::ConnectionSocketPtr&& upstream_socket,
+                                      uint64_t listener_tag) PURE;
+
+  /**
+   * @return the thread local registry.
+   */
+  virtual Network::LocalRevConnRegistry& reverseConnRegistry() const PURE;
+
   /**
    * Used by ConnectionHandler to manage listeners.
    */
@@ -303,6 +319,28 @@ class InternalListener : public virtual ConnectionHandler::ActiveListener {
 using InternalListenerPtr = std::unique_ptr<InternalListener>;
 using InternalListenerOptRef = OptRef<InternalListener>;
 
+/**
+ * Reverse connection listener callbacks.
+ */
+class ReverseConnectionListener : public virtual ConnectionHandler::ActiveListener {
+public:
+
+  /**
+   * Helper method that triggers the reverse connection workflow.
+   * @param dispatcher the thread local dispatcher.
+   * @param conn_handler the thread local connection handler.
+   * @param config the listener config that triggers the reverse connection.
+   */
+  virtual void startRCWorkflow(Event::Dispatcher& dispatcher, Network::ConnectionHandler& conn_handler, Network::ListenerConfig& config) PURE;
+
+  /**
+   * 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 ReverseConnectionListenerPtr = std::unique_ptr<ReverseConnectionListener>;
+
 /**
  * The query interface of the registered internal listener callbacks.
  */
@@ -351,5 +389,35 @@ class InternalListenerRegistry {
   virtual LocalInternalListenerRegistry* getLocalRegistry() PURE;
 };
 
+// The thread local registry.
+class LocalRevConnRegistry {
+public:
+  virtual ~LocalRevConnRegistry() = default;
+
+  virtual Network::ReverseConnectionListenerPtr
+  createActiveReverseConnectionListener(Network::ConnectionHandler& conn_handler,
+                                        Event::Dispatcher& dispatcher,
+                                        Network::ListenerConfig& config) PURE;
+};
+
+// The central reverse conn registry interface providing the thread local accessor.
+class RevConnRegistry {
+public:
+  virtual ~RevConnRegistry() = default;
+
+  /**
+   * @return The thread local registry.
+   */
+  virtual LocalRevConnRegistry* getLocalRegistry() PURE;
+
+  /**
+   * Helper function to create a ReverseConnectionListenerConfig from a google.protobuf.Any.
+   * @param config is the reverse connection listener config.
+   * @return the ReverseConnectionListenerConfig object.
+   */
+  virtual absl::StatusOr<Network::ReverseConnectionListenerConfigPtr>
+  fromAnyConfig(const google::protobuf::Any& config) PURE;
+};
+
 } // namespace Network
 } // namespace Envoy

envoy/network/filter.h

@@ -383,6 +383,12 @@ class ListenerFilter {
    */
   virtual FilterStatus onData(Network::ListenerFilterBuffer& buffer) PURE;
 
+  /**
+   * Called when the connection is closed. Only the current filter that has stopped filter
+   * chain iteration will get the callback.
+   */
+  virtual void onClose(){};
+
   /**
    * Return the size of data the filter want to inspect from the connection.
    * The size can be increased after filter need to inspect more data.

envoy/network/listen_socket.h

@@ -94,7 +94,7 @@ class ConnectionSocket : public virtual Socket {
   virtual void dumpState(std::ostream& os, int indent_level = 0) const PURE;
 };
 
-using ConnectionSocketPtr = std::unique_ptr<ConnectionSocket>;
+using ConnectionSocketPtr = std::shared_ptr<ConnectionSocket>;
 
 } // namespace Network
 } // namespace Envoy

envoy/network/listener.h

@@ -134,6 +134,42 @@ class InternalListenerConfig {
 
 using InternalListenerConfigOptRef = OptRef<InternalListenerConfig>;
 
+class RevConnRegistry;
+
+/**
+ * Configuration for a reverse connection listener.
+ */
+class ReverseConnectionListenerConfig {
+public:
+  virtual ~ReverseConnectionListenerConfig() = default;
+
+  /**
+   * Encapsulates the source node, cluster and tenant IDs of the local envoy. These are used when
+   * the listener creates reverse connections.
+   */
+  struct ReverseConnParams {
+    std::string src_node_id_;
+    std::string src_cluster_id_;
+    std::string src_tenant_id_;
+    absl::flat_hash_map<std::string, uint32_t> remote_cluster_to_conn_count_map_;
+  };
+  using ReverseConnParamsPtr = std::unique_ptr<ReverseConnParams>;
+  /**
+   * @return the private ReverseConnParams object, containing
+   * the params identifying the local envoy.
+   */
+  virtual ReverseConnParamsPtr& getReverseConnParams() PURE;
+
+  /**
+   * @return the global reverse conn registry.
+   */
+  virtual RevConnRegistry& reverseConnRegistry() PURE;
+};
+
+using ReverseConnectionListenerConfigOptRef = OptRef<ReverseConnectionListenerConfig>;
+using ReverseConnectionListenerConfigPtr = std::unique_ptr<ReverseConnectionListenerConfig>;
+using ReverseConnParamsPtr = Network::ReverseConnectionListenerConfig::ReverseConnParamsPtr;
+
 /**
  * Description of the listener.
  */
@@ -260,6 +296,17 @@ class ListenerConfig {
    */
   virtual InternalListenerConfigOptRef internalListenerConfig() PURE;
 
+  /**
+   * @return the reverse connection configuration for the listener IFF it is an reverse connection
+   * listener.
+   */
+  virtual ReverseConnectionListenerConfigOptRef reverseConnectionListenerConfig() const PURE;
+
+  /**
+   * @return the listener's version.
+   */
+  virtual const std::string& versionInfo() const PURE;
+
   /**
    * @param address is used for query the address specific connection balancer.
    * @return the connection balancer for this listener. All listeners have a connection balancer,

envoy/upstream/host_description.h

@@ -235,6 +235,17 @@ class HostDescription {
    */
   virtual absl::optional<MonotonicTime> lastHcPassTime() const PURE;
 
+  /**
+   * @return host-id to be used to retrieve reverse connection sockets from
+   * reverse connection handler.
+   */
+  virtual const absl::string_view getHostId() const PURE;
+
+  /*
+   * Set the current host-id.
+   */
+  virtual void setHostId(const std::string host_id) PURE;
+
   /**
    * Set the timestamp of when the host has transitioned from unhealthy to healthy state via an
    * active health checking.

source/common/http/codec_client.h

@@ -143,6 +143,8 @@ class CodecClient : protected Logger::Loggable<Logger::Id::client>,
    */
   void connect();
 
+  Network::ClientConnectionPtr& connection() { return connection_; }
+
   bool connectCalled() const { return connect_called_; }
 
 protected:

source/common/http/conn_pool_base.cc

@@ -189,10 +189,10 @@ uint64_t MultiplexedActiveClientBase::maxStreamsPerConnection(uint64_t max_strea
 MultiplexedActiveClientBase::MultiplexedActiveClientBase(
     HttpConnPoolImplBase& parent, uint32_t effective_concurrent_streams,
     uint32_t max_configured_concurrent_streams, Stats::Counter& cx_total,
-    OptRef<Upstream::Host::CreateConnectionData> data)
+    OptRef<Upstream::Host::CreateConnectionData> data, CreateConnectionDataFn connection_fn)
     : Envoy::Http::ActiveClient(
           parent, maxStreamsPerConnection(parent.host()->cluster().maxRequestsPerConnection()),
-          effective_concurrent_streams, max_configured_concurrent_streams, data) {
+          effective_concurrent_streams, max_configured_concurrent_streams, data, connection_fn) {
   codec_client_->setCodecClientCallbacks(*this);
   codec_client_->setCodecConnectionCallbacks(*this);
   cx_total.inc();

source/common/http/conn_pool_base.h

@@ -102,25 +102,31 @@ class HttpConnPoolImplBase : public Envoy::ConnectionPool::ConnPoolImplBase,
   absl::optional<HttpServerPropertiesCache::Origin> origin_;
 };
 
+using CreateConnectionDataFn =
+    std::function<Upstream::Host::CreateConnectionData(HttpConnPoolImplBase& pool)>;
 // An implementation of Envoy::ConnectionPool::ActiveClient for HTTP/1.1 and HTTP/2
 class ActiveClient : public Envoy::ConnectionPool::ActiveClient {
 public:
-  ActiveClient(HttpConnPoolImplBase& parent, uint64_t lifetime_stream_limit,
-               uint64_t effective_concurrent_stream_limit,
-               uint64_t configured_concurrent_stream_limit,
-               OptRef<Upstream::Host::CreateConnectionData> opt_data)
+  ActiveClient(
+      HttpConnPoolImplBase& parent, uint64_t lifetime_stream_limit,
+      uint64_t effective_concurrent_stream_limit, uint64_t configured_concurrent_stream_limit,
+      OptRef<Upstream::Host::CreateConnectionData> opt_data,
+      CreateConnectionDataFn connection_fn =
+          [](HttpConnPoolImplBase& parent) {
+            return static_cast<Envoy::ConnectionPool::ConnPoolImplBase*>(&parent)
+                ->host()
+                ->createConnection(parent.dispatcher(), parent.socketOptions(),
+                                   parent.transportSocketOptions());
+          })
       : Envoy::ConnectionPool::ActiveClient(parent, lifetime_stream_limit,
                                             effective_concurrent_stream_limit,
                                             configured_concurrent_stream_limit) {
     if (opt_data.has_value()) {
       initialize(opt_data.value(), parent);
       return;
     }
-    // The static cast makes sure we call the base class host() and not
-    // HttpConnPoolImplBase::host which is of a different type.
-    Upstream::Host::CreateConnectionData data =
-        static_cast<Envoy::ConnectionPool::ConnPoolImplBase*>(&parent)->host()->createConnection(
-            parent.dispatcher(), parent.socketOptions(), parent.transportSocketOptions());
+    ENVOY_LOG(debug, "Creating CreateConnectionData");
+    Upstream::Host::CreateConnectionData data = connection_fn(parent);
     initialize(data, parent);
   }
 
@@ -204,7 +210,8 @@ class MultiplexedActiveClientBase : public CodecClientCallbacks,
 public:
   MultiplexedActiveClientBase(HttpConnPoolImplBase& parent, uint32_t effective_concurrent_streams,
                               uint32_t max_configured_concurrent_streams, Stats::Counter& cx_total,
-                              OptRef<Upstream::Host::CreateConnectionData> data);
+                              OptRef<Upstream::Host::CreateConnectionData> data,
+                              CreateConnectionDataFn connection_fn = nullptr);
   ~MultiplexedActiveClientBase() override = default;
   // Caps max streams per connection below 2^31 to prevent overflow.
   static uint64_t maxStreamsPerConnection(uint64_t max_streams_config);

source/common/http/filter_manager.cc

@@ -1000,10 +1000,12 @@ void DownstreamFilterManager::sendLocalReply(
       // route refreshment in the response filter chain.
       cb->route(nullptr);
     }
-
+
+    bool reverse_conn_force_local_reply = 
+      Runtime::runtimeFeatureEnabled("envoy.reloadable_features.reverse_conn_force_local_reply");
     // We only prepare a local reply to execute later if we're actively
     // invoking filters to avoid re-entrant in filters.
-    if (state_.filter_call_state_ & FilterCallState::IsDecodingMask) {
+    if (!reverse_conn_force_local_reply && state_.filter_call_state_ & FilterCallState::IsDecodingMask) {
       prepareLocalReplyViaFilterChain(is_grpc_request, code, body, modify_headers, is_head_request,
                                       grpc_status, details);
     } else {

source/common/http/headers.h

@@ -238,6 +238,8 @@ class HeaderValues {
   const LowerCaseString XContentTypeOptions{"x-content-type-options"};
   const LowerCaseString XSquashDebug{"x-squash-debug"};
   const LowerCaseString EarlyData{"early-data"};
+  const LowerCaseString EnvoyDstNodeUUID{"x-remote-node-id"};
+  const LowerCaseString EnvoyDstClusterUUID{"x-dst-cluster-uuid"};
 
   struct {
     const std::string Close{"close"};

source/common/http/http2/BUILD

@@ -83,6 +83,7 @@ envoy_cc_library(
     srcs = ["conn_pool.cc"],
     hdrs = ["conn_pool.h"],
     deps = [
+        "//envoy/config:typed_config_interface",
         "//envoy/event:dispatcher_interface",
         "//envoy/upstream:upstream_interface",
         "//source/common/http:codec_client_lib",

source/common/http/http2/conn_pool.cc

@@ -2,8 +2,8 @@
 
 #include <cstdint>
 
+#include "envoy/config/typed_config.h"
 #include "envoy/event/dispatcher.h"
-#include "envoy/upstream/upstream.h"
 
 #include "source/common/http/http2/codec_impl.h"
 #include "source/common/runtime/runtime_features.h"
@@ -37,11 +37,12 @@ uint32_t ActiveClient::calculateInitialStreamsLimit(
 }
 
 ActiveClient::ActiveClient(HttpConnPoolImplBase& parent,
-                           OptRef<Upstream::Host::CreateConnectionData> data)
+                           OptRef<Upstream::Host::CreateConnectionData> data,
+                           CreateConnectionDataFn connection_fn)
     : MultiplexedActiveClientBase(
           parent, calculateInitialStreamsLimit(parent.cache(), parent.origin(), parent.host()),
           parent.host()->cluster().http2Options().max_concurrent_streams().value(),
-          parent.host()->cluster().trafficStats()->upstream_cx_http2_total_, data) {}
+          parent.host()->cluster().trafficStats()->upstream_cx_http2_total_, data, connection_fn) {}
 
 ConnectionPool::InstancePtr
 allocateConnPool(Event::Dispatcher& dispatcher, Random::RandomGenerator& random_generator,