diff --git a/libcockatrice_network/libcockatrice/network/server/remote/game/server_abstract_player.cpp b/libcockatrice_network/libcockatrice/network/server/remote/game/server_abstract_player.cpp index 957a89792..6b4101a99 100644 --- a/libcockatrice_network/libcockatrice/network/server/remote/game/server_abstract_player.cpp +++ b/libcockatrice_network/libcockatrice/network/server/remote/game/server_abstract_player.cpp @@ -66,6 +66,15 @@ Server_AbstractPlayer::Server_AbstractPlayer(Server_Game *_game, Server_AbstractPlayer::~Server_AbstractPlayer() = default; +int Server_AbstractPlayer::getCardCount() const +{ + int result = 0; + for (auto *zone : zones) { + result += zone->getCards().size(); + } + return result; +} + void Server_AbstractPlayer::prepareDestroy() { delete deck; diff --git a/libcockatrice_network/libcockatrice/network/server/remote/game/server_abstract_player.h b/libcockatrice_network/libcockatrice/network/server/remote/game/server_abstract_player.h index 85fbc0557..4cc79c5fe 100644 --- a/libcockatrice_network/libcockatrice/network/server/remote/game/server_abstract_player.h +++ b/libcockatrice_network/libcockatrice/network/server/remote/game/server_abstract_player.h @@ -43,6 +43,8 @@ public: Server_AbstractUserInterface *_handler); ~Server_AbstractPlayer() override; void prepareDestroy() override; + /// Total cards across all of this player's zones. The caller must hold the game's mutex. + int getCardCount() const; const DeckList *getDeckList() const { return deck; diff --git a/libcockatrice_network/libcockatrice/network/server/remote/game/server_game.cpp b/libcockatrice_network/libcockatrice/network/server/remote/game/server_game.cpp index 43209e994..799b1e7ee 100644 --- a/libcockatrice_network/libcockatrice/network/server/remote/game/server_game.cpp +++ b/libcockatrice_network/libcockatrice/network/server/remote/game/server_game.cpp @@ -32,6 +32,7 @@ #include "server_spectator.h" #include +#include #include #include #include @@ -238,6 +239,17 @@ int Server_Game::getPlayerCount() const return participants.size() - getSpectatorCount(); } +int Server_Game::getCardsInGame() const +{ + QMutexLocker locker(&gameMutex); + + int result = 0; + for (auto *player : getPlayers()) { + result += player->getCardCount(); + } + return result; +} + int Server_Game::getSpectatorCount() const { QMutexLocker locker(&gameMutex); @@ -330,6 +342,9 @@ void Server_Game::doStartGameIfReady(bool forceStartGame) } } + // Only actual starts are timed. The early returns above are no-ops. + QElapsedTimer startupTimer; + startupTimer.start(); players = getPlayers(); // players could have been kicked, get new list of players if (lifecycleStrategy->onGameStarting(this) == Server_GameLifecycleStrategy::StartAction::Handled) { locker.unlock(); @@ -373,6 +388,7 @@ void Server_Game::doStartGameIfReady(bool forceStartGame) activePlayer = -1; nextTurn(); + room->getServer()->observeGameStartDurationMs(startupTimer.nsecsElapsed() / 1000000); locker.unlock(); diff --git a/libcockatrice_network/libcockatrice/network/server/remote/game/server_game.h b/libcockatrice_network/libcockatrice/network/server/remote/game/server_game.h index 1b9f651bd..1ed4fe4ca 100644 --- a/libcockatrice_network/libcockatrice/network/server/remote/game/server_game.h +++ b/libcockatrice_network/libcockatrice/network/server/remote/game/server_game.h @@ -123,6 +123,8 @@ public: return gameStarted; } int getPlayerCount() const; + /// Total cards across all players' zones. Takes gameMutex itself. + int getCardsInGame() const; int getSpectatorCount() const; QMap getPlayers() const; Server_AbstractPlayer *getPlayer(int id) const; diff --git a/libcockatrice_network/libcockatrice/network/server/remote/server.h b/libcockatrice_network/libcockatrice/network/server/remote/server.h index 0ded27afa..3d27f4210 100644 --- a/libcockatrice_network/libcockatrice/network/server/remote/server.h +++ b/libcockatrice_network/libcockatrice/network/server/remote/server.h @@ -180,6 +180,11 @@ public: { return false; } + /// Called once per actual game start with how long bringing every player's + /// zones online took, so servers can spot deck sizes that wedge threads. + virtual void observeGameStartDurationMs(qint64 /* elapsedMs */) + { + } Server_DatabaseInterface *getDatabaseInterface() const; int getNextLocalGameId() diff --git a/libcockatrice_network/libcockatrice/network/server/remote/server_protocolhandler.h b/libcockatrice_network/libcockatrice/network/server/remote/server_protocolhandler.h index d62213188..2c8efe50e 100644 --- a/libcockatrice_network/libcockatrice/network/server/remote/server_protocolhandler.h +++ b/libcockatrice_network/libcockatrice/network/server/remote/server_protocolhandler.h @@ -136,7 +136,7 @@ public: return timeRunning - lastDataReceived; } bool addSaidMessageSize(int size); - void processCommandContainer(const CommandContainer &cont); + virtual void processCommandContainer(const CommandContainer &cont); void sendProtocolItem(const Response &item); void sendProtocolItem(const SessionEvent &item); diff --git a/servatrice/CMakeLists.txt b/servatrice/CMakeLists.txt index 5d8089ad1..68e422d8c 100644 --- a/servatrice/CMakeLists.txt +++ b/servatrice/CMakeLists.txt @@ -6,7 +6,9 @@ project(Servatrice VERSION "${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${ set(servatrice_SOURCES src/email_parser.cpp + src/event_loop_watchdog.cpp src/main.cpp + src/metrics_registry.cpp src/servatrice.cpp src/servatrice_connection_pool.cpp src/servatrice_database_interface.cpp diff --git a/servatrice/servatrice.ini.example b/servatrice/servatrice.ini.example index c1940c22f..7e0789073 100644 --- a/servatrice/servatrice.ini.example +++ b/servatrice/servatrice.ini.example @@ -382,6 +382,17 @@ max_reports_per_day=10 ; Maximum number of report comments a single user can post per hour; default is 30; set to 0 to disable the limit max_comments_per_hour=30 +[metrics] +; Command containers that take longer than this many milliseconds are logged +; as slow commands. Set to 0 to disable the log line. +slow_command_ms=500 + +; Each socket pool thread runs a watchdog heartbeat. If a heartbeat arrives +; this many milliseconds late, the stall is logged and exposed as +; servatrice_eventloop_* metrics in the Developer tab. Set to 0 to disable +; the watchdogs. +stall_warn_ms=2000 + [logging] ; Admin/Moderators can query the stored logs for information when looking up reports by various players. This ; option can allow or disallow them from doing so. diff --git a/servatrice/src/event_loop_watchdog.cpp b/servatrice/src/event_loop_watchdog.cpp new file mode 100644 index 000000000..e50bd4201 --- /dev/null +++ b/servatrice/src/event_loop_watchdog.cpp @@ -0,0 +1,34 @@ +/** + * @file event_loop_watchdog.cpp + * @ingroup Servatrice + */ + +#include "event_loop_watchdog.h" + +#include "servatrice.h" + +#include + +EventLoopWatchdog::EventLoopWatchdog(Servatrice *_servatrice, QString _threadName) + : QObject(nullptr), servatrice(_servatrice), threadName(std::move(_threadName)) +{ +} + +void EventLoopWatchdog::start() +{ + heartbeatTimer = new QTimer(this); + sinceLastTick.start(); + connect(heartbeatTimer, &QTimer::timeout, this, &EventLoopWatchdog::checkHeartbeat); + heartbeatTimer->start(HeartbeatIntervalMs); +} + +void EventLoopWatchdog::checkHeartbeat() +{ + const qint64 elapsedMs = sinceLastTick.restart(); + const qint64 overshootMs = qMax(0, elapsedMs - HeartbeatIntervalMs); + if (overshootMs < servatrice->getMetricsStallWarnMs()) { + return; + } + + servatrice->observeEventLoopStall(threadName, overshootMs); +} diff --git a/servatrice/src/event_loop_watchdog.h b/servatrice/src/event_loop_watchdog.h new file mode 100644 index 000000000..b9061ff97 --- /dev/null +++ b/servatrice/src/event_loop_watchdog.h @@ -0,0 +1,50 @@ +/** + * @file event_loop_watchdog.h + * @ingroup Servatrice + */ + +#ifndef EVENT_LOOP_WATCHDOG_H +#define EVENT_LOOP_WATCHDOG_H + +#include +#include +#include + +class Servatrice; +class QTimer; + +/** + * @brief Detects blocked or overloaded worker event loops. + * + * One instance lives in each socket pool thread. A heartbeat timer tick that + * arrives late means the loop spent that time elsewhere: busy work, a queued + * slot, or a hard wedge. Overshoots past the configured threshold bump + * lock-free counters on the metrics registry and log one warning per stall, + * so a stuck pool thread becomes visible instead of silent lag. + */ +class EventLoopWatchdog : public QObject +{ + Q_OBJECT +public: + /// How often the heartbeat expects to fire. Small enough to catch short stalls. + static constexpr int HeartbeatIntervalMs = 500; + + EventLoopWatchdog(Servatrice *_servatrice, QString _threadName); + + /** + * Starts the heartbeat timer. Must be invoked queued after the instance + * was moved to its target thread so the timer lives there too. + */ + void start(); + +private slots: + void checkHeartbeat(); + +private: + Servatrice *servatrice; + QString threadName; + QElapsedTimer sinceLastTick; + QTimer *heartbeatTimer = nullptr; +}; + +#endif diff --git a/servatrice/src/metrics_registry.cpp b/servatrice/src/metrics_registry.cpp new file mode 100644 index 000000000..8e3769954 --- /dev/null +++ b/servatrice/src/metrics_registry.cpp @@ -0,0 +1,142 @@ +#include "metrics_registry.h" + +#include + +int MetricsRegistry::bucketIndexFor(qint64 elapsedMs) +{ + const int lastFiniteBucket = static_cast(BucketBounds.size()) - 1; + int bucket = 0; + while (bucket < lastFiniteBucket && elapsedMs > BucketBounds[static_cast(bucket)]) { + ++bucket; + } + return bucket; +} + +void MetricsRegistry::appendCumulativeBuckets(QString &out, + const QString &bucketLine, + const std::array, BucketCount> &buckets) +{ + // Cumulative buckets are required by the Prometheus histogram convention. + qint64 cumulative = 0; + for (int bucket = 0; bucket < static_cast(BucketBounds.size()); ++bucket) { + cumulative += buckets[static_cast(bucket)].load(std::memory_order_relaxed); + out += QStringLiteral("%1,le=\"%2\"} %3\n") + .arg(bucketLine) + .arg(BucketBounds[static_cast(bucket)]) + .arg(cumulative); + } + cumulative += buckets[BucketCount - 1].load(std::memory_order_relaxed); + out += QStringLiteral("%1,le=\"+Inf\"} %2\n").arg(bucketLine).arg(cumulative); +} + +void MetricsRegistry::observeCommand(int typeId, qint64 elapsedMs) +{ + if (typeId < 0 || typeId >= MaxTypes) { + typeId = MaxTypes - 1; // overflow slot keeps misrouted ids visible + } + if (elapsedMs < 0) { + elapsedMs = 0; + } + + TypeStats &stats = slotFor(typeId); + stats.count.fetch_add(1, std::memory_order_relaxed); + stats.totalMs.fetch_add(elapsedMs, std::memory_order_relaxed); + totalCommandsCounter.fetch_add(1, std::memory_order_relaxed); + totalTimeCounter.fetch_add(elapsedMs, std::memory_order_relaxed); + stats.buckets[static_cast(bucketIndexFor(elapsedMs))].fetch_add(1, std::memory_order_relaxed); +} + +void MetricsRegistry::observeGameStartDurationMs(qint64 elapsedMs) +{ + if (elapsedMs < 0) { + elapsedMs = 0; + } + + gameStartStats.count.fetch_add(1, std::memory_order_relaxed); + gameStartStats.totalMs.fetch_add(elapsedMs, std::memory_order_relaxed); + gameStartStats.buckets[static_cast(bucketIndexFor(elapsedMs))].fetch_add(1, std::memory_order_relaxed); +} + +MetricsRegistry::TypeStats &MetricsRegistry::slotFor(int typeId) +{ + return typeSlots[static_cast(typeId)]; +} + +const MetricsRegistry::TypeStats &MetricsRegistry::slotFor(int typeId) const +{ + return typeSlots[static_cast(typeId)]; +} + +int MetricsRegistry::activeTypeCount() const +{ + int active = 0; + for (int type = 0; type < MaxTypes; ++type) { + if (slotFor(type).count.load(std::memory_order_relaxed) > 0) { + ++active; + } + } + return active; +} + +QString MetricsRegistry::toPrometheusText(const std::function &nameForType, + const QHash &gauges) const +{ + QString out; + out.reserve(4096); + + for (auto it = gauges.constBegin(); it != gauges.constEnd(); ++it) { + out += QStringLiteral("# TYPE %1 gauge\n").arg(it.key()); + out += QStringLiteral("%1 %2\n").arg(it.key()).arg(it.value()); + } + + // Cumulative buckets are required by the Prometheus histogram convention. + out += QLatin1String("# TYPE servatrice_commands_duration_ms histogram\n"); + + for (int type = 0; type < MaxTypes; ++type) { + const TypeStats &stats = slotFor(type); + const qint64 count = stats.count.load(std::memory_order_relaxed); + if (count == 0) { + continue; + } + + const QString label = nameForType ? nameForType(type) : QString::number(type); + appendCumulativeBuckets(out, QStringLiteral("servatrice_commands_duration_ms_bucket{command=\"%1\"").arg(label), + stats.buckets); + + out += QStringLiteral("servatrice_commands_duration_ms_sum{command=\"%1\"} %2\n") + .arg(label) + .arg(stats.totalMs.load(std::memory_order_relaxed)); + out += QStringLiteral("servatrice_commands_duration_ms_count{command=\"%1\"} %2\n").arg(label).arg(count); + } + + const qint64 startCount = gameStartStats.count.load(std::memory_order_relaxed); + if (startCount > 0) { + out += QLatin1String("# TYPE servatrice_game_start_duration_ms histogram\n"); + appendCumulativeBuckets(out, QLatin1String("servatrice_game_start_duration_ms_bucket"), gameStartStats.buckets); + out += QStringLiteral("servatrice_game_start_duration_ms_sum %1\n") + .arg(gameStartStats.totalMs.load(std::memory_order_relaxed)); + out += QStringLiteral("servatrice_game_start_duration_ms_count %1\n").arg(startCount); + } + + return out; +} + +QList MetricsRegistry::collectActiveStats() const +{ + QList result; + for (int type = 0; type < MaxTypes; ++type) { + const TypeStats &stats = slotFor(type); + const qint64 count = stats.count.load(std::memory_order_relaxed); + if (count == 0) { + continue; + } + result.append({type, count, stats.totalMs.load(std::memory_order_relaxed)}); + } + return result; +} + +MetricsRegistry::GameStartSnapshot MetricsRegistry::getGameStartSnapshot() const +{ + return {gameStartStats.count.load(std::memory_order_relaxed), + gameStartStats.totalMs.load(std::memory_order_relaxed)}; +} diff --git a/servatrice/src/metrics_registry.h b/servatrice/src/metrics_registry.h new file mode 100644 index 000000000..cbca2a2c6 --- /dev/null +++ b/servatrice/src/metrics_registry.h @@ -0,0 +1,135 @@ +/** + * @file metrics_registry.h + * @ingroup Servatrice + */ + +#ifndef METRICS_REGISTRY_H +#define METRICS_REGISTRY_H + +#include +#include +#include +#include +#include + +/** + * @brief Lock-free accumulation of command processing statistics. + * + * observeCommand() is called once per processed command from whichever socket + * thread handled it. It uses relaxed atomic adds on preallocated storage only, + * so it introduces no locks, allocations, or shared cache-line ping-pong + * beyond the unavoidable counter updates. + * + * Reading happens rarely (metrics scraping), accepts momentary tears between + * related counters, and therefore also needs no synchronization. + */ +class MetricsRegistry +{ +public: + /** + * Extension numbers are only unique per command kind, so recorded ids + * combine the kind index with the protobuf extension number. + */ + static constexpr int KindStride = 2048; + + static constexpr int NumKinds = 5; + + static constexpr const char *KindNames[NumKinds] = {"session", "room", "game", "moderator", "admin"}; + + /// Upper bound on distinct command type ids (see typeIdFor). + static constexpr int MaxTypes = NumKinds * KindStride; + + /// Histogram bucket upper bounds in milliseconds. Anything above the last + /// bound lands in the trailing +Inf bucket. Constexpr so the recording + /// hot path never allocates. + static constexpr std::array BucketBounds{1, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000}; + + static int typeIdFor(int kindIndex, int extensionNumber) + { + return kindIndex * KindStride + extensionNumber; + } + + void observeCommand(int typeId, qint64 elapsedMs); + + /** + * Records how long one game start took to bring every player's zones + * online. Kept separate from command timings because it is triggered by + * the server itself and can dwarf any single command when decks are huge. + */ + void observeGameStartDurationMs(qint64 elapsedMs); + + /// Total number of observed commands across all types. + qint64 totalCommands() const + { + return totalCommandsCounter.load(std::memory_order_relaxed); + } + + /// Cumulative processing milliseconds across all types. + qint64 totalTimeMs() const + { + return totalTimeCounter.load(std::memory_order_relaxed); + } + + /// Number of distinct type slots that have seen at least one sample. + int activeTypeCount() const; + + struct ActiveTypeStats + { + int typeId; + qint64 count; + qint64 totalMs; + }; + + /** + * Returns stats for every type slot that has seen at least one sample. + * The caller-provided @p labelForType maps a numeric type id to a stable + * human-readable label. Pass nullptr to skip label resolution. + */ + QList collectActiveStats() const; + + struct GameStartSnapshot + { + qint64 count; + qint64 totalMs; + }; + + GameStartSnapshot getGameStartSnapshot() const; + + /** + * @brief Renders all recorded data in Prometheus text exposition format. + * + * @param nameForType maps a numeric command type id to a stable label + * value. Ids without a mapping are rendered as their number. + * @param gauges simple name/value pairs emitted as gauge samples. + */ + QString toPrometheusText(const std::function &nameForType, + const QHash &gauges) const; + +private: + static constexpr int BucketCount = static_cast(BucketBounds.size()) + 1; ///< bounds + the +Inf bucket + + struct TypeStats + { + std::atomic count{0}; + std::atomic totalMs{0}; + std::array, BucketCount> buckets{}; + }; + + TypeStats &slotFor(int typeId); + const TypeStats &slotFor(int typeId) const; + + /// Index of the histogram bucket the sample falls into. The last index is +Inf. + static int bucketIndexFor(qint64 elapsedMs); + + /// Appends one series of cumulative +Inf-terminated buckets to @p out. + static void appendCumulativeBuckets(QString &out, + const QString &bucketLine, + const std::array, BucketCount> &buckets); + + std::array typeSlots{}; + TypeStats gameStartStats{}; + std::atomic totalCommandsCounter{0}; + std::atomic totalTimeCounter{0}; +}; + +#endif diff --git a/servatrice/src/servatrice.cpp b/servatrice/src/servatrice.cpp index db8751658..0929af97d 100644 --- a/servatrice/src/servatrice.cpp +++ b/servatrice/src/servatrice.cpp @@ -20,6 +20,7 @@ #include "servatrice.h" #include "email_parser.h" +#include "event_loop_watchdog.h" #include "isl_interface.h" #include "main.h" #include "servatrice_connection_pool.h" @@ -38,6 +39,7 @@ #include #include #include +#include #include #include #include @@ -63,6 +65,7 @@ Servatrice_GameServer::Servatrice_GameServer(Servatrice *_server, server->addDatabaseInterface(newThread, newDatabaseInterface); newThread->start(); + server->watchWorkerThread(newThread); QMetaObject::invokeMethod(newDatabaseInterface, "initDatabase", Qt::BlockingQueuedConnection, Q_ARG(QSqlDatabase, _sqlDatabase)); @@ -131,6 +134,7 @@ Servatrice_WebsocketGameServer::Servatrice_WebsocketGameServer(Servatrice *_serv server->addDatabaseInterface(newThread, newDatabaseInterface); newThread->start(); + server->watchWorkerThread(newThread); QMetaObject::invokeMethod(newDatabaseInterface, "initDatabase", Qt::BlockingQueuedConnection, Q_ARG(QSqlDatabase, _sqlDatabase)); @@ -470,9 +474,60 @@ bool Servatrice::initServer() } setRequiredFeatures(getRequiredFeatures()); + + // METRICS (always active. Slow-command logging and stall watchdogs are + // controlled by their respective thresholds below) + metricsSlowCommandMs = settingsCache->value("metrics/slow_command_ms", 500).toInt(); + metricsStallWarnMs = qMax(0, settingsCache->value("metrics/stall_warn_ms", 2000).toInt()); + return true; } +void Servatrice::observeGameStartDurationMs(qint64 elapsedMs) +{ + metricsRegistry.observeGameStartDurationMs(elapsedMs); +} + +void Servatrice::observeEventLoopStall(const QString &threadName, qint64 overshootMs) +{ + eventLoopStallsTotal.fetch_add(1, std::memory_order_relaxed); + eventLoopLastStallMs.store(overshootMs, std::memory_order_relaxed); + qint64 prevMax = eventLoopMaxStallMs.load(std::memory_order_relaxed); + while (overshootMs > prevMax && + !eventLoopMaxStallMs.compare_exchange_weak(prevMax, overshootMs, std::memory_order_relaxed)) { + // retry until the max is at least as high as the new sample + } + + qWarning() << "Event loop stall in" << threadName << "- heartbeat overshot by" << overshootMs << "ms"; +} + +void Servatrice::watchWorkerThread(QThread *thread) +{ + if (metricsStallWarnMs <= 0) { + return; // watchdogs disabled via metrics/stall_warn_ms = 0 + } + + auto *watchdog = new EventLoopWatchdog(this, thread->objectName()); + connect(thread, &QThread::finished, watchdog, &QObject::deleteLater); + watchdog->moveToThread(thread); + QMetaObject::invokeMethod(watchdog, &EventLoopWatchdog::start, Qt::QueuedConnection); +} + +qint64 Servatrice::getCardsInGamesTotal() const +{ + qint64 total = 0; + QReadLocker roomsLocker(&roomsLock); // locking order: roomsLock before gamesLock/gameMutex + QMapIterator roomIterator(rooms); + while (roomIterator.hasNext()) { + Server_Room *room = roomIterator.next().value(); + QReadLocker gamesLocker(&room->gamesLock); + for (auto *game : room->getGames()) { + total += game->getCardsInGame(); + } + } + return total; +} + void Servatrice::addDatabaseInterface(QThread *thread, Servatrice_DatabaseInterface *databaseInterface) { databaseInterfaces.insert(thread, databaseInterface); @@ -728,6 +783,7 @@ void Servatrice::incTxBytes(quint64 num) txBytesMutex.lock(); txBytes += num; txBytesMutex.unlock(); + txBytesTotal.fetch_add(num, std::memory_order_relaxed); } void Servatrice::incRxBytes(quint64 num) @@ -735,6 +791,7 @@ void Servatrice::incRxBytes(quint64 num) rxBytesMutex.lock(); rxBytes += num; rxBytesMutex.unlock(); + rxBytesTotal.fetch_add(num, std::memory_order_relaxed); } void Servatrice::shutdownTimeout() diff --git a/servatrice/src/servatrice.h b/servatrice/src/servatrice.h index 8b0f5ad60..cfb5ff943 100644 --- a/servatrice/src/servatrice.h +++ b/servatrice/src/servatrice.h @@ -20,6 +20,8 @@ #ifndef SERVATRICE_H #define SERVATRICE_H +#include "metrics_registry.h" + #include #include #include @@ -30,6 +32,7 @@ #include #include #include +#include #include #include #include @@ -170,6 +173,14 @@ private: int uptime; QMutex txBytesMutex, rxBytesMutex; quint64 txBytes, rxBytes; + std::atomic txBytesTotal{0}; ///< cumulative bytes sent since process start + std::atomic rxBytesTotal{0}; ///< cumulative bytes received since process start + MetricsRegistry metricsRegistry; + int metricsSlowCommandMs = 500; + int metricsStallWarnMs = 2000; + std::atomic eventLoopStallsTotal{0}; ///< heartbeat overshoots past the warn threshold + std::atomic eventLoopLastStallMs{0}; ///< overshoot of the most recent stall + std::atomic eventLoopMaxStallMs{0}; ///< worst overshoot seen since process start QString shutdownReason; int shutdownMinutes; @@ -286,6 +297,58 @@ public: void incRxBytes(quint64 num); void addDatabaseInterface(QThread *thread, Servatrice_DatabaseInterface *databaseInterface); + // Metrics (see [metrics] section in servatrice.ini.example) + MetricsRegistry &getMetricsRegistry() + { + return metricsRegistry; + } + quint64 getTxBytesTotal() const + { + return txBytesTotal.load(std::memory_order_relaxed); + } + quint64 getRxBytesTotal() const + { + return rxBytesTotal.load(std::memory_order_relaxed); + } + int getUptimeSeconds() const + { + return uptime; + } + /** + * Sums cards across all zones of all running games. Locks rooms and games + * briefly per level, so scrape-time cost grows with live game count only. + */ + qint64 getCardsInGamesTotal() const; + int getMetricsSlowCommandMs() const + { + return metricsSlowCommandMs; + } + /// Heartbeat overshoot that counts as a stall. A value of 0 disables the watchdogs. + int getMetricsStallWarnMs() const + { + return metricsStallWarnMs; + } + void observeGameStartDurationMs(qint64 elapsedMs) override; + qint64 getEventLoopStallsTotal() const + { + return eventLoopStallsTotal.load(std::memory_order_relaxed); + } + qint64 getEventLoopLastStallMs() const + { + return eventLoopLastStallMs.load(std::memory_order_relaxed); + } + qint64 getEventLoopMaxStallMs() const + { + return eventLoopMaxStallMs.load(std::memory_order_relaxed); + } + /// Records one heartbeat overshoot and logs a single warning for it. + void observeEventLoopStall(const QString &threadName, qint64 overshootMs); + /** + * Installs an EventLoopWatchdog in @p thread. Called once per socket pool + * thread right after it starts. + */ + void watchWorkerThread(QThread *thread); + bool islConnectionExists(int _serverId) const; void addIslInterface(int _serverId, IslInterface *interface); void removeIslInterface(int _serverId); diff --git a/servatrice/src/serversocketinterface.cpp b/servatrice/src/serversocketinterface.cpp index 4b1502a15..55c9716c7 100644 --- a/servatrice/src/serversocketinterface.cpp +++ b/servatrice/src/serversocketinterface.cpp @@ -30,6 +30,7 @@ #include #include +#include #include #include #include @@ -39,8 +40,10 @@ #include #include #include +#include #include #include +#include #include #include #include @@ -192,6 +195,41 @@ void AbstractServerSocketInterface::logDebugMessage(const QString &message) logger->logMessage(message, this); } +void AbstractServerSocketInterface::processCommandContainer(const CommandContainer &cont) +{ + QElapsedTimer timer; + timer.start(); + Server_ProtocolHandler::processCommandContainer(cont); + const qint64 elapsedMs = timer.nsecsElapsed() / 1000000; + + // A container usually holds a single command. When several are batched, + // each is attributed the container's total processing time. + for (const auto &cmd : cont.session_command()) { + servatrice->getMetricsRegistry().observeCommand(MetricsRegistry::typeIdFor(0, getPbExtension(cmd)), elapsedMs); + } + for (const auto &cmd : cont.room_command()) { + servatrice->getMetricsRegistry().observeCommand(MetricsRegistry::typeIdFor(1, getPbExtension(cmd)), elapsedMs); + } + for (const auto &cmd : cont.game_command()) { + servatrice->getMetricsRegistry().observeCommand(MetricsRegistry::typeIdFor(2, getPbExtension(cmd)), elapsedMs); + } + for (const auto &cmd : cont.moderator_command()) { + servatrice->getMetricsRegistry().observeCommand(MetricsRegistry::typeIdFor(3, getPbExtension(cmd)), elapsedMs); + } + for (const auto &cmd : cont.admin_command()) { + servatrice->getMetricsRegistry().observeCommand(MetricsRegistry::typeIdFor(4, getPbExtension(cmd)), elapsedMs); + } + + const int slowCommandMs = servatrice->getMetricsSlowCommandMs(); + if (slowCommandMs > 0 && elapsedMs >= slowCommandMs) { + const ServerInfo_User *info = getUserInfo(); + const QString user = authState == PasswordRight && info ? QString::fromStdString(info->name()) + : QStringLiteral("unauthenticated"); + qCWarning(AbstractServerSocketInterfaceLog) << "slow command container from" << user << "processed in" + << elapsedMs << "ms (" << cont.ByteSizeLong() << "bytes)"; + } +} + Response::ResponseCode AbstractServerSocketInterface::processExtendedSessionCommand(int cmdType, const SessionCommand &cmd, ResponseContainer &rc) @@ -1704,6 +1742,51 @@ Response::ResponseCode AbstractServerSocketInterface::cmdGetServerStats(const Co re->set_uptime_secs(snapshot.uptimeSecs); re->set_timest(snapshot.timest); + // Live metrics from the in-process MetricsRegistry (resets on server restart) + re->set_cards_in_games(static_cast(servatrice->getCardsInGamesTotal())); + re->set_eventloop_stalls_total(static_cast(servatrice->getEventLoopStallsTotal())); + re->set_eventloop_last_stall_ms(static_cast(servatrice->getEventLoopLastStallMs())); + re->set_eventloop_max_stall_ms(static_cast(servatrice->getEventLoopMaxStallMs())); + re->set_total_commands(static_cast(servatrice->getMetricsRegistry().totalCommands())); + re->set_total_command_time_ms( + static_cast(servatrice->getMetricsRegistry().totalTimeMs())); + re->set_active_command_types(servatrice->getMetricsRegistry().activeTypeCount()); + + const auto gameStart = servatrice->getMetricsRegistry().getGameStartSnapshot(); + re->set_game_start_count(static_cast(gameStart.count)); + re->set_game_start_total_ms(static_cast(gameStart.totalMs)); + + // Per-command breakdown: resolve protobuf extension names via the descriptor pool + static const char *messageNames[] = {"SessionCommand", "RoomCommand", "GameCommand", "ModeratorCommand", + "AdminCommand"}; + const auto activeStats = servatrice->getMetricsRegistry().collectActiveStats(); + for (const auto &stat : activeStats) { + const int kind = stat.typeId / MetricsRegistry::KindStride; + const int number = stat.typeId % MetricsRegistry::KindStride; + + QString label; + if (kind >= 0 && kind < MetricsRegistry::NumKinds) { + const google::protobuf::DescriptorPool *pool = google::protobuf::DescriptorPool::generated_pool(); + const google::protobuf::Descriptor *message = pool->FindMessageTypeByName(messageNames[kind]); + const google::protobuf::FieldDescriptor *extension = + message ? pool->FindExtensionByNumber(message, number) : nullptr; + if (extension) { + label = QString::fromLatin1(MetricsRegistry::KindNames[kind]) + QStringLiteral("/") + + QString::fromStdString(std::string(extension->name())); + } + } + if (label.isEmpty()) { + label = QString::number(stat.typeId); + } + + CommandStats *cs = re->add_command_stats(); + cs->set_kind_index(static_cast(kind)); + cs->set_extension_number(static_cast(number)); + cs->set_command_name(label.toStdString()); + cs->set_count(static_cast(stat.count)); + cs->set_total_ms(static_cast(stat.totalMs)); + } + rc.setResponseExtension(re); return Response::RespOk; } diff --git a/servatrice/src/serversocketinterface.h b/servatrice/src/serversocketinterface.h index c7516b405..b464e6a9b 100644 --- a/servatrice/src/serversocketinterface.h +++ b/servatrice/src/serversocketinterface.h @@ -81,6 +81,7 @@ signals: protected: void logDebugMessage(const QString &message) override; bool tooManyRegistrationAttempts(const QString &ipAddress); + void processCommandContainer(const CommandContainer &cont) override; virtual void writeToSocket(QByteArray &data) = 0; virtual void flushSocket() = 0; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index b7e712e3a..4f5dc88eb 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -15,6 +15,7 @@ add_test(NAME server_developer_role_test COMMAND server_developer_role_test) add_test(NAME warning_categories_test COMMAND warning_categories_test) add_test(NAME lag_monitor_test COMMAND lag_monitor_test) add_test(NAME latency_tracker_test COMMAND latency_tracker_test) +add_test(NAME metrics_registry_test COMMAND metrics_registry_test) add_test(NAME deck_hash_performance_test COMMAND deck_hash_performance_test) set_tests_properties(deck_hash_performance_test PROPERTIES TIMEOUT 15) @@ -36,6 +37,7 @@ add_executable(warning_categories_test warning_categories_test.cpp) add_executable(lag_monitor_test ${CMAKE_SOURCE_DIR}/cockatrice/src/client/lag_monitor.cpp lag_monitor_test.cpp) target_include_directories(lag_monitor_test PRIVATE ${CMAKE_SOURCE_DIR}/cockatrice/src) add_executable(latency_tracker_test latency_tracker_test.cpp) +add_executable(metrics_registry_test ../servatrice/src/metrics_registry.cpp metrics_registry_test.cpp) find_package(GTest) @@ -76,6 +78,7 @@ if(NOT GTEST_FOUND) add_dependencies(warning_categories_test gtest) add_dependencies(lag_monitor_test gtest) add_dependencies(latency_tracker_test gtest) + add_dependencies(metrics_registry_test gtest) endif() include_directories(${GTEST_INCLUDE_DIRS}) @@ -118,6 +121,8 @@ target_link_libraries(lag_monitor_test Threads::Threads ${GTEST_BOTH_LIBRARIES} target_link_libraries( latency_tracker_test libcockatrice_network Threads::Threads ${GTEST_BOTH_LIBRARIES} ${TEST_QT_MODULES} ) +target_include_directories(metrics_registry_test PRIVATE ${CMAKE_SOURCE_DIR}/servatrice/src) +target_link_libraries(metrics_registry_test ${TEST_QT_MODULES} Threads::Threads ${GTEST_BOTH_LIBRARIES}) add_subdirectory(card_zone_algorithms) add_subdirectory(carddatabase) diff --git a/tests/metrics_registry_test.cpp b/tests/metrics_registry_test.cpp new file mode 100644 index 000000000..40eea4250 --- /dev/null +++ b/tests/metrics_registry_test.cpp @@ -0,0 +1,131 @@ +#include +#include +#include + +TEST(MetricsRegistryTest, EmptyRegistryProducesNoHistogramLines) +{ + MetricsRegistry registry; + + EXPECT_EQ(0, registry.totalCommands()); + EXPECT_EQ(0, registry.totalTimeMs()); + EXPECT_EQ(0, registry.activeTypeCount()); + + const QString text = registry.toPrometheusText([](int) { return QString("x"); }, {}); + // The family TYPE declaration may stand alone. What must not exist is a + // histogram sample without data behind it. + EXPECT_FALSE(text.contains(QRegularExpression("servatrice_commands_duration_ms_(bucket|sum|count)"))); +} + +TEST(MetricsRegistryTest, SingleSampleIsRecordedInTotalsAndBuckets) +{ + MetricsRegistry registry; + registry.observeCommand(MetricsRegistry::typeIdFor(0, 1000), 7); + + EXPECT_EQ(1, registry.totalCommands()); + EXPECT_EQ(7, registry.totalTimeMs()); + EXPECT_EQ(1, registry.activeTypeCount()); + + const QString text = registry.toPrometheusText( + [](int typeId) { + return QString("%1/%2").arg(typeId / MetricsRegistry::KindStride).arg(typeId % MetricsRegistry::KindStride); + }, + {}); + // 7ms falls into the le="10" bucket. Smaller buckets stay empty + EXPECT_TRUE(text.contains("# TYPE servatrice_commands_duration_ms histogram\n")); + EXPECT_TRUE(text.contains(",le=\"10\"} 1")); + EXPECT_TRUE(text.contains(",le=\"5\"} 0")); + EXPECT_TRUE(text.contains("_sum{command=\"0/1000\"} 7")); + EXPECT_TRUE(text.contains("_count{command=\"0/1000\"} 1")); +} + +TEST(MetricsRegistryTest, BucketsAreCumulative) +{ + MetricsRegistry registry; + registry.observeCommand(MetricsRegistry::typeIdFor(0, 1000), 2); + registry.observeCommand(MetricsRegistry::typeIdFor(0, 1000), 30); + + const QString text = registry.toPrometheusText([](int) { return QString("cmd"); }, {}); + + // cumulative counts: <=25 -> 1 sample, <=50 -> 2 samples + EXPECT_TRUE(text.contains(",le=\"25\"} 1\n")); + EXPECT_TRUE(text.contains(",le=\"50\"} 2\n")); + EXPECT_TRUE(text.contains(",le=\"+Inf\"} 2\n")); +} + +TEST(MetricsRegistryTest, KindEncodingSeparatesSameExtensionNumber) +{ + MetricsRegistry registry; + const int sessionPing = MetricsRegistry::typeIdFor(0, 1000); + const int roomLeaveRoom = MetricsRegistry::typeIdFor(1, 1000); + ASSERT_NE(sessionPing, roomLeaveRoom); + + registry.observeCommand(sessionPing, 1); + registry.observeCommand(roomLeaveRoom, 5000); + + EXPECT_EQ(2, registry.activeTypeCount()); +} + +TEST(MetricsRegistryTest, OutOfRangeIdsLandInOverflowSlot) +{ + MetricsRegistry registry; + registry.observeCommand(-1, 4); + registry.observeCommand(MetricsRegistry::MaxTypes + 12345, 4); + + EXPECT_EQ(2, registry.totalCommands()); + EXPECT_EQ(1, registry.activeTypeCount()); // both collapsed into one slot + EXPECT_EQ(8, registry.totalTimeMs()); +} + +TEST(MetricsRegistryTest, NegativeDurationsAreClamped) +{ + MetricsRegistry registry; + registry.observeCommand(MetricsRegistry::typeIdFor(0, 1000), -50); + + EXPECT_EQ(0, registry.totalTimeMs()); +} + +TEST(MetricsRegistryTest, GaugesAndLabelEscapingAreRendered) +{ + MetricsRegistry registry; + + QHash gauges; + gauges.insert("servatrice_users_current", 42); + + const QString text = registry.toPrometheusText(nullptr, gauges); + EXPECT_TRUE(text.contains("# TYPE servatrice_users_current gauge\n")); + EXPECT_TRUE(text.contains("servatrice_users_current 42\n")); +} + +TEST(MetricsRegistryTest, UnnamedTypesFallBackToNumericLabel) +{ + MetricsRegistry registry; + registry.observeCommand(MetricsRegistry::typeIdFor(2, 1042), 9); + + const QString text = registry.toPrometheusText(nullptr, {}); + EXPECT_TRUE(text.contains("{command=\"" + QString::number(MetricsRegistry::typeIdFor(2, 1042)) + "\"}")); +} + +TEST(MetricsRegistryTest, GameStartHistogramOnlyAppearsAfterSamples) +{ + MetricsRegistry registry; + EXPECT_FALSE(registry.toPrometheusText(nullptr, {}).contains("servatrice_game_start_duration_ms")); + + registry.observeGameStartDurationMs(120); + const QString text = registry.toPrometheusText(nullptr, {}); + // 120ms falls into the le="250" bucket + EXPECT_TRUE(text.contains("# TYPE servatrice_game_start_duration_ms histogram\n")); + EXPECT_TRUE(text.contains(",le=\"100\"} 0\n")); + EXPECT_TRUE(text.contains(",le=\"250\"} 1\n")); + EXPECT_TRUE(text.contains("servatrice_game_start_duration_ms_sum 120\n")); + EXPECT_TRUE(text.contains("servatrice_game_start_duration_ms_count 1\n")); +} + +TEST(MetricsRegistryTest, GameStartHistogramIsSeparateFromCommandTotals) +{ + MetricsRegistry registry; + registry.observeGameStartDurationMs(10); + + EXPECT_EQ(0, registry.totalCommands()); + EXPECT_EQ(0, registry.totalTimeMs()); + EXPECT_EQ(0, registry.activeTypeCount()); +}