WebSocket: allow configuring idle timeout and reconnects. (envoyproxy…

…#3277)

* WebSocket: allow configuring idle timeout and reconnects.

This also resolves a TODO in TcpProxy by always passing it a
configuration, even in the WebSocket case.

Signed-off-by: Greg Greenway <[email protected]>

api/envoy/api/v2/route/route.proto

@@ -305,6 +305,7 @@ message CorsPolicy {
   google.protobuf.BoolValue enabled = 7;
 }
 
+// [#comment:next free field: 23]
 message RouteAction {
   oneof cluster_specifier {
     option (validate.required) = true;
@@ -553,24 +554,48 @@ message RouteAction {
   // backend).
   repeated HashPolicy hash_policy = 15;
 
-  // Indicates that a HTTP/1.1 client connection to this particular route
-  // should be allowed (and expected) to upgrade to a WebSocket connection. The
-  // default is false.
+  // Indicates that a HTTP/1.1 client connection to this particular route is allowed to
+  // upgrade to a WebSocket connection. The default is false.
   //
   // .. attention::
   //
-  //   If set to true, Envoy will expect the first request matching this route to
-  //   contain WebSocket upgrade headers. If the headers are not present, the
-  //   connection will be rejected. If set to true, Envoy will setup plain TCP
+  //   If a connection is upgraded to a WebSocket connection, Envoy will set up plain TCP
   //   proxying between the client and the upstream server. Hence, an upstream
   //   server that rejects the WebSocket upgrade request is also responsible for
   //   closing the associated connection. Until then, Envoy will continue to
   //   proxy data from the client to the upstream server.
   //
-  //   Redirects, timeouts and retries are not supported on routes where websocket upgrades are
-  //   allowed.
+  //   Redirects are not supported on routes where WebSocket upgrades are allowed.
   google.protobuf.BoolValue use_websocket = 16;
 
+  message WebSocketProxyConfig {
+    // See :ref:`stat_prefix <envoy_api_field_config.filter.network.tcp_proxy.v2.TcpProxy.stat_prefix>`.
+    // If the parameter is not specified, the default value of "websocket" is used.
+    //
+    // WebSocket connections support the :ref:`downstream statistics
+    // <config_network_filters_tcp_proxy_stats>` for TCP proxy, except for the following, which are reported
+    // in the :ref:`HTTP Connection Manager statistics <config_http_conn_man_stats>`:
+    // - downstream_cx_tx_bytes_total
+    // - downstream_cx_tx_bytes_buffered
+    // - downstream_cx_rx_bytes_total
+    // - downstream_cx_rx_bytes_buffered
+    string stat_prefix = 1;
+
+    // See :ref:`idle_timeout <envoy_api_field_config.filter.network.tcp_proxy.v2.TcpProxy.idle_timeout>`.
+    // This timeout is only in effect after the WebSocket upgrade request is received by Envoy. It does
+    // not cover the initial part of the HTTP request.
+    google.protobuf.Duration idle_timeout = 2
+      [(validate.rules).duration.gt = {}, (gogoproto.stdduration) = true];
+
+    // See :ref:`max_connect_attempts
+    // <envoy_api_field_config.filter.network.tcp_proxy.v2.TcpProxy.max_connect_attempts>`.
+    google.protobuf.UInt32Value max_connect_attempts = 3 [(validate.rules).uint32.gte = 1];
+  }
+
+  // Proxy configuration used for WebSocket connections. If unset, the default values as specified
+  // in :ref:`TcpProxy <envoy_api_msg_config.filter.network.tcp_proxy.v2.TcpProxy>` are used.
+  WebSocketProxyConfig websocket_config = 22;
+
   // Indicates that the route has a CORS policy.
   CorsPolicy cors = 17;
 

docs/root/configuration/network_filters/tcp_proxy_filter.rst

@@ -21,8 +21,13 @@ statistics are rooted at *tcp.<stat_prefix>.* with the following statistics:
   :widths: 1, 1, 2
 
   downstream_cx_total, Counter, Total number of connections handled by the filter
-  downstream_cx_no_route, Counter, Number of connections for which no matching route was found
+  downstream_cx_no_route, Counter, Number of connections for which no matching route was found or the cluster for the route was not found
   downstream_cx_tx_bytes_total, Counter, Total bytes written to the downstream connection
   downstream_cx_tx_bytes_buffered, Gauge, Total bytes currently buffered to the downstream connection
+  downstream_cx_rx_bytes_total, Counter, Total bytes read from the downstream connection
+  downstream_cx_rx_bytes_buffered, Gauge, Total bytes currently buffered from the downstream connection
   downstream_flow_control_paused_reading_total, Counter, Total number of times flow control paused reading from downstream
   downstream_flow_control_resumed_reading_total, Counter, Total number of times flow control resumed reading from downstream
+  idle_timeout, Counter, Total number of connections closed due to idle timeout
+  upstream_flush_total, Counter, Total number of connections that continued to flush upstream data after the downstream connection was closed
+  upstream_flush_active, Gauge, Total connections currently continuing to flush upstream data after the downstream connection was closed

docs/root/intro/version_history.rst

@@ -74,6 +74,8 @@ Version history
 * tracing: the sampling decision is now delegated to the tracers, allowing the tracer to decide when and if
   to use it. For example, if the :ref:`x-b3-sampled <config_http_conn_man_headers_x-b3-sampled>` header
   is supplied with the client request, its value will override any sampling decision made by the Envoy proxy.
+* websocket: support configuring
+  :ref:`idle_timeout and max_connect_attempts <envoy_api_field_route.RouteAction.websocket_config>`.
 
 1.6.0 (March 20, 2018)
 ======================

include/envoy/http/BUILD

@@ -58,7 +58,7 @@ envoy_cc_library(
 )
 
 envoy_cc_library(
-    name = "wshandler_callback_interface",
+    name = "websocket_interface",
     hdrs = ["websocket.h"],
     deps = [
         ":header_map_interface",

include/envoy/http/websocket.h

@@ -1,16 +1,18 @@
 #pragma once
 
+#include "envoy/buffer/buffer.h"
 #include "envoy/http/header_map.h"
+#include "envoy/network/filter.h"
 
 namespace Envoy {
 namespace Http {
 
 /**
  * Callback interface for WebSocket connection management.
  */
-class WsHandlerCallbacks {
+class WebSocketProxyCallbacks {
 public:
-  virtual ~WsHandlerCallbacks() {}
+  virtual ~WebSocketProxyCallbacks() {}
 
   /**
    * Used by a WebSocket implementation to send HTTP error codes back to the
@@ -19,5 +21,19 @@ class WsHandlerCallbacks {
   virtual void sendHeadersOnlyResponse(HeaderMap& headers) PURE;
 };
 
+/**
+ * An instance of a WebSocketProxy.
+ */
+class WebSocketProxy {
+public:
+  virtual ~WebSocketProxy() {}
+
+  /**
+   * @see Network::Filter::onData
+   */
+  virtual Network::FilterStatus onData(Buffer::Instance& data, bool end_stream) PURE;
+};
+typedef std::unique_ptr<WebSocketProxy> WebSocketProxyPtr;
+
 } // namespace Http
 } // namespace Envoy

include/envoy/router/BUILD

@@ -41,10 +41,12 @@ envoy_cc_library(
         "//include/envoy/http:codec_interface",
         "//include/envoy/http:codes_interface",
         "//include/envoy/http:header_map_interface",
+        "//include/envoy/http:websocket_interface",
         "//include/envoy/tracing:http_tracer_interface",
         "//include/envoy/upstream:resource_manager_interface",
         "//source/common/protobuf",
         "//source/common/protobuf:utility_lib",
+        "@envoy_api//envoy/api/v2:rds_cc",
     ],
 )
 

include/envoy/router/router.h

@@ -13,6 +13,7 @@
 #include "envoy/http/codec.h"
 #include "envoy/http/codes.h"
 #include "envoy/http/header_map.h"
+#include "envoy/http/websocket.h"
 #include "envoy/tracing/http_tracer.h"
 #include "envoy/upstream/resource_manager.h"
 
@@ -22,6 +23,11 @@
 #include "absl/types/optional.h"
 
 namespace Envoy {
+
+namespace Upstream {
+class ClusterManager;
+}
+
 namespace Router {
 
 /**
@@ -476,6 +482,19 @@ class RouteEntry : public ResponseEntry {
    */
   virtual bool useWebSocket() const PURE;
 
+  /**
+   * Create an instance of a WebSocketProxy, using the configuration in this route.
+   *
+   * This may only be called if useWebSocket() returns true on this RouteEntry.
+   *
+   * @return WebSocketProxyPtr An instance of a WebSocketProxy with the configuration specified
+   *         in this route.
+   */
+  virtual Http::WebSocketProxyPtr createWebSocketProxy(
+      Http::HeaderMap& request_headers, const RequestInfo::RequestInfo& request_info,
+      Http::WebSocketProxyCallbacks& callbacks, Upstream::ClusterManager& cluster_manager,
+      Network::ReadFilterCallbacks* read_callbacks) const PURE;
+
   /**
    * @return MetadataMatchCriteria* the metadata that a subset load balancer should match when
    * selecting an upstream host

source/common/http/BUILD

@@ -115,7 +115,7 @@ envoy_cc_library(
         "//include/envoy/http:codec_interface",
         "//include/envoy/http:filter_interface",
         "//include/envoy/http:header_map_interface",
-        "//include/envoy/http:wshandler_callback_interface",
+        "//include/envoy/http:websocket_interface",
         "//include/envoy/local_info:local_info_interface",
         "//include/envoy/network:connection_interface",
         "//include/envoy/network:drain_decision_interface",
@@ -137,7 +137,6 @@ envoy_cc_library(
         "//source/common/common:utility_lib",
         "//source/common/http/http1:codec_lib",
         "//source/common/http/http2:codec_lib",
-        "//source/common/http/websocket:ws_handler_lib",
         "//source/common/network:utility_lib",
         "//source/common/request_info:request_info_lib",
         "//source/common/runtime:uuid_util_lib",

source/common/http/async_client_impl.h

@@ -199,6 +199,12 @@ class AsyncStreamImpl : public AsyncClient::Stream,
     const Router::VirtualHost& virtualHost() const override { return virtual_host_; }
     bool autoHostRewrite() const override { return false; }
     bool useWebSocket() const override { return false; }
+    Http::WebSocketProxyPtr createWebSocketProxy(Http::HeaderMap&, const RequestInfo::RequestInfo&,
+                                                 Http::WebSocketProxyCallbacks&,
+                                                 Upstream::ClusterManager&,
+                                                 Network::ReadFilterCallbacks*) const override {
+      NOT_IMPLEMENTED;
+    }
     bool includeVirtualHostRateLimits() const override { return true; }
     const envoy::api::v2::core::Metadata& metadata() const override { return metadata_; }
     const Router::PathMatchCriterion& pathMatchCriterion() const override {

source/common/http/conn_manager_impl.cc

@@ -569,10 +569,10 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
     if (websocket_requested && websocket_allowed) {
       ENVOY_STREAM_LOG(debug, "found websocket connection. (end_stream={}):", *this, end_stream);
 
-      connection_manager_.ws_connection_.reset(new WebSocket::WsHandlerImpl(
-          *request_headers_, request_info_, *route_entry, *this,
-          connection_manager_.cluster_manager_, connection_manager_.read_callbacks_));
-      connection_manager_.ws_connection_->onNewConnection();
+      connection_manager_.ws_connection_ = route_entry->createWebSocketProxy(
+          *request_headers_, request_info_, *this, connection_manager_.cluster_manager_,
+          connection_manager_.read_callbacks_);
+      ASSERT(connection_manager_.ws_connection_ != nullptr);
       connection_manager_.stats_.named_.downstream_cx_websocket_active_.inc();
       connection_manager_.stats_.named_.downstream_cx_http1_active_.dec();
       connection_manager_.stats_.named_.downstream_cx_websocket_total_.inc();

source/common/http/conn_manager_impl.h

@@ -27,7 +27,6 @@
 #include "common/common/linked_object.h"
 #include "common/http/conn_manager_config.h"
 #include "common/http/user_agent.h"
-#include "common/http/websocket/ws_handler_impl.h"
 #include "common/request_info/request_info_impl.h"
 #include "common/tracing/http_tracer_impl.h"
 
@@ -241,7 +240,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
                         public StreamCallbacks,
                         public StreamDecoder,
                         public FilterChainFactoryCallbacks,
-                        public WsHandlerCallbacks,
+                        public WebSocketProxyCallbacks,
                         public Tracing::Config {
     ActiveStream(ConnectionManagerImpl& connection_manager);
     ~ActiveStream();
@@ -289,7 +288,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     }
     void addAccessLogHandler(AccessLog::InstanceSharedPtr handler) override;
 
-    // Http::WsHandlerCallbacks
+    // Http::WebSocketProxyCallbacks
     void sendHeadersOnlyResponse(HeaderMap& headers) override {
       encodeHeaders(nullptr, headers, true);
     }
@@ -417,7 +416,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
   Runtime::Loader& runtime_;
   const LocalInfo::LocalInfo& local_info_;
   Upstream::ClusterManager& cluster_manager_;
-  WebSocket::WsHandlerImplPtr ws_connection_{};
+  WebSocketProxyPtr ws_connection_;
   Network::ReadFilterCallbacks* read_callbacks_{};
   ConnectionManagerListenerStats& listener_stats_;
 };

source/common/http/websocket/BUILD

@@ -15,7 +15,7 @@ envoy_cc_library(
     deps = [
         "//include/envoy/http:codec_interface",
         "//include/envoy/http:header_map_interface",
-        "//include/envoy/http:wshandler_callback_interface",
+        "//include/envoy/http:websocket_interface",
         "//include/envoy/network:connection_interface",
         "//include/envoy/network:filter_interface",
         "//include/envoy/router:router_interface",

source/common/http/websocket/ws_handler_impl.cc

@@ -13,16 +13,52 @@ namespace Envoy {
 namespace Http {
 namespace WebSocket {
 
+Extensions::NetworkFilters::TcpProxy::TcpProxyConfigSharedPtr
+tcpProxyConfig(const envoy::api::v2::route::RouteAction& route_config,
+               Server::Configuration::FactoryContext& factory_context) {
+  envoy::config::filter::network::tcp_proxy::v2::TcpProxy tcp_config;
+
+  // Set the default value. This may be overwritten below.
+  tcp_config.set_stat_prefix("websocket");
+
+  if (route_config.has_websocket_config()) {
+    // WebSocket has its own TcpProxy config type because some of the fields
+    // in envoy::config::filter::network::tcp_proxy::v2::TcpProxy don't apply, and some
+    // are duplicated in the route config (such as the upstream cluster).
+    const envoy::api::v2::route::RouteAction::WebSocketProxyConfig& ws_config =
+        route_config.websocket_config();
+
+    if (!ws_config.stat_prefix().empty()) {
+      tcp_config.set_stat_prefix(ws_config.stat_prefix());
+    }
+
+    if (ws_config.has_idle_timeout()) {
+      *tcp_config.mutable_idle_timeout() = ws_config.idle_timeout();
+    }
+
+    if (ws_config.has_max_connect_attempts()) {
+      *tcp_config.mutable_max_connect_attempts() = ws_config.max_connect_attempts();
+    }
+  }
+  return std::make_shared<Extensions::NetworkFilters::TcpProxy::TcpProxyConfig>(tcp_config,
+                                                                                factory_context);
+}
+
 WsHandlerImpl::WsHandlerImpl(HeaderMap& request_headers,
                              const RequestInfo::RequestInfo& request_info,
-                             const Router::RouteEntry& route_entry, WsHandlerCallbacks& callbacks,
+                             const Router::RouteEntry& route_entry,
+                             WebSocketProxyCallbacks& callbacks,
                              Upstream::ClusterManager& cluster_manager,
-                             Network::ReadFilterCallbacks* read_callbacks)
-    : Extensions::NetworkFilters::TcpProxy::TcpProxyFilter(nullptr, cluster_manager),
+                             Network::ReadFilterCallbacks* read_callbacks,
+                             Extensions::NetworkFilters::TcpProxy::TcpProxyConfigSharedPtr config)
+    : Extensions::NetworkFilters::TcpProxy::TcpProxyFilter(config, cluster_manager),
       request_headers_(request_headers), request_info_(request_info), route_entry_(route_entry),
       ws_callbacks_(callbacks) {
 
-  initializeReadFilterCallbacks(*read_callbacks);
+  // set_connection_stats == false because the http connection manager has already set them
+  // and they will be inaccurate if we change them now.
+  initialize(*read_callbacks, false);
+  onNewConnection();
 }
 
 void WsHandlerImpl::onInitFailure(UpstreamFailureReason failure_reason) {

source/common/http/websocket/ws_handler_impl.h

@@ -19,6 +19,14 @@ namespace Envoy {
 namespace Http {
 namespace WebSocket {
 
+/**
+ * @return Extensions::NetworkFilters::TcpProxy::TcpProxyConfigSharedPtr to use in creating
+ * instances of WsHandlerImpl.
+ */
+Extensions::NetworkFilters::TcpProxy::TcpProxyConfigSharedPtr
+tcpProxyConfig(const envoy::api::v2::route::RouteAction& route_config,
+               Server::Configuration::FactoryContext& factory_context);
+
 /**
  * An implementation of a WebSocket proxy based on TCP proxy. This will be used for
  * handling client connection only after a WebSocket upgrade request succeeds
@@ -27,12 +35,14 @@ namespace WebSocket {
  * All data will be proxied back and forth between the two connections, without any
  * knowledge of the underlying WebSocket protocol.
  */
-class WsHandlerImpl : public Extensions::NetworkFilters::TcpProxy::TcpProxyFilter {
+class WsHandlerImpl : public Extensions::NetworkFilters::TcpProxy::TcpProxyFilter,
+                      public Http::WebSocketProxy {
 public:
   WsHandlerImpl(HeaderMap& request_headers, const RequestInfo::RequestInfo& request_info,
-                const Router::RouteEntry& route_entry, WsHandlerCallbacks& callbacks,
+                const Router::RouteEntry& route_entry, WebSocketProxyCallbacks& callbacks,
                 Upstream::ClusterManager& cluster_manager,
-                Network::ReadFilterCallbacks* read_callbacks);
+                Network::ReadFilterCallbacks* read_callbacks,
+                Extensions::NetworkFilters::TcpProxy::TcpProxyConfigSharedPtr config);
 
   // Upstream::LoadBalancerContext
   const Router::MetadataMatchCriteria* metadataMatchCriteria() override {
@@ -59,7 +69,7 @@ class WsHandlerImpl : public Extensions::NetworkFilters::TcpProxy::TcpProxyFilte
   HeaderMap& request_headers_;
   const RequestInfo::RequestInfo& request_info_;
   const Router::RouteEntry& route_entry_;
-  WsHandlerCallbacks& ws_callbacks_;
+  WebSocketProxyCallbacks& ws_callbacks_;
   NullHttpConnectionCallbacks http_conn_callbacks_;
   Buffer::OwnedImpl queued_data_;
   bool queued_end_stream_{false};