udp: add new key based hash policy (#15967)

Signed-off-by: Kornel David <[email protected]>

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

@@ -33,6 +33,13 @@ message UdpProxyConfig {
 
       // The source IP will be used to compute the hash used by hash-based load balancing algorithms.
       bool source_ip = 1 [(validate.rules).bool = {const: true}];
+
+      // A given key will be used to compute the hash used by hash-based load balancing algorithms.
+      // In certain cases there is a need to direct different UDP streams jointly towards the selected set of endpoints.
+      // A possible use-case is VoIP telephony, where media (RTP) and its corresponding control (RTCP) belong to the same logical session,
+      // although they travel in separate streams. To ensure that these pair of streams are load-balanced on session level
+      // (instead of individual stream level), dynamically created listeners can use the same hash key for each stream in the session.
+      string key = 2 [(validate.rules).string = {min_len: 1}];
     }
   }
 

docs/root/version_history/current.rst

@@ -48,6 +48,7 @@ New Features
 ------------
 
 * metric service: added support for sending metric tags as labels. This can be enabled by setting the :ref:`emit_tags_as_labels <envoy_v3_api_field_config.metrics.v3.MetricsServiceConfig.emit_tags_as_labels>` field to true.
+* udp_proxy: added :ref:`key <envoy_v3_api_msg_extensions.filters.udp.udp_proxy.v3.UdpProxyConfig.HashPolicy>` as another hash policy to support hash based routing on any given key.
 
 Deprecated
 ----------

source/extensions/filters/udp/udp_proxy/hash_policy_impl.cc

@@ -1,6 +1,7 @@
 #include "extensions/filters/udp/udp_proxy/hash_policy_impl.h"
 
 #include "common/common/assert.h"
+#include "common/common/macros.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -20,13 +21,32 @@ class SourceIpHashMethod : public HashPolicyImpl::HashMethod {
   }
 };
 
+class KeyHashMethod : public HashPolicyImpl::HashMethod {
+public:
+  explicit KeyHashMethod(const std::string& key) : hash_{HashUtil::xxHash64(key)} {
+    ASSERT(!key.empty());
+  }
+
+  absl::optional<uint64_t>
+  evaluate(const Network::Address::Instance& downstream_addr) const override {
+    UNREFERENCED_PARAMETER(downstream_addr);
+    return hash_;
+  }
+
+private:
+  const uint64_t hash_;
+};
+
 HashPolicyImpl::HashPolicyImpl(
     const absl::Span<const UdpProxyConfig::HashPolicy* const>& hash_policies) {
   ASSERT(hash_policies.size() == 1);
   switch (hash_policies[0]->policy_specifier_case()) {
   case UdpProxyConfig::HashPolicy::PolicySpecifierCase::kSourceIp:
     hash_impl_ = std::make_unique<SourceIpHashMethod>();
     break;
+  case UdpProxyConfig::HashPolicy::PolicySpecifierCase::kKey:
+    hash_impl_ = std::make_unique<KeyHashMethod>(hash_policies[0]->key());
+    break;
   default:
     NOT_REACHED_GCOVR_EXCL_LINE;
   }

test/extensions/filters/udp/udp_proxy/hash_policy_impl_test.cc

@@ -1,3 +1,5 @@
+#include <string>
+
 #include "envoy/extensions/filters/udp/udp_proxy/v3/udp_proxy.pb.h"
 
 #include "common/common/hash.h"
@@ -50,6 +52,15 @@ class HashPolicyImplSourceIpTest : public HashPolicyImplBaseTest {
   const Network::Address::InstanceConstSharedPtr pipe_address_;
 };
 
+class HashPolicyImplKeyTest : public HashPolicyImplBaseTest {
+public:
+  HashPolicyImplKeyTest() : key_("key") {}
+
+  void additionalSetup() override { hash_policy_config_->set_key(key_); }
+
+  const std::string key_;
+};
+
 // Check invalid policy type
 TEST_F(HashPolicyImplBaseTest, NotSupportedPolicy) {
   EXPECT_DEATH(setup(), ".*panic: not reached.*");
@@ -74,6 +85,16 @@ TEST_F(HashPolicyImplSourceIpTest, SourceIpWithUnixDomainSocketType) {
   EXPECT_FALSE(hash.has_value());
 }
 
+// Check if generate correct hash
+TEST_F(HashPolicyImplKeyTest, KeyHash) {
+  setup();
+
+  auto generated_hash = HashUtil::xxHash64(key_);
+  auto hash = hash_policy_->generateHash(*peer_address_);
+
+  EXPECT_EQ(generated_hash, hash.value());
+}
+
 } // namespace
 } // namespace UdpProxy
 } // namespace UdpFilters

test/extensions/filters/udp/udp_proxy/udp_proxy_filter_test.cc

@@ -796,6 +796,61 @@ cluster: fake_cluster
   test_sessions_[0].recvDataFromUpstream("world");
 }
 
+// Make sure hash policy with key is created.
+TEST_F(UdpProxyFilterTest, HashPolicyWithKey) {
+  InSequence s;
+
+  setup(R"EOF(
+stat_prefix: foo
+cluster: fake_cluster
+hash_policies:
+- key: "key"
+  )EOF");
+
+  EXPECT_NE(nullptr, config_->hashPolicy());
+}
+
+// Make sure validation fails if key is an empty string.
+TEST_F(UdpProxyFilterTest, ValidateHashPolicyWithKey) {
+  InSequence s;
+  auto config = R"EOF(
+stat_prefix: foo
+cluster: fake_cluster
+hash_policies:
+- key: ""
+  )EOF";
+
+  EXPECT_THROW_WITH_REGEX(setup(config), EnvoyException,
+                          "caused by HashPolicyValidationError\\.Key");
+}
+
+// Expect correct hash is created if hash_policy with key is mentioned.
+TEST_F(UdpProxyFilterTest, HashWithKey) {
+  InSequence s;
+
+  setup(R"EOF(
+stat_prefix: foo
+cluster: fake_cluster
+hash_policies:
+- key: "key"
+  )EOF");
+
+  auto host = createHost(upstream_address_);
+  auto generated_hash = HashUtil::xxHash64("key");
+  EXPECT_CALL(cluster_manager_.thread_local_cluster_.lb_, chooseHost(_))
+      .WillOnce(Invoke([host, generated_hash](
+                           Upstream::LoadBalancerContext* context) -> Upstream::HostConstSharedPtr {
+        auto hash = context->computeHashKey();
+        EXPECT_TRUE(hash.has_value());
+        EXPECT_EQ(generated_hash, hash.value());
+        return host;
+      }));
+  expectSessionCreate(upstream_address_);
+  test_sessions_[0].expectWriteToUpstream("hello");
+  recvDataFromDownstream("10.0.0.1:1000", "10.0.0.2:80", "hello");
+  test_sessions_[0].recvDataFromUpstream("world");
+}
+
 } // namespace
 } // namespace UdpProxy
 } // namespace UdpFilters

tools/spelling/spelling_dictionary.txt

@@ -34,6 +34,8 @@ HEXDIG
 HEXDIGIT
 OWS
 Preconnecting
+RTCP
+RTP
 STATNAME
 SkyWalking
 TIDs