WIP: Consistent hashing for balancer

.gitmodules

@@ -4,3 +4,7 @@
 [submodule "subprojects/json"]
 	path = subprojects/json
 	url = https://github.com/nlohmann/json.git
+[submodule "subprojects/chash"]
+	path = subprojects/chash
+	url = https://github.com/yanet-platform/chash.git
+	branch = master

autotest/autotest.cpp

@@ -291,6 +291,7 @@ bool readTimeLimited(int fd, u_char* buf, ssize_t len, std::chrono::system_clock
 		switch (ret)
 		{
 			case 0:
+				YANET_LOG_ERROR("Failed to read packets: end of file\n");
 				return false;
 			case -1:
 				if ((errno == EAGAIN) || (errno = EWOULDBLOCK))
@@ -299,6 +300,7 @@ bool readTimeLimited(int fd, u_char* buf, ssize_t len, std::chrono::system_clock
 				}
 				else
 				{
+					YANET_LOG_ERROR("Failed to read packets: %s\n", strerror(errno));
 					return false;
 				}
 				break;

autotest/units/001_one_port/055_balancer_wlc/autotest.yaml

@@ -14,9 +14,9 @@ steps:
     YANET_FORMAT_COLUMNS=module,virtual_ip,proto,virtual_port,scheduler,real_ip,real_port,enabled,weight,connections,packets,bytes balancer real any
     module     virtual_ip  proto  virtual_port  scheduler  real_ip  real_port  enabled  weight  connections  packets  bytes
     ---------  ----------  -----  ------------  ---------  -------  ---------  -------  ------  -----------  -------  -----
-    balancer0  10.1.0.55   tcp    443           wlc        2443::1  443        true     2       16           16       1568
+    balancer0  10.1.0.55   tcp    443           wlc        2443::1  443        true     2       15           15       1470
     balancer0  10.1.0.55   tcp    443           wlc        2443::2  443        true     1       8            8        784
-    balancer0  10.1.0.55   tcp    443           wlc        2443::3  443        true     1       8            8        784
+    balancer0  10.1.0.55   tcp    443           wlc        2443::3  443        true     1       9            9        882
     balancer0  10.1.0.55   tcp    443           wlc        2443::4  443        false    4       0            0        0
 
 - cli:
@@ -30,10 +30,10 @@ steps:
     YANET_FORMAT_COLUMNS=module,virtual_ip,proto,virtual_port,scheduler,real_ip,real_port,enabled,weight,connections,packets,bytes balancer real any
     module     virtual_ip  proto  virtual_port  scheduler  real_ip  real_port  enabled  weight  connections  packets  bytes
     ---------  ----------  -----  ------------  ---------  -------  ---------  -------  ------  -----------  -------  -----
-    balancer0  10.1.0.55   tcp    443           wlc        2443::1  443        true     2       17           17       1666
-    balancer0  10.1.0.55   tcp    443           wlc        2443::2  443        true     1       10           10       980
+    balancer0  10.1.0.55   tcp    443           wlc        2443::1  443        true     2       16           16       1568
+    balancer0  10.1.0.55   tcp    443           wlc        2443::2  443        true     1       8            8        784
     balancer0  10.1.0.55   tcp    443           wlc        2443::3  443        true     1       9            9        882
-    balancer0  10.1.0.55   tcp    443           wlc        2443::4  443        true     4       28           28       2744
+    balancer0  10.1.0.55   tcp    443           wlc        2443::4  443        true     4       31           31       3038
 
 - sleep: 2
 - sendPackets:
@@ -45,6 +45,6 @@ steps:
     module     virtual_ip  proto  virtual_port  scheduler  real_ip  real_port  enabled  weight  connections  packets  bytes
     ---------  ----------  -----  ------------  ---------  -------  ---------  -------  ------  -----------  -------  -----
     balancer0  10.1.0.55   tcp    443           wlc        2443::1  443        true     2       24           24       2352
-    balancer0  10.1.0.55   tcp    443           wlc        2443::2  443        true     1       11           11       1078
+    balancer0  10.1.0.55   tcp    443           wlc        2443::2  443        true     1       12           12       1176
     balancer0  10.1.0.55   tcp    443           wlc        2443::3  443        true     1       12           12       1176
-    balancer0  10.1.0.55   tcp    443           wlc        2443::4  443        true     4       49           49       4802
+    balancer0  10.1.0.55   tcp    443           wlc        2443::4  443        true     4       48           48       4704

common/controlplaneconfig.h

@@ -206,7 +206,7 @@ using service_t = std::tuple<balancer_service_id_t,
                              std::optional<uint16_t>, ///< vport
                              std::optional<std::string>, ///< version
                              ::balancer::scheduler,
-                             ::balancer::scheduler_params,
+                             uint32_t, ///< wlc power
                              ::balancer::forwarding_method,
                              uint8_t, ///< flags: mss_fix|ops
                              std::optional<common::ipv4_prefix_t>, ///< ipv4_outer_source_network

common/deferer.h

@@ -0,0 +1,23 @@
+namespace utils
+{
+
+template<typename F>
+class Deferer
+{
+	F action_;
+
+public:
+	Deferer(F&& action) :
+	        action_{std::move(action)}
+	{}
+	Deferer(const Deferer& other) = delete;
+	Deferer(Deferer&& other) = delete;
+	Deferer& operator=(const Deferer& other) = delete;
+	Deferer& operator=(Deferer&& other) = delete;
+	~Deferer()
+	{
+		action_();
+	}
+};
+
+} // namespace utils

common/define.h

@@ -98,6 +98,8 @@ extern LogPriority logPriority;
 #define YANET_BALANCER_DEFAULT_MSS_SIZE 536
 #define YANET_BALANCER_FIX_MSS_SIZE 1220
 #define YANET_BALANCER_FIX_MSS_FLAG ((uint8_t)(1u << 0))
+static constexpr size_t YANET_DEFAULT_BALANCER_REAL_MAPPINGS_LIMIT = 20000;
+static constexpr size_t YANET_DEFAULT_BALANCER_CELLS_PER_WEIGHT_UNIT = 20;
 
 #define YANET_BALANCER_OPS_FLAG ((uint8_t)(1u << 1))
 #define YANET_BALANCER_PURE_L3 ((uint8_t)(1u << 2))

common/idp.h

@@ -283,7 +283,6 @@ using service = std::tuple<balancer_service_id_t, /// service id
                            tCounterId, ///< size 4
                            balancer::scheduler,
                            balancer::forwarding_method, // tunneling method (default ipip)
-                           uint32_t, /// default_wlc_power
                            uint32_t, ///< real_start
                            uint32_t, ///< real_size
                            std::optional<common::ipv4_prefix_t>, ///< ipv4_outer_source_network

common/scheduler.h

@@ -1,4 +1,5 @@
 #pragma once
+#include <variant>
 
 #include <cstdint>
 namespace balancer
@@ -9,34 +10,24 @@ enum class scheduler : uint8_t
 	rr,
 	wrr,
 	wlc,
-};
-
-class scheduler_params
-{
-public:
-	scheduler_params() = default;
-	uint32_t wlc_power;
+	chash
 };
 
 [[maybe_unused]] constexpr const char* to_string(const scheduler& scheduler)
 {
 	switch (scheduler)
 	{
 		case scheduler::rr:
-		{
 			return "rr";
-		}
 		case scheduler::wrr:
-		{
 			return "wrr";
-		}
 		case scheduler::wlc:
-		{
 			return "wlc";
-		}
+		case scheduler::chash:
+			return "chash";
+		default:
+			return "unknown";
 	}
-
-	return "unknown";
 }
 
 }

controlplane/balancer.cpp

@@ -757,14 +757,14 @@ void balancer_t::compile(common::idp::updateGlobalBase::request& globalbase,
 		                  virtual_port,
 		                  version,
 		                  scheduler,
-		                  scheduler_params,
+		                  wlc_power,
 		                  forwarding_method,
 		                  flags,
 		                  ipv4_outer_source_network,
 		                  ipv6_outer_source_network,
 		                  reals] : balancer.services)
 		{
-			YANET_GCC_BUG_UNUSED(scheduler_params);
+			YANET_GCC_BUG_UNUSED(wlc_power);
 			YANET_GCC_BUG_UNUSED(version);
 
 			if (service_id >= YANET_CONFIG_BALANCER_SERVICES_SIZE)
@@ -808,7 +808,6 @@ void balancer_t::compile(common::idp::updateGlobalBase::request& globalbase,
 			        counter_id,
 			        scheduler,
 			        forwarding_method,
-			        balancer.default_wlc_power, // todo use scheduler_params.wlc_power when other services will be able to set it
 			        (uint32_t)real_start,
 			        (uint32_t)(req_reals.size() - real_start),
 			        ipv4_outer_source_network,
@@ -883,6 +882,12 @@ void balancer_t::flush_reals(common::idp::updateGlobalBaseBalancer::request& bal
 					}
 				}
 
+				auto reals_wlc = reals_wlc_weight.find(key);
+				if (reals_wlc != reals_wlc_weight.end() && reals_wlc->second.has_value())
+				{
+					effective_weight = reals_wlc->second.value();
+				}
+
 				uint32_t real_unordered_id = 0;
 				{
 					std::lock_guard<std::mutex> guard(reals_unordered_mutex);
@@ -926,8 +931,153 @@ void balancer_t::reconfigure_wlc_thread()
 {
 	while (!flagStop)
 	{
-		balancer_real_flush();
+		if (balancer_t::reconfigure_wlc())
+		{
+			balancer_real_flush();
+		}
 
 		std::this_thread::sleep_for(std::chrono::seconds(YANET_CONFIG_BALANCER_WLC_RECONFIGURE));
 	}
 }
+
+bool balancer_t::reconfigure_wlc()
+{
+	bool wlc_weight_changed = false;
+
+	common::idp::updateGlobalBaseBalancer::request balancer;
+	const auto balancer_real_connections = dataplane.balancer_real_connections();
+
+	std::lock_guard<std::mutex> guard(reals_enabled_mutex);
+
+	for (const auto& [module_name, balancer] : generations_config.current().config_balancers)
+	{
+
+		for (const auto& [service_id,
+		                  virtual_ip,
+		                  proto,
+		                  virtual_port,
+		                  version,
+		                  scheduler,
+		                  requested_wlc_power,
+		                  forwarding_method,
+		                  flags,
+		                  ipv4_outer_source_network,
+		                  ipv6_outer_source_network,
+		                  reals] : balancer.services)
+		{
+			YANET_GCC_BUG_UNUSED(flags);
+			YANET_GCC_BUG_UNUSED(version);
+			YANET_GCC_BUG_UNUSED(forwarding_method);
+			YANET_GCC_BUG_UNUSED(ipv4_outer_source_network);
+			YANET_GCC_BUG_UNUSED(ipv6_outer_source_network);
+
+			if (scheduler != ::balancer::scheduler::wlc)
+			{
+				continue;
+			}
+
+			if (service_id >= YANET_CONFIG_BALANCER_SERVICES_SIZE)
+			{
+				continue;
+			}
+
+			std::vector<std::tuple<balancer::real_key_global_t, uint32_t, uint32_t>> service_reals_usage_info;
+			service_reals_usage_info.reserve(reals.size());
+			uint32_t connection_sum = 0;
+			uint32_t weight_sum = 0;
+
+			for (const auto& [real_ip, real_port, weight] : reals)
+			{
+				balancer::real_key_global_t key = {module_name, {virtual_ip, proto, virtual_port}, {real_ip, real_port}};
+				uint32_t effective_weight = weight;
+				{
+					auto it = reals_enabled.find(key);
+					if (it != reals_enabled.end())
+					{
+						if (it->second.has_value())
+						{
+							effective_weight = it->second.value();
+						}
+					}
+				}
+
+				weight_sum += effective_weight;
+
+				// don`t count connections for disabled reals - it can make other reals "feel" underloaded
+				if (effective_weight == 0)
+				{
+					continue;
+				}
+
+				common::idp::balancer_real_connections::real_key_t real_connections_key = {balancer.balancer_id,
+				                                                                           virtual_ip,
+				                                                                           proto,
+				                                                                           virtual_port.value(),
+				                                                                           real_ip,
+				                                                                           real_port.value()};
+				uint32_t connections = 0;
+				for (auto& [socket_id, real_connections] : balancer_real_connections)
+				{
+					YANET_GCC_BUG_UNUSED(socket_id);
+
+					if (auto it = real_connections.find(real_connections_key); it != real_connections.end())
+					{
+						connections += it->second;
+						continue;
+						connections += it->second;
+					}
+				}
+
+				connection_sum += connections;
+
+				service_reals_usage_info.emplace_back(key,
+				                                      effective_weight,
+				                                      connections);
+			}
+
+			for (auto [key,
+			           effective_weight,
+			           connections] : service_reals_usage_info)
+			{
+				uint32_t wlc_power = requested_wlc_power;
+				if (wlc_power < 1 || wlc_power > 100)
+				{
+					wlc_power = YANET_CONFIG_BALANCER_WLC_DEFAULT_POWER;
+				}
+
+				effective_weight = calculate_wlc_weight(effective_weight, connections, weight_sum, connection_sum, wlc_power);
+
+				if (reals_wlc_weight[key] != effective_weight)
+				{
+					reals_wlc_weight[key] = effective_weight;
+					real_updates.insert(key);
+					if (in_reload)
+					{
+						real_reload_updates.insert(key);
+					}
+					wlc_weight_changed = true;
+				}
+			}
+		}
+	}
+
+	return wlc_weight_changed;
+}
+
+uint32_t balancer_t::calculate_wlc_weight(uint32_t weight, uint32_t connections, uint32_t weight_sum, uint32_t connection_sum, uint32_t wlc_power)
+{
+	if (weight == 0 || weight_sum == 0 || connection_sum < weight_sum)
+	{
+		return weight;
+	}
+
+	auto wlc_ratio = std::max(1.0, wlc_power * (1 - 1.0 * connections * weight_sum / connection_sum / weight));
+	auto wlc_weight = (uint32_t)(weight * wlc_ratio);
+
+	if (wlc_weight > YANET_CONFIG_BALANCER_REAL_WEIGHT_MAX)
+	{
+		wlc_weight = YANET_CONFIG_BALANCER_REAL_WEIGHT_MAX;
+	}
+
+	return wlc_weight;
+}

controlplane/balancer.h

@@ -113,6 +113,8 @@ class balancer_t : public module_t, common::icp_proto::BalancerService
 
 	std::map<balancer::real_key_global_t, std::optional<uint32_t>> reals_enabled;
 
+	std::map<balancer::real_key_global_t, std::optional<uint32_t>> reals_wlc_weight;
+
 	// The set contains all reals touched after the last one flush operation
 	std::set<balancer::real_key_global_t> real_updates;
 
@@ -133,7 +135,11 @@ class balancer_t : public module_t, common::icp_proto::BalancerService
 	friend class telegraf_t;
 	counter_t<balancer::service_counter_key_t, (size_t)balancer::service_counter::size> service_counters;
 	counter_t<balancer::real_counter_key_t, (size_t)balancer::real_counter::size> real_counters;
+
 	void RealFind(google::protobuf::RpcController* controller, const common::icp_proto::BalancerRealFindRequest* request, common::icp_proto::BalancerRealFindResponse* response, google::protobuf::Closure* done) override;
 	void Real(google::protobuf::RpcController* controller, const ::common::icp_proto::BalancerRealRequest* request, ::common::icp_proto::Empty* response, ::google::protobuf::Closure* done) override;
 	void RealFlush(google::protobuf::RpcController* controller, const ::common::icp_proto::Empty* request, ::common::icp_proto::Empty* response, ::google::protobuf::Closure* done) override;
+
+	bool reconfigure_wlc();
+	uint32_t calculate_wlc_weight(uint32_t weight, uint32_t connections, uint32_t weight_sum, uint32_t connection_sum, uint32_t wlc_power);
 };