path: root/src/zenserver/hub/zenhubserver.cpp

// Copyright Epic Games, Inc. All Rights Reserved.

#include "zenhubserver.h"

#include "frontend/frontend.h"
#include "httphubservice.h"
#include "hub.h"

#include <zencore/config.h>
#include <zencore/fmtutils.h>
#include <zencore/memory/llm.h>
#include <zencore/memory/memorytrace.h>
#include <zencore/memory/tagtrace.h>
#include <zencore/scopeguard.h>
#include <zencore/sentryintegration.h>
#include <zencore/windows.h>
#include <zenhttp/httpapiservice.h>
#include <zenutil/service.h>
#include <zenutil/workerpools.h>

ZEN_THIRD_PARTY_INCLUDES_START
#include <cxxopts.hpp>
ZEN_THIRD_PARTY_INCLUDES_END

namespace zen {

const std::string&
GetDefaultConsulTokenEnvVariableName()
{
	static const std::string Name = "CONSUL_HTTP_TOKEN";
	return Name;
}

void
ZenHubServerConfigurator::AddCliOptions(cxxopts::Options& Options)
{
	const char* DefaultInstanceHttp = "asio";

#if ZEN_WITH_HTTPSYS
	if (!windows::IsRunningOnWine())
	{
		DefaultInstanceHttp = "httpsys";
	}
#endif

	Options.add_option("hub",
					   "",
					   "upstream-notification-endpoint",
					   "Endpoint URL for upstream notifications",
					   cxxopts::value<std::string>(m_ServerOptions.UpstreamNotificationEndpoint)->default_value(""),
					   "");

	Options.add_option("hub",
					   "",
					   "instance-id",
					   "Instance ID for use in notifications",
					   cxxopts::value<std::string>(m_ServerOptions.InstanceId)->default_value(""),
					   "");

	Options.add_option("hub",
					   "",
					   "consul-endpoint",
					   "Consul endpoint URL for service registration (empty = disabled)",
					   cxxopts::value<std::string>(m_ServerOptions.ConsulEndpoint)->default_value(""),
					   "");

	Options.add_option("hub",
					   "",
					   "consul-token-env",
					   fmt::format("Name of environment variable that holds the consul access token (defaults to '{}')",
								   GetDefaultConsulTokenEnvVariableName()),
					   cxxopts::value<std::string>(m_ServerOptions.ConsulTokenEnv)->default_value(""),
					   "<envvariable>");

	Options.add_option("hub",
					   "",
					   "consul-health-interval-seconds",
					   "Interval in seconds between Consul health checks",
					   cxxopts::value<uint32_t>(m_ServerOptions.ConsulHealthIntervalSeconds)->default_value("10"),
					   "<seconds>");

	Options.add_option("hub",
					   "",
					   "consul-deregister-after-seconds",
					   "Seconds after which Consul deregisters an unhealthy service",
					   cxxopts::value<uint32_t>(m_ServerOptions.ConsulDeregisterAfterSeconds)->default_value("30"),
					   "<seconds>");

	Options.add_option("hub",
					   "",
					   "hub-base-port-number",
					   "Base port number for provisioned instances",
					   cxxopts::value<uint16_t>(m_ServerOptions.HubBasePortNumber)->default_value("21000"),
					   "");

	Options.add_option("hub",
					   "",
					   "hub-instance-limit",
					   "Maximum number of provisioned instances for this hub",
					   cxxopts::value<int>(m_ServerOptions.HubInstanceLimit)->default_value("1000"),
					   "");

	Options.add_option("hub",
					   "",
					   "hub-instance-http",
					   "Select HTTP server implementation for provisioned instances (asio|"
#if ZEN_WITH_HTTPSYS
					   "httpsys|"
#endif
					   "null)",
					   cxxopts::value<std::string>(m_ServerOptions.HubInstanceHttpClass)->default_value(DefaultInstanceHttp),
					   "<instance http class>");

	Options.add_option("hub",
					   "",
					   "hub-instance-http-threads",
					   "Number of http server connection threads for provisioned instances",
					   cxxopts::value<unsigned int>(m_ServerOptions.HubInstanceHttpThreadCount),
					   "<instance http threads>");
	Options.add_option("hub",
					   "",
					   "hub-instance-corelimit",
					   "Limit concurrency of provisioned instances",
					   cxxopts::value(m_ServerOptions.HubInstanceCoreLimit),
					   "<instance core limit>");

	Options.add_option("hub",
					   "",
					   "hub-instance-config",
					   "Path to Lua config file for provisioned instances",
					   cxxopts::value(m_ServerOptions.HubInstanceConfigPath),
					   "<instance config>");

	Options.add_option("hub",
					   "",
					   "hub-hydration-target-spec",
					   "Specification for hydration target. 'file://<path>' prefix indicates file storage at <path>. Defaults to "
					   "<data-dir>/servers/hydration_storage",
					   cxxopts::value(m_ServerOptions.HydrationTargetSpecification),
					   "<hydration-target-spec>");

#if ZEN_PLATFORM_WINDOWS
	Options.add_option("hub",
					   "",
					   "hub-use-job-object",
					   "Enable the use of a Windows Job Object for child process management",
					   cxxopts::value<bool>(m_ServerOptions.HubUseJobObject)->default_value("true"),
					   "");
#endif	// ZEN_PLATFORM_WINDOWS
}

void
ZenHubServerConfigurator::AddConfigOptions(LuaConfig::Options& Options)
{
	ZEN_UNUSED(Options);
}

void
ZenHubServerConfigurator::ApplyOptions(cxxopts::Options& Options)
{
	ZEN_UNUSED(Options);
}

void
ZenHubServerConfigurator::OnConfigFileParsed(LuaConfig::Options& LuaOptions)
{
	ZEN_UNUSED(LuaOptions);
}

void
ZenHubServerConfigurator::ValidateOptions()
{
}

///////////////////////////////////////////////////////////////////////////

ZenHubServer::ZenHubServer()
{
}

ZenHubServer::~ZenHubServer()
{
	Cleanup();
}

void
ZenHubServer::OnModuleStateChanged(std::string_view					 HubInstanceId,
								   std::string_view					 ModuleId,
								   const HubProvisionedInstanceInfo& Info,
								   HubInstanceState					 PreviousState,
								   HubInstanceState					 NewState)
{
	ZEN_UNUSED(PreviousState);
	if (!m_ConsulClient)
	{
		return;
	}

	if (NewState == HubInstanceState::Provisioning || NewState == HubInstanceState::Provisioned)
	{
		consul::ServiceRegistrationInfo ServiceInfo{
			.ServiceId				= std::string(ModuleId),
			.ServiceName			= "zen-storage",
			.Port					= Info.Port,
			.HealthEndpoint			= "health",
			.Tags					= std::vector<std::pair<std::string, std::string>>{std::make_pair("module", std::string(ModuleId)),
																	   std::make_pair("zen-hub", std::string(HubInstanceId)),
																	   std::make_pair("version", std::string(ZEN_CFG_VERSION))},
			.HealthIntervalSeconds	= NewState == HubInstanceState::Provisioning
										  ? 0u
										  : m_ConsulHealthIntervalSeconds,	// Disable health checks while not finished provisioning
			.DeregisterAfterSeconds = NewState == HubInstanceState::Provisioning
										  ? 0u
										  : m_ConsulDeregisterAfterSeconds};  // Disable health checks while not finished provisioning

		if (!m_ConsulClient->RegisterService(ServiceInfo))
		{
			ZEN_WARN("Failed to register storage server instance for module '{}' with Consul, continuing anyway", ModuleId);
		}
		else
		{
			ZEN_INFO("Registered storage server instance for module '{}' at port {} with Consul as '{}'",
					 ModuleId,
					 Info.Port,
					 ServiceInfo.ServiceName);
		}
	}
	else if (NewState == HubInstanceState::Unprovisioned)
	{
		if (!m_ConsulClient->DeregisterService(ModuleId))
		{
			ZEN_WARN("Failed to deregister storage server instance for module '{}' at port {} from Consul, continuing anyway",
					 ModuleId,
					 Info.Port);
		}
		else
		{
			ZEN_INFO("Deregistered storage server instance for module '{}' at port {} from Consul", ModuleId, Info.Port);
		}
	}
	// Transitional states (Deprovisioning, Hibernating, Waking, Recovering, Crashed)
	// and Hibernated are intentionally ignored.
}

int
ZenHubServer::Initialize(const ZenHubServerConfig& ServerConfig, ZenServerState::ZenServerEntry* ServerEntry)
{
	ZEN_TRACE_CPU("ZenHubServer::Initialize");
	ZEN_MEMSCOPE(GetZenserverTag());

	ZEN_INFO(ZEN_APP_NAME " initializing in HUB server mode");

	const int EffectiveBasePort = ZenServerBase::Initialize(ServerConfig, ServerEntry);
	if (EffectiveBasePort < 0)
	{
		return EffectiveBasePort;
	}

	// This is a workaround to make sure we can have automated tests. Without
	// this the ranges for different child zen hub processes could overlap with
	// the main test range.
	ZenServerEnvironment::SetBaseChildId(1000);

	m_DebugOptionForcedCrash = ServerConfig.ShouldCrash;

	InitializeState(ServerConfig);
	InitializeConsulRegistration(ServerConfig, EffectiveBasePort);
	InitializeServices(ServerConfig);
	RegisterServices(ServerConfig);

	ZenServerBase::Finalize();

	return EffectiveBasePort;
}

void
ZenHubServer::Cleanup()
{
	ZEN_TRACE_CPU("ZenHubServer::Cleanup");
	ZEN_INFO(ZEN_APP_NAME " cleaning up");
	try
	{
		m_IoContext.stop();
		if (m_IoRunner.joinable())
		{
			m_IoRunner.join();
		}

		ShutdownServices();
		if (m_Http)
		{
			m_Http->Close();
		}

		if (m_Hub)
		{
			m_Hub->Shutdown();
		}

		m_FrontendService.reset();
		m_HubService.reset();
		m_ApiService.reset();
		m_Hub.reset();

		m_ConsulRegistration.reset();
		m_ConsulClient.reset();
	}
	catch (const std::exception& Ex)
	{
		ZEN_ERROR("exception thrown during Cleanup() in {}: '{}'", ZEN_APP_NAME, Ex.what());
	}
}

void
ZenHubServer::InitializeState(const ZenHubServerConfig& ServerConfig)
{
	ZEN_UNUSED(ServerConfig);
}

void
ZenHubServer::InitializeServices(const ZenHubServerConfig& ServerConfig)
{
	ZEN_UNUSED(ServerConfig);

	ZEN_INFO("instantiating Hub");
	m_Hub = std::make_unique<Hub>(
		Hub::Configuration{.UseJobObject				 = ServerConfig.HubUseJobObject,
						   .BasePortNumber				 = ServerConfig.HubBasePortNumber,
						   .InstanceLimit				 = ServerConfig.HubInstanceLimit,
						   .InstanceHttpThreadCount		 = ServerConfig.HubInstanceHttpThreadCount,
						   .InstanceCoreLimit			 = ServerConfig.HubInstanceCoreLimit,
						   .InstanceConfigPath			 = ServerConfig.HubInstanceConfigPath,
						   .HydrationTargetSpecification = ServerConfig.HydrationTargetSpecification},
		ZenServerEnvironment(ZenServerEnvironment::Hub,
							 ServerConfig.DataDir / "hub",
							 ServerConfig.DataDir / "servers",
							 ServerConfig.HubInstanceHttpClass),
		&GetMediumWorkerPool(EWorkloadType::Background),
		m_ConsulClient ? Hub::AsyncModuleStateChangeCallbackFunc{[this, HubInstanceId = fmt::format("zen-hub-{}", ServerConfig.InstanceId)](
																	 std::string_view				   ModuleId,
																	 const HubProvisionedInstanceInfo& Info,
																	 HubInstanceState				   PreviousState,
																	 HubInstanceState				   NewState) {
			OnModuleStateChanged(HubInstanceId, ModuleId, Info, PreviousState, NewState);
		}}
					   : Hub::AsyncModuleStateChangeCallbackFunc{});

	ZEN_INFO("instantiating API service");
	m_ApiService = std::make_unique<zen::HttpApiService>(*m_Http);

	ZEN_INFO("instantiating hub service");
	m_HubService = std::make_unique<HttpHubService>(*m_Hub);
	m_HubService->SetNotificationEndpoint(ServerConfig.UpstreamNotificationEndpoint, ServerConfig.InstanceId);

	m_FrontendService = std::make_unique<HttpFrontendService>(m_ContentRoot, m_StatusService);
}

void
ZenHubServer::RegisterServices(const ZenHubServerConfig& ServerConfig)
{
	ZEN_UNUSED(ServerConfig);

	if (m_HubService)
	{
		m_Http->RegisterService(*m_HubService);
	}

	if (m_ApiService)
	{
		m_Http->RegisterService(*m_ApiService);
	}

	if (m_FrontendService)
	{
		m_Http->RegisterService(*m_FrontendService);
	}
}

void
ZenHubServer::InitializeConsulRegistration(const ZenHubServerConfig& ServerConfig, int EffectivePort)
{
	if (ServerConfig.ConsulEndpoint.empty())
	{
		ZEN_INFO("Consul registration disabled (no endpoint configured)");
		return;
	}

	ZEN_INFO("Initializing Consul registration with endpoint: {}", ServerConfig.ConsulEndpoint);

	std::string ConsulAccessTokenEnvName =
		ServerConfig.ConsulTokenEnv.empty() ? GetDefaultConsulTokenEnvVariableName() : ServerConfig.ConsulTokenEnv;
	std::string ConsulAccessToken = GetEnvVariable(ConsulAccessTokenEnvName);
	if (ConsulAccessToken.empty())
	{
		if (!ServerConfig.ConsulTokenEnv.empty())
		{
			ZEN_WARN("Consul token environment variable '{}' is not set or empty", ServerConfig.ConsulTokenEnv);
		}
	}
	else
	{
		ZEN_INFO("Consul token read from environment variable '{}'", ConsulAccessTokenEnvName);
	}

	try
	{
		m_ConsulClient				   = std::make_unique<consul::ConsulClient>(ServerConfig.ConsulEndpoint, ConsulAccessToken);
		m_ConsulHealthIntervalSeconds  = ServerConfig.ConsulHealthIntervalSeconds;
		m_ConsulDeregisterAfterSeconds = ServerConfig.ConsulDeregisterAfterSeconds;

		consul::ServiceRegistrationInfo Info;
		Info.ServiceId	 = fmt::format("zen-hub-{}", ServerConfig.InstanceId);
		Info.ServiceName = "zen-hub";
		// Info.Address		= "localhost";	// Let the consul agent figure out out external address // TODO: Info.BaseUri?
		Info.Port			= static_cast<uint16_t>(EffectivePort);
		Info.HealthEndpoint = "hub/health";
		Info.Tags			= std::vector<std::pair<std::string, std::string>>{
			  std::make_pair("zen-hub", Info.ServiceId),
			  std::make_pair("version", std::string(ZEN_CFG_VERSION)),
			  std::make_pair("base-port-number", fmt::format("{}", ServerConfig.HubBasePortNumber)),
			  std::make_pair("instance-limit", fmt::format("{}", ServerConfig.HubInstanceLimit)),
			  std::make_pair("use-job-object", fmt::format("{}", ServerConfig.HubUseJobObject))};
		Info.HealthIntervalSeconds	= ServerConfig.ConsulHealthIntervalSeconds;
		Info.DeregisterAfterSeconds = ServerConfig.ConsulDeregisterAfterSeconds;

		m_ConsulRegistration = std::make_unique<consul::ServiceRegistration>(m_ConsulClient.get(), Info);

		ZEN_INFO("Consul service registration initiated for service ID: {}", Info.ServiceId);
	}
	catch (const std::exception& Ex)
	{
		// REQ-F-12: Hub should start successfully even if Consul registration fails
		ZEN_WARN("Failed to initialize Consul registration (hub will continue without it): {}", Ex.what());
		m_ConsulRegistration.reset();
		m_ConsulClient.reset();
	}
}

void
ZenHubServer::Run()
{
	if (m_ProcessMonitor.IsActive())
	{
		CheckOwnerPid();
	}

	if (!m_TestMode)
	{
		// clang-format off
		ZEN_INFO(R"(__________                ___ ___      ___.    )" "\n"
				 R"(\____    /____   ____    /   |   \ __ _\_ |__  )" "\n"
				 R"(  /     // __ \ /    \  /    ~    \  |  \ __ \ )" "\n"
				 R"( /     /\  ___/|   |  \ \    Y    /  |  / \_\ \)" "\n"
				 R"(/_______ \___  >___|  /  \___|_  /|____/|___  /)" "\n"
				 R"(        \/   \/     \/         \/           \/ )");
		// clang-format on

		ExtendableStringBuilder<256> BuildOptions;
		GetBuildOptions(BuildOptions, '\n');
		ZEN_INFO("Build options ({}/{}, {}):\n{}", GetOperatingSystemName(), GetCpuName(), GetCompilerName(), BuildOptions);
	}

	ZEN_INFO(ZEN_APP_NAME " now running as HUB (pid: {})", GetCurrentProcessId());

#if ZEN_PLATFORM_WINDOWS
	if (zen::windows::IsRunningOnWine())
	{
		ZEN_INFO("detected Wine session - " ZEN_APP_NAME " is not formally tested on Wine and may therefore not work or perform well");
	}
#endif

#if ZEN_USE_SENTRY
	ZEN_INFO("sentry crash handler {}", m_UseSentry ? "ENABLED" : "DISABLED");
	if (m_UseSentry)
	{
		SentryIntegration::ClearCaches();
	}
#endif

	if (m_DebugOptionForcedCrash)
	{
		ZEN_DEBUG_BREAK();
	}

	const bool IsInteractiveMode = IsInteractiveSession();	// &&!m_TestMode;

	if (m_ConsulRegistration)
	{
		if (!m_ConsulRegistration->IsRegistered())
		{
			ZEN_INFO("Waiting for consul integration to register...");
			m_ConsulRegistration->WaitForReadyEvent(2000);
		}
		if (!m_ConsulRegistration->IsRegistered())
		{
			m_ConsulClient.reset();
			m_ConsulRegistration.reset();
			ZEN_WARN("Consul registration failed, running without consul integration");
		}
	}

	SetNewState(kRunning);

	OnReady();

	m_Http->Run(IsInteractiveMode);

	SetNewState(kShuttingDown);

	ZEN_INFO(ZEN_APP_NAME " exiting");
}

//////////////////////////////////////////////////////////////////////////////////

ZenHubServerMain::ZenHubServerMain(ZenHubServerConfig& ServerOptions) : ZenServerMain(ServerOptions), m_ServerOptions(ServerOptions)
{
}

void
ZenHubServerMain::DoRun(ZenServerState::ZenServerEntry* Entry)
{
	ZenHubServer Server;
	Server.SetServerMode("Hub");
	Server.SetDataRoot(m_ServerOptions.DataDir);
	Server.SetContentRoot(m_ServerOptions.ContentDir);
	Server.SetTestMode(m_ServerOptions.IsTest);
	Server.SetDedicatedMode(m_ServerOptions.IsDedicated);
	Server.SetAllowPortProbing(!m_ServerOptions.IsDedicated && m_ServerOptions.AllowPortProbing);

	const int EffectiveBasePort = Server.Initialize(m_ServerOptions, Entry);
	if (EffectiveBasePort == -1)
	{
		// Server.Initialize has already logged what the issue is - just exit with failure code here.
		std::exit(1);
	}

	Entry->EffectiveListenPort = uint16_t(EffectiveBasePort);
	if (EffectiveBasePort != m_ServerOptions.BasePort)
	{
		ZEN_INFO(ZEN_APP_NAME " - relocated to base port {}", EffectiveBasePort);
		m_ServerOptions.BasePort = EffectiveBasePort;
	}

	std::unique_ptr<std::thread> ShutdownThread;
	std::unique_ptr<NamedEvent>	 ShutdownEvent;

	ExtendableStringBuilder<64> ShutdownEventName;
	ShutdownEventName << "Zen_" << m_ServerOptions.BasePort << "_Shutdown";
	ShutdownEvent.reset(new NamedEvent{ShutdownEventName});

	// Monitor shutdown signals

	ShutdownThread.reset(new std::thread{[&] {
		SetCurrentThreadName("shutdown_mon");

		ZEN_INFO("shutdown monitor thread waiting for shutdown signal '{}' for process {}", ShutdownEventName, zen::GetCurrentProcessId());

		if (ShutdownEvent->Wait())
		{
			ZEN_INFO("shutdown signal for pid {} received", zen::GetCurrentProcessId());
			Server.RequestExit(0);
		}
		else
		{
			ZEN_INFO("shutdown signal wait() failed");
		}
	}});

	auto CleanupShutdown = MakeGuard([&ShutdownEvent, &ShutdownThread] {
		ReportServiceStatus(ServiceStatus::Stopping);

		if (ShutdownEvent)
		{
			ShutdownEvent->Set();
		}
		if (ShutdownThread && ShutdownThread->joinable())
		{
			ShutdownThread->join();
		}
	});

	// If we have a parent process, establish the mechanisms we need
	// to be able to communicate readiness with the parent

	Server.SetIsReadyFunc([&] {
		std::error_code Ec;
		m_LockFile.Update(MakeLockData(true), Ec);
		ReportServiceStatus(ServiceStatus::Running);
		NotifyReady();
	});

	Server.Run();
}

}  // namespace zen