diff --git a/CMakeLists.txt b/CMakeLists.txt index c1e3f5d25..13f95e409 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -17,6 +17,14 @@ endif() # ON by default with on-demand mode - zero overhead until a profiler connects. option(FW_PROFILING "Enable Tracy profiler instrumentation" ON) +set(FW_METRICS_BACKEND "prometheus" CACHE STRING "Metrics backend") +set_property(CACHE FW_METRICS_BACKEND PROPERTY STRINGS prometheus none) +string(TOLOWER "${FW_METRICS_BACKEND}" FW_METRICS_BACKEND) +if(NOT FW_METRICS_BACKEND STREQUAL "prometheus" AND NOT FW_METRICS_BACKEND STREQUAL "none") + message(FATAL_ERROR "Unsupported FW_METRICS_BACKEND: ${FW_METRICS_BACKEND}") +endif() +message(STATUS "Metrics backend: ${FW_METRICS_BACKEND}") + if (WIN32) if (MSVC) add_compile_options( diff --git a/code/framework/CMakeLists.txt b/code/framework/CMakeLists.txt index b7d261357..f352ea3fe 100644 --- a/code/framework/CMakeLists.txt +++ b/code/framework/CMakeLists.txt @@ -1,6 +1,14 @@ include(UpdateGitHash) update_git_version(FW "${CMAKE_CURRENT_SOURCE_DIR}/src/utils/version.cpp.in" "${CMAKE_CURRENT_BINARY_DIR}/version.cpp") +if(FW_METRICS_BACKEND STREQUAL "prometheus") + set(FRAMEWORK_METRICS_BACKEND_SOURCE src/metrics/backends/prometheus.cpp) +elseif(FW_METRICS_BACKEND STREQUAL "none") + set(FRAMEWORK_METRICS_BACKEND_SOURCE src/metrics/backends/none.cpp) +else() + message(FATAL_ERROR "Unsupported FW_METRICS_BACKEND: ${FW_METRICS_BACKEND}") +endif() + # List the source files set(FRAMEWORK_SRC src/logging/logger.cpp @@ -54,6 +62,9 @@ set(FRAMEWORK_SRC src/jobs/job_system.cpp src/jobs/io_tasks.cpp + + src/metrics/registry.cpp + ${FRAMEWORK_METRICS_BACKEND_SOURCE} ) set(FRAMEWORK_SERVER_SRC @@ -303,6 +314,9 @@ target_compile_features(FrameworkV8Engine PRIVATE cxx_std_20) # Link the solutions based on the platform link_shared_deps(Framework) target_link_libraries(Framework v8 v8pp) +if(FW_METRICS_BACKEND STREQUAL "prometheus") + target_link_libraries(Framework prometheus-cpp-lite) +endif() if(LIBNODE_AVAILABLE) target_link_libraries(Framework libnode) endif() diff --git a/code/framework/src/http/webserver.cpp b/code/framework/src/http/webserver.cpp index c32c55c3e..760caeff7 100644 --- a/code/framework/src/http/webserver.cpp +++ b/code/framework/src/http/webserver.cpp @@ -10,8 +10,60 @@ #include #include +#include + +#include namespace Framework::HTTP { + namespace { + Metrics::Counter *RouteStatusCounter(const std::string &route, const char *method, const char *codeClass) { + return Metrics::Registry::Get().RegisterCounter("fw_http_requests_total", "HTTP requests by route, method, and status class", {{"route", route}, {"method", method}, {"code", codeClass}}); + } + + struct RouteMetrics { + Metrics::Counter *c2xx = nullptr; + Metrics::Counter *c4xx = nullptr; + Metrics::Counter *c5xx = nullptr; + Metrics::Counter *cother = nullptr; + Metrics::Histogram *duration = nullptr; + Metrics::Gauge *inFlight = nullptr; + }; + + RouteMetrics MakeRouteMetrics(const std::string &route, const char *method) { + auto ® = Metrics::Registry::Get(); + RouteMetrics metrics { + RouteStatusCounter(route, method, "2xx"), + RouteStatusCounter(route, method, "4xx"), + RouteStatusCounter(route, method, "5xx"), + RouteStatusCounter(route, method, "other"), + reg.RegisterHistogram("fw_http_request_duration_seconds", "HTTP route handler duration", Metrics::Buckets::Exponential(0.0005, 2.0, 13), {{"route", route}, {"method", method}}), + reg.RegisterGauge("fw_http_requests_in_flight", "HTTP requests currently executing a route handler", {{"route", route}, {"method", method}}), + }; + return metrics; + } + + void RecordRouteStatus(const RouteMetrics &c, int status) { + Metrics::Counter *sel = status >= 200 && status < 300 ? c.c2xx : status >= 400 && status < 500 ? c.c4xx : status >= 500 && status < 600 ? c.c5xx : c.cother; + if (sel) { + sel->Inc(); + } + } + + int EffectiveStatus(const httplib::Response &res) { + return res.status == -1 ? 200 : res.status; + } + + void FinishRoute(const RouteMetrics &metrics, std::chrono::steady_clock::time_point startedAt, int status) { + RecordRouteStatus(metrics, status); + if (metrics.duration) { + metrics.duration->Observe(std::chrono::duration(std::chrono::steady_clock::now() - startedAt).count()); + } + if (metrics.inFlight) { + metrics.inFlight->Add(-1.0); + } + } + } + Webserver::Webserver() { _server = std::make_shared(); } @@ -77,17 +129,51 @@ namespace Framework::HTTP { void Webserver::RegisterRequest(const std::string &path, const RequestCallback &callback) const { if (!_running) return; - if (!path.empty() && callback) { - _server->Get(path, callback); - } + if (path.empty() || !callback) + return; + + const RouteMetrics metrics = MakeRouteMetrics(path, "GET"); + RequestCallback userCb = callback; + auto wrapped = [userCb, metrics](const httplib::Request &req, httplib::Response &res) { + const auto startedAt = std::chrono::steady_clock::now(); + if (metrics.inFlight) { + metrics.inFlight->Add(1.0); + } + try { + userCb(req, res); + FinishRoute(metrics, startedAt, EffectiveStatus(res)); + } + catch (...) { + FinishRoute(metrics, startedAt, 500); + throw; + } + }; + _server->Get(path, wrapped); } void Webserver::RegisterPostRequest(const std::string &path, const PostCallback &callback) const { if (!_running) return; - if (!path.empty() && callback) { - _server->Post(path, callback); - } + if (path.empty() || !callback) + return; + + const RouteMetrics metrics = MakeRouteMetrics(path, "POST"); + PostCallback userCb = callback; + auto wrapped = [userCb, metrics](const httplib::Request &req, httplib::Response &res, const httplib::ContentReader &reader) { + const auto startedAt = std::chrono::steady_clock::now(); + if (metrics.inFlight) { + metrics.inFlight->Add(1.0); + } + try { + userCb(req, res, reader); + FinishRoute(metrics, startedAt, EffectiveStatus(res)); + } + catch (...) { + FinishRoute(metrics, startedAt, 500); + throw; + } + }; + _server->Post(path, wrapped); } void Webserver::ServeDirectory(const std::string &dir) { diff --git a/code/framework/src/integrations/server/instance.cpp b/code/framework/src/integrations/server/instance.cpp index 8819f293d..4e870ba24 100644 --- a/code/framework/src/integrations/server/instance.cpp +++ b/code/framework/src/integrations/server/instance.cpp @@ -9,8 +9,8 @@ #include "instance.h" #include -#include #include +#include #include #include "core_modules.h" @@ -37,8 +37,8 @@ #include "utils/command_processor.h" #include "utils/path.h" #include "utils/profiler.h" -#include "utils/version.h" #include "utils/time.h" +#include "utils/version.h" #include "cxxopts.hpp" #include @@ -48,6 +48,7 @@ namespace Framework::Integrations::Server { Instance::Instance(): _shuttingDown(false) { + _processStart = std::chrono::steady_clock::now(); _networkingEngine = std::make_unique(); _webServer = std::make_unique(); _fileConfig = std::make_unique(); @@ -187,6 +188,8 @@ namespace Framework::Integrations::Server { // Register the default endpoints InitEndpoints(); + InitMetrics(); + // Initialize default messages InitNetworkingMessages(); @@ -195,7 +198,7 @@ namespace Framework::Integrations::Server { // Initialize mod subsystems PostInit(); - + const auto sdkCallback = [this](Framework::Scripting::Engine *engine) { this->RegisterScriptingBuiltins(engine); }; @@ -281,6 +284,63 @@ namespace Framework::Integrations::Server { Logging::GetLogger(FRAMEWORK_INNER_HTTP)->debug("All core endpoints have been registered!"); } + void Instance::InitMetrics() { + auto ® = Metrics::Registry::Get(); + + const std::string configLabel = +#ifdef NDEBUG + "Release" +#else + "Debug" +#endif + ; + auto *buildInfo = reg.RegisterGauge("fw_build_info", "Framework build version and configuration", {{"version", Utils::Version::rel}, {"config", configLabel}}); + buildInfo->Set(1.0); + + _uptimeGauge = reg.RegisterGauge("fw_uptime_seconds", "Process uptime in seconds"); + _uptimeGauge->Set(0.0); + + _tickDurationHist = reg.RegisterHistogram("fw_server_tick_duration_seconds", "Server tick wall-clock duration in seconds", Metrics::Buckets::Exponential(0.0005, 2.0, 10)); + _tickLatenessHist = reg.RegisterHistogram("fw_server_tick_lateness_seconds", "Delay between the scheduled and actual tick start", Metrics::Buckets::Exponential(0.0001, 2.0, 12)); + _tickIntervalHist = reg.RegisterHistogram("fw_server_tick_interval_seconds", "Elapsed wall time between consecutive tick starts", Metrics::Buckets::Exponential(0.001, 2.0, 11)); + _tickRateGauge = reg.RegisterGauge("fw_server_tick_rate_hz", "Instantaneous achieved server tick rate"); + _tickTargetRateGauge = reg.RegisterGauge("fw_server_tick_target_hz", "Configured target server tick rate"); + _tickRateGauge->Set(0.0); + _tickTargetRateGauge->Set(_opts.worldConfig.tickInterval > 0.0f ? 1.0 / static_cast(_opts.worldConfig.tickInterval) : 0.0); + + _tickOverrunsCounter = reg.RegisterCounter("fw_server_tick_budget_overruns_total", "Ticks whose duration exceeded the configured tick interval"); + + _connFailAuth = reg.RegisterCounter("fw_net_connection_failures_total", "Connection failures by handshake stage", {{"stage", "auth"}}); + + _masterlistUpdates = reg.RegisterCounter("fw_masterlist_connector_updates_total", "Server-info updates submitted to the masterlist connector"); + _masterlistUpdateErrors = reg.RegisterCounter("fw_masterlist_connector_update_errors_total", "Synchronous masterlist connector update failures"); + + if (_opts.metrics.enabled && reg.HasExporter() && _opts.webServerEnabled && _webServer) { + const std::string token = _opts.metrics.token; + const std::string path = _opts.metrics.path.empty() ? std::string("/metrics") : _opts.metrics.path; + const std::string contentType(reg.ContentType()); + _webServer->RegisterRequest(path, [token, contentType](const httplib::Request &req, httplib::Response &res) { + if (!token.empty()) { + const std::string auth = req.get_header_value("Authorization"); + if (auth != "Bearer " + token) { + res.status = 401; + res.set_content("unauthorized\n", "text/plain"); + return; + } + } + thread_local std::string buf; + Metrics::Registry::Get().Render(buf); + res.set_content(buf, contentType); + res.status = 200; + }); + if (token.empty()) { + Logging::GetLogger(FRAMEWORK_INNER_HTTP)->warn("Metrics exporter mounted at {} without authentication", path); + } + else { + Logging::GetLogger(FRAMEWORK_INNER_HTTP)->info("Metrics exporter mounted at {}", path); + } + } + } bool Instance::LoadConfigFromJSON() { const auto configHandle = cppfs::fs::open(_opts.modConfigFile); @@ -307,6 +367,19 @@ namespace Framework::Integrations::Server { _opts.bindMapName = _fileConfig->Get("map"); _opts.maxPlayers = _fileConfig->Get("maxplayers"); _opts.bindSecretKey = _fileConfig->Get("server-token"); + + if (const auto *doc = _fileConfig->GetDocument(); doc && doc->contains("metrics") && (*doc)["metrics"].is_object()) { + const auto &m = (*doc)["metrics"]; + if (m.contains("enabled") && m["enabled"].is_boolean()) { + _opts.metrics.enabled = m["enabled"].get(); + } + if (m.contains("path") && m["path"].is_string()) { + _opts.metrics.path = m["path"].get(); + } + if (m.contains("token") && m["token"].is_string()) { + _opts.metrics.token = m["token"].get(); + } + } } catch (const std::exception &ex) { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->critical("JSON config has missing fields: {}", ex.what()); @@ -357,10 +430,13 @@ namespace Framework::Integrations::Server { const auto guid = packet->guid; if (!net->IsAuthenticated(guid)) { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->warn("Ignoring identity from unauthenticated peer {}", guid.g); + if (_connFailAuth) { + _connFailAuth->Inc(); + } return; } - auto *replication = net->GetReplicationManager(); + auto *replication = net->GetReplicationManager(); const auto peerGuid = MafiaNet::ToPeerGuid(guid); if (replication && (replication->GetConnectionByGUID(guid) || replication->GetViewer(peerGuid))) { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->warn("Ignoring duplicate identity from {}", guid.g); @@ -397,6 +473,7 @@ namespace Framework::Integrations::Server { // Gate opens: this connection now starts receiving the replicated world. net->PushReplicationConnection(guid); + net->MarkClientReady(guid); // Arm our half of the spawn barrier (the client armed its half before sending identity). const int eventId = Framework::Networking::NetworkServer::ReadyEventId(guid); @@ -488,8 +565,7 @@ namespace Framework::Integrations::Server { v8::TryCatch tryCatch(isolate); v8::Local jsonStr; v8::Local parsed; - if (!v8::String::NewFromUtf8(isolate, payloadJson.c_str()).ToLocal(&jsonStr) || - !v8::JSON::Parse(context, jsonStr).ToLocal(&parsed)) { + if (!v8::String::NewFromUtf8(isolate, payloadJson.c_str()).ToLocal(&jsonStr) || !v8::JSON::Parse(context, jsonStr).ToLocal(&parsed)) { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->warn("Dropping client event '{}' from {}: malformed JSON payload", eventName, senderNetworkId); return; } @@ -509,8 +585,8 @@ namespace Framework::Integrations::Server { const auto net = GetNetworkingEngine()->GetNetworkServer(); const auto streamer = net->GetAssetStreamer(); - const auto scripting = GetScriptingModule(); - const auto resourcesPath = scripting->GetResourcesPath(); + const auto scripting = GetScriptingModule(); + const auto resourcesPath = scripting->GetResourcesPath(); const std::string assetsPath = Framework::Utils::GetAbsolutePathA(resourcesPath); Logging::GetLogger(FRAMEWORK_INNER_SERVER)->debug("Resources directory: {}", assetsPath); @@ -521,11 +597,13 @@ namespace Framework::Integrations::Server { if (resourceManager) { for (const auto &resourceName : resourceManager->GetAllResourceNames()) { const auto resource = resourceManager->GetResource(resourceName); - if (!resource) continue; + if (!resource) + continue; // Only process resources with client entry points const auto &clientEntryRelative = resource->GetManifest().GetMafiaHubConfig().client; - if (clientEntryRelative.empty()) continue; + if (clientEntryRelative.empty()) + continue; const auto resourcePath = resource->GetPath(); @@ -635,11 +713,11 @@ namespace Framework::Integrations::Server { void Instance::InitCommandListener() { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->debug("Setting up command listener and processor..."); - + _commandListener->SetCommandCallback([this](const std::string &command) { this->HandleCommand(command); }); - + _commandProcessor->RegisterCommand( "help", {}, [this](cxxopts::ParseResult &) { @@ -650,7 +728,7 @@ namespace Framework::Integrations::Server { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info("Available commands:\n{}", ss.str()); }, "Show this help message"); - + _commandProcessor->RegisterCommand( "quit", {}, [this](cxxopts::ParseResult &) { @@ -676,7 +754,8 @@ namespace Framework::Integrations::Server { auto res = rm->StopResource(args[0]); if (res) { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info("Stopped resource '{}'", args[0]); - } else { + } + else { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->error("Failed to stop '{}': {}", args[0], res.GetError()); } }, @@ -688,9 +767,9 @@ namespace Framework::Integrations::Server { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info("Server status:"); Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info(" Name: {}", _opts.modName); Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info(" Host: {}:{}", _opts.bindHost, _opts.bindPort); - + if (_networkingEngine) { - const auto net = _networkingEngine->GetNetworkServer(); + const auto net = _networkingEngine->GetNetworkServer(); const auto peer = net->GetPeer(); Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info(" Players: {}/{}", peer->NumberOfConnections(), _opts.maxPlayers); } @@ -713,35 +792,36 @@ namespace Framework::Integrations::Server { auto res = op(rm, args[0]); if (res) { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info("{}: '{}'", verb, args[0]); - } else { + } + else { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->error("Failed to {} '{}': {}", verb, args[0], res.GetError()); } }; }; - _commandProcessor->RegisterCommand( - "start", {}, - resourceCommand("start", [](Framework::Scripting::ResourceManager *rm, const std::string &n) { - return rm->StartResource(n); - }), + _commandProcessor->RegisterCommand("start", {}, + resourceCommand("start", + [](Framework::Scripting::ResourceManager *rm, const std::string &n) { + return rm->StartResource(n); + }), "Start a resource: start "); - _commandProcessor->RegisterCommand( - "restart", {}, - resourceCommand("restart", [](Framework::Scripting::ResourceManager *rm, const std::string &n) { - if (!rm->IsResourceRunning(n)) { - return Framework::Scripting::ResourceOperationResult(std::string("resource is not running (use start)")); - } - return rm->RestartResource(n); - }), + _commandProcessor->RegisterCommand("restart", {}, + resourceCommand("restart", + [](Framework::Scripting::ResourceManager *rm, const std::string &n) { + if (!rm->IsResourceRunning(n)) { + return Framework::Scripting::ResourceOperationResult(std::string("resource is not running (use start)")); + } + return rm->RestartResource(n); + }), "Reload a running resource's code: restart "); // Start-or-reload — the canonical verb FiveM/MTASA operators expect. - _commandProcessor->RegisterCommand( - "ensure", {}, - resourceCommand("ensure", [](Framework::Scripting::ResourceManager *rm, const std::string &n) { - return rm->IsResourceRunning(n) ? rm->RefreshResource(n) : rm->StartResource(n); - }), + _commandProcessor->RegisterCommand("ensure", {}, + resourceCommand("ensure", + [](Framework::Scripting::ResourceManager *rm, const std::string &n) { + return rm->IsResourceRunning(n) ? rm->RefreshResource(n) : rm->StartResource(n); + }), "Start or reload a resource: ensure "); // Re-scan for new/changed resources (manifests), without restarting. @@ -767,7 +847,8 @@ namespace Framework::Integrations::Server { auto res = rm->RefreshAll(); if (res) { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info("Refreshed all resources ({} affected)", res.GetValue().size()); - } else { + } + else { Logging::GetLogger(FRAMEWORK_INNER_SERVER)->error("Failed to refresh all resources: {}", res.GetError()); } }, @@ -781,28 +862,21 @@ namespace Framework::Integrations::Server { auto result = _commandProcessor->ProcessCommand(command); if (result.GetError() != Utils::CommandProcessorError::COMMAND_NONE) { switch (result.GetError()) { - case Utils::CommandProcessorError::COMMAND_PRINT_HELP: - Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info("{}", result.GetValue()); - break; - case Utils::CommandProcessorError::COMMAND_UNKNOWN: { - // Not a built-in command; hand it to the mod override and the scripting layer. - std::vector tokens = Utils::CommandProcessor::Tokenize(command); - if (!tokens.empty()) { - const std::string name = std::move(tokens.front()); - tokens.erase(tokens.begin()); - OnConsoleCommand(std::string(command), name, tokens); - EmitConsoleCommand(name, tokens); - } - break; + case Utils::CommandProcessorError::COMMAND_PRINT_HELP: Logging::GetLogger(FRAMEWORK_INNER_SERVER)->info("{}", result.GetValue()); break; + case Utils::CommandProcessorError::COMMAND_UNKNOWN: { + // Not a built-in command; hand it to the mod override and the scripting layer. + std::vector tokens = Utils::CommandProcessor::Tokenize(command); + if (!tokens.empty()) { + const std::string name = std::move(tokens.front()); + tokens.erase(tokens.begin()); + OnConsoleCommand(std::string(command), name, tokens); + EmitConsoleCommand(name, tokens); } - case Utils::CommandProcessorError::COMMAND_EMPTY_INPUT: - break; - case Utils::CommandProcessorError::COMMAND_INTERNAL_ERROR: - Logging::GetLogger(FRAMEWORK_INNER_SERVER)->error("Error processing command ({}): {}", command, result.GetValue()); - break; - default: - Logging::GetLogger(FRAMEWORK_INNER_SERVER)->error("Error processing command ({}): {}", command, static_cast(result.GetError())); - break; + break; + } + case Utils::CommandProcessorError::COMMAND_EMPTY_INPUT: break; + case Utils::CommandProcessorError::COMMAND_INTERNAL_ERROR: Logging::GetLogger(FRAMEWORK_INNER_SERVER)->error("Error processing command ({}): {}", command, result.GetValue()); break; + default: Logging::GetLogger(FRAMEWORK_INNER_SERVER)->error("Error processing command ({}): {}", command, static_cast(result.GetError())); break; } } } @@ -885,10 +959,25 @@ namespace Framework::Integrations::Server { } void Instance::Update() { - const auto start = std::chrono::high_resolution_clock::now(); + const auto start = std::chrono::steady_clock::now(); if (_nextTick <= start) { FW_PROFILE_SCOPE_N("Server::Tick"); + if (_hasLastTickStart && _tickLatenessHist) { + _tickLatenessHist->Observe(std::chrono::duration(start - _nextTick).count()); + } + if (_hasLastTickStart) { + const double intervalSeconds = std::chrono::duration(start - _lastTickStart).count(); + if (_tickIntervalHist) { + _tickIntervalHist->Observe(intervalSeconds); + } + if (_tickRateGauge && intervalSeconds > 0.0) { + _tickRateGauge->Set(1.0 / intervalSeconds); + } + } + _lastTickStart = start; + _hasLastTickStart = true; + if (_networkingEngine) { FW_PROFILE_SCOPE_N("Server::Networking"); _networkingEngine->Update(); @@ -912,17 +1001,40 @@ namespace Framework::Integrations::Server { info.version = Utils::Version::rel; info.maxPlayers = _opts.maxPlayers; info.currentPlayers = _networkingEngine->GetNetworkServer()->GetPeer()->NumberOfConnections(); - _masterlist->Ping(info); + try { + _masterlist->Ping(info); + if (_masterlistUpdates) { + _masterlistUpdates->Inc(); + } + } + catch (...) { + if (_masterlistUpdateErrors) { + _masterlistUpdateErrors->Inc(); + } + Logging::GetLogger(FRAMEWORK_INNER_SERVER)->warn("Masterlist connector update failed"); + } } { FW_PROFILE_SCOPE_N("Server::PostUpdate"); PostUpdate(); } + + const auto tickEnd = std::chrono::steady_clock::now(); + const double tickSeconds = std::chrono::duration(tickEnd - start).count(); + if (_uptimeGauge) { + _uptimeGauge->Set(std::chrono::duration(tickEnd - _processStart).count()); + } + if (_tickDurationHist) { + _tickDurationHist->Observe(tickSeconds); + } + if (_tickOverrunsCounter && tickSeconds > _opts.worldConfig.tickInterval) { + _tickOverrunsCounter->Inc(); + } FW_PROFILE_FRAME(); - _nextTick = std::chrono::high_resolution_clock::now() + std::chrono::milliseconds(static_cast(Utils::Time::SecondsToMs(_opts.worldConfig.tickInterval))); + _nextTick = std::chrono::steady_clock::now() + std::chrono::duration_cast(std::chrono::duration(_opts.worldConfig.tickInterval)); } else { std::this_thread::sleep_for(std::chrono::milliseconds(1)); diff --git a/code/framework/src/integrations/server/instance.h b/code/framework/src/integrations/server/instance.h index dd49b79bc..e2440a637 100644 --- a/code/framework/src/integrations/server/instance.h +++ b/code/framework/src/integrations/server/instance.h @@ -15,16 +15,17 @@ #include "http/webserver.h" #include "logging/logger.h" +#include "metrics/registry.h" #include "networking/engine.h" #include "scripting/module.h" #include -#include #include "services/masterlist.h" -#include "utils/config.h" #include "utils/command_listener.h" #include "utils/command_processor.h" +#include "utils/config.h" +#include #include @@ -46,7 +47,6 @@ namespace v8 { namespace Framework::Integrations::Server { struct InstanceOptions { - std::string modSlug; std::string modHelpText; std::string modName; @@ -83,7 +83,7 @@ namespace Framework::Integrations::Server { // MafiaHub Services struct Services { - std::string apiUrl = "https://api.mafiahub.dev"; + std::string apiUrl = "https://api.mafiahub.dev"; std::string masterlistUrl = ""; } services; @@ -94,6 +94,12 @@ namespace Framework::Integrations::Server { int32_t maxPlayers; std::string httpServeDir; + struct MetricsConfig { + bool enabled = false; + std::string path = "/metrics"; + std::string token; + } metrics; + bool enableSignals; // update intervals and streaming @@ -111,7 +117,6 @@ namespace Framework::Integrations::Server { // args int argc; char **argv; - }; // Connection metadata handed to the player-connect callback so the game can create and fully @@ -126,12 +131,26 @@ namespace Framework::Integrations::Server { std::string discordId; }; - class Instance : public Framework::Lifecycle { + class Instance: public Framework::Lifecycle { private: std::atomic _shuttingDown; // Set after the initial StartAll; gates runtime broadcasts to clients. bool _resourcesBooted = false; - std::chrono::time_point _nextTick; + std::chrono::time_point _nextTick; + std::chrono::time_point _processStart; + std::chrono::time_point _lastTickStart; + bool _hasLastTickStart = false; + + Metrics::Histogram *_tickDurationHist = nullptr; + Metrics::Histogram *_tickLatenessHist = nullptr; + Metrics::Histogram *_tickIntervalHist = nullptr; + Metrics::Gauge *_tickRateGauge = nullptr; + Metrics::Gauge *_tickTargetRateGauge = nullptr; + Metrics::Gauge *_uptimeGauge = nullptr; + Metrics::Counter *_tickOverrunsCounter = nullptr; + Metrics::Counter *_connFailAuth = nullptr; + Metrics::Counter *_masterlistUpdates = nullptr; + Metrics::Counter *_masterlistUpdateErrors = nullptr; InstanceOptions _opts; @@ -145,6 +164,7 @@ namespace Framework::Integrations::Server { std::unique_ptr _crashReporter; void InitEndpoints(); + void InitMetrics(); void InitNetworkingMessages(); void InitAssetStreamer(); // Re-sync a hot-reloaded/started client resource to connected clients. @@ -154,7 +174,7 @@ namespace Framework::Integrations::Server { void InitCommandListener(); bool LoadConfigFromJSON(); void RegisterScriptingBuiltins(Framework::Scripting::Engine *); - + void HandleCommand(std::string_view command); void EmitConsoleCommand(const std::string &command, const std::vector &args); diff --git a/code/framework/src/jobs/job_system.cpp b/code/framework/src/jobs/job_system.cpp index 6d1a50e98..062527bf1 100644 --- a/code/framework/src/jobs/job_system.cpp +++ b/code/framework/src/jobs/job_system.cpp @@ -10,38 +10,143 @@ #include +#include + namespace Framework::Jobs { + namespace { + Metrics::Counter *g_jobsSchedHigh = nullptr; + Metrics::Counter *g_jobsSchedNormal = nullptr; + Metrics::Counter *g_jobsExceptions = nullptr; + Metrics::Gauge *g_jobsQueueHigh = nullptr; + Metrics::Gauge *g_jobsQueueNormal = nullptr; + Metrics::Gauge *g_jobsActive = nullptr; + Metrics::Histogram *g_jobsWaitHigh = nullptr; + Metrics::Histogram *g_jobsWaitNormal = nullptr; + Metrics::Histogram *g_jobsExecHigh = nullptr; + Metrics::Histogram *g_jobsExecNormal = nullptr; + std::once_flag g_jobMetricsOnce; + + void EnsureJobCounters() { + std::call_once(g_jobMetricsOnce, [] { + auto ® = Metrics::Registry::Get(); + g_jobsSchedHigh = reg.RegisterCounter("fw_jobs_scheduled_total", "Jobs scheduled", {{"priority", "high"}}); + g_jobsSchedNormal = reg.RegisterCounter("fw_jobs_scheduled_total", "Jobs scheduled", {{"priority", "normal"}}); + g_jobsExceptions = reg.RegisterCounter("fw_jobs_exceptions_total", "Jobs that threw an uncaught exception"); + g_jobsQueueHigh = reg.RegisterGauge("fw_jobs_queue_depth", "Jobs queued but not yet executing", {{"priority", "high"}}); + g_jobsQueueNormal = reg.RegisterGauge("fw_jobs_queue_depth", "Jobs queued but not yet executing", {{"priority", "normal"}}); + g_jobsActive = reg.RegisterGauge("fw_jobs_active", "Jobs that have begun but not yet completed"); + g_jobsWaitHigh = reg.RegisterHistogram("fw_jobs_queue_wait_duration_seconds", "Time from scheduling until execution begins", Metrics::Buckets::Exponential(0.00005, 4.0, 9), {{"priority", "high"}}); + g_jobsWaitNormal = reg.RegisterHistogram("fw_jobs_queue_wait_duration_seconds", "Time from scheduling until execution begins", Metrics::Buckets::Exponential(0.00005, 4.0, 9), {{"priority", "normal"}}); + g_jobsExecHigh = reg.RegisterHistogram("fw_jobs_execution_duration_seconds", "Job execution wall time", Metrics::Buckets::Exponential(0.00005, 4.0, 9), {{"priority", "high"}}); + g_jobsExecNormal = reg.RegisterHistogram("fw_jobs_execution_duration_seconds", "Job execution wall time", Metrics::Buckets::Exponential(0.00005, 4.0, 9), {{"priority", "normal"}}); + g_jobsQueueHigh->Set(0.0); + g_jobsQueueNormal->Set(0.0); + g_jobsActive->Set(0.0); + }); + } + + Metrics::Gauge *QueueGauge(ftl::TaskPriority priority) { + return priority == ftl::TaskPriority::High ? g_jobsQueueHigh : g_jobsQueueNormal; + } + + Metrics::Histogram *WaitHistogram(ftl::TaskPriority priority) { + return priority == ftl::TaskPriority::High ? g_jobsWaitHigh : g_jobsWaitNormal; + } + + Metrics::Histogram *ExecutionHistogram(ftl::TaskPriority priority) { + return priority == ftl::TaskPriority::High ? g_jobsExecHigh : g_jobsExecNormal; + } + } // namespace + + namespace Detail { + void RecordTaskScheduled(ftl::TaskPriority priority) noexcept { + auto *counter = priority == ftl::TaskPriority::High ? g_jobsSchedHigh : g_jobsSchedNormal; + if (counter) { + counter->Inc(); + } + if (auto *queue = QueueGauge(priority)) { + queue->Add(1.0); + } + } + + void RecordTaskStarted(ftl::TaskPriority priority, std::chrono::steady_clock::time_point enqueuedAt, std::chrono::steady_clock::time_point startedAt) noexcept { + if (auto *queue = QueueGauge(priority)) { + queue->Add(-1.0); + } + if (g_jobsActive) { + g_jobsActive->Add(1.0); + } + if (auto *wait = WaitHistogram(priority)) { + wait->Observe(std::chrono::duration(startedAt - enqueuedAt).count()); + } + } + + void RecordTaskFinished(ftl::TaskPriority priority, std::chrono::steady_clock::time_point startedAt) noexcept { + if (auto *execution = ExecutionHistogram(priority)) { + execution->Observe(std::chrono::duration(std::chrono::steady_clock::now() - startedAt).count()); + } + if (g_jobsActive) { + g_jobsActive->Add(-1.0); + } + } + + void ReportTaskException(const std::exception &exception) { + if (g_jobsExceptions) { + g_jobsExceptions->Inc(); + } + spdlog::error("Task threw exception: {}", exception.what()); + } + + void ReportUnknownTaskException() { + if (g_jobsExceptions) { + g_jobsExceptions->Inc(); + } + spdlog::error("Task threw unknown exception"); + } + } // namespace Detail // Helper struct to wrap fu2::function for FTL's C-style function pointer struct TaskWrapper { fu2::function func; + ftl::TaskPriority priority = ftl::TaskPriority::Normal; + std::chrono::steady_clock::time_point enqueuedAt; }; static void TaskWrapperFunc(ftl::TaskScheduler * /*scheduler*/, void *arg) { - auto *wrapper = static_cast(arg); + auto *wrapper = static_cast(arg); + const auto startedAt = std::chrono::steady_clock::now(); + Detail::RecordTaskStarted(wrapper->priority, wrapper->enqueuedAt, startedAt); try { wrapper->func(); - } catch (const std::exception &e) { - spdlog::error("Task threw exception: {}", e.what()); - } catch (...) { - spdlog::error("Task threw unknown exception"); } + catch (const std::exception &e) { + Detail::ReportTaskException(e); + } + catch (...) { + Detail::ReportUnknownTaskException(); + } + Detail::RecordTaskFinished(wrapper->priority, startedAt); delete wrapper; } - JobSystem::JobSystem(const JobSystemConfig &config) : _profilingEnabled(config.enableProfiling), _config(config) { + JobSystem::JobSystem(const JobSystemConfig &config): _profilingEnabled(config.enableProfiling), _config(config) { _scheduler = std::make_unique(); } JobSystemError JobSystem::Init() { ftl::TaskSchedulerInitOptions options; - options.FiberPoolSize = _config.fiberPoolSize; + options.FiberPoolSize = _config.fiberPoolSize; options.ThreadPoolSize = _config.workerThreadCount; - options.Behavior = _config.emptyQueueBehavior; + options.Behavior = _config.emptyQueueBehavior; + + EnsureJobCounters(); + _blockingCalls = Metrics::Registry::Get().RegisterCounter("fw_jobs_blocking_calls_total", "BlockingCall invocations (detached-thread pressure indicator)"); + _callbackQueueDepth = Metrics::Registry::Get().RegisterGauge("fw_jobs_callback_queue_depth", "Pending completed-callback queue depth"); + _callbackQueueDepth->Set(0.0); if (_config.enableProfiling) { // Profiling callbacks can be added here for Tracy/Remotery integration - options.Callbacks.Context = this; + options.Callbacks.Context = this; options.Callbacks.OnFiberAttached = [](void * /*context*/, unsigned /*fiberIndex*/) { // Hook for profiler: fiber attached to thread }; @@ -73,12 +178,13 @@ namespace Framework::Jobs { } void JobSystem::Schedule(fu2::function task, ftl::TaskPriority priority) { - auto *wrapper = new TaskWrapper{std::move(task)}; + auto *wrapper = new TaskWrapper {std::move(task), priority, std::chrono::steady_clock::now()}; ftl::Task ftlTask; ftlTask.Function = TaskWrapperFunc; - ftlTask.ArgData = wrapper; + ftlTask.ArgData = wrapper; + Detail::RecordTaskScheduled(priority); _scheduler->AddTask(ftlTask, priority); } @@ -89,21 +195,29 @@ namespace Framework::Jobs { } void JobSystem::Schedule(ftl::WaitGroup *waitGroup, fu2::function task, ftl::TaskPriority priority) { - auto *wrapper = new TaskWrapper{std::move(task)}; + auto *wrapper = new TaskWrapper {std::move(task), priority, std::chrono::steady_clock::now()}; ftl::Task ftlTask; ftlTask.Function = TaskWrapperFunc; - ftlTask.ArgData = wrapper; + ftlTask.ArgData = wrapper; + Detail::RecordTaskScheduled(priority); _scheduler->AddTask(ftlTask, priority, waitGroup); } + void JobSystem::RecordBlockingCall() { + if (_blockingCalls) { + _blockingCalls->Inc(); + } + } + void JobSystem::ScheduleWithCallback(fu2::function task, fu2::function onSuccess, fu2::function onError, ftl::TaskPriority priority) { auto wrappedTask = [this, task = std::move(task), onSuccess = std::move(onSuccess), onError = std::move(onError)]() mutable { std::exception_ptr exception; try { task(); - } catch (...) { + } + catch (...) { exception = std::current_exception(); } @@ -112,16 +226,20 @@ namespace Framework::Jobs { QueueCallback([onError = std::move(onError), exception]() mutable { onError(exception); }); - } else { + } + else { try { std::rethrow_exception(exception); - } catch (const std::exception &e) { + } + catch (const std::exception &e) { spdlog::error("Task failed with unhandled exception: {}", e.what()); - } catch (...) { + } + catch (...) { spdlog::error("Task failed with unknown exception"); } } - } else if (onSuccess) { + } + else if (onSuccess) { QueueCallback(std::move(onSuccess)); } }; @@ -135,15 +253,20 @@ namespace Framework::Jobs { { std::scoped_lock lock(_callbackMutex); std::swap(callbacks, _completedCallbacks); + if (_callbackQueueDepth) { + _callbackQueueDepth->Set(0.0); + } } while (!callbacks.empty()) { auto &cb = callbacks.front(); try { cb.callback(); - } catch (const std::exception &e) { + } + catch (const std::exception &e) { spdlog::error("Callback threw exception: {}", e.what()); - } catch (...) { + } + catch (...) { spdlog::error("Callback threw unknown exception"); } callbacks.pop(); @@ -152,7 +275,10 @@ namespace Framework::Jobs { void JobSystem::QueueCallback(fu2::function callback) { std::scoped_lock lock(_callbackMutex); - _completedCallbacks.push(CompletedCallback{std::move(callback)}); + _completedCallbacks.push(CompletedCallback {std::move(callback)}); + if (_callbackQueueDepth) { + _callbackQueueDepth->Set(static_cast(_completedCallbacks.size())); + } } } // namespace Framework::Jobs diff --git a/code/framework/src/jobs/job_system.h b/code/framework/src/jobs/job_system.h index 4fab24955..f85338a9f 100644 --- a/code/framework/src/jobs/job_system.h +++ b/code/framework/src/jobs/job_system.h @@ -10,13 +10,16 @@ #include "errors.h" +#include #include #include -#include -#include #include +#include +#include +#include +#include #include #include #include @@ -54,11 +57,11 @@ namespace Framework::Jobs { bool success = true; static TaskResult Success(T val) { - return TaskResult{std::move(val), "", true}; + return TaskResult {std::move(val), "", true}; } static TaskResult Failure(std::string err) { - return TaskResult{T{}, std::move(err), false}; + return TaskResult {T {}, std::move(err), false}; } }; @@ -68,11 +71,11 @@ namespace Framework::Jobs { bool success = true; static TaskResult Success() { - return TaskResult{"", true}; + return TaskResult {"", true}; } static TaskResult Failure(std::string err) { - return TaskResult{std::move(err), false}; + return TaskResult {std::move(err), false}; } }; @@ -83,6 +86,14 @@ namespace Framework::Jobs { fu2::function callback; }; + namespace Detail { + void RecordTaskScheduled(ftl::TaskPriority priority) noexcept; + void RecordTaskStarted(ftl::TaskPriority priority, std::chrono::steady_clock::time_point enqueuedAt, std::chrono::steady_clock::time_point startedAt) noexcept; + void RecordTaskFinished(ftl::TaskPriority priority, std::chrono::steady_clock::time_point startedAt) noexcept; + void ReportTaskException(const std::exception &exception); + void ReportUnknownTaskException(); + } // namespace Detail + /** * @brief Fiber-based job system built on FTL (Fiber Tasking Library) * @@ -111,7 +122,7 @@ namespace Framework::Jobs { * @brief Construct a new JobSystem * @param config Configuration options */ - explicit JobSystem(const JobSystemConfig &config = JobSystemConfig{}); + explicit JobSystem(const JobSystemConfig &config = JobSystemConfig {}); ~JobSystem(); @@ -121,10 +132,10 @@ namespace Framework::Jobs { */ [[nodiscard]] JobSystemError Init(); - JobSystem(const JobSystem &) = delete; + JobSystem(const JobSystem &) = delete; JobSystem &operator=(const JobSystem &) = delete; - JobSystem(JobSystem &&) = delete; - JobSystem &operator=(JobSystem &&) = delete; + JobSystem(JobSystem &&) = delete; + JobSystem &operator=(JobSystem &&) = delete; /** * @brief Schedule a task to run on a worker fiber @@ -169,7 +180,58 @@ namespace Framework::Jobs { */ template void ScheduleBatch(std::vector &items, Func &&func, size_t batchSize = 1, ftl::TaskPriority priority = ftl::TaskPriority::Normal) { - ftl::ParallelFor(_scheduler.get(), items.begin(), items.end(), batchSize, std::forward(func), priority); + if (items.empty()) { + return; + } + if (batchSize == 0) { + batchSize = 1; + } + + using Callable = std::remove_reference_t; + struct BatchTask { + T *items = nullptr; + size_t count = 0; + Callable *func = nullptr; + ftl::TaskPriority priority = ftl::TaskPriority::Normal; + std::chrono::steady_clock::time_point enqueuedAt; + }; + + const size_t batchCount = ((items.size() - 1) / batchSize) + 1; + std::vector batches(batchCount); + ftl::WaitGroup waitGroup(_scheduler.get()); + for (size_t batchIndex = 0; batchIndex < batchCount; ++batchIndex) { + const size_t begin = batchIndex * batchSize; + auto &batch = batches[batchIndex]; + batch.items = items.data() + begin; + batch.count = std::min(batchSize, items.size() - begin); + batch.func = std::addressof(func); + batch.priority = priority; + batch.enqueuedAt = std::chrono::steady_clock::now(); + + ftl::Task task {}; + task.ArgData = &batch; + task.Function = [](ftl::TaskScheduler *scheduler, void *arg) { + auto *batchTask = static_cast(arg); + const auto startedAt = std::chrono::steady_clock::now(); + Detail::RecordTaskStarted(batchTask->priority, batchTask->enqueuedAt, startedAt); + try { + for (size_t i = 0; i < batchTask->count; ++i) { + (*batchTask->func)(scheduler, &batchTask->items[i]); + } + } + catch (const std::exception &e) { + Detail::ReportTaskException(e); + } + catch (...) { + Detail::ReportUnknownTaskException(); + } + Detail::RecordTaskFinished(batchTask->priority, startedAt); + }; + + Detail::RecordTaskScheduled(priority); + _scheduler->AddTask(task, priority, &waitGroup); + } + waitGroup.Wait(); } /** @@ -185,8 +247,9 @@ namespace Framework::Jobs { * FOOTGUN: detached thread, no timeout; capture owned types by value, not refs to locals. */ template ()())> - requires (!std::is_void_v) + requires(!std::is_void_v) ReturnType BlockingCall(Func &&func) { + RecordBlockingCall(); // For blocking I/O, we use a WaitGroup to yield the fiber ftl::WaitGroup wg(_scheduler.get()); wg.Add(1); @@ -198,7 +261,8 @@ namespace Framework::Jobs { std::thread([&]() { try { result = func(); - } catch (...) { + } + catch (...) { exception = std::current_exception(); } wg.Done(); @@ -220,6 +284,7 @@ namespace Framework::Jobs { template ()())> requires std::is_void_v void BlockingCall(Func &&func) { + RecordBlockingCall(); // For blocking I/O, we use a WaitGroup to yield the fiber ftl::WaitGroup wg(_scheduler.get()); wg.Add(1); @@ -230,7 +295,8 @@ namespace Framework::Jobs { std::thread([&]() { try { func(); - } catch (...) { + } + catch (...) { exception = std::current_exception(); } wg.Done(); @@ -292,6 +358,11 @@ namespace Framework::Jobs { return std::make_unique(_scheduler.get()); } + private: + Metrics::Counter *_blockingCalls = nullptr; + Metrics::Gauge *_callbackQueueDepth = nullptr; + void RecordBlockingCall(); + private: std::unique_ptr _scheduler; diff --git a/code/framework/src/metrics/backend.h b/code/framework/src/metrics/backend.h new file mode 100644 index 000000000..ae3d1627d --- /dev/null +++ b/code/framework/src/metrics/backend.h @@ -0,0 +1,48 @@ +/* + * MafiaHub OSS license + * Copyright (c) 2021-2026, MafiaHub. All rights reserved. + * + * This file comes from MafiaHub, hosted at https://github.com/MafiaHub/Framework. + * See LICENSE file in the source repository for information regarding licensing. + */ + +#pragma once + +#include "metrics/registry.h" + +#include + +namespace Framework::Metrics::Detail { + + struct CounterHandle { + void *impl = nullptr; + void (*increment)(void *, std::uint64_t) noexcept = nullptr; + }; + + struct GaugeHandle { + void *impl = nullptr; + void (*set)(void *, double) noexcept = nullptr; + void (*add)(void *, double) noexcept = nullptr; + }; + + struct HistogramHandle { + void *impl = nullptr; + void (*observe)(void *, double) noexcept = nullptr; + }; + + class Backend { + public: + virtual ~Backend() = default; + + virtual CounterHandle RegisterCounter(std::string_view name, std::string_view help, std::initializer_list