http: alpn upstream (#13922)

Signed-off-by: Alyssa Wilk <alyssar@chromium.org>

api/envoy/extensions/upstreams/http/v3/http_protocol_options.proto

@@ -56,6 +56,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 //             http2_protocol_options:
 //               max_concurrent_streams: 100
 //        .... [further cluster config]
+// [#next-free-field: 6]
 message HttpProtocolOptions {
   // If this is used, the cluster will only operate on one of the possible upstream protocols (HTTP/1.1, HTTP/2).
   // Note that HTTP/2 should generally be used for upstream clusters doing gRPC.
@@ -77,6 +78,21 @@ message HttpProtocolOptions {
     config.core.v3.Http2ProtocolOptions http2_protocol_options = 2;
   }
 
+  // If this is used, the cluster can use either HTTP/1 or HTTP/2, and will use whichever
+  // protocol is negotiated by ALPN with the upstream.
+  // Clusters configured with *AutoHttpConfig* will use the highest available
+  // protocol; HTTP/2 if supported, otherwise HTTP/1.
+  // If the upstream does not support ALPN, *AutoHttpConfig* will fail over to HTTP/1.
+  // This can only be used with transport sockets which support ALPN. Using a
+  // transport socket which does not support ALPN will result in configuration
+  // failure. The transport layer may be configured with custom ALPN, but the default ALPN
+  // for the cluster (or if custom ALPN fails) will be "h2,http/1.1".
+  message AutoHttpConfig {
+    config.core.v3.Http1ProtocolOptions http_protocol_options = 1;
+
+    config.core.v3.Http2ProtocolOptions http2_protocol_options = 2;
+  }
+
   // This contains options common across HTTP/1 and HTTP/2
   config.core.v3.HttpProtocolOptions common_http_protocol_options = 1;
 
@@ -94,5 +110,8 @@ message HttpProtocolOptions {
     // This allows switching on protocol based on what protocol the downstream
     // connection used.
     UseDownstreamHttpConfig use_downstream_protocol_config = 4;
+
+    // This allows switching on protocol based on ALPN
+    AutoHttpConfig auto_config = 5;
   }
 }

docs/root/version_history/current.rst

@@ -69,6 +69,7 @@ New Features
 * health_check: added option to use :ref:`no_traffic_healthy_interval <envoy_v3_api_field_config.core.v3.HealthCheck.no_traffic_healthy_interval>` which allows a different no traffic interval when the host is healthy.
 * http: added HCM :ref:`timeout config field <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.request_headers_timeout>` to control how long a downstream has to finish sending headers before the stream is cancelled.
 * http: added frame flood and abuse checks to the upstream HTTP/2 codec. This check is off by default and can be enabled by setting the `envoy.reloadable_features.upstream_http2_flood_checks` runtime key to true.
+* http: clusters now support selecting HTTP/1 or HTTP/2 based on ALPN, configurable via :ref:`alpn_config <envoy_v3_api_field_extensions.upstreams.http.v3.HttpProtocolOptions.auto_config>` in the :ref:`http_protocol_options <envoy_v3_api_msg_extensions.upstreams.http.v3.HttpProtocolOptions>` message.
 * jwt_authn: added support for :ref:`per-route config <envoy_v3_api_msg_extensions.filters.http.jwt_authn.v3.PerRouteConfig>`.
 * kill_request: added new :ref:`HTTP kill request filter <config_http_filters_kill_request>`.
 * listener: added an optional :ref:`default filter chain <envoy_v3_api_field_config.listener.v3.Listener.default_filter_chain>`. If this field is supplied, and none of the :ref:`filter_chains <envoy_v3_api_field_config.listener.v3.Listener.filter_chains>` matches, this default filter chain is used to serve the connection.

include/envoy/network/transport_socket.h

@@ -241,6 +241,12 @@ class TransportSocketFactory {
    * @return bool whether the transport socket will use proxy protocol options.
    */
   virtual bool usesProxyProtocolOptions() const PURE;
+
+  /**
+   * Returns true if the transport socket created by this factory supports some form of ALPN
+   * negotiation.
+   */
+  virtual bool supportsAlpn() const { return false; }
 };
 
 using TransportSocketFactoryPtr = std::unique_ptr<TransportSocketFactory>;

include/envoy/upstream/cluster_manager.h

@@ -353,7 +353,7 @@ class ClusterManagerFactory {
    */
   virtual Http::ConnectionPool::InstancePtr
   allocateConnPool(Event::Dispatcher& dispatcher, HostConstSharedPtr host,
-                   ResourcePriority priority, Http::Protocol protocol,
+                   ResourcePriority priority, std::vector<Http::Protocol>& protocol,
                    const Network::ConnectionSocket::OptionsSharedPtr& options,
                    const Network::TransportSocketOptionsSharedPtr& transport_socket_options,
                    ClusterConnectivityState& state) PURE;

include/envoy/upstream/upstream.h

@@ -717,6 +717,9 @@ class ClusterInfo {
     static const uint64_t USE_DOWNSTREAM_PROTOCOL = 0x2;
     // Whether connections should be immediately closed upon health failure.
     static const uint64_t CLOSE_CONNECTIONS_ON_HOST_HEALTH_FAILURE = 0x4;
+    // If USE_ALPN and HTTP2 are true, the upstream protocol will be negotiated using ALPN.
+    // If ALPN is attempted but not supported by the upstream HTTP/1.1 is used.
+    static const uint64_t USE_ALPN = 0x8;
   };
 
   virtual ~ClusterInfo() = default;
@@ -966,9 +969,9 @@ class ClusterInfo {
   virtual void createNetworkFilterChain(Network::Connection& connection) const PURE;
 
   /**
-   * Calculate upstream protocol based on features.
+   * Calculate upstream protocol(s) based on features.
    */
-  virtual Http::Protocol
+  virtual std::vector<Http::Protocol>
   upstreamHttpProtocol(absl::optional<Http::Protocol> downstream_protocol) const PURE;
 
   /**

source/common/conn_pool/conn_pool_base.cc

@@ -329,6 +329,11 @@ void ConnPoolImplBase::onConnectionEvent(ActiveClient& client, absl::string_view
     connecting_stream_capacity_ -= client.effectiveConcurrentStreamLimit();
   }
 
+  if (client.connect_timer_) {
+    client.connect_timer_->disableTimer();
+    client.connect_timer_.reset();
+  }
+
   if (event == Network::ConnectionEvent::RemoteClose ||
       event == Network::ConnectionEvent::LocalClose) {
     state_.decrConnectingStreamCapacity(client.currentUnusedCapacity());
@@ -387,18 +392,15 @@ void ConnPoolImplBase::onConnectionEvent(ActiveClient& client, absl::string_view
   } else if (event == Network::ConnectionEvent::Connected) {
     client.conn_connect_ms_->complete();
     client.conn_connect_ms_.reset();
-
     ASSERT(client.state_ == ActiveClient::State::CONNECTING);
     transitionActiveClientState(client, ActiveClient::State::READY);
 
+    // At this point, for the mixed ALPN pool, the client may be deleted. Do not
+    // refer to client after this point.
+    onConnected(client);
     onUpstreamReady();
     checkForDrained();
   }
-
-  if (client.connect_timer_) {
-    client.connect_timer_->disableTimer();
-    client.connect_timer_.reset();
-  }
 }
 
 PendingStream::PendingStream(ConnPoolImplBase& parent) : parent_(parent) {

source/common/conn_pool/conn_pool_base.h

@@ -49,6 +49,8 @@ class ActiveClient : public LinkedObject<ActiveClient>,
     return std::min(remaining_streams_, concurrent_stream_limit_);
   }
 
+  // Returns the application protocol, or absl::nullopt for TCP.
+  virtual absl::optional<Http::Protocol> protocol() const PURE;
   uint32_t currentUnusedCapacity() const {
     return std::min(remaining_streams_, concurrent_stream_limit_ - numActiveStreams());
   }
@@ -183,6 +185,8 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   bool hasPendingStreams() const { return !pending_streams_.empty(); }
 
 protected:
+  virtual void onConnected(Envoy::ConnectionPool::ActiveClient&) {}
+
   // Creates up to 3 connections, based on the prefetch ratio.
   void tryCreateNewConnections();
 
@@ -211,8 +215,13 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
 
   bool hasActiveStreams() const { return num_active_streams_ > 0; }
 
-  void decrConnectingStreamCapacity(int32_t delta) {
+  void incrConnectingStreamCapacity(uint32_t delta) {
+    state_.incrConnectingStreamCapacity(delta);
+    connecting_stream_capacity_ += delta;
+  }
+  void decrConnectingStreamCapacity(uint32_t delta) {
     state_.decrConnectingStreamCapacity(delta);
+    ASSERT(connecting_stream_capacity_ > delta);
     connecting_stream_capacity_ -= delta;
   }
 
@@ -241,16 +250,16 @@ class ConnPoolImplBase : protected Logger::Loggable<Logger::Id::pool> {
   // Clients that are not ready to handle additional streams because they are CONNECTING.
   std::list<ActiveClientPtr> connecting_clients_;
 
+  // The number of streams that can be immediately dispatched
+  // if all CONNECTING connections become connected.
+  uint32_t connecting_stream_capacity_{0};
+
 private:
   std::list<PendingStreamPtr> pending_streams_;
 
   // The number of streams currently attached to clients.
   uint32_t num_active_streams_{0};
 
-  // The number of streams that can be immediately dispatched
-  // if all CONNECTING connections become connected.
-  uint32_t connecting_stream_capacity_{0};
-
   void onUpstreamReady();
   Event::SchedulableCallbackPtr upstream_ready_cb_;
 };

source/common/http/BUILD

@@ -141,6 +141,18 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "mixed_conn_pool",
+    srcs = ["mixed_conn_pool.cc"],
+    hdrs = ["mixed_conn_pool.h"],
+    deps = [
+        ":conn_pool_base_lib",
+        "//source/common/http/http1:conn_pool_lib",
+        "//source/common/http/http2:conn_pool_lib",
+        "//source/common/tcp:conn_pool_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "conn_manager_config_interface",
     hdrs = ["conn_manager_config.h"],

source/common/http/codec_client.cc

@@ -36,8 +36,15 @@ CodecClient::CodecClient(Type type, Network::ClientConnectionPtr&& connection,
   connection_->addConnectionCallbacks(*this);
   connection_->addReadFilter(Network::ReadFilterSharedPtr{new CodecReadFilter(*this)});
 
-  ENVOY_CONN_LOG(debug, "connecting", *connection_);
-  connection_->connect();
+  // In general, codecs are handed new not-yet-connected connections, but in the
+  // case of ALPN, the codec may be handed an already connected connection.
+  if (!connection_->connecting()) {
+    ASSERT(connection_->state() == Network::Connection::State::Open);
+    connected_ = true;
+  } else {
+    ENVOY_CONN_LOG(debug, "connecting", *connection_);
+    connection_->connect();
+  }
 
   if (idle_timeout_) {
     idle_timer_ = dispatcher.createTimer([this]() -> void { onIdleTimeout(); });

source/common/http/conn_pool_base.h

@@ -101,6 +101,13 @@ class ActiveClient : public Envoy::ConnectionPool::ActiveClient {
     initialize(data, parent);
   }
 
+  ActiveClient(HttpConnPoolImplBase& parent, uint64_t lifetime_stream_limit,
+               uint64_t concurrent_stream_limit, Upstream::Host::CreateConnectionData& data)
+      : Envoy::ConnectionPool::ActiveClient(parent, lifetime_stream_limit,
+                                            concurrent_stream_limit) {
+    initialize(data, parent);
+  }
+
   void initialize(Upstream::Host::CreateConnectionData& data, HttpConnPoolImplBase& parent) {
     real_host_description_ = data.host_description_;
     codec_client_ = parent.createCodecClient(data);
@@ -113,6 +120,7 @@ class ActiveClient : public Envoy::ConnectionPool::ActiveClient {
          &parent_.host()->cluster().stats().bind_errors_, nullptr});
   }
 
+  absl::optional<Http::Protocol> protocol() const override { return codec_client_->protocol(); }
   void close() override { codec_client_->close(); }
   virtual Http::RequestEncoder& newStreamEncoder(Http::ResponseDecoder& response_decoder) PURE;
   void onEvent(Network::ConnectionEvent event) override {

source/common/http/http1/conn_pool.cc

@@ -33,7 +33,7 @@ ActiveClient::StreamWrapper::~StreamWrapper() {
   // Upstream connection might be closed right after response is complete. Setting delay=true
   // here to attach pending requests in next dispatcher loop to handle that case.
   // https://github.com/envoyproxy/envoy/issues/2715
-  parent_.parent().onStreamClosed(parent_, true);
+  parent_.parent_.onStreamClosed(parent_, true);
 }
 
 void ActiveClient::StreamWrapper::onEncodeComplete() { encode_complete_ = true; }
@@ -95,6 +95,14 @@ ActiveClient::ActiveClient(HttpConnPoolImplBase& parent)
   parent.host()->cluster().stats().upstream_cx_http1_total_.inc();
 }
 
+ActiveClient::ActiveClient(HttpConnPoolImplBase& parent, Upstream::Host::CreateConnectionData& data)
+    : Envoy::Http::ActiveClient(
+          parent, parent.host()->cluster().maxRequestsPerConnection(),
+          1, // HTTP1 always has a concurrent-request-limit of 1 per connection.
+          data) {
+  parent.host()->cluster().stats().upstream_cx_http1_total_.inc();
+}
+
 bool ActiveClient::closingWithIncompleteStream() const {
   return (stream_wrapper_ != nullptr) && (!stream_wrapper_->decode_complete_);
 }

source/common/http/http1/conn_pool.h

@@ -17,6 +17,7 @@ namespace Http1 {
 class ActiveClient : public Envoy::Http::ActiveClient {
 public:
   ActiveClient(HttpConnPoolImplBase& parent);
+  ActiveClient(HttpConnPoolImplBase& parent, Upstream::Host::CreateConnectionData& data);
 
   // ConnPoolImplBase::ActiveClient
   bool closingWithIncompleteStream() const override;