mirror of
https://github.com/DarkflameUniverse/DarkflameServer.git
synced 2026-10-22 20:52:49 +00:00
feat(dashboard): Diagnostics page with live traffic of every server
Keeps every server's traffic reports (last hour at one second, per-minute rows to server_traffic once a minute, pruned after traffic_days) and shows packets, bytes, HTTP requests and latency per second, per server, with the busiest message types and HTTP routes. Live over the traffic WebSocket topic; 1 hour, 24 hours and 7 days ranges. The same counters are in /metrics. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Aaron Kimbrell
@@ -27,6 +27,8 @@ set(DASHBOARDROUTES_SOURCES
|
||||
"MissionTools.cpp"
|
||||
"CharacterProgress.cpp"
|
||||
"ServerRoutes.cpp"
|
||||
"Traffic.cpp"
|
||||
"TrafficHistory.cpp"
|
||||
"LeaderboardRoutes.cpp"
|
||||
"VanityRoutes.cpp"
|
||||
"ChatRoutes.cpp"
|
||||
|
||||
@@ -465,6 +465,7 @@ void RegisterDashboardRoutes() {
|
||||
SimplePage("/vanity", Perm("vanity_manage"), "vanity.jinja2", "vanity", "Vanity: the vanity files and their NPCs, plaque texts, vanity events and a preview of what the worlds load");
|
||||
SimplePage("/leaderboards", Perm("leaderboards_view"), "leaderboards.jinja2", "leaderboards", "Leaderboards for every activity");
|
||||
SimplePage("/health", Perm("health_view"), "health.jinja2", "health", "Player counts, worlds, uptime and memory over time; crash dumps");
|
||||
SimplePage("/diagnostics", Perm("health_view"), "diagnostics.jinja2", "diagnostics", "Packets, bytes and HTTP requests per second of every server");
|
||||
SimplePage("/players", Perm("players_view"), "players.jinja2", "players", "Who is online, with kick, rescue and teleport");
|
||||
SimplePage("/backups", Perm("backups"), "backups.jinja2", "backups", "Database backups");
|
||||
SimplePage("/permissions", Perm("permissions_manage"), "permissions.jinja2", "permissions", "What each GM level may do");
|
||||
|
||||
@@ -215,6 +215,7 @@ namespace {
|
||||
std::tuple{ std::optional(eLog::LOGIN_ADDRESS), "log_login_address_days", 90 },
|
||||
std::tuple{ std::optional<eLog>(), "log_task_days", 90 },
|
||||
std::tuple{ std::optional<eLog>(), "health_days", 30 },
|
||||
std::tuple{ std::optional<eLog>(), "traffic_days", 30 },
|
||||
std::tuple{ std::optional<eLog>(), "position_history_days", 3 } }) {
|
||||
const auto days = Setting(key, fallback);
|
||||
run->Log(std::string(key) + " = " + (days > 0 ? std::to_string(days) : "0 (keep everything)"));
|
||||
@@ -227,6 +228,7 @@ namespace {
|
||||
if (days <= 0) continue;
|
||||
const auto before = now - days * DAY_SECONDS;
|
||||
pruned[key] = log ? db.PruneLog(*log, before) : key == std::string("health_days") ? db.PruneHealthSamples(before)
|
||||
: key == std::string("traffic_days") ? db.PruneTrafficMinutes(before)
|
||||
: key == std::string("position_history_days") ? db.PrunePositionSamples(before) : db.PruneTaskRuns(before);
|
||||
}
|
||||
return pruned;
|
||||
|
||||
@@ -73,13 +73,21 @@ namespace MetricsFormat {
|
||||
public:
|
||||
// type: counter or gauge. A counter's name should end in _total.
|
||||
void Add(const std::string& name, const std::string& help, const std::string& type, const Labels& labels, double value) {
|
||||
const auto metric = Name(name);
|
||||
AddSample(name, name, help, type, labels, value);
|
||||
}
|
||||
|
||||
/**
|
||||
* A sample named differently from its family, for histograms: the family "x_seconds" (type histogram) holds
|
||||
* x_seconds_bucket{le=...}, x_seconds_sum and x_seconds_count.
|
||||
*/
|
||||
void AddSample(const std::string& family, const std::string& sample, const std::string& help, const std::string& type, const Labels& labels, double value) {
|
||||
const auto metric = Name(family);
|
||||
auto it = m_Index.find(metric);
|
||||
if (it == m_Index.end()) {
|
||||
it = m_Index.emplace(metric, m_Families.size()).first;
|
||||
m_Families.push_back({ metric, help, type, {} });
|
||||
}
|
||||
std::string line = metric;
|
||||
std::string line = Name(sample);
|
||||
if (!labels.empty()) {
|
||||
line += '{';
|
||||
for (size_t i = 0; i < labels.size(); i++) {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
#include "PrometheusMetrics.h"
|
||||
#include "MetricsFormat.h"
|
||||
#include "Traffic.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <chrono>
|
||||
@@ -266,6 +267,7 @@ namespace {
|
||||
if (!g_Text.empty() && now - g_BuiltAt < std::chrono::seconds(cacheSeconds)) return g_Text;
|
||||
MetricsFormat::Writer w;
|
||||
AddLive(w);
|
||||
Traffic::AddMetrics(w);
|
||||
AddDatabase(w);
|
||||
g_Text = w.Text();
|
||||
g_BuiltAt = now;
|
||||
|
||||
@@ -386,6 +386,7 @@ namespace {
|
||||
c.Add(Days("log_chat_days", "Chat log", "", "90"));
|
||||
c.Add(Days("log_login_address_days", "Login addresses", "Addresses an account hasn't logged in from for this long are forgotten.", "90"));
|
||||
c.Add(Days("health_days", "Server health history", "Minute-by-minute player counts and uptime.", "30"));
|
||||
c.Add(Days("traffic_days", "Server traffic history", "Minute-by-minute packets, bytes and HTTP requests of every server (Diagnostics). The last hour at one second is only kept in memory.", "30"));
|
||||
|
||||
c.AddSection("Player movement", "Where players went, for replays on the 3D world view (staff with players_history only).");
|
||||
c.Add(Bool(DASHBOARD, "position_history", "Record player movement", "Keeps a player's position every few seconds while they move, and every 30 seconds while they stand still.", true));
|
||||
|
||||
408
dDashboardServer/routes/Traffic.cpp
Normal file
408
dDashboardServer/routes/Traffic.cpp
Normal file
@@ -0,0 +1,408 @@
|
||||
#include "Traffic.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <chrono>
|
||||
#include <cmath>
|
||||
#include <map>
|
||||
#include <mutex>
|
||||
|
||||
#include "Background.h"
|
||||
#include "Database.h"
|
||||
#include "Game.h"
|
||||
#include "Logger.h"
|
||||
#include "MessageIdentifiers.h"
|
||||
#include "MetricsFormat.h"
|
||||
#include "Permissions.h"
|
||||
#include "RouteUtils.h"
|
||||
#include "ServerState.h"
|
||||
#include "ServiceType.h"
|
||||
#include "TrafficHistory.h"
|
||||
#include "TrafficStats.h"
|
||||
#include "Web.h"
|
||||
#include "Workers.h"
|
||||
#include "dServer.h"
|
||||
#include "eHTTPMethod.h"
|
||||
#include "magic_enum.hpp"
|
||||
#include "master/ServerTraffic.h"
|
||||
#include "MessageType/Auth.h"
|
||||
#include "MessageType/Chat.h"
|
||||
#include "MessageType/Client.h"
|
||||
#include "MessageType/Game.h"
|
||||
#include "MessageType/Master.h"
|
||||
#include "MessageType/Server.h"
|
||||
#include "MessageType/World.h"
|
||||
|
||||
using namespace RouteUtils;
|
||||
|
||||
namespace {
|
||||
constexpr const char* PERMISSION = "health_view";
|
||||
constexpr const char* TOPIC = "traffic";
|
||||
constexpr auto PUSH_INTERVAL = std::chrono::seconds(2);
|
||||
constexpr auto WRITE_INTERVAL = std::chrono::seconds(15);
|
||||
constexpr int64_t ONLINE_SECONDS = 20; // a server that reported this recently is shown as up
|
||||
|
||||
TrafficHistory g_History;
|
||||
bool g_Changed = false;
|
||||
std::chrono::steady_clock::time_point g_NextPush{};
|
||||
std::chrono::steady_clock::time_point g_NextWrite{};
|
||||
// Minutes the background writer couldn't take yet (it was still busy)
|
||||
std::vector<IServerTraffic::TrafficMinute> g_Unwritten;
|
||||
|
||||
template<typename Enum>
|
||||
std::string EnumName(uint32_t value) {
|
||||
const auto name = magic_enum::enum_name(static_cast<Enum>(value));
|
||||
return name.empty() ? "#" + std::to_string(value) : std::string(name);
|
||||
}
|
||||
|
||||
std::string RakNetName(uint32_t id) {
|
||||
switch (id) {
|
||||
case ID_REPLICA_MANAGER_CONSTRUCTION: return "REPLICA_CONSTRUCTION";
|
||||
case ID_REPLICA_MANAGER_SERIALIZE: return "REPLICA_SERIALIZE";
|
||||
case ID_REPLICA_MANAGER_DESTRUCTION: return "REPLICA_DESTRUCTION";
|
||||
case ID_NEW_INCOMING_CONNECTION: return "NEW_INCOMING_CONNECTION";
|
||||
case ID_CONNECTION_REQUEST_ACCEPTED: return "CONNECTION_REQUEST_ACCEPTED";
|
||||
case ID_DISCONNECTION_NOTIFICATION: return "DISCONNECTION_NOTIFICATION";
|
||||
case ID_CONNECTION_LOST: return "CONNECTION_LOST";
|
||||
default: return "RAKNET_" + std::to_string(id);
|
||||
}
|
||||
}
|
||||
|
||||
// {service, packet, game message} names of a message key
|
||||
struct MessageNames {
|
||||
std::string service;
|
||||
std::string packet;
|
||||
std::string gameMessage; // empty unless a game message
|
||||
};
|
||||
|
||||
MessageNames NamesOf(const TrafficStats::MessageKey& key) {
|
||||
if (key.service == TrafficStats::MessageKey::RAKNET) return { "RAKNET", RakNetName(key.packet), "" };
|
||||
MessageNames names;
|
||||
const auto service = static_cast<ServiceType>(key.service);
|
||||
const auto serviceName = magic_enum::enum_name(service);
|
||||
names.service = serviceName.empty() ? "SERVICE_" + std::to_string(key.service) : std::string(serviceName);
|
||||
switch (service) {
|
||||
case ServiceType::COMMON: names.packet = EnumName<MessageType::Server>(key.packet); break;
|
||||
case ServiceType::AUTH: names.packet = EnumName<MessageType::Auth>(key.packet); break;
|
||||
case ServiceType::CHAT: names.packet = EnumName<MessageType::Chat>(key.packet); break;
|
||||
case ServiceType::WORLD: names.packet = EnumName<MessageType::World>(key.packet); break;
|
||||
case ServiceType::CLIENT: names.packet = EnumName<MessageType::Client>(key.packet); break;
|
||||
case ServiceType::MASTER: names.packet = EnumName<MessageType::Master>(key.packet); break;
|
||||
default: names.packet = "#" + std::to_string(key.packet); break;
|
||||
}
|
||||
const bool gameMessage = (service == ServiceType::WORLD && key.packet == static_cast<uint32_t>(MessageType::World::GAME_MSG)) ||
|
||||
(service == ServiceType::CLIENT && key.packet == static_cast<uint32_t>(MessageType::Client::GAME_MSG));
|
||||
if (gameMessage) names.gameMessage = EnumName<MessageType::Game>(key.gameMessage);
|
||||
return names;
|
||||
}
|
||||
|
||||
// "World 1100 Avant Gardens #3" for world:1100:3, else the key capitalised
|
||||
std::string LabelOf(const TrafficHistory::Server& server) {
|
||||
switch (static_cast<ServiceType>(server.type)) {
|
||||
case ServiceType::MASTER: return "Master";
|
||||
case ServiceType::AUTH: return "Auth";
|
||||
case ServiceType::CHAT: return "Chat";
|
||||
case ServiceType::DASHBOARD: return "Dashboard";
|
||||
case ServiceType::UGC: return "UGC";
|
||||
case ServiceType::WORLD: {
|
||||
std::string zoneName;
|
||||
{
|
||||
std::lock_guard lock(ServerState::g_StatusMutex);
|
||||
for (const auto& world : ServerState::g_WorldInstances) {
|
||||
if (world.mapID == server.zoneId && world.instanceID == server.instanceId) zoneName = world.zoneName;
|
||||
}
|
||||
}
|
||||
return "World " + std::to_string(server.zoneId) + (zoneName.empty() ? "" : " " + zoneName) + " #" + std::to_string(server.instanceId);
|
||||
}
|
||||
default: return server.key;
|
||||
}
|
||||
}
|
||||
|
||||
double Rate(uint64_t value, int64_t seconds) {
|
||||
return std::round(static_cast<double>(value) / static_cast<double>(std::max<int64_t>(seconds, 1)) * 100.0) / 100.0;
|
||||
}
|
||||
|
||||
double Millis(uint64_t microseconds) {
|
||||
return std::round(static_cast<double>(microseconds) / 10.0) / 100.0;
|
||||
}
|
||||
|
||||
// One server's chart lines: per-second rates of each bucket, latency in ms
|
||||
struct Lines {
|
||||
nlohmann::json packetsIn = nlohmann::json::array(), packetsOut = nlohmann::json::array(), bytesIn = nlohmann::json::array(), bytesOut = nlohmann::json::array();
|
||||
nlohmann::json http = nlohmann::json::array(), httpErrors = nlohmann::json::array(), p50 = nlohmann::json::array(), p95 = nlohmann::json::array(), p99 = nlohmann::json::array();
|
||||
bool anyHttp = false;
|
||||
|
||||
void Add(uint64_t pin, uint64_t pout, uint64_t bin, uint64_t bout, uint64_t requests, uint64_t errors, double lat50, double lat95, double lat99, int64_t step) {
|
||||
packetsIn.push_back(Rate(pin, step));
|
||||
packetsOut.push_back(Rate(pout, step));
|
||||
bytesIn.push_back(Rate(bin, step));
|
||||
bytesOut.push_back(Rate(bout, step));
|
||||
http.push_back(Rate(requests, step));
|
||||
httpErrors.push_back(Rate(errors, step));
|
||||
p50.push_back(requests ? nlohmann::json(lat50) : nlohmann::json(nullptr));
|
||||
p95.push_back(requests ? nlohmann::json(lat95) : nlohmann::json(nullptr));
|
||||
p99.push_back(requests ? nlohmann::json(lat99) : nlohmann::json(nullptr));
|
||||
anyHttp = anyHttp || requests > 0;
|
||||
}
|
||||
|
||||
nlohmann::json Json() const {
|
||||
nlohmann::json out = { {"packets_in", packetsIn}, {"packets_out", packetsOut}, {"bytes_in", bytesIn}, {"bytes_out", bytesOut} };
|
||||
if (anyHttp) {
|
||||
out["http"] = http;
|
||||
out["http_errors"] = httpErrors;
|
||||
out["p50"] = p50;
|
||||
out["p95"] = p95;
|
||||
out["p99"] = p99;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
};
|
||||
|
||||
nlohmann::json ServerInfo(const std::string& key, int64_t now) {
|
||||
const auto& servers = g_History.Servers();
|
||||
const auto it = servers.find(key);
|
||||
nlohmann::json info = { {"key", key}, {"label", key}, {"online", false} };
|
||||
if (it == servers.end()) {
|
||||
// Only in the database (e.g. a world that has since closed)
|
||||
if (key.starts_with("world:")) info["label"] = "World " + key.substr(6);
|
||||
return info;
|
||||
}
|
||||
const auto& server = it->second;
|
||||
info["label"] = LabelOf(server);
|
||||
info["type"] = std::string(magic_enum::enum_name(static_cast<ServiceType>(server.type)));
|
||||
info["online"] = now - server.lastSeen <= ONLINE_SECONDS;
|
||||
info["last_seen"] = server.lastSeen;
|
||||
info["link"] = { {"connections", server.link.connections}, {"ping_ms", server.link.averagePingMs}, {"resend_queue", server.link.resendQueue},
|
||||
{"resends", server.link.resends}, {"datagrams_sent", server.link.datagramsSent}, {"datagrams_received", server.link.datagramsReceived} };
|
||||
nlohmann::json gauges = nlohmann::json::object();
|
||||
for (const auto& [name, value] : server.gauges) gauges[name] = value;
|
||||
info["gauges"] = gauges;
|
||||
return info;
|
||||
}
|
||||
|
||||
// The chart data of a range: live (5m) and 1h from memory, 24h and 7d from the database plus the minutes not written yet
|
||||
nlohmann::json Series(const std::string& range, int64_t now) {
|
||||
int64_t span = 300, step = 1;
|
||||
if (range == "1h") { span = 3600; step = 10; }
|
||||
else if (range == "24h") { span = 86400; step = 300; }
|
||||
else if (range == "7d") { span = 7 * 86400; step = 1800; }
|
||||
const int64_t to = now - (now % step); // the current bucket isn't complete yet
|
||||
const int64_t from = to - span;
|
||||
const auto count = static_cast<size_t>(span / step);
|
||||
|
||||
nlohmann::json times = nlohmann::json::array();
|
||||
for (size_t i = 0; i < count; i++) times.push_back(from + static_cast<int64_t>(i) * step);
|
||||
|
||||
std::map<std::string, Lines> lines;
|
||||
if (step < 60) {
|
||||
for (const auto& [key, points] : g_History.Buckets(from, to, step)) {
|
||||
auto& line = lines[key];
|
||||
for (const auto& p : points) {
|
||||
const auto latency = p.Latency();
|
||||
line.Add(p.packetsIn, p.packetsOut, p.bytesIn, p.bytesOut, p.httpRequests, p.http4xx + p.http5xx,
|
||||
Millis(latency.Percentile(0.5)), Millis(latency.Percentile(0.95)), Millis(latency.Percentile(0.99)), step);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Database rows per bucket and server, then the minutes still in memory
|
||||
std::map<std::string, std::vector<IServerTraffic::TrafficMinute>> buckets;
|
||||
const auto add = [&](const IServerTraffic::TrafficMinute& row) {
|
||||
if (row.time < from || row.time >= to + step) return;
|
||||
auto& list = buckets[row.server];
|
||||
if (list.empty()) {
|
||||
list.resize(count);
|
||||
for (size_t i = 0; i < count; i++) list[i].time = from + static_cast<int64_t>(i) * step;
|
||||
}
|
||||
const auto index = std::min(count - 1, static_cast<size_t>((row.time - from) / step));
|
||||
auto& b = list[index];
|
||||
b.packetsIn += row.packetsIn; b.packetsOut += row.packetsOut; b.bytesIn += row.bytesIn; b.bytesOut += row.bytesOut;
|
||||
b.httpRequests += row.httpRequests; b.http4xx += row.http4xx; b.http5xx += row.http5xx;
|
||||
// Several minutes in one bucket: the worst minute's percentiles
|
||||
b.latencyP50Us = std::max(b.latencyP50Us, row.latencyP50Us);
|
||||
b.latencyP95Us = std::max(b.latencyP95Us, row.latencyP95Us);
|
||||
b.latencyP99Us = std::max(b.latencyP99Us, row.latencyP99Us);
|
||||
};
|
||||
try {
|
||||
for (const auto& row : Database::Get()->GetTrafficMinutes(from, to + step, step)) add(row);
|
||||
} catch (const std::exception& ex) {
|
||||
LOG_DEBUG("Traffic: reading server_traffic failed: %s", ex.what());
|
||||
}
|
||||
for (const auto& row : g_Unwritten) add(row);
|
||||
for (const auto& [key, server] : g_History.Servers()) {
|
||||
for (const auto& minute : server.minutes) {
|
||||
const auto latency = minute.point.Latency();
|
||||
add({ .time = minute.point.time, .server = key, .packetsIn = minute.point.packetsIn, .packetsOut = minute.point.packetsOut,
|
||||
.bytesIn = minute.point.bytesIn, .bytesOut = minute.point.bytesOut, .httpRequests = minute.point.httpRequests,
|
||||
.http4xx = minute.point.http4xx, .http5xx = minute.point.http5xx,
|
||||
.latencyP50Us = static_cast<uint32_t>(latency.Percentile(0.5)), .latencyP95Us = static_cast<uint32_t>(latency.Percentile(0.95)),
|
||||
.latencyP99Us = static_cast<uint32_t>(latency.Percentile(0.99)) });
|
||||
}
|
||||
}
|
||||
for (const auto& [key, list] : buckets) {
|
||||
auto& line = lines[key];
|
||||
for (const auto& b : list) {
|
||||
line.Add(b.packetsIn, b.packetsOut, b.bytesIn, b.bytesOut, b.httpRequests, b.http4xx + b.http5xx,
|
||||
Millis(b.latencyP50Us), Millis(b.latencyP95Us), Millis(b.latencyP99Us), step);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
nlohmann::json servers = nlohmann::json::array();
|
||||
for (const auto& [key, line] : lines) {
|
||||
auto info = ServerInfo(key, now);
|
||||
info["series"] = line.Json();
|
||||
servers.push_back(std::move(info));
|
||||
}
|
||||
// Servers that are up but had nothing in the range yet
|
||||
for (const auto& [key, server] : g_History.Servers()) {
|
||||
if (!lines.contains(key) && now - server.lastSeen <= ONLINE_SECONDS) servers.push_back(ServerInfo(key, now));
|
||||
}
|
||||
|
||||
const int64_t since = step < 60 ? from : std::max<int64_t>(from, now - 86400);
|
||||
nlohmann::json messages = nlohmann::json::array();
|
||||
for (const auto& count : g_History.TopMessages("", since, 15)) {
|
||||
const auto names = NamesOf(count.key);
|
||||
messages.push_back({ {"direction", count.key.outbound ? "out" : "in"}, {"service", names.service}, {"packet", names.packet},
|
||||
{"game_message", names.gameMessage}, {"count", count.count}, {"bytes", count.bytes} });
|
||||
}
|
||||
nlohmann::json routes = nlohmann::json::array();
|
||||
for (const auto& [server, stats] : g_History.Routes(std::max<int64_t>(from, now - 3600))) {
|
||||
if (routes.size() >= 30) break;
|
||||
routes.push_back({ {"server", server}, {"route", stats.route}, {"count", stats.count}, {"status_4xx", stats.status[3]}, {"status_5xx", stats.status[4]},
|
||||
{"bytes", stats.bytesOut}, {"p50", Millis(stats.latency.Percentile(0.5))}, {"p95", Millis(stats.latency.Percentile(0.95))},
|
||||
{"p99", Millis(stats.latency.Percentile(0.99))} });
|
||||
}
|
||||
return { {"range", range}, {"from", from}, {"to", to}, {"step", step}, {"times", times}, {"servers", servers}, {"messages", messages},
|
||||
{"messages_since", since}, {"routes", routes} };
|
||||
}
|
||||
|
||||
// What open pages get every few seconds: each server's rates over its last five reported seconds
|
||||
nlohmann::json Summary(int64_t now) {
|
||||
nlohmann::json servers = nlohmann::json::object();
|
||||
for (const auto& [key, server] : g_History.Servers()) {
|
||||
if (now - server.lastSeen > ONLINE_SECONDS || server.seconds.empty()) continue;
|
||||
TrafficHistory::Point p;
|
||||
const auto first = server.seconds.back().time - 4;
|
||||
for (auto it = server.seconds.rbegin(); it != server.seconds.rend() && it->time >= first; ++it) p.Add(*it);
|
||||
servers[key] = { {"packets_in", Rate(p.packetsIn, 5)}, {"packets_out", Rate(p.packetsOut, 5)}, {"bytes_in", Rate(p.bytesIn, 5)},
|
||||
{"bytes_out", Rate(p.bytesOut, 5)}, {"http", Rate(p.httpRequests, 5)} };
|
||||
}
|
||||
return { {"time", now}, {"servers", servers} };
|
||||
}
|
||||
|
||||
void Write() {
|
||||
auto rows = g_History.TakeFinishedMinutes(TrafficStats::Now());
|
||||
g_Unwritten.insert(g_Unwritten.end(), rows.begin(), rows.end());
|
||||
if (g_Unwritten.empty()) return;
|
||||
// A day of a busy universe at most, should the database be away for long
|
||||
if (g_Unwritten.size() > 100000) g_Unwritten.erase(g_Unwritten.begin(), g_Unwritten.begin() + static_cast<std::ptrdiff_t>(g_Unwritten.size() - 100000));
|
||||
auto batch = std::move(g_Unwritten);
|
||||
g_Unwritten.clear();
|
||||
const bool queued = Background::Run("traffic_minutes", [batch](GameDatabase& db) -> nlohmann::json {
|
||||
db.InsertTrafficMinutes(batch);
|
||||
return nlohmann::json::object();
|
||||
}, [](nlohmann::json, const std::string& error) {
|
||||
if (!error.empty()) LOG("Traffic: writing minutes failed: %s", error.c_str());
|
||||
});
|
||||
if (!queued) g_Unwritten = std::move(batch); // still writing the last batch: next time
|
||||
}
|
||||
}
|
||||
|
||||
namespace Traffic {
|
||||
void Ingest(const ServerTraffic& report) {
|
||||
g_History.Ingest(static_cast<uint16_t>(report.serverType), report.zoneId, report.instanceId, report.report, TrafficStats::Now());
|
||||
g_Changed = true;
|
||||
}
|
||||
|
||||
void Update() {
|
||||
const auto now = std::chrono::steady_clock::now();
|
||||
if (now >= g_NextWrite) {
|
||||
g_NextWrite = now + WRITE_INTERVAL;
|
||||
Write();
|
||||
g_History.Forget(TrafficStats::Now());
|
||||
}
|
||||
if (g_Changed && now >= g_NextPush) {
|
||||
g_NextPush = now + PUSH_INTERVAL;
|
||||
g_Changed = false;
|
||||
auto summary = Summary(TrafficStats::Now());
|
||||
Web::SendWSMessage(TOPIC, summary);
|
||||
}
|
||||
}
|
||||
|
||||
void AddMetrics(MetricsFormat::Writer& w) {
|
||||
using MetricsFormat::Labels;
|
||||
w.Declare("darkflame_net_packets_total", "Packets a server sent and received (LU packets and RakNet messages), since the dashboard started", "counter");
|
||||
w.Declare("darkflame_net_bytes_total", "Bytes of the packets a server sent and received, since the dashboard started", "counter");
|
||||
w.Declare("darkflame_net_messages_total", "Packets by type (the busiest types each server reports), since the dashboard started", "counter");
|
||||
w.Declare("darkflame_net_datagrams_total", "UDP datagrams (acknowledgements and resends included, as RakNet counts them), since the dashboard started", "counter");
|
||||
w.Declare("darkflame_net_resends_total", "Messages RakNet had to send again, since the dashboard started", "counter");
|
||||
w.Declare("darkflame_net_connections", "Open RakNet connections of a server", "gauge");
|
||||
w.Declare("darkflame_net_ping_milliseconds", "Average ping over a server's connections", "gauge");
|
||||
w.Declare("darkflame_http_requests_total", "HTTP requests answered, by route pattern and status class, since the dashboard started", "counter");
|
||||
w.Declare("darkflame_http_response_bytes_total", "HTTP response bodies sent, since the dashboard started", "counter");
|
||||
w.Declare("darkflame_http_request_duration_seconds", "HTTP request latency (deferred requests until their answer went out)", "histogram");
|
||||
const auto now = TrafficStats::Now();
|
||||
for (const auto& [key, server] : g_History.Servers()) {
|
||||
const auto& t = server.totals;
|
||||
const auto labels = [&key](std::initializer_list<std::pair<std::string, std::string>> more) {
|
||||
Labels out{ {"server", key} };
|
||||
out.insert(out.end(), more.begin(), more.end());
|
||||
return out;
|
||||
};
|
||||
w.Add("darkflame_net_packets_total", "", "counter", labels({ {"direction", "in"} }), static_cast<double>(t.packetsIn));
|
||||
w.Add("darkflame_net_packets_total", "", "counter", labels({ {"direction", "out"} }), static_cast<double>(t.packetsOut));
|
||||
w.Add("darkflame_net_bytes_total", "", "counter", labels({ {"direction", "in"} }), static_cast<double>(t.bytesIn));
|
||||
w.Add("darkflame_net_bytes_total", "", "counter", labels({ {"direction", "out"} }), static_cast<double>(t.bytesOut));
|
||||
w.Add("darkflame_net_datagrams_total", "", "counter", labels({ {"direction", "in"} }), static_cast<double>(t.datagramsReceived));
|
||||
w.Add("darkflame_net_datagrams_total", "", "counter", labels({ {"direction", "out"} }), static_cast<double>(t.datagramsSent));
|
||||
w.Add("darkflame_net_resends_total", "", "counter", labels({}), static_cast<double>(t.resends));
|
||||
for (const auto& [_, count] : t.messages) {
|
||||
const auto names = NamesOf(count.key);
|
||||
w.Add("darkflame_net_messages_total", "", "counter", labels({ {"direction", count.key.outbound ? "out" : "in"}, {"service", names.service},
|
||||
{"message", names.gameMessage.empty() ? names.packet : names.gameMessage} }), static_cast<double>(count.count));
|
||||
}
|
||||
if (now - server.lastSeen <= ONLINE_SECONDS) {
|
||||
w.Add("darkflame_net_connections", "", "gauge", labels({}), server.link.connections);
|
||||
w.Add("darkflame_net_ping_milliseconds", "", "gauge", labels({}), server.link.averagePingMs);
|
||||
for (const auto& [name, value] : server.gauges) {
|
||||
w.Add("darkflame_server_" + name, "Reported by a server with its traffic (see docs/Dashboard.md, Traffic diagnostics)", "gauge", labels({}), value);
|
||||
}
|
||||
}
|
||||
static constexpr const char* CLASSES[] = { "1xx", "2xx", "3xx", "4xx", "5xx" };
|
||||
for (const auto& [route, stats] : t.routes) {
|
||||
for (size_t i = 0; i < 5; i++) {
|
||||
if (stats.status[i]) w.Add("darkflame_http_requests_total", "", "counter", labels({ {"route", route}, {"status", CLASSES[i]} }), static_cast<double>(stats.status[i]));
|
||||
}
|
||||
w.Add("darkflame_http_response_bytes_total", "", "counter", labels({ {"route", route} }), static_cast<double>(stats.bytesOut));
|
||||
// Buckets at every doubling from 0.1 ms (every third histogram bucket), cumulative
|
||||
uint64_t cumulative = 0;
|
||||
const std::string family = "darkflame_http_request_duration_seconds";
|
||||
for (size_t i = 0; i + 1 < TrafficStats::Histogram::BUCKETS; i++) {
|
||||
cumulative += stats.latency.At(i);
|
||||
if (i % 3 != 0) continue;
|
||||
w.AddSample(family, family + "_bucket", "", "histogram", labels({ {"route", route}, {"le", MetricsFormat::Value(static_cast<double>(TrafficStats::Histogram::UpperBound(i)) / 1e6)} }), static_cast<double>(cumulative));
|
||||
}
|
||||
w.AddSample(family, family + "_bucket", "", "histogram", labels({ {"route", route}, {"le", "+Inf"} }), static_cast<double>(stats.latency.Count()));
|
||||
w.AddSample(family, family + "_sum", "", "histogram", labels({ {"route", route} }), static_cast<double>(stats.latency.Sum()) / 1e6);
|
||||
w.AddSample(family, family + "_count", "", "histogram", labels({ {"route", route} }), static_cast<double>(stats.latency.Count()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void RegisterRoutes() {
|
||||
Game::web.RegisterWSSubscription(TOPIC, std::function<uint8_t()>([] { return Permissions::Level(PERMISSION); }));
|
||||
|
||||
// The dashboard's own report stays here; the worker pool is what its deferred requests wait for
|
||||
Game::server->SetTrafficSink([](ServerTraffic& report) { Ingest(report); });
|
||||
TrafficStats::Local().SetGauge("workers_busy", [] { return static_cast<double>(Workers::Pool().Active()); });
|
||||
TrafficStats::Local().SetGauge("workers_queued", [] { return static_cast<double>(Workers::Pool().Queued()); });
|
||||
TrafficStats::Local().SetGauge("workers_threads", [] { return static_cast<double>(Workers::Pool().Threads()); });
|
||||
|
||||
Route(eHTTPMethod::GET, "/api/diagnostics/traffic", Perm(PERMISSION),
|
||||
"Packets, bytes and HTTP requests per second of every server. Query: ?range=5m|1h|24h|7d. Series are per-second rates of each step, latency in ms",
|
||||
[](HTTPReply& reply, const HTTPContext& context) {
|
||||
auto range = QueryValue(context.queryString, "range");
|
||||
if (range != "1h" && range != "24h" && range != "7d") range = "5m";
|
||||
JsonSuccess(reply, Series(range, TrafficStats::Now()));
|
||||
});
|
||||
}
|
||||
}
|
||||
21
dDashboardServer/routes/Traffic.h
Normal file
21
dDashboardServer/routes/Traffic.h
Normal file
@@ -0,0 +1,21 @@
|
||||
#pragma once
|
||||
|
||||
struct ServerTraffic;
|
||||
namespace MetricsFormat { class Writer; }
|
||||
|
||||
/**
|
||||
* Traffic diagnostics: every server's packets, bytes and HTTP requests (see ServerTraffic.h and TrafficHistory.h).
|
||||
* The last hour is kept in memory at one second; finished minutes go to the server_traffic table once a minute.
|
||||
*/
|
||||
namespace Traffic {
|
||||
void RegisterRoutes();
|
||||
|
||||
// SERVER_TRAFFIC from a server (via master), or the dashboard's own report
|
||||
void Ingest(const ServerTraffic& report);
|
||||
|
||||
// Main loop: write finished minutes, tell open pages there is news
|
||||
void Update();
|
||||
|
||||
// Counters for /metrics
|
||||
void AddMetrics(MetricsFormat::Writer& w);
|
||||
}
|
||||
260
dDashboardServer/routes/TrafficHistory.cpp
Normal file
260
dDashboardServer/routes/TrafficHistory.cpp
Normal file
@@ -0,0 +1,260 @@
|
||||
#include "TrafficHistory.h"
|
||||
|
||||
#include <algorithm>
|
||||
|
||||
#include "ServiceType.h"
|
||||
|
||||
namespace {
|
||||
using Sparse = std::vector<std::pair<uint8_t, uint32_t>>;
|
||||
|
||||
void MergeSparse(Sparse& into, const Sparse& from) {
|
||||
if (from.empty()) return;
|
||||
if (into.empty()) {
|
||||
into = from;
|
||||
return;
|
||||
}
|
||||
auto histogram = TrafficStats::Histogram::FromSparse(into, 0);
|
||||
histogram.Merge(TrafficStats::Histogram::FromSparse(from, 0));
|
||||
into = histogram.Sparse();
|
||||
}
|
||||
|
||||
int64_t Floor(int64_t time, int64_t step) {
|
||||
return time - ((time % step) + step) % step;
|
||||
}
|
||||
|
||||
template<typename Counts>
|
||||
void AddCounts(TrafficHistory::MessageCounts& into, const Counts& from, size_t cap = SIZE_MAX) {
|
||||
for (const auto& count : from) {
|
||||
const auto packed = count.key.Packed();
|
||||
auto it = into.find(packed);
|
||||
if (it == into.end()) {
|
||||
if (into.size() >= cap) continue;
|
||||
it = into.emplace(packed, TrafficStats::MessageCount{ count.key }).first;
|
||||
}
|
||||
it->second.count += count.count;
|
||||
it->second.bytes += count.bytes;
|
||||
}
|
||||
}
|
||||
|
||||
void AddRoutes(std::map<std::string, TrafficStats::RouteStats>& into, const std::vector<TrafficStats::RouteStats>& from, size_t cap) {
|
||||
for (const auto& route : from) {
|
||||
auto it = into.find(route.route);
|
||||
if (it == into.end()) {
|
||||
if (into.size() >= cap) continue;
|
||||
it = into.emplace(route.route, TrafficStats::RouteStats{ .route = route.route }).first;
|
||||
}
|
||||
it->second.Merge(route);
|
||||
}
|
||||
}
|
||||
|
||||
// The entry of `time` at the back of a time-ordered deque, added when it isn't there yet
|
||||
template<typename Value>
|
||||
Value& Slot(std::deque<std::pair<int64_t, Value>>& slots, int64_t time, size_t keep) {
|
||||
if (slots.empty() || slots.back().first < time) {
|
||||
slots.emplace_back(time, Value{});
|
||||
while (slots.size() > keep) slots.pop_front();
|
||||
return slots.back().second;
|
||||
}
|
||||
for (auto it = slots.rbegin(); it != slots.rend(); ++it) {
|
||||
if (it->first == time) return it->second;
|
||||
}
|
||||
return slots.back().second; // older than everything kept: count it with the newest
|
||||
}
|
||||
}
|
||||
|
||||
void TrafficHistory::Point::Add(const TrafficStats::Second& second) {
|
||||
packetsIn += second.packetsIn;
|
||||
packetsOut += second.packetsOut;
|
||||
bytesIn += second.bytesIn;
|
||||
bytesOut += second.bytesOut;
|
||||
httpRequests += second.httpRequests;
|
||||
http4xx += second.httpStatus[3];
|
||||
http5xx += second.httpStatus[4];
|
||||
httpBytesOut += second.httpBytesOut;
|
||||
if (!second.httpLatency.Empty()) {
|
||||
MergeSparse(latency, second.httpLatency.Sparse());
|
||||
latencySum += second.httpLatency.Sum();
|
||||
}
|
||||
}
|
||||
|
||||
void TrafficHistory::Point::Add(const Point& other) {
|
||||
packetsIn += other.packetsIn;
|
||||
packetsOut += other.packetsOut;
|
||||
bytesIn += other.bytesIn;
|
||||
bytesOut += other.bytesOut;
|
||||
httpRequests += other.httpRequests;
|
||||
http4xx += other.http4xx;
|
||||
http5xx += other.http5xx;
|
||||
httpBytesOut += other.httpBytesOut;
|
||||
MergeSparse(latency, other.latency);
|
||||
latencySum += other.latencySum;
|
||||
}
|
||||
|
||||
TrafficStats::Histogram TrafficHistory::Point::Latency() const {
|
||||
return TrafficStats::Histogram::FromSparse(latency, latencySum);
|
||||
}
|
||||
|
||||
std::string TrafficHistory::KeyFor(uint16_t serviceType, uint32_t zoneId, uint32_t instanceId) {
|
||||
switch (static_cast<ServiceType>(serviceType)) {
|
||||
case ServiceType::MASTER: return "master";
|
||||
case ServiceType::AUTH: return "auth";
|
||||
case ServiceType::CHAT: return "chat";
|
||||
case ServiceType::DASHBOARD: return "dashboard";
|
||||
case ServiceType::UGC: return "ugc";
|
||||
case ServiceType::WORLD: return "world:" + std::to_string(zoneId) + ":" + std::to_string(instanceId);
|
||||
default: return "service:" + std::to_string(serviceType) + ":" + std::to_string(zoneId) + ":" + std::to_string(instanceId);
|
||||
}
|
||||
}
|
||||
|
||||
void TrafficHistory::Ingest(uint16_t serviceType, uint32_t zoneId, uint32_t instanceId, const TrafficStats::Report& report, int64_t now) {
|
||||
const auto key = KeyFor(serviceType, zoneId, instanceId);
|
||||
auto [it, added] = m_Servers.try_emplace(key);
|
||||
auto& server = it->second;
|
||||
if (added) {
|
||||
server.key = key;
|
||||
server.type = serviceType;
|
||||
server.zoneId = zoneId;
|
||||
server.instanceId = instanceId;
|
||||
server.firstSeen = now;
|
||||
// Minutes before the first report were never seen, so nothing before it is written
|
||||
server.writtenUntil = Floor(now, 60) - 60;
|
||||
}
|
||||
server.lastSeen = now;
|
||||
server.link = report.link;
|
||||
server.gauges = report.gauges;
|
||||
|
||||
auto& totals = server.totals;
|
||||
int64_t lastSecond = now;
|
||||
for (const auto& second : report.seconds) {
|
||||
// A server whose clock is far off is still shown, at the dashboard's time
|
||||
if (second.time < now - SECONDS_KEPT || second.time > now + 60) continue;
|
||||
lastSecond = second.time;
|
||||
totals.packetsIn += second.packetsIn;
|
||||
totals.packetsOut += second.packetsOut;
|
||||
totals.bytesIn += second.bytesIn;
|
||||
totals.bytesOut += second.bytesOut;
|
||||
|
||||
if (server.seconds.empty() || server.seconds.back().time < second.time) {
|
||||
server.seconds.push_back(Point{ .time = second.time });
|
||||
server.seconds.back().Add(second);
|
||||
} else {
|
||||
const auto at = std::lower_bound(server.seconds.begin(), server.seconds.end(), second.time, [](const Point& p, int64_t t) { return p.time < t; });
|
||||
if (at != server.seconds.end() && at->time == second.time) at->Add(second);
|
||||
else server.seconds.insert(at, Point{ .time = second.time })->Add(second);
|
||||
}
|
||||
|
||||
const auto minute = Floor(second.time, 60);
|
||||
if (minute < server.writtenUntil) continue; // that minute is in the database already
|
||||
auto slot = std::find_if(server.minutes.rbegin(), server.minutes.rend(), [minute](const Minute& m) { return m.point.time <= minute; });
|
||||
if (slot == server.minutes.rend() || slot->point.time != minute) {
|
||||
const auto at = server.minutes.insert(slot.base(), Minute{ .point = Point{ .time = minute } });
|
||||
at->point.Add(second);
|
||||
} else {
|
||||
slot->point.Add(second);
|
||||
}
|
||||
}
|
||||
while (!server.seconds.empty() && server.seconds.front().time < now - SECONDS_KEPT) server.seconds.pop_front();
|
||||
|
||||
const auto& link = report.link;
|
||||
totals.datagramsSent += link.datagramsSent;
|
||||
totals.datagramsReceived += link.datagramsReceived;
|
||||
totals.linkBytesSent += link.bytesSent;
|
||||
totals.linkBytesReceived += link.bytesReceived;
|
||||
totals.resends += link.resends;
|
||||
if (link.resends && !server.minutes.empty()) {
|
||||
const auto minute = Floor(lastSecond, 60);
|
||||
auto slot = std::find_if(server.minutes.rbegin(), server.minutes.rend(), [minute](const Minute& m) { return m.point.time == minute; });
|
||||
(slot != server.minutes.rend() ? *slot : server.minutes.back()).resends += link.resends;
|
||||
}
|
||||
|
||||
AddCounts(totals.messages, report.messages, MAX_TOTAL_MESSAGES);
|
||||
AddCounts(Slot(server.messageMinutes, Floor(now, 60), MESSAGE_MINUTES), report.messages);
|
||||
AddCounts(Slot(server.messageHours, Floor(now, 3600), MESSAGE_HOURS), report.messages);
|
||||
|
||||
AddRoutes(totals.routes, report.routes, MAX_TOTAL_ROUTES);
|
||||
if (!report.routes.empty()) AddRoutes(Slot(server.routeMinutes, Floor(now, 60), ROUTE_MINUTES), report.routes, MAX_TOTAL_ROUTES);
|
||||
}
|
||||
|
||||
std::vector<IServerTraffic::TrafficMinute> TrafficHistory::TakeFinishedMinutes(int64_t now) {
|
||||
std::vector<IServerTraffic::TrafficMinute> rows;
|
||||
for (auto& [key, server] : m_Servers) {
|
||||
while (!server.minutes.empty() && server.minutes.front().point.time + 60 + MINUTE_SLACK <= now) {
|
||||
const auto& minute = server.minutes.front();
|
||||
const auto& p = minute.point;
|
||||
const auto latency = p.Latency();
|
||||
rows.push_back({ .time = p.time, .server = key, .packetsIn = p.packetsIn, .packetsOut = p.packetsOut, .bytesIn = p.bytesIn, .bytesOut = p.bytesOut,
|
||||
.resends = minute.resends, .httpRequests = p.httpRequests, .http4xx = p.http4xx, .http5xx = p.http5xx, .httpBytesOut = p.httpBytesOut,
|
||||
.latencyP50Us = static_cast<uint32_t>(std::min<uint64_t>(latency.Percentile(0.50), UINT32_MAX)),
|
||||
.latencyP95Us = static_cast<uint32_t>(std::min<uint64_t>(latency.Percentile(0.95), UINT32_MAX)),
|
||||
.latencyP99Us = static_cast<uint32_t>(std::min<uint64_t>(latency.Percentile(0.99), UINT32_MAX)) });
|
||||
server.writtenUntil = std::max(server.writtenUntil, p.time + 60);
|
||||
server.minutes.pop_front();
|
||||
}
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
void TrafficHistory::Forget(int64_t now) {
|
||||
for (auto it = m_Servers.begin(); it != m_Servers.end();) {
|
||||
auto& seconds = it->second.seconds;
|
||||
while (!seconds.empty() && seconds.front().time < now - SECONDS_KEPT) seconds.pop_front();
|
||||
if (it->second.lastSeen < now - FORGET_AFTER && it->second.minutes.empty()) it = m_Servers.erase(it);
|
||||
else ++it;
|
||||
}
|
||||
}
|
||||
|
||||
std::map<std::string, std::vector<TrafficHistory::Point>> TrafficHistory::Buckets(int64_t from, int64_t to, int64_t step) const {
|
||||
std::map<std::string, std::vector<Point>> out;
|
||||
if (step <= 0 || to <= from) return out;
|
||||
const auto count = static_cast<size_t>((to - from + step - 1) / step);
|
||||
for (const auto& [key, server] : m_Servers) {
|
||||
auto it = std::lower_bound(server.seconds.begin(), server.seconds.end(), from, [](const Point& p, int64_t t) { return p.time < t; });
|
||||
if (it == server.seconds.end() || it->time >= to) continue;
|
||||
auto& points = out[key];
|
||||
points.resize(count);
|
||||
for (size_t i = 0; i < count; i++) points[i].time = from + static_cast<int64_t>(i) * step;
|
||||
for (; it != server.seconds.end() && it->time < to; ++it) points[static_cast<size_t>((it->time - from) / step)].Add(*it);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
std::vector<TrafficStats::MessageCount> TrafficHistory::TopMessages(const std::string& serverKey, int64_t since, size_t limit) const {
|
||||
MessageCounts counts;
|
||||
for (const auto& [key, server] : m_Servers) {
|
||||
if (!serverKey.empty() && key != serverKey) continue;
|
||||
const bool minutes = !server.messageMinutes.empty() && server.messageMinutes.front().first <= Floor(since, 60);
|
||||
const auto& slots = minutes || server.messageHours.empty() ? server.messageMinutes : server.messageHours;
|
||||
const auto start = minutes ? Floor(since, 60) : Floor(since, 3600);
|
||||
for (const auto& [time, slot] : slots) {
|
||||
if (time < start) continue;
|
||||
for (const auto& [packed, count] : slot) {
|
||||
auto& total = counts[packed];
|
||||
total.key = count.key;
|
||||
total.count += count.count;
|
||||
total.bytes += count.bytes;
|
||||
}
|
||||
}
|
||||
}
|
||||
std::vector<TrafficStats::MessageCount> list;
|
||||
list.reserve(counts.size());
|
||||
for (const auto& [_, count] : counts) list.push_back(count);
|
||||
return TrafficStats::Top(list, limit);
|
||||
}
|
||||
|
||||
std::vector<std::pair<std::string, TrafficStats::RouteStats>> TrafficHistory::Routes(int64_t since) const {
|
||||
std::vector<std::pair<std::string, TrafficStats::RouteStats>> out;
|
||||
for (const auto& [key, server] : m_Servers) {
|
||||
std::map<std::string, TrafficStats::RouteStats> routes;
|
||||
for (const auto& [time, slot] : server.routeMinutes) {
|
||||
if (time < Floor(since, 60)) continue;
|
||||
for (const auto& [name, stats] : slot) {
|
||||
auto& total = routes[name];
|
||||
total.route = name;
|
||||
total.Merge(stats);
|
||||
}
|
||||
}
|
||||
for (auto& [_, stats] : routes) out.emplace_back(key, std::move(stats));
|
||||
}
|
||||
std::sort(out.begin(), out.end(), [](const auto& a, const auto& b) { return a.second.count != b.second.count ? a.second.count > b.second.count : a.second.route < b.second.route; });
|
||||
return out;
|
||||
}
|
||||
102
dDashboardServer/routes/TrafficHistory.h
Normal file
102
dDashboardServer/routes/TrafficHistory.h
Normal file
@@ -0,0 +1,102 @@
|
||||
#pragma once
|
||||
|
||||
#include <array>
|
||||
#include <cstdint>
|
||||
#include <deque>
|
||||
#include <map>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include "IServerTraffic.h"
|
||||
#include "TrafficStats.h"
|
||||
|
||||
/**
|
||||
* The traffic reports of every server, kept by the dashboard: the last hour at one second, the minutes that are not in
|
||||
* the database yet, the busiest message types of the last hour and day, and totals since the dashboard started (for
|
||||
* Prometheus). Pure (no database, network or clock), so it is unit tested; Traffic.cpp feeds it and writes the minutes.
|
||||
*/
|
||||
class TrafficHistory {
|
||||
public:
|
||||
static constexpr int64_t SECONDS_KEPT = 3600;
|
||||
static constexpr int64_t MINUTE_SLACK = 20; // seconds a minute waits for late reports before it is final
|
||||
static constexpr int64_t FORGET_AFTER = 86400; // a server silent this long is dropped
|
||||
static constexpr size_t MESSAGE_MINUTES = 60; // per-minute message counts kept
|
||||
static constexpr size_t MESSAGE_HOURS = 24; // per-hour message counts kept
|
||||
static constexpr size_t MAX_TOTAL_MESSAGES = 256; // message types a server's totals keep
|
||||
static constexpr size_t MAX_TOTAL_ROUTES = 128; // routes a server's totals keep
|
||||
static constexpr size_t ROUTE_MINUTES = 60; // per-minute route stats kept
|
||||
|
||||
// One second (or a bucket of them) of one server
|
||||
struct Point {
|
||||
int64_t time{};
|
||||
uint64_t packetsIn{}, packetsOut{}, bytesIn{}, bytesOut{};
|
||||
uint64_t httpRequests{}, http4xx{}, http5xx{}, httpBytesOut{};
|
||||
std::vector<std::pair<uint8_t, uint32_t>> latency; // sparse HTTP latency histogram, empty without requests
|
||||
uint64_t latencySum{};
|
||||
|
||||
void Add(const TrafficStats::Second& second);
|
||||
void Add(const Point& other);
|
||||
TrafficStats::Histogram Latency() const;
|
||||
};
|
||||
|
||||
struct Minute {
|
||||
Point point;
|
||||
uint64_t resends{};
|
||||
};
|
||||
|
||||
using MessageCounts = std::map<uint64_t, TrafficStats::MessageCount>; // by MessageKey::Packed
|
||||
|
||||
struct Totals {
|
||||
uint64_t packetsIn{}, packetsOut{}, bytesIn{}, bytesOut{};
|
||||
uint64_t datagramsSent{}, datagramsReceived{}, linkBytesSent{}, linkBytesReceived{}, resends{};
|
||||
MessageCounts messages;
|
||||
std::map<std::string, TrafficStats::RouteStats> routes;
|
||||
};
|
||||
|
||||
struct Server {
|
||||
std::string key;
|
||||
uint16_t type{}; // ServiceType
|
||||
uint32_t zoneId{};
|
||||
uint32_t instanceId{};
|
||||
int64_t firstSeen{};
|
||||
int64_t lastSeen{};
|
||||
std::deque<Point> seconds; // oldest first, up to SECONDS_KEPT
|
||||
std::deque<Minute> minutes; // not written yet
|
||||
int64_t writtenUntil{}; // minutes before this were handed out by TakeFinishedMinutes
|
||||
TrafficStats::Link link; // the last report's
|
||||
std::vector<std::pair<std::string, double>> gauges;
|
||||
Totals totals;
|
||||
std::deque<std::pair<int64_t, MessageCounts>> messageMinutes;
|
||||
std::deque<std::pair<int64_t, MessageCounts>> messageHours;
|
||||
std::deque<std::pair<int64_t, std::map<std::string, TrafficStats::RouteStats>>> routeMinutes;
|
||||
};
|
||||
|
||||
// "master", "auth", "chat", "dashboard", "ugc", "world:<zone>:<instance>"
|
||||
static std::string KeyFor(uint16_t serviceType, uint32_t zoneId, uint32_t instanceId);
|
||||
|
||||
void Ingest(uint16_t serviceType, uint32_t zoneId, uint32_t instanceId, const TrafficStats::Report& report, int64_t now);
|
||||
|
||||
// Minutes that are final (and not handed out before), as database rows
|
||||
std::vector<IServerTraffic::TrafficMinute> TakeFinishedMinutes(int64_t now);
|
||||
|
||||
// Drop servers silent for FORGET_AFTER and seconds older than SECONDS_KEPT
|
||||
void Forget(int64_t now);
|
||||
|
||||
const std::map<std::string, Server>& Servers() const { return m_Servers; }
|
||||
|
||||
/**
|
||||
* The seconds from `from` (inclusive) to `to` (exclusive) summed into buckets of `step` seconds, per server:
|
||||
* index i is the bucket starting at from + i * step. Servers with nothing in the range are left out.
|
||||
*/
|
||||
std::map<std::string, std::vector<Point>> Buckets(int64_t from, int64_t to, int64_t step) const;
|
||||
|
||||
// The busiest message types since `since` (minute counts for the last hour, hour counts before), one server or all ("")
|
||||
std::vector<TrafficStats::MessageCount> TopMessages(const std::string& server, int64_t since, size_t limit) const;
|
||||
|
||||
// The HTTP routes since `since` (the last hour at most) of every server with a web server, as (server, route); busiest first
|
||||
std::vector<std::pair<std::string, TrafficStats::RouteStats>> Routes(int64_t since) const;
|
||||
|
||||
private:
|
||||
std::map<std::string, Server> m_Servers;
|
||||
};
|
||||
Reference in New Issue
Block a user