socket: add SO_KEEPALIVE option for upstream connections.

api/envoy/api/v2/cds.proto

@@ -41,7 +41,7 @@ service ClusterDiscoveryService {
 // [#protodoc-title: Clusters]
 
 // Configuration for a single upstream cluster.
-// [#comment:next free field: 30]
+// [#comment:next free field: 31]
 message Cluster {
   // Supplies the name of the cluster which must be unique across all clusters.
   // The cluster name is used when emitting
@@ -428,6 +428,9 @@ message Cluster {
 
   // Determines how Envoy selects the protocol used to speak to upstream hosts.
   ClusterProtocolSelection protocol_selection = 26;
+
+  // Optional options for upstream connections.
+  envoy.api.v2.UpstreamConnectionOptions upstream_connection_options = 30;
 }
 
 // An extensible structure containing the address Envoy should bind to when
@@ -436,3 +439,8 @@ message UpstreamBindConfig {
   // The address Envoy should bind to when establishing upstream connections.
   core.Address source_address = 1;
 }
+
+message UpstreamConnectionOptions {
+  // If set then set SO_KEEPALIVE on the socket to enable TCP Keepalives.
+  core.TcpKeepalive tcp_keepalive = 1;
+}

api/envoy/api/v2/core/address.proto

@@ -58,6 +58,20 @@ message SocketAddress {
   bool ipv4_compat = 6;
 }
 
+message TcpKeepalive {
+  // Maximum number of keepalive probes to send without response before deciding
+  // the connection is dead. Default is to use the OS level configuration (unless
+  // overridden, Linux defaults to 9.)
+  google.protobuf.UInt32Value keepalive_probes = 1;
+  // The number of seconds a connection needs to be idle before keep-alive probes
+  // start being sent. Default is to use the OS level configuration (unless
+  // overridden, Linux defaults to 7200s (ie 2 hours.)
+  google.protobuf.UInt32Value keepalive_time = 2;
+  // The number of seconds between keep-alive probes. Default is to use the OS
+  // level configuration (unless overridden, Linux defaults to 75s.)
+  google.protobuf.UInt32Value keepalive_interval = 3;
+}
+
 message BindConfig {
   // The address to bind to when creating a socket.
   SocketAddress source_address = 1

docs/root/intro/version_history.rst

@@ -49,6 +49,8 @@ Version history
   :ref:`cluster specific <envoy_api_field_Cluster.upstream_bind_config>` options.
 * sockets: added `IP_TRANSPARENT` socket option support for :ref:`listeners
   <envoy_api_field_Listener.transparent>`.
+* sockets: added `SO_KEEPALIVE` socket option for upstream connections
+  :ref:`per cluster <envoy_api_field_Cluster.upstream_connection_options>`.
 * 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.

include/envoy/network/listen_socket.h

@@ -68,13 +68,35 @@ class Socket {
   typedef std::vector<OptionConstSharedPtr> Options;
   typedef std::shared_ptr<Options> OptionsSharedPtr;
 
+  static OptionsSharedPtr& appendOptions(OptionsSharedPtr& to, const OptionsSharedPtr& from) {
+    to->insert(to->end(), from->begin(), from->end());
+    return to;
+  }
+
+  static bool applyOptions(const OptionsSharedPtr& options, Socket& socket, SocketState state) {
+    if (options == nullptr) {
+      return true;
+    }
+    for (const auto& option : *options) {
+      if (!option->setOption(socket, state)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   /**
    * Add a socket option visitor for later retrieval with options().
    */
   virtual void addOption(const OptionConstSharedPtr&) PURE;
 
   /**
-   * @return the socket options stored earlier with addOption() calls, if any.
+   * Add socket option visitors for later retrieval with options().
+   */
+  virtual void addOptions(const OptionsSharedPtr&) PURE;
+
+  /**
+   * @return the socket options stored earlier with addOption() and addOptions() calls, if any.
    */
   virtual const OptionsSharedPtr& options() const PURE;
 };

include/envoy/server/filter_config.h

@@ -136,6 +136,8 @@ class ListenerFactoryContext : public FactoryContext {
    * Store socket options to be set on the listen socket before listening.
    */
   virtual void addListenSocketOption(const Network::Socket::OptionConstSharedPtr& option) PURE;
+
+  virtual void addListenSocketOptions(const Network::Socket::OptionsSharedPtr& options) PURE;
 };
 
 /**

include/envoy/upstream/upstream.h

@@ -397,8 +397,6 @@ class ClusterInfo {
     // Use the downstream protocol (HTTP1.1, HTTP2) for upstream connections as well, if available.
     // This is used when creating connection pools.
     static const uint64_t USE_DOWNSTREAM_PROTOCOL = 0x2;
-    // Use IP_FREEBIND socket option when binding.
-    static const uint64_t FREEBIND = 0x4;
   };
 
   virtual ~ClusterInfo() {}
@@ -521,6 +519,13 @@ class ClusterInfo {
    * @return const envoy::api::v2::core::Metadata& the configuration metadata for this cluster.
    */
   virtual const envoy::api::v2::core::Metadata& metadata() const PURE;
+
+  /**
+   *
+   * @return const Network::ConnectionSocket::OptionsSharedPtr& socket options for all
+   *         connections for this cluster.
+   */
+  virtual const Network::ConnectionSocket::OptionsSharedPtr& clusterSocketOptions() const PURE;
 };
 
 typedef std::shared_ptr<const ClusterInfo> ClusterInfoConstSharedPtr;

source/common/network/BUILD

@@ -187,6 +187,35 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "addr_family_aware_socket_option_lib",
+    srcs = ["addr_family_aware_socket_option_impl.cc"],
+    hdrs = ["addr_family_aware_socket_option_impl.h"],
+    external_deps = ["abseil_optional"],
+    deps = [
+        ":address_lib",
+        ":socket_option_lib",
+        "//include/envoy/network:listen_socket_interface",
+        "//source/common/api:os_sys_calls_lib",
+        "//source/common/common:assert_lib",
+        "//source/common/common:logger_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "socket_option_factory_lib",
+    srcs = ["socket_option_factory.cc"],
+    hdrs = ["socket_option_factory.h"],
+    external_deps = ["abseil_optional"],
+    deps = [
+        ":addr_family_aware_socket_option_lib",
+        ":address_lib",
+        ":socket_option_lib",
+        "//include/envoy/network:listen_socket_interface",
+        "//source/common/common:logger_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "utility_lib",
     srcs = ["utility.cc"],

source/common/network/addr_family_aware_socket_option_impl.cc

@@ -0,0 +1,58 @@
+#include "common/network/addr_family_aware_socket_option_impl.h"
+
+#include "envoy/common/exception.h"
+
+#include "common/api/os_sys_calls_impl.h"
+#include "common/common/assert.h"
+#include "common/network/address_impl.h"
+#include "common/network/socket_option_impl.h"
+
+namespace Envoy {
+namespace Network {
+
+bool AddrFailyAwareSocketOptionImpl::setOption(Socket& socket, Socket::SocketState state) const {
+  return setIpSocketOption(socket, state, ipv4_option_, ipv6_option_);
+}
+
+bool AddrFailyAwareSocketOptionImpl::setIpSocketOption(
+    Socket& socket, Socket::SocketState state, const std::unique_ptr<SocketOptionImpl>& ipv4_option,
+    const std::unique_ptr<SocketOptionImpl>& ipv6_option) {
+  // If this isn't IP, we're out of luck.
+  Address::InstanceConstSharedPtr address;
+  const Address::Ip* ip = nullptr;
+  try {
+    // We have local address when the socket is used in a listener but have to
+    // infer the IP from the socket FD when initiating connections.
+    // TODO(htuch): Figure out a way to obtain a consistent interface for IP
+    // version from socket.
+    if (socket.localAddress()) {
+      ip = socket.localAddress()->ip();
+    } else {
+      address = Address::addressFromFd(socket.fd());
+      ip = address->ip();
+    }
+  } catch (const EnvoyException&) {
+    // Ignore, we get here because we failed in getsockname().
+    // TODO(htuch): We should probably clean up this logic to avoid relying on exceptions.
+  }
+  if (ip == nullptr) {
+    ENVOY_LOG(warn, "Failed to set IP socket option on non-IP socket");
+    return false;
+  }
+
+  // If the FD is v4, we can only try the IPv4 variant.
+  if (ip->version() == Network::Address::IpVersion::v4) {
+    return ipv4_option->setOption(socket, state);
+  }
+
+  // If the FD is v6, we first try the IPv6 variant if the platfrom supports it and fallback to the
+  // IPv4 variant otherwise.
+  ASSERT(ip->version() == Network::Address::IpVersion::v6);
+  if (ipv6_option->isSupported()) {
+    return ipv6_option->setOption(socket, state);
+  }
+  return ipv4_option->setOption(socket, state);
+}
+
+} // namespace Network
+} // namespace Envoy

source/common/network/addr_family_aware_socket_option_impl.h

@@ -0,0 +1,55 @@
+#pragma once
+
+#include <netinet/in.h>
+#include <netinet/ip.h>
+#include <sys/socket.h>
+
+#include "envoy/network/listen_socket.h"
+
+#include "common/common/logger.h"
+#include "common/network/socket_option_impl.h"
+
+#include "absl/types/optional.h"
+
+namespace Envoy {
+namespace Network {
+
+class AddrFailyAwareSocketOptionImpl : public Socket::Option,
+                                       Logger::Loggable<Logger::Id::connection> {
+public:
+  AddrFailyAwareSocketOptionImpl(Socket::SocketState in_state, SocketOptionName ipv4_optname,
+                                 SocketOptionName ipv6_optname, int value)
+      : ipv4_option_(absl::make_unique<SocketOptionImpl>(in_state, ipv4_optname, value)),
+        ipv6_option_(absl::make_unique<SocketOptionImpl>(in_state, ipv6_optname, value)) {}
+
+  // Socket::Option
+  bool setOption(Socket& socket, Socket::SocketState state) const override;
+  // The common socket options don't require a hash key.
+  void hashKey(std::vector<uint8_t>&) const override {}
+
+  /**
+   * Set a socket option that applies at both IPv4 and IPv6 socket levels. When the underlying FD
+   * is IPv6, this function will attempt to set at IPv6 unless the platform only supports the
+   * option at the IPv4 level.
+   * @param socket.
+   * @param ipv4_optname SocketOptionName for IPv4 level. Set to empty if not supported on
+   * platform.
+   * @param ipv6_optname SocketOptionName for IPv6 level. Set to empty if not supported on
+   * platform.
+   * @param optval as per setsockopt(2).
+   * @param optlen as per setsockopt(2).
+   * @return int as per setsockopt(2). ENOTSUP is returned if the option is not supported on the
+   * platform for fd after the above option level fallback semantics are taken into account or the
+   *         socket is non-IP.
+   */
+  static bool setIpSocketOption(Socket& socket, Socket::SocketState state,
+                                const std::unique_ptr<SocketOptionImpl>& ipv4_option,
+                                const std::unique_ptr<SocketOptionImpl>& ipv6_option);
+
+private:
+  const std::unique_ptr<SocketOptionImpl> ipv4_option_;
+  const std::unique_ptr<SocketOptionImpl> ipv6_option_;
+};
+
+} // namespace Network
+} // namespace Envoy

source/common/network/connection_impl.cc

@@ -536,18 +536,15 @@ ClientConnectionImpl::ClientConnectionImpl(
     const Network::ConnectionSocket::OptionsSharedPtr& options)
     : ConnectionImpl(dispatcher, std::make_unique<ClientSocketImpl>(remote_address),
                      std::move(transport_socket), false) {
-  if (options) {
-    for (const auto& option : *options) {
-      if (!option->setOption(*socket_, Socket::SocketState::PreBind)) {
-        // Set a special error state to ensure asynchronous close to give the owner of the
-        // ConnectionImpl a chance to add callbacks and detect the "disconnect".
-        immediate_error_event_ = ConnectionEvent::LocalClose;
-        // Trigger a write event to close this connection out-of-band.
-        file_event_->activate(Event::FileReadyType::Write);
-        return;
-      }
-    }
+  if (!Network::Socket::applyOptions(options, *socket_, Socket::SocketState::PreBind)) {
+    // Set a special error state to ensure asynchronous close to give the owner of the
+    // ConnectionImpl a chance to add callbacks and detect the "disconnect".
+    immediate_error_event_ = ConnectionEvent::LocalClose;
+    // Trigger a write event to close this connection out-of-band.
+    file_event_->activate(Event::FileReadyType::Write);
+    return;
   }
+
   if (source_address != nullptr) {
     const int rc = source_address->bind(fd());
     if (rc < 0) {

source/common/network/listen_socket_impl.cc

@@ -30,12 +30,8 @@ void ListenSocketImpl::doBind() {
 }
 
 void ListenSocketImpl::setListenSocketOptions(const Network::Socket::OptionsSharedPtr& options) {
-  if (options) {
-    for (const auto& option : *options) {
-      if (!option->setOption(*this, SocketState::PreBind)) {
-        throw EnvoyException("ListenSocket: Setting socket options failed");
-      }
-    }
+  if (!Network::Socket::applyOptions(options, *this, Socket::SocketState::PreBind)) {
+    throw EnvoyException("ListenSocket: Setting socket options failed");
   }
 }
 

source/common/network/listen_socket_impl.h

@@ -27,12 +27,19 @@ class SocketImpl : public virtual Socket {
       fd_ = -1;
     }
   }
-  void addOption(const OptionConstSharedPtr& option) override {
+  void ensureOptions() {
     if (!options_) {
       options_ = std::make_shared<std::vector<OptionConstSharedPtr>>();
     }
+  }
+  void addOption(const OptionConstSharedPtr& option) override {
+    ensureOptions();
     options_->emplace_back(std::move(option));
   }
+  void addOptions(const OptionsSharedPtr& options) override {
+    ensureOptions();
+    Network::Socket::appendOptions(options_, options);
+  }
   const OptionsSharedPtr& options() const override { return options_; }
 
 protected:

source/common/network/listener_impl.cc

@@ -66,15 +66,9 @@ ListenerImpl::ListenerImpl(Event::DispatcherImpl& dispatcher, Socket& socket, Li
           fmt::format("cannot listen on socket: {}", socket.localAddress()->asString()));
     }
 
-    Socket::OptionsSharedPtr options = socket.options();
-    if (options != nullptr) {
-      for (auto& option : *options) {
-        if (!option->setOption(socket, Socket::SocketState::Listening)) {
-          throw CreateListenerException(
-              fmt::format("cannot set post-listen socket option on socket: {}",
-                          socket.localAddress()->asString()));
-        }
-      }
+    if (!Network::Socket::applyOptions(socket.options(), socket, Socket::SocketState::Listening)) {
+      throw CreateListenerException(fmt::format(
+          "cannot set post-listen socket option on socket: {}", socket.localAddress()->asString()));
     }
 
     evconnlistener_set_error_cb(listener_.get(), errorCallback);