Added support for choosing best ZEN upstream endpoint based on latency.

5 files changed, 112 insertions, 16 deletions

diff --git a/zenserver/config.cpp b/zenserver/config.cpp
index 759534d58..7a1efe6e8 100644
--- a/zenserver/config.cpp
+++ b/zenserver/config.cpp
@@ -213,8 +213,8 @@ ParseGlobalCliOptions(int argc, char* argv[], ZenServerOptions& GlobalOptions, Z
 	options.add_option("cache",
 					   "",
 					   "upstream-zen-url",
-					   "URL to a remote Zen server instance",
-					   cxxopts::value<std::string>(ServiceConfig.UpstreamCacheConfig.ZenConfig.Url)->default_value(""),
+					   "URL to remote Zen server. Use a comma separated list to choose the one with the best latency.",
+					   cxxopts::value<std::vector<std::string>>(ServiceConfig.UpstreamCacheConfig.ZenConfig.Urls)->default_value(""),
 					   "");
 
 	options.add_option("cache",
@@ -367,9 +367,17 @@ ParseServiceConfig(const std::filesystem::path& DataRoot, ZenServiceConfig& Serv
 
 				if (auto ZenConfig = UpstreamConfig->get<sol::optional<sol::table>>("zen"))
 				{
-					UpdateStringValueFromConfig(ZenConfig.value(),
-												std::string_view("url"),
-												ServiceConfig.UpstreamCacheConfig.ZenConfig.Url);
+					if (auto Url = ZenConfig.value().get<sol::optional<std::string>>("url"))
+					{
+						ServiceConfig.UpstreamCacheConfig.ZenConfig.Urls.push_back(Url.value());
+					}
+					else if (auto Urls = ZenConfig.value().get<sol::optional<sol::table>>("url"))
+					{
+						for (const auto& Kv : Urls.value())
+						{
+							ServiceConfig.UpstreamCacheConfig.ZenConfig.Urls.push_back(Kv.second.as<std::string>());
+						}
+					}
 				}
 			}
 		}
diff --git a/zenserver/config.h b/zenserver/config.h
index af1a24455..ec6d7340a 100644
--- a/zenserver/config.h
+++ b/zenserver/config.h
@@ -35,7 +35,7 @@ struct ZenUpstreamJupiterConfig
 
 struct ZenUpstreamZenConfig
 {
-	std::string Url;
+	std::vector<std::string> Urls;
 };
 
 enum class UpstreamCachePolicy : uint8_t
diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index b1966e299..a183e7ebf 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -111,6 +111,8 @@ namespace detail {
 
 		virtual ~JupiterUpstreamEndpoint() = default;
 
+		virtual UpstreamEndpointHealth Initialize() override { return CheckHealth(); }
+
 		virtual bool IsHealthy() const override { return m_HealthOk.load(); }
 
 		virtual UpstreamEndpointHealth CheckHealth() override
@@ -400,22 +402,70 @@ namespace detail {
 
 	class ZenUpstreamEndpoint final : public UpstreamEndpoint
 	{
+		struct ZenEndpoint
+		{
+			std::string Url;
+			std::string Reason;
+			double		Latency{};
+			bool		Ok = false;
+
+			bool operator<(const ZenEndpoint& RHS) const { return Ok && RHS.Ok ? Latency < RHS.Latency : Ok; }
+		};
+
 	public:
-		ZenUpstreamEndpoint(std::string_view ServiceUrl)
+		ZenUpstreamEndpoint(std::span<std::string const> Urls) : m_Log(zen::logging::Get("upstream")), m_DisplayName("ZEN")
 		{
-			using namespace fmt::literals;
-			m_DisplayName = "Zen - {}"_format(ServiceUrl);
-			m_Client	  = new ZenStructuredCacheClient(ServiceUrl);
+			for (const auto& Url : Urls)
+			{
+				m_Endpoints.push_back({.Url = Url});
+			}
 		}
 
 		~ZenUpstreamEndpoint() = default;
 
+		virtual UpstreamEndpointHealth Initialize() override
+		{
+			using namespace fmt::literals;
+
+			const ZenEndpoint& Ep = GetEndpoint();
+			if (Ep.Ok)
+			{
+				m_ServiceUrl  = Ep.Url;
+				m_DisplayName = "ZEN - {}"_format(m_ServiceUrl);
+				m_Client	  = new ZenStructuredCacheClient(m_ServiceUrl);
+
+				m_HealthOk = true;
+				return {.Ok = true};
+			}
+
+			m_HealthOk = false;
+			return {.Reason = Ep.Reason};
+		}
+
 		virtual bool IsHealthy() const override { return m_HealthOk; }
 
 		virtual UpstreamEndpointHealth CheckHealth() override
 		{
+			using namespace fmt::literals;
+
 			try
 			{
+				if (m_Client.IsNull())
+				{
+					const ZenEndpoint& Ep = GetEndpoint();
+					if (Ep.Ok)
+					{
+						m_ServiceUrl  = Ep.Url;
+						m_DisplayName = "ZEN - {}"_format(m_ServiceUrl);
+						m_Client	  = new ZenStructuredCacheClient(m_ServiceUrl);
+
+						m_HealthOk = true;
+						return {.Ok = true};
+					}
+
+					return {.Reason = Ep.Reason};
+				}
+
 				ZenStructuredCacheSession Session(*m_Client);
 				ZenCacheResult			  Result;
 
@@ -591,6 +641,42 @@ namespace detail {
 		virtual UpstreamEndpointStats& Stats() override { return m_Stats; }
 
 	private:
+		const ZenEndpoint& GetEndpoint()
+		{
+			for (ZenEndpoint& Ep : m_Endpoints)
+			{
+				ZenStructuredCacheClient  Client(Ep.Url);
+				ZenStructuredCacheSession Session(Client);
+				const int32_t			  SampleCount = 2;
+
+				Ep.Ok	   = false;
+				Ep.Latency = {};
+
+				for (int32_t Sample = 0; Sample < SampleCount; ++Sample)
+				{
+					ZenCacheResult Result = Session.CheckHealth();
+					Ep.Ok				  = Result.Success;
+					Ep.Reason			  = std::move(Result.Reason);
+					Ep.Latency += Result.ElapsedSeconds;
+				}
+				Ep.Latency /= double(SampleCount);
+			}
+
+			std::sort(std::begin(m_Endpoints), std::end(m_Endpoints));
+
+			for (const auto& Ep : m_Endpoints)
+			{
+				ZEN_INFO("ping ZEN endpoint '{}' latency '{:.3}s' {}", Ep.Url, Ep.Latency, Ep.Ok ? "OK" : Ep.Reason);
+			}
+
+			return m_Endpoints.front();
+		}
+
+		spdlog::logger& Log() { return m_Log; }
+
+		spdlog::logger&					 m_Log;
+		std::string						 m_ServiceUrl;
+		std::vector<ZenEndpoint>		 m_Endpoints;
 		std::string						 m_DisplayName;
 		RefPtr<ZenStructuredCacheClient> m_Client;
 		UpstreamEndpointStats			 m_Stats;
@@ -711,7 +797,7 @@ public:
 	{
 		for (auto& Endpoint : m_Endpoints)
 		{
-			const UpstreamEndpointHealth Health = Endpoint->CheckHealth();
+			const UpstreamEndpointHealth Health = Endpoint->Initialize();
 			if (Health.Ok)
 			{
 				ZEN_INFO("initialize endpoint '{}' OK", Endpoint->DisplayName());
@@ -993,9 +1079,9 @@ MakeJupiterUpstreamEndpoint(const CloudCacheClientOptions& Options)
 }
 
 std::unique_ptr<UpstreamEndpoint>
-MakeZenUpstreamEndpoint(std::string_view Url)
+MakeZenUpstreamEndpoint(std::span<std::string const> Urls)
 {
-	return std::make_unique<detail::ZenUpstreamEndpoint>(Url);
+	return std::make_unique<detail::ZenUpstreamEndpoint>(Urls);
 }
 
 }  // namespace zen
diff --git a/zenserver/upstream/upstreamcache.h b/zenserver/upstream/upstreamcache.h
index 08f379b11..edc995da6 100644
--- a/zenserver/upstream/upstreamcache.h
+++ b/zenserver/upstream/upstreamcache.h
@@ -96,6 +96,8 @@ class UpstreamEndpoint
 public:
 	virtual ~UpstreamEndpoint() = default;
 
+	virtual UpstreamEndpointHealth Initialize() = 0;
+
 	virtual bool IsHealthy() const = 0;
 
 	virtual UpstreamEndpointHealth CheckHealth() = 0;
@@ -143,6 +145,6 @@ std::unique_ptr<UpstreamCache> MakeUpstreamCache(const UpstreamCacheOptions& Opt
 
 std::unique_ptr<UpstreamEndpoint> MakeJupiterUpstreamEndpoint(const CloudCacheClientOptions& Options);
 
-std::unique_ptr<UpstreamEndpoint> MakeZenUpstreamEndpoint(std::string_view Url);
+std::unique_ptr<UpstreamEndpoint> MakeZenUpstreamEndpoint(std::span<std::string const> Urls);
 
 }  // namespace zen
diff --git a/zenserver/zenserver.cpp b/zenserver/zenserver.cpp
index db1be9dea..f9b4d5677 100644
--- a/zenserver/zenserver.cpp
+++ b/zenserver/zenserver.cpp
@@ -193,9 +193,9 @@ public:
 
 				UpstreamCache = zen::MakeUpstreamCache(UpstreamOptions, *m_CacheStore, *m_CidStore);
 
-				if (!UpstreamConfig.ZenConfig.Url.empty())
+				if (!UpstreamConfig.ZenConfig.Urls.empty())
 				{
-					std::unique_ptr<zen::UpstreamEndpoint> ZenEndpoint = zen::MakeZenUpstreamEndpoint(UpstreamConfig.ZenConfig.Url);
+					std::unique_ptr<zen::UpstreamEndpoint> ZenEndpoint = zen::MakeZenUpstreamEndpoint(UpstreamConfig.ZenConfig.Urls);
 					UpstreamCache->RegisterEndpoint(std::move(ZenEndpoint));
 				}