Add routing capabilities to tcp_proxy

configs/envoy_service_to_service.template.json

@@ -245,8 +245,14 @@
       "type": "read",
       "name": "tcp_proxy",
       "config": {
-        "cluster": "mongo_{{ key }}",
-        "stat_prefix": "mongo_{{ key }}"
+        "stat_prefix": "mongo_{{ key }}",
+        "route_config": {
+          "routes": [
+            {
+              "cluster": "mongo_{{ key }}"
+            }
+          ]
+        }
       }
     }]
   }{% if not loop.last %},{% endif -%}

docs/configuration/network_filters/tcp_proxy_filter.rst

@@ -11,19 +11,114 @@ TCP proxy :ref:`architecture overview <arch_overview_tcp_proxy>`.
     "type": "read",
     "name": "tcp_proxy",
     "config": {
-      "cluster": "...",
-      "stat_prefix": "..."
+      "stat_prefix": "...",
+      "route_config": "{...}"
     }
   }
 
-cluster
-  *(required, string)* The :ref:`cluster manager <arch_overview_cluster_manager>` cluster to connect
-  to when a new downstream network connection is received.
+:ref:`route_config <config_network_filters_tcp_proxy_route_config>`
+  *(required, object)* The route table for the filter.
+  All filter instances must have a route table, even if it is empty.
 
 stat_prefix
   *(required, string)* The prefix to use when emitting :ref:`statistics
   <config_network_filters_tcp_proxy_stats>`.
 
+.. _config_network_filters_tcp_proxy_route_config:
+
+Route Configuration
+-------------------
+
+.. code-block:: json
+
+  {
+    "routes": []
+  }
+
+:ref:`routes <config_network_filters_tcp_proxy_route>`
+  *(required, array)* An array of route entries that make up the route table.
+
+.. _config_network_filters_tcp_proxy_route:
+
+Route
+-----
+
+A TCP proxy route consists of a set of optional L4 criteria and the name of a
+:ref:`cluster <config_cluster_manager_cluster>`. If a downstream connection matches
+all the specified criteria, the cluster in the route is used for the corresponding upstream
+connection. Routes are tried in the order specified until a match is found. If no match is
+found, the connection is closed. A route with no criteria is valid and always produces a match.
+
+.. code-block:: json
+
+  {
+    "cluster": "...",
+    "destination_ip_list": [],
+    "destination_ports": "...",
+    "source_ip_list": [],
+    "source_ports": "..."
+  }
+
+cluster
+  *(required, string)* The :ref:`cluster <config_cluster_manager_cluster>` to connect
+  to when a the downstream network connection matches the specified criteria.
+
+destination_ip_list
+  *(optional, array)*  An optional list of IPv4 subnets in the form "a.b.c.d/xx".
+  The criteria is satisfied if the destination IP address of the downstream connection is
+  contained in at least one of the specified subnets.
+  If the parameter is not specified or the list is empty, the destination IP address is ignored.
+  The destination IP address of the downstream connection might be different from the addresses
+  on which the proxy is listening if the connection has been redirected.  Example:
+
+ .. code-block:: json
+
+    [
+      "192.168.3.0/24",
+      "50.1.2.3/32",
+      "10.15.0.0/16"
+    ]
+
+destination_ports
+  *(optional, string)* An optional string containing a comma-separated list of port numbers or
+  ranges. The criteria is satisfied if the destination port of the downstream connection
+  is contained in at least one of the specified ranges.
+  If the parameter is not specified or the list is empty, the destination port is ignored.
+  The destination port address of the downstream connection might be different from the port
+  on which the proxy is listening if the connection has been redirected. Example:
+
+ .. code-block:: json
+
+  {
+    "destination_ports": "1-1024,2048-4096,12345"
+  }
+
+source_ip_list
+  *(optional, array)*  An optional list of IPv4 subnets in the form "a.b.c.d/xx".
+  The criteria is satisfied if the source IP address of the downstream connection is contained
+  in at least one of the specified subnets. If the parameter is not specified or the list is empty,
+  the source IP address is ignored. Example:
+
+ .. code-block:: json
+
+    [
+      "192.168.3.0/24",
+      "50.1.2.3/32",
+      "10.15.0.0/16"
+    ]
+
+source_ports
+  *(optional, string)* An optional string containing a comma-separated list of port numbers or
+  ranges. The criteria is satisfied if the source port of the downstream connection is contained
+  in at least one of the specified ranges. If the parameter is not specified or the list is empty,
+  the source port is ignored.  Example:
+
+ .. code-block:: json
+
+  {
+    "source_ports": "1-1024,2048-4096,12345"
+  }
+
 .. _config_network_filters_tcp_proxy_stats:
 
 Statistics
@@ -37,5 +132,8 @@ statistics are rooted at *tcp.<stat_prefix>.* with the following statistics:
   :header: Name, Type, Description
   :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_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.
+

include/envoy/network/connection.h

@@ -115,6 +115,13 @@ class Connection : public Event::DeferredDeletable, public FilterManager {
    */
   virtual const std::string& remoteAddress() PURE;
 
+  /**
+   * @return The address the remote client is trying to connect to.
+   * It can be different from the proxy address if the downstream connection
+   * has been redirected or the proxy is operating in transparent mode.
+   */
+  virtual const std::string destinationAddress() PURE;
+
   /**
    * Set the buffer stats to update when the connection's read/write buffers change. Note that
    * for performance reasons these stats are eventually consistent and may not always accurately

source/common/filter/tcp_proxy.cc

@@ -9,20 +9,84 @@
 
 #include "common/common/assert.h"
 #include "common/json/config_schemas.h"
+#include "common/common/empty_string.h"
 #include "common/json/json_loader.h"
 
 namespace Filter {
 
+TcpProxyConfig::Route::Route(const Json::Object& config) {
+  if (config.hasObject("cluster")) {
+    cluster_name_ = config.getString("cluster");
+  } else {
+    throw EnvoyException("tcp proxy: route without cluster");
+  }
+
+  if (config.hasObject("source_ip_list")) {
+    source_ips_ = Network::IpList(config.getStringArray("source_ip_list"));
+  }
+
+  if (config.hasObject("source_ports")) {
+    Network::Utility::parsePortRangeList(config.getString("source_ports"), source_port_ranges_);
+  }
+
+  if (config.hasObject("destination_ip_list")) {
+    destination_ips_ = Network::IpList(config.getStringArray("destination_ip_list"));
+  }
+
+  if (config.hasObject("destination_ports")) {
+    Network::Utility::parsePortRangeList(config.getString("destination_ports"),
+                                         destination_port_ranges_);
+  }
+}
+
 TcpProxyConfig::TcpProxyConfig(const Json::Object& config,
                                Upstream::ClusterManager& cluster_manager, Stats::Store& stats_store)
-    : cluster_name_(config.getString("cluster")),
-      stats_(generateStats(config.getString("stat_prefix"), stats_store)) {
-
+    : stats_(generateStats(config.getString("stat_prefix"), stats_store)) {
   config.validateSchema(Json::Schema::TCP_PROXY_NETWORK_FILTER_SCHEMA);
 
-  if (!cluster_manager.get(cluster_name_)) {
-    throw EnvoyException(fmt::format("tcp proxy: unknown cluster '{}'", cluster_name_));
+  for (const Json::ObjectPtr& route_desc :
+       config.getObject("route_config")->getObjectArray("routes")) {
+    routes_.emplace_back(Route(*route_desc));
+
+    if (!cluster_manager.get(route_desc->getString("cluster"))) {
+      throw EnvoyException(fmt::format("tcp proxy: unknown cluster '{}' in TCP route",
+                                       route_desc->getString("cluster")));
+    }
+  }
+}
+
+const std::string& TcpProxyConfig::getRouteFromEntries(Network::Connection& connection) {
+  for (const TcpProxyConfig::Route& route : routes_) {
+    if (!route.source_port_ranges_.empty() &&
+        !Network::Utility::portInRangeList(
+            Network::Utility::portFromUrl(connection.remoteAddress()), route.source_port_ranges_)) {
+      continue; // no match, try next route
+    }
+
+    if (!route.source_ips_.empty() &&
+        !route.source_ips_.contains(Network::Utility::hostFromUrl(connection.remoteAddress()))) {
+      continue; // no match, try next route
+    }
+
+    if (!route.destination_port_ranges_.empty() &&
+        !Network::Utility::portInRangeList(
+            Network::Utility::portFromUrl(connection.destinationAddress()),
+            route.destination_port_ranges_)) {
+      continue; // no match, try next route
+    }
+
+    if (!route.destination_ips_.empty() &&
+        !route.destination_ips_.contains(
+            Network::Utility::hostFromUrl(connection.destinationAddress()))) {
+      continue; // no match, try next route
+    }
+
+    // if we made it past all checks, the route matches
+    return route.cluster_name_;
   }
+
+  // no match, no more routes to try
+  return EMPTY_STRING;
 }
 
 TcpProxy::TcpProxy(TcpProxyConfigPtr config, Upstream::ClusterManager& cluster_manager)
@@ -52,6 +116,7 @@ TcpProxyStats TcpProxyConfig::generateStats(const std::string& name, Stats::Stor
 void TcpProxy::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {
   read_callbacks_ = &callbacks;
   conn_log_info("new tcp proxy session", read_callbacks_->connection());
+  config_->stats().downstream_cx_total_.inc();
   read_callbacks_->connection().addConnectionCallbacks(downstream_callbacks_);
   read_callbacks_->connection().setBufferStats({config_->stats().downstream_cx_rx_bytes_total_,
                                                 config_->stats().downstream_cx_rx_bytes_buffered_,
@@ -60,14 +125,25 @@ void TcpProxy::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callb
 }
 
 Network::FilterStatus TcpProxy::initializeUpstreamConnection() {
-  Upstream::ClusterInfoPtr cluster = cluster_manager_.get(config_->clusterName());
+  const std::string& cluster_name = config_->getRouteFromEntries(read_callbacks_->connection());
+
+  Upstream::ClusterInfoPtr cluster = cluster_manager_.get(cluster_name);
+
+  if (cluster) {
+    conn_log_debug("Creating connection to cluster {}", read_callbacks_->connection(),
+                   cluster_name);
+  } else {
+    config_->stats().downstream_cx_no_route_.inc();
+    read_callbacks_->connection().close(Network::ConnectionCloseType::NoFlush);
+    return Network::FilterStatus::StopIteration;
+  }
+
   if (!cluster->resourceManager(Upstream::ResourcePriority::Default).connections().canCreate()) {
     cluster->stats().upstream_cx_overflow_.inc();
     read_callbacks_->connection().close(Network::ConnectionCloseType::NoFlush);
     return Network::FilterStatus::StopIteration;
   }
-  Upstream::Host::CreateConnectionData conn_info =
-      cluster_manager_.tcpConnForCluster(config_->clusterName());
+  Upstream::Host::CreateConnectionData conn_info = cluster_manager_.tcpConnForCluster(cluster_name);
 
   upstream_connection_ = std::move(conn_info.connection_);
   read_callbacks_->upstreamHost(conn_info.host_description_);

source/common/filter/tcp_proxy.h

@@ -9,6 +9,7 @@
 #include "common/common/logger.h"
 #include "common/json/json_loader.h"
 #include "common/network/filter_impl.h"
+#include "common/network/utility.h"
 
 namespace Filter {
 
@@ -20,7 +21,9 @@ namespace Filter {
   COUNTER(downstream_cx_rx_bytes_total)                                                            \
   GAUGE  (downstream_cx_rx_bytes_buffered)                                                         \
   COUNTER(downstream_cx_tx_bytes_total)                                                            \
-  GAUGE  (downstream_cx_tx_bytes_buffered)
+  GAUGE  (downstream_cx_tx_bytes_buffered)                                                         \
+  COUNTER(downstream_cx_total)                                                                     \
+  COUNTER(downstream_cx_no_route)
 // clang-format on
 
 /**
@@ -38,13 +41,32 @@ class TcpProxyConfig {
   TcpProxyConfig(const Json::Object& config, Upstream::ClusterManager& cluster_manager,
                  Stats::Store& stats_store);
 
-  const std::string& clusterName() { return cluster_name_; }
+  /**
+   * Find out which cluster an upstream connection should be opened to based on the
+   * parameters of a downstream connection.
+   * @param connection supplies the parameters of the downstream connection for
+   * which the proxy needs to open the corresponding upstream.
+   * @return the cluster name to be used for the upstream connection.
+     If no route applies, returns the empty string.
+   */
+  const std::string& getRouteFromEntries(Network::Connection& connection);
+
   const TcpProxyStats& stats() { return stats_; }
 
 private:
+  struct Route {
+    Route(const Json::Object& config);
+
+    Network::IpList source_ips_;
+    Network::PortRangeList source_port_ranges_;
+    Network::IpList destination_ips_;
+    Network::PortRangeList destination_port_ranges_;
+    std::string cluster_name_;
+  };
+
   static TcpProxyStats generateStats(const std::string& name, Stats::Store& store);
 
-  std::string cluster_name_;
+  std::vector<Route> routes_;
   const TcpProxyStats stats_;
 };
 

source/common/json/config_schemas.cc

@@ -128,11 +128,44 @@ const std::string Json::Schema::REDIS_PROXY_NETWORK_FILTER_SCHEMA(R"EOF(
 const std::string Json::Schema::TCP_PROXY_NETWORK_FILTER_SCHEMA(R"EOF(
   {
       "$schema": "http://json-schema.org/schema#",
-      "properties":{
+      "properties": {
         "stat_prefix" : {"type" : "string"},
-        "cluster" : {"type" : "string"}
+        "route_config" : {
+          "type" : "object",
+          "properties": {
+            "routes" : {
+              "type" : "array",
+              "items" : {
+                "cluster": {
+                  "type" : "string"
+                },
+                "source_ip_list" : {
+                  "type" : "array",
+                  "items" : {
+                    "type" : "string"
+                  }
+                },
+                "source__ports": {
+                  "type" : "string"
+                },
+                "destination_ip_list" : {
+                  "type" : "array",
+                  "items" : {
+                    "type" : "string"
+                  }
+                },
+                "destination_ports": {
+                  "type" : "string"
+                }
+              },
+              "required": ["cluster"],
+              "additionalProperties": false
+            }
+          },
+          "additionalProperties": false
+        }
       },
-      "required": ["stat_prefix", "cluster"],
+      "required": ["stat_prefix", "route_config"],
       "additionalProperties": false
   }
   )EOF");