udp: add router for UDP proxy

api/envoy/extensions/filters/udp/udp_proxy/v3/BUILD

@@ -6,7 +6,10 @@ licenses(["notice"])  # Apache 2
 
 api_proto_package(
     deps = [
+        "//envoy/annotations:pkg",
         "//envoy/config/core/v3:pkg",
         "@com_github_cncf_udpa//udpa/annotations:pkg",
+        "@com_github_cncf_udpa//xds/annotations/v3:pkg",
+        "@com_github_cncf_udpa//xds/type/matcher/v3:pkg",
     ],
 )

api/envoy/extensions/filters/udp/udp_proxy/v3/route.proto

@@ -0,0 +1,20 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.udp.udp_proxy.v3;
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.udp.udp_proxy.v3";
+option java_outer_classname = "RouteProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/udp/udp_proxy/v3;udp_proxyv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: UDP proxy route configuration]
+// UDP proxy :ref:`configuration overview <config_udp_listener_filters_udp_proxy>`.
+
+message Route {
+  // Indicates the upstream cluster to which the request should be routed.
+  string cluster = 1 [(validate.rules).string = {min_len: 1}];
+}

api/envoy/extensions/filters/udp/udp_proxy/v3/udp_proxy.proto

@@ -6,6 +6,10 @@ import "envoy/config/core/v3/udp_socket_config.proto";
 
 import "google/protobuf/duration.proto";
 
+import "xds/annotations/v3/status.proto";
+import "xds/type/matcher/v3/matcher.proto";
+
+import "envoy/annotations/deprecation.proto";
 import "udpa/annotations/status.proto";
 import "udpa/annotations/versioning.proto";
 import "validate/validate.proto";
@@ -21,7 +25,7 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // [#extension: envoy.filters.udp_listener.udp_proxy]
 
 // Configuration for the UDP proxy filter.
-// [#next-free-field: 8]
+// [#next-free-field: 9]
 message UdpProxyConfig {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.filter.udp.udp_proxy.v2alpha.UdpProxyConfig";
@@ -51,7 +55,15 @@ message UdpProxyConfig {
     option (validate.required) = true;
 
     // The upstream cluster to connect to.
-    string cluster = 2 [(validate.rules).string = {min_len: 1}];
+    string cluster = 2 [
+      deprecated = true,
+      (validate.rules).string = {min_len: 1},
+      (envoy.annotations.deprecated_at_minor_version) = "3.0"
+    ];
+
+    // The match tree to use when resolving route actions for incoming requests.
+    xds.type.matcher.v3.Matcher matcher = 8
+        [(xds.annotations.v3.field_status).work_in_progress = true];
   }
 
   // The idle timeout for sessions. Idle is defined as no datagrams between received or sent by

docs/root/configuration/listeners/udp_filters/_include/udp-proxy.yaml

@@ -20,7 +20,13 @@ static_resources:
       typed_config:
         '@type': type.googleapis.com/envoy.extensions.filters.udp.udp_proxy.v3.UdpProxyConfig
         stat_prefix: service
-        cluster: service_udp
+        matcher:
+          on_no_match:
+            action:
+              name: route
+              typed_config:
+                '@type': type.googleapis.com/envoy.extensions.filters.udp.udp_proxy.v3.Route
+                cluster: service_udp
         upstream_socket_config:
           max_rx_datagram_size: 9000
   clusters:

docs/root/version_history/current.rst

@@ -81,6 +81,7 @@ New Features
   useful in support systems such as CI, CD, etc. The
   :ref:`schema validator check tool <install_tools_schema_validator_check_tool>` has been added
   to the tools image.
+* udp_proxy: added :ref:`matcher <envoy_v3_api_field_extensions.filters.udp.udp_proxy.v3.UdpProxyConfig.matcher>` to support matching and routing to different clusters.
 
 Deprecated
 ----------

examples/udp/envoy.yaml

@@ -11,7 +11,13 @@ static_resources:
       typed_config:
         '@type': type.googleapis.com/envoy.extensions.filters.udp.udp_proxy.v3.UdpProxyConfig
         stat_prefix: service
-        cluster: service_udp
+        matcher:
+          on_no_match:
+            action:
+              name: route
+              typed_config:
+                '@type': type.googleapis.com/envoy.extensions.filters.udp.udp_proxy.v3.Route
+                cluster: service_udp
 
   clusters:
   - name: service_udp

source/extensions/filters/udp/udp_proxy/BUILD

@@ -33,10 +33,12 @@ envoy_cc_library(
         "//envoy/network:listener_interface",
         "//envoy/upstream:cluster_manager_interface",
         "//source/common/api:os_sys_calls_lib",
+        "//source/common/common:empty_string",
         "//source/common/network:socket_lib",
         "//source/common/network:socket_option_factory_lib",
         "//source/common/network:utility_lib",
         "//source/common/upstream:load_balancer_lib",
+        "//source/extensions/filters/udp/udp_proxy/router:router_lib",
         "@envoy_api//envoy/extensions/filters/udp/udp_proxy/v3:pkg_cc_proto",
     ],
 )

source/extensions/filters/udp/udp_proxy/config.h

@@ -23,6 +23,7 @@ class UdpProxyFilterConfigFactory
                                Server::Configuration::ListenerFactoryContext& context) override {
     auto shared_config = std::make_shared<UdpProxyFilterConfig>(
         context.clusterManager(), context.timeSource(), context.scope(),
+        context.getServerFactoryContext(),
         MessageUtil::downcastAndValidate<
             const envoy::extensions::filters::udp::udp_proxy::v3::UdpProxyConfig&>(
             config, context.messageValidationVisitor()));

source/extensions/filters/udp/udp_proxy/router/BUILD

@@ -0,0 +1,29 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_library(
+    name = "router_interface",
+    hdrs = ["router.h"],
+)
+
+envoy_cc_library(
+    name = "router_lib",
+    srcs = ["router_impl.cc"],
+    hdrs = ["router_impl.h"],
+    deps = [
+        ":router_interface",
+        "//source/common/common:empty_string",
+        "//source/common/matcher:matcher_lib",
+        "//source/common/matcher:validation_visitor_lib",
+        "//source/common/network/matching:inputs_lib",
+        "@envoy_api//envoy/extensions/filters/udp/udp_proxy/v3:pkg_cc_proto",
+        "@envoy_api//envoy/type/matcher/v3:pkg_cc_proto",
+    ],
+)

source/extensions/filters/udp/udp_proxy/router/router.h

@@ -0,0 +1,45 @@
+#pragma once
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "envoy/common/pure.h"
+#include "envoy/network/address.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace UdpFilters {
+namespace UdpProxy {
+namespace Router {
+
+/**
+ * The router which holds all clusters and determines the cluster which UDP data should be routed
+ * to.
+ */
+class Router {
+public:
+  virtual ~Router() = default;
+
+  /**
+   * Based on the source address of the incoming UDP data, determine the target route for the data.
+   * @return the cluster name or empty string if there is not matching route for the data.
+   */
+  virtual const std::string
+  route(Network::Address::InstanceConstSharedPtr source_address) const PURE;
+
+  /**
+   * Returns all cluster names in the router. The UDP proxy filter requires every cluster names for
+   * initialization which will call this method on construction.
+   * @return vector of all cluster names.
+   */
+  virtual const std::vector<std::string>& allClusterNames() const PURE;
+};
+
+using RouterConstSharedPtr = std::shared_ptr<const Router>;
+
+} // namespace Router
+} // namespace UdpProxy
+} // namespace UdpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/udp/udp_proxy/router/router_impl.cc

@@ -0,0 +1,93 @@
+#include "source/extensions/filters/udp/udp_proxy/router/router_impl.h"
+
+#include "envoy/extensions/filters/udp/udp_proxy/v3/route.pb.h"
+#include "envoy/extensions/filters/udp/udp_proxy/v3/route.pb.validate.h"
+#include "envoy/type/matcher/v3/network_inputs.pb.h"
+
+#include "source/common/common/empty_string.h"
+#include "source/common/network/matching/data_impl.h"
+#include "source/common/network/matching/inputs.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace UdpFilters {
+namespace UdpProxy {
+namespace Router {
+
+Matcher::ActionFactoryCb RouteMatchActionFactory::createActionFactoryCb(
+    const Protobuf::Message& config, RouteActionContext& context,
+    ProtobufMessage::ValidationVisitor& validation_visitor) {
+  const auto& route_config = MessageUtil::downcastAndValidate<
+      const envoy::extensions::filters::udp::udp_proxy::v3::Route&>(config, validation_visitor);
+  const auto& cluster = route_config.cluster();
+
+  // Emplace cluster names to context to get all cluster names
+  context.cluster_name_.emplace(cluster);
+
+  return [cluster]() { return std::make_unique<RouteMatchAction>(cluster); };
+}
+
+REGISTER_FACTORY(RouteMatchActionFactory, Matcher::ActionFactory<RouteActionContext>);
+
+absl::Status RouteActionValidationVisitor::performDataInputValidation(
+    const Matcher::DataInputFactory<Network::MatchingData>&, absl::string_view type_url) {
+  static std::string source_ip_input_name = TypeUtil::descriptorFullNameToTypeUrl(
+      envoy::type::matcher::v3::SourceIpMatchInput::descriptor()->full_name());
+  if (type_url == source_ip_input_name) {
+    return absl::OkStatus();
+  }
+
+  return absl::InvalidArgumentError(
+      fmt::format("Route table can only match on source IP, saw {}", type_url));
+}
+
+RouterImpl::RouterImpl(const envoy::extensions::filters::udp::udp_proxy::v3::UdpProxyConfig& config,
+                       Server::Configuration::ServerFactoryContext& factory_context) {
+  if (config.has_cluster()) {
+    cluster_ = config.cluster();
+    cluster_names_.push_back(config.cluster());
+  } else {
+    RouteActionContext context{};
+    RouteActionValidationVisitor validation_visitor;
+    Matcher::MatchTreeFactory<Network::MatchingData, RouteActionContext> factory(
+        context, factory_context, validation_visitor);
+    matcher_ = factory.create(config.matcher())();
+
+    if (!validation_visitor.errors().empty()) {
+      // TODO(snowp): Output all violations.
+      throw EnvoyException(fmt::format("requirement violation while creating route match tree: {}",
+                                       validation_visitor.errors()[0]));
+    }
+
+    // Copy all clusters names
+    cluster_names_.insert(cluster_names_.end(), context.cluster_name_.begin(),
+                          context.cluster_name_.end());
+  }
+}
+
+const std::string RouterImpl::route(Network::Address::InstanceConstSharedPtr source_address) const {
+  if (cluster_.has_value()) {
+    return cluster_.value();
+  }
+
+  if (source_address->ip()) {
+    Network::Matching::MatchingDataImpl data(source_address->ip());
+
+    auto result = matcher_->match(data);
+    if (result.match_state_ == Matcher::MatchState::MatchComplete) {
+      if (result.on_match_.has_value()) {
+        return result.on_match_.value().action_cb_()->getTyped<RouteMatchAction>().cluster();
+      }
+    }
+  }
+
+  return EMPTY_STRING;
+}
+
+const std::vector<std::string>& RouterImpl::allClusterNames() const { return cluster_names_; }
+
+} // namespace Router
+} // namespace UdpProxy
+} // namespace UdpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/udp/udp_proxy/router/router_impl.h

@@ -0,0 +1,71 @@
+#pragma once
+
+#include "envoy/extensions/filters/udp/udp_proxy/v3/route.pb.h"
+#include "envoy/extensions/filters/udp/udp_proxy/v3/udp_proxy.pb.h"
+#include "envoy/network/filter.h"
+#include "envoy/server/factory_context.h"
+
+#include "source/common/matcher/matcher.h"
+#include "source/common/matcher/validation_visitor.h"
+#include "source/extensions/filters/udp/udp_proxy/router/router.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace UdpFilters {
+namespace UdpProxy {
+namespace Router {
+
+struct RouteActionContext {
+  absl::flat_hash_set<std::string> cluster_name_;
+};
+
+class RouteMatchAction
+    : public Matcher::ActionBase<envoy::extensions::filters::udp::udp_proxy::v3::Route> {
+public:
+  explicit RouteMatchAction(const std::string& cluster) : cluster_(cluster) {}
+
+  const std::string& cluster() const { return cluster_; }
+
+private:
+  const std::string cluster_;
+};
+
+class RouteMatchActionFactory : public Matcher::ActionFactory<RouteActionContext> {
+public:
+  Matcher::ActionFactoryCb
+  createActionFactoryCb(const Protobuf::Message& config, RouteActionContext& context,
+                        ProtobufMessage::ValidationVisitor& validation_visitor) override;
+  std::string name() const override { return "route"; }
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return std::make_unique<envoy::extensions::filters::udp::udp_proxy::v3::Route>();
+  }
+};
+
+class RouteActionValidationVisitor
+    : public Matcher::MatchTreeValidationVisitor<Network::MatchingData> {
+public:
+  absl::Status
+  performDataInputValidation(const Matcher::DataInputFactory<Network::MatchingData>& data_input,
+                             absl::string_view type_url) override;
+};
+
+class RouterImpl : public Router {
+public:
+  RouterImpl(const envoy::extensions::filters::udp::udp_proxy::v3::UdpProxyConfig& config,
+             Server::Configuration::ServerFactoryContext& context);
+
+  // Router::Router
+  const std::string route(Network::Address::InstanceConstSharedPtr source_address) const override;
+  const std::vector<std::string>& allClusterNames() const override;
+
+private:
+  absl::optional<std::string> cluster_;
+  Matcher::MatchTreeSharedPtr<Network::MatchingData> matcher_;
+  std::vector<std::string> cluster_names_;
+};
+
+} // namespace Router
+} // namespace UdpProxy
+} // namespace UdpFilters
+} // namespace Extensions
+} // namespace Envoy