reverse connection: reverse connection bootstrap extension

api/BUILD

@@ -72,6 +72,7 @@ proto_library(
     name = "v3_protos",
     visibility = ["//visibility:public"],
     deps = [
+        "//contrib/envoy/extensions/bootstrap/reverse_connection/v3alpha:pkg",
         "//contrib/envoy/extensions/compression/qatzip/compressor/v3alpha:pkg",
         "//contrib/envoy/extensions/compression/qatzstd/compressor/v3alpha:pkg",
         "//contrib/envoy/extensions/filters/http/checksum/v3alpha:pkg",

api/contrib/envoy/extensions/bootstrap/reverse_connection/v3alpha/BUILD

@@ -0,0 +1,9 @@
+# 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_xds//udpa/annotations:pkg"],
+)

api/contrib/envoy/extensions/bootstrap/reverse_connection/v3alpha/reverse_connection.proto

@@ -0,0 +1,22 @@
+syntax = "proto3";
+
+package envoy.extensions.bootstrap.reverse_connection.v3alpha;
+
+import "udpa/annotations/status.proto";
+import "google/protobuf/wrappers.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.bootstrap.reverse_connection.v3alpha";
+option java_outer_classname = "ReverseConnectionProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/bootstrap/reverse_connection/v3alpha";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: Bootstrap settings for Reverse Connections]
+// [#extension: envoy.bootstrap.reverse_connection]
+
+// All the future reverse connection settings should be added here.
+message ReverseConnection {
+    // Stat prefix to be used for reverse connection stats.
+    string stat_prefix = 1;
+}

api/versioning/BUILD

@@ -9,6 +9,7 @@ proto_library(
     name = "active_protos",
     visibility = ["//visibility:public"],
     deps = [
+        "//contrib/envoy/extensions/bootstrap/reverse_connection/v3alpha:pkg",
         "//contrib/envoy/extensions/compression/qatzip/compressor/v3alpha:pkg",
         "//contrib/envoy/extensions/compression/qatzstd/compressor/v3alpha:pkg",
         "//contrib/envoy/extensions/config/v3alpha:pkg",

contrib/contrib_build_config.bzl

@@ -29,6 +29,11 @@ CONTRIB_EXTENSIONS = {
     "envoy.filters.network.rocketmq_proxy":                     "//contrib/rocketmq_proxy/filters/network/source:config",
     "envoy.filters.network.golang":                             "//contrib/golang/filters/network/source:config",
 
+    #
+    # Reverse Connection
+    #
+    "envoy.bootstrap.reverse_connection":                          "//contrib/reverse_connection/bootstrap/source:reverse_conn_global_registry",
+
     #
     # Sip proxy
     #

contrib/extensions_metadata.yaml

@@ -88,6 +88,13 @@ envoy.tls.key_providers.qat:
   - envoy.tls.key_providers
   security_posture: robust_to_untrusted_downstream
   status: alpha
+envoy.bootstrap.reverse_connection:
+  categories:
+  - envoy.bootstrap
+  security_posture: unknown
+  status: wip
+  type_urls:
+  - envoy.extensions.bootstrap.reverse_connection.v3alpha.ReverseConnection
 envoy.bootstrap.vcl:
   categories:
   - envoy.bootstrap

contrib/reverse_connection/bootstrap/source/BUILD

@@ -0,0 +1,103 @@
+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 = "reverse_connection_lib",
+    srcs = [
+        "reverse_conn_thread_local_registry.cc",
+        "reverse_connection_handler.cc",
+        "reverse_connection_manager.cc",
+    ],
+    deps = [
+        ":reverse_connection_includes",
+        ":reverse_connection_initiator_lib",
+        "//contrib/reverse_connection/reverse_connection_listener_config/source:active_reverse_connection_listener_lib",
+        "//envoy/upstream:cluster_manager_interface",
+        "//source/common/api:os_sys_calls_lib",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/common:lock_guard_lib",
+        "//source/common/common:random_generator_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "reverse_connection_includes",
+    hdrs = [
+        "reverse_conn_thread_local_registry.h",
+        "reverse_connection_handler.h",
+        "reverse_connection_manager.h",
+    ],
+    deps = [
+        "//envoy/event:dispatcher_interface",
+        "//envoy/thread_local:thread_local_object",
+    ],
+)
+
+envoy_cc_library(
+    name = "reverse_connection_initiator_lib",
+    srcs = ["reverse_connection_initiator.cc"],
+    hdrs = ["reverse_connection_initiator.h"],
+    deps = [
+        ":reverse_connection_includes",
+        "//envoy/network:listener_interface",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/http:headers_lib",
+        "//source/common/http:utility_lib",
+        "//source/common/network:filter_lib",
+        "//source/common/upstream:load_balancer_context_base_lib",
+        "@envoy_api//contrib/envoy/extensions/filters/http/reverse_conn/v3alpha:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_extension(
+    name = "reverse_conn_global_registry",
+    srcs = [
+        "reverse_conn_global_registry.cc",
+    ],
+    hdrs = [
+        "reverse_conn_global_registry.h",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":reverse_connection_includes",
+        ":reverse_connection_lib",
+        "//contrib/reverse_connection/reverse_connection_listener_config/source:reverse_connection_listener_config_lib",
+        "//envoy/server:bootstrap_extension_config_interface",
+        "@envoy_api//contrib/envoy/extensions/bootstrap/reverse_connection/v3alpha:pkg_cc_proto",
+        "@envoy_api//contrib/envoy/extensions/reverse_connection/reverse_connection_listener_config/v3alpha:pkg_cc_proto",
+    ],
+    alwayslink = 1,
+)
+
+envoy_cc_library(
+    name = "conn_pool_lib",
+    srcs = ["conn_pool.cc"],
+    hdrs = ["conn_pool.h"],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":reversed_connection_lib",
+        "//envoy/event:dispatcher_interface",
+        "//source/common/http:codec_client_lib",
+        "//source/common/network:connection_impl",
+        "//source/common/http/http2:conn_pool_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "reversed_connection_lib",
+    srcs = ["reversed_connection_impl.cc"],
+    hdrs = ["reversed_connection_impl.h"],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":reverse_conn_global_registry",
+        "//source/common/network:connection_impl",
+    ],
+)

contrib/reverse_connection/bootstrap/source/conn_pool.cc

@@ -0,0 +1,108 @@
+#include "contrib/reverse_connection/bootstrap/source/conn_pool.h"
+
+#include <cstdint>
+
+#include "source/common/http/http2/codec_impl.h"
+#include "contrib/reverse_connection/bootstrap/source/reversed_connection_impl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Bootstrap {
+namespace ReverseConnection {
+
+ActiveClient::ActiveClient(Http::HttpConnPoolImplBase& parent,
+                           OptRef<Upstream::Host::CreateConnectionData> data,
+                           Http::CreateConnectionDataFn connection_fn)
+    : Envoy::Http::Http2::ActiveClient(parent, data, connection_fn) {}
+
+Http::ConnectionPool::InstancePtr
+ReverseConnPoolFactoryImpl::allocateConnPool(Event::Dispatcher& dispatcher, Random::RandomGenerator& random_generator,
+                 Singleton::Manager& singleton_manager,
+                 Upstream::HostConstSharedPtr host, Upstream::ResourcePriority priority,
+                 const Network::ConnectionSocket::OptionsSharedPtr& options,
+                 const Network::TransportSocketOptionsConstSharedPtr& transport_socket_options,
+                 Upstream::ClusterConnectivityState& state,
+                 absl::optional<Http::HttpServerPropertiesCache::Origin> origin,
+                 Http::HttpServerPropertiesCacheSharedPtr cache) {
+  ENVOY_LOG_MISC(debug, "Creating custom ActiveClient for reverse connections for host {}",
+                 host->getHostId());
+  return std::make_unique<Http::FixedHttpConnPoolImpl>(
+      host, priority, dispatcher, options, transport_socket_options, random_generator, state,
+      [&singleton_manager](Http::HttpConnPoolImplBase* pool) {
+        return std::make_unique<ActiveClient>(
+            *pool, absl::nullopt, [&singleton_manager](Http::HttpConnPoolImplBase& pool) {
+                const Upstream::HostConstSharedPtr& host =
+                    static_cast<Envoy::ConnectionPool::ConnPoolImplBase*>(&pool)->host();
+                const std::string host_id = static_cast<std::string>(host->getHostId());
+                const Upstream::ClusterInfo& cluster = host->cluster();
+                if (cluster.type() != envoy::config::cluster::v3::Cluster_DiscoveryType::
+                                        Cluster_DiscoveryType_REVERSE_CONNECTION) {
+                ENVOY_LOG_MISC(debug, "not a reverse connection cluster");
+                return host->createConnection(pool.dispatcher(), pool.socketOptions(),
+                                              pool.transportSocketOptions());
+                }
+                ENVOY_LOG_MISC(trace, "reverse connection cluster: obtaining cached socket");
+
+                // Retrieve the RevConnRegistry singleton and access the thread local slot
+                std::shared_ptr<ReverseConnRegistry> reverse_conn_registry =
+                        singleton_manager.getTyped<ReverseConnRegistry>("reverse_conn_registry_singleton");
+                if (reverse_conn_registry == nullptr) {
+                    throw EnvoyException(
+                        "Cannot access cached reverse connection socket. Reverse connection registry not found");
+                }
+                Envoy::Extensions::Bootstrap::ReverseConnection::RCThreadLocalRegistry* thread_local_registry = reverse_conn_registry->getLocalRegistry();
+                if (thread_local_registry == nullptr) {
+                    throw EnvoyException("Cannot access cached reverse connection socket. Thread local reverse connection registry is null");
+                }
+
+                std::pair<Network::ConnectionSocketPtr, bool> host_socket_pair =
+                    thread_local_registry->getRCHandler().getConnectionSocket(host_id, false);
+
+                ENVOY_LOG_MISC(trace, "reverse connection cluster: obtained cached socket");
+                Network::ConnectionSocketPtr&& host_socket = std::move(host_socket_pair.first);
+                if (!host_socket) {
+                    // fallback - try to create connection by connecting directly using address
+                    ENVOY_LOG_MISC(
+                        debug, "Could not find existing socket for host {}. Creating new connection",
+                        host_id);
+                    return host->createConnection(pool.dispatcher(), pool.socketOptions(),
+                                                pool.transportSocketOptions());
+                }
+                const bool expects_proxy_protocol = host_socket_pair.second;
+                ENVOY_LOG_MISC(debug,
+                                "Found existing socket for host {}. Using reverse connection. "
+                                "expects_proxy_protocol:{}",
+                                host_id, expects_proxy_protocol);
+                Network::TransportSocketPtr&& transport_socket =
+                    host->transportSocketFactory().createTransportSocket(
+                        pool.transportSocketOptions(), host);
+                Network::Address::InstanceConstSharedPtr source_address =
+                    cluster.getUpstreamLocalAddressSelector()
+                        ->getUpstreamLocalAddress(host->address(), pool.socketOptions())
+                        .address_;
+                ENVOY_LOG_MISC(debug, "creating ReversedClientConnectionImpl over cached socket");
+                Network::ClientConnectionPtr connection =
+                    std::make_unique<ReversedClientConnectionImpl>(
+                        host->address(), source_address, pool.dispatcher(),
+                        std::move(transport_socket), std::move(host_socket),
+                        *thread_local_registry, expects_proxy_protocol);
+                connection->setBufferLimits(cluster.perConnectionBufferLimitBytes());
+                cluster.createNetworkFilterChain(*connection);
+                return Upstream::Host::CreateConnectionData{std::move(connection), std::move(host)};
+                });
+      },
+      [](Upstream::Host::CreateConnectionData& data, Http::HttpConnPoolImplBase* pool) {
+        Http::CodecClientPtr codec{new Http::CodecClientProd(
+            Http::CodecType::HTTP2, std::move(data.connection_), data.host_description_,
+            pool->dispatcher(), pool->randomGenerator(), pool->transportSocketOptions())};
+        return codec;
+      },
+      std::vector<Http::Protocol>{Http::Protocol::Http2}, origin, cache);
+}
+
+REGISTER_FACTORY(ReverseConnPoolFactoryImpl, Http::Http2::ReverseConnPoolFactory);
+
+} // namespace ReverseConnection
+} // namespace Bootstrap
+} // namespace Extensions
+} // namespace Envoy

contrib/reverse_connection/bootstrap/source/conn_pool.h

@@ -0,0 +1,44 @@
+#pragma once
+
+#include <cstdint>
+
+#include "envoy/event/dispatcher.h"
+#include "envoy/upstream/upstream.h"
+#include "source/common/http/http2/conn_pool.h"
+
+
+namespace Envoy {
+namespace Extensions {
+namespace Bootstrap {
+namespace ReverseConnection {
+
+/**
+ * Implementation of an active client for Reverse connections
+ */
+class ActiveClient : public Http::Http2::ActiveClient {
+public:
+  ActiveClient(Envoy::Http::HttpConnPoolImplBase& parent,
+               OptRef<Upstream::Host::CreateConnectionData> data,
+               Http::CreateConnectionDataFn connection_fn = nullptr);
+};
+
+class ReverseConnPoolFactoryImpl : public Http::Http2::ReverseConnPoolFactory {
+public:
+  Http::ConnectionPool::InstancePtr allocateConnPool(
+      Event::Dispatcher& dispatcher, Random::RandomGenerator& random_generator,
+      Singleton::Manager& singleton_manager, Upstream::HostConstSharedPtr host,
+      Upstream::ResourcePriority priority, const Network::ConnectionSocket::OptionsSharedPtr& options,
+      const Network::TransportSocketOptionsConstSharedPtr& transport_socket_options,
+      Upstream::ClusterConnectivityState& state,
+      absl::optional<Http::HttpServerPropertiesCache::Origin> origin = absl::nullopt,
+      Http::HttpServerPropertiesCacheSharedPtr http_server_properties_cache = nullptr) override;
+
+      std::string name() const override { return "envoy.http.reverse_conn.default"; }
+};
+
+DECLARE_FACTORY(ReverseConnPoolFactoryImpl);
+
+} // namespace ReverseConnection
+} // namespace Bootstrap
+} // namespace Extensions
+} // namespace Envoy