diff --git a/common/arg.cpp b/common/arg.cpp
index be415ee0..68920618 100644
--- a/common/arg.cpp
+++ b/common/arg.cpp
@@ -1783,6 +1783,29 @@ common_params_context common_params_parser_init(common_params & params, llama_ex
params.scr = value;
}
).set_env("LLAMA_ARG_SCR").set_examples({LLAMA_EXAMPLE_SERVER}));
+ add_opt(common_arg(
+ {"--telemetry-dir"}, "PATH",
+ "record request telemetry (GET /stats) and append it to JSONL files in PATH\n"
+ "(default: <--cache-disk>/telemetry, off without --cache-disk)",
+ [](common_params & params, const std::string & value) {
+ params.telemetry_dir = value;
+ }
+ ).set_env("LLAMA_ARG_TELEMETRY_DIR").set_examples({LLAMA_EXAMPLE_SERVER}));
+ add_opt(common_arg(
+ {"--telemetry"},
+ {"--no-telemetry"},
+ "request telemetry, on by default where it has a directory (see --telemetry-dir)",
+ [](common_params & params, bool value) {
+ params.telemetry = value;
+ }
+ ).set_env("LLAMA_ARG_TELEMETRY").set_examples({LLAMA_EXAMPLE_SERVER}));
+ add_opt(common_arg(
+ {"--telemetry-size"}, "N",
+ string_format("size cap of the telemetry files in MiB, the oldest are deleted (default: %d)", params.telemetry_mib),
+ [](common_params & params, int value) {
+ params.telemetry_mib = std::max(value, 1);
+ }
+ ).set_env("LLAMA_ARG_TELEMETRY_SIZE").set_examples({LLAMA_EXAMPLE_SERVER}));
add_opt(common_arg(
{"-kvu", "--kv-unified"},
{"-no-kvu", "--no-kv-unified"},
diff --git a/common/common.h b/common/common.h
index cc89d4b1..a7dc6224 100644
--- a/common/common.h
+++ b/common/common.h
@@ -629,6 +629,9 @@ struct common_params {
std::string cache_classifier; // URL that classifies new prompt cache entries, empty = off
int32_t cache_classifier_timeout = 20; // seconds
bool scr = false; // suffix cache reuse for requests that do not set "scr"
+ bool telemetry = true; // request telemetry, see telemetry_dir
+ std::string telemetry_dir; // request telemetry JSONL directory, empty = <cache_disk_path>/telemetry (off without a disk tier)
+ int32_t telemetry_mib = 200; // size cap of the telemetry files in MiB
std::string hostname = "127.0.0.1";
std::string public_path = ""; // NOLINT
diff --git a/tools/server/CMakeLists.txt b/tools/server/CMakeLists.txt
index 1f6ad1f2..5606a966 100644
--- a/tools/server/CMakeLists.txt
+++ b/tools/server/CMakeLists.txt
@@ -25,6 +25,8 @@ add_library(${TARGET} STATIC
server-mcp.h
server-schema.cpp
server-schema.h
+ server-telemetry.cpp
+ server-telemetry.h
)
if (BUILD_SHARED_LIBS)
diff --git a/tools/server/server-context.cpp b/tools/server/server-context.cpp
index 0609fa71..f37bed58 100644
--- a/tools/server/server-context.cpp
+++ b/tools/server/server-context.cpp
@@ -16,6 +16,7 @@
#include "sampling.h"
#include "speculative.h"
#include "scr.h"
+#include "server-telemetry.h"
#include "../../src/llama-ext.h" // suffix cache reuse primitives
#include "mtmd.h"
#include "mtmd-helper.h"
@@ -268,6 +269,19 @@ struct server_slot {
std::unique_ptr<scr_splice> scr;
+ // telemetry of the current task. mutable: send_error() records its message on a const slot
+ struct telemetry_state {
+ std::string tier = "slot"; // where the exact reuse comes from, set when the slot is chosen
+ int32_t n_images_encoded = 0;
+ int32_t n_images_cached = 0; // embeddings from the image embedding cache, the encoder skipped
+ double image_ms = 0.0; // encode (or cache read) and the decode of the image tokens
+ double image_encode_ms = 0.0; // the vision encoder alone
+ bool final_sent = false;
+ std::string error;
+ };
+
+ mutable telemetry_state tele;
+
// drops a splice that did not finish: the moved runs beyond the prefill position go, the prefix stays
void scr_abort() {
if (!scr) {
@@ -837,6 +851,7 @@ static int process_mtmd_chunk(const server_slot & slot, mtmd::batch_ptr & mbatch
return -1;
}
SLT_INF(slot, "encoded %d media chunk(s) from idx = %zu in %.1f ms\n", n_added, idx, (ggml_time_us() - t_encode) / 1e3);
+ slot.tele.image_encode_ms += (ggml_time_us() - t_encode) / 1e3;
return try_decode();
}
@@ -860,6 +875,9 @@ public:
server_queue queue_tasks;
server_response queue_results;
+ // request telemetry; its readers are thread-safe (GET /stats is answered from the HTTP threads)
+ server_telemetry telemetry;
+
// note: chat_params must not be refreshed upon existing sleeping state
server_chat_params chat_params;
@@ -987,6 +1005,111 @@ private:
mctx = nullptr;
}
+ // the slots and cache tiers as the telemetry minute samples see them
+ void telemetry_gauges() {
+ if (!telemetry.enabled()) {
+ return;
+ }
+ server_telemetry_gauges g;
+ g.slots_total = (int32_t) slots.size();
+ for (const auto & slot : slots) {
+ g.slots_busy += slot.is_processing();
+ }
+ if (prompt_cache) {
+ g.ram_bytes = (int64_t) prompt_cache->size();
+ g.ram_limit = (int64_t) prompt_cache->limit_size;
+ size_t disk_bytes = 0, disk_limit = 0;
+ if (prompt_cache->disk_usage(disk_bytes, disk_limit)) {
+ g.disk_bytes = (int64_t) disk_bytes;
+ g.disk_limit = (int64_t) disk_limit;
+ telemetry_disk = { g.disk_bytes, g.disk_limit };
+ } else {
+ // the disk tier is busy: its last known size
+ g.disk_bytes = telemetry_disk.first;
+ g.disk_limit = telemetry_disk.second;
+ }
+ }
+ telemetry.set_gauges(g);
+ }
+
+ // hands the finished task of a slot to telemetry; called before the slot resets (every release path)
+ void telemetry_record(const server_slot & slot) {
+ const auto tele = std::move(slot.tele);
+ slot.tele = {};
+
+ if (!telemetry.enabled() || !slot.task || slot.task->is_child() || slot.task->tele.t_admit_us == 0) {
+ return;
+ }
+ const auto & task = *slot.task;
+ const auto & st = slot.stats;
+ const int64_t t_admit = task.tele.t_admit_us;
+
+ server_telemetry_record r;
+ r.t_wall_ms = task.tele.t_admit_wall_ms;
+ r.id_slot = slot.id;
+ r.id = task.params.oaicompat_cmpl_id;
+ r.endpoint = task.tele.endpoint;
+ r.user_agent = task.tele.user_agent;
+ r.excerpt = task.tele.excerpt;
+
+ if (!tele.error.empty()) {
+ r.status = "error";
+ r.reason = tele.error;
+ } else if (tele.final_sent) {
+ r.status = "completed";
+ r.reason = slot.stop == STOP_TYPE_EOS ? "stop" : slot.stop == STOP_TYPE_WORD ? "word" : slot.stop == STOP_TYPE_LIMIT ? "length" : "none";
+ if (slot.truncated) {
+ r.reason += ", truncated";
+ }
+ } else {
+ r.status = "cancelled";
+ r.reason = st.t_start == 0 ? "queued" : st.n_gen == 0 ? "prefill" : "decode";
+ }
+
+ r.n_prompt = task.n_tokens();
+ r.n_cached = (int32_t) st.n_prompt_cached;
+ r.tier = r.n_cached > 0 ? tele.tier : "none";
+ r.n_scr = (int32_t) std::max<int64_t>(0, st.n_prompt_scr);
+ r.n_prefilled = (int32_t) st.n_prompt_processed;
+ r.scr = task.params.scr;
+
+ r.n_images = 0;
+ if (task.tokens.has_mtmd) {
+ for (size_t i = 0; i < task.tokens.size(); ) {
+ if (task.tokens[i] == LLAMA_TOKEN_NULL) {
+ r.n_images++;
+ i += mtmd_input_chunk_get_n_tokens(task.tokens.find_chunk(i).get());
+ } else {
+ i++;
+ }
+ }
+ }
+ r.n_images_encoded = tele.n_images_encoded;
+ r.n_images_cached = tele.n_images_cached;
+ r.image_ms = tele.image_ms;
+ r.image_encode_ms = tele.image_encode_ms;
+
+ r.queue_ms = ((st.t_start > 0 ? st.t_start : ggml_time_us()) - t_admit) / 1000.0;
+ r.prefill_ms = st.t_prompt_ms();
+ r.prefill_tps = st.n_prompt_tps();
+ r.ttft_ms = st.n_gen > 0 ? r.queue_ms + r.prefill_ms : -1.0;
+ r.n_gen = (int32_t) st.n_gen;
+ r.decode_ms = st.t_gen_ms();
+ r.decode_tps = st.n_gen_tps();
+ r.n_draft = (int32_t) st.n_draft_tokens;
+ r.n_draft_accepted = (int32_t) st.n_draft_accepted;
+ r.n_cmpl = task.params.n_cmpl;
+
+ r.cache_class = cache_class_name(slot.cache_skip ? CACHE_CLASS_SKIP : slot.prompt.meta.cls);
+ r.cache_source = cache_source_name(slot.prompt.meta.source);
+ r.family = slot.prompt.meta.family;
+
+ telemetry.push(std::move(r));
+ }
+
+ // the disk tier's bytes and limit at the last reading that got its lock
+ std::pair<int64_t, int64_t> telemetry_disk = { -1, 0 };
+
std::vector<cache_meta *> slot_metas() {
std::vector<cache_meta *> res;
for (auto & slot : slots) {
@@ -1348,9 +1471,11 @@ private:
slot.callback_on_release = [this](int id_slot) {
queue_tasks.pop_deferred_task(id_slot);
+ telemetry_gauges();
};
slot.callback_on_reset = [this](const server_slot & slot) {
+ telemetry_record(slot);
// flush the generated token stats before reset()
if (slot.stats.n_gen > 0) {
metrics_on_prediction(slot);
@@ -1449,6 +1574,15 @@ private:
} else {
SRV_TRC("%s", "prompt cache is disabled - use `--cache-ram N` to enable it\n");
}
+
+ if (params_base.telemetry) {
+ // without an explicit directory, next to the prompt cache's disk tier
+ const std::string dir = !params_base.telemetry_dir.empty() ? params_base.telemetry_dir
+ : !params_base.cache_disk_path.empty() ? (std::filesystem::path(params_base.cache_disk_path) / "telemetry").string() : "";
+ telemetry.start(dir, params_base.telemetry_mib, ggml_backend_dev_by_type(GGML_BACKEND_DEVICE_TYPE_GPU));
+ }
+ telemetry_gauges();
+
SRV_TRC("%s", "for more info see https://github.com/ggml-org/llama.cpp/pull/16391\n");
if (params_base.n_ctx_checkpoints > 0) {
@@ -1734,6 +1868,11 @@ private:
}
if (ret) {
+ // telemetry: a prompt that changes below came from the RAM or the disk tier
+ const uint64_t id_before = ret->prompt.meta.id;
+ const int n_before = ret->prompt.n_tokens();
+ const uint64_t n_disk_before = prompt_cache ? prompt_cache->n_disk_reads : 0;
+
update_cache = update_cache && prompt_cache;
// cache prompts only for completion tasks
@@ -1770,6 +1909,11 @@ private:
if (prompt_cache && task.params.scr && task.type == SERVER_TASK_TYPE_COMPLETION) {
scr_load_entry(*ret, task, update_cache);
}
+
+ if (!ret->is_processing()) {
+ const bool loaded = ret->prompt.meta.id != id_before || ret->prompt.n_tokens() != n_before;
+ ret->tele.tier = !loaded ? "slot" : prompt_cache && prompt_cache->n_disk_reads != n_disk_before ? "disk" : "ram";
+ }
}
return ret;
@@ -1949,6 +2093,8 @@ private:
// reset server kill-switch counter
n_empty_consecutive = 0;
+ telemetry_gauges();
+
SLT_INF(slot, "processing task, is_child = %d\n", slot.task->is_child());
return true;
}
@@ -2146,6 +2292,7 @@ private:
}
void send_error(const server_slot & slot, const std::string & error, const enum error_type type = ERROR_TYPE_SERVER) {
+ slot.tele.error = error;
send_error(slot.task->id, error, type, slot.task->n_tokens(), slot.n_ctx);
}
@@ -2210,6 +2357,8 @@ private:
}
void send_final_response(server_slot & slot) {
+ slot.tele.final_sent = true;
+
auto res = std::make_unique<server_task_result_cmpl_final>();
res->id = slot.task->id;
@@ -3081,6 +3230,7 @@ private:
if (prompt_cache) {
prompt_cache->maintain(slot_metas());
}
+ telemetry_gauges();
} break;
case SERVER_TASK_TYPE_CACHE_GET:
case SERVER_TASK_TYPE_CACHE_CLEAR:
@@ -3303,6 +3453,8 @@ private:
#endif
void update_slots() {
+ telemetry.note_gpu();
+
#ifdef DEBUG_TIMINGS
static int64_t t_prev = 0;
int64_t t_start = ggml_time_us();
@@ -4032,9 +4184,17 @@ private:
// encode on the worker thread, so we can still handle metrics tasks
size_t n_tokens_out = 0;
int32_t res = 0;
+ const int64_t t_mtmd = ggml_time_us();
+ const uint64_t n_hits = slot.embd_cache ? slot.embd_cache->n_hits() : 0;
queue_tasks.yield_to_queue([&]() {
res = process_mtmd_chunk(slot, slot.mbatch, cur_token_idx, n_tokens_out);
});
+ if (slot.embd_cache && slot.embd_cache->n_hits() != n_hits) {
+ slot.tele.n_images_cached++;
+ } else {
+ slot.tele.n_images_encoded++;
+ }
+ slot.tele.image_ms += (ggml_time_us() - t_mtmd) / 1000.0;
if (res != 0) {
SLT_ERR(slot, "failed to process mtmd chunk, res = %d\n", res);
@@ -4863,6 +5023,10 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
task.id_slot = json_value(data, "id_slot", -1);
sse_ping_interval = task.params.sse_ping_interval;
+ if (ctx_server.telemetry.enabled()) {
+ task.tele = server_telemetry_hint_make(req.headers, req.path, prompt.is_string() ? prompt.get<std::string>() : std::string());
+ }
+
// OAI-compat
task.params.res_type = res_type;
task.params.oaicompat_cmpl_id = completion_id;
@@ -5318,6 +5482,32 @@ void server_routes::init_routes() {
return handle_cache(req, SERVER_TASK_TYPE_CACHE_CLEAR, scope.empty() ? "all" : scope);
};
+ // GET /stats?window=1h|24h|7d|SECONDS: request telemetry aggregates; answered here, not by the server loop
+ this->get_stats = [this](const server_http_req & req) {
+ auto res = create_response(true);
+ const int64_t window = server_telemetry_parse_window(req.get_param("window", "24h"));
+ if (window <= 0) {
+ res->error(format_error_response("window must be 1h, 24h, 7d or a number of seconds", ERROR_TYPE_INVALID_REQUEST));
+ return res;
+ }
+ res->ok(ctx_server.telemetry.stats(window));
+ return res;
+ };
+
+ // GET /stats/requests?limit=N: the newest request records
+ this->get_stats_requests = [this](const server_http_req & req) {
+ auto res = create_response(true);
+ int32_t limit = 50;
+ try {
+ limit = std::clamp(std::stoi(req.get_param("limit", "50")), 0, 2000);
+ } catch (const std::exception &) {
+ res->error(format_error_response("limit must be a number", ERROR_TYPE_INVALID_REQUEST));
+ return res;
+ }
+ res->ok(ctx_server.telemetry.requests(limit));
+ return res;
+ };
+
// POST /cache/entry?id=ID&class=CLASS[&ttl=SECONDS] or ?id=ID&action=delete
this->post_cache_entry = [this](const server_http_req & req) {
return handle_cache(req, SERVER_TASK_TYPE_CACHE_ENTRY, "");
diff --git a/tools/server/server-context.h b/tools/server/server-context.h
index fa342d3a..3d75753a 100644
--- a/tools/server/server-context.h
+++ b/tools/server/server-context.h
@@ -135,6 +135,8 @@ struct server_routes {
server_http_context::handler_t get_cache;
server_http_context::handler_t post_cache_clear;
server_http_context::handler_t post_cache_entry;
+ server_http_context::handler_t get_stats;
+ server_http_context::handler_t get_stats_requests;
server_http_context::handler_t get_props;
server_http_context::handler_t post_props;
server_http_context::handler_t post_infill;
diff --git a/tools/server/server-embd-cache.h b/tools/server/server-embd-cache.h
index f5ce9488..cc7c9565 100644
--- a/tools/server/server-embd-cache.h
+++ b/tools/server/server-embd-cache.h
@@ -24,6 +24,12 @@ struct server_embd_cache {
json info() const;
+ // RAM and disk hits so far
+ uint64_t n_hits() const {
+ std::lock_guard<std::mutex> lock(mutex);
+ return n_hits_ram + n_hits_disk;
+ }
+
// drops both tiers; returns the number of entries dropped
size_t clear();
diff --git a/tools/server/server-task.cpp b/tools/server/server-task.cpp
index 044e2480..f9a4b636 100644
--- a/tools/server/server-task.cpp
+++ b/tools/server/server-task.cpp
@@ -2411,6 +2411,7 @@ std::list<server_prompt_cache_state>::iterator server_prompt_cache::disk_promote
SRV_INF("prompt cache: read %zu tokens, %lld images from disk in %.2f s\n", state.prompt.tokens.size(),
(long long) std::max<int64_t>(0, media_count(state.prompt.tokens)), (ggml_time_us() - t_start) / 1e6);
+ n_disk_reads++;
if (it != disk.end()) {
// the file stays: it records the use, and is most recently used, here and after a restart
it->meta.last_used = unix_now();
@@ -3460,6 +3461,22 @@ json server_prompt_cache::info() {
return res;
}
+bool server_prompt_cache::disk_usage(size_t & bytes, size_t & limit) const {
+ if (!disk_thread.joinable()) {
+ return false;
+ }
+ std::unique_lock<std::mutex> lock(disk_mutex, std::try_to_lock);
+ if (!lock.owns_lock()) {
+ return false;
+ }
+ bytes = 0;
+ for (const auto & e : disk) {
+ bytes += e.size;
+ }
+ limit = disk_limit;
+ return true;
+}
+
size_t server_prompt_cache::clear_disk(bool include_pinned) {
if (!disk_thread.joinable()) {
return 0;
diff --git a/tools/server/server-task.h b/tools/server/server-task.h
index 2575c9cf..e428a5a4 100644
--- a/tools/server/server-task.h
+++ b/tools/server/server-task.h
@@ -16,6 +16,7 @@
// TODO: prevent including the whole server-common.h as we only use server_tokens
#include "server-common.h"
+#include "server-telemetry.h"
enum server_task_type {
@@ -224,6 +225,9 @@ struct server_task {
// used by SERVER_TASK_TYPE_SET_LORA
std::map<int, float> set_lora; // mapping adapter ID -> scale
+ // request telemetry: admission time, endpoint, client
+ server_telemetry_hint tele;
+
server_task() = default;
server_task(server_task_type type) : type(type) {}
@@ -790,6 +794,10 @@ struct server_prompt_cache {
const cache_policy & policy() const { return pol; }
+ // telemetry: disk entries read into RAM so far; the disk tier's bytes and limit, false if it is off or busy
+ uint64_t n_disk_reads = 0;
+ bool disk_usage(size_t & bytes, size_t & limit) const;
+
private:
struct disk_entry {
std::string path;
diff --git a/tools/server/server-telemetry.cpp b/tools/server/server-telemetry.cpp
new file mode 100644
index 00000000..547a6744
--- /dev/null
+++ b/tools/server/server-telemetry.cpp
@@ -0,0 +1,829 @@
+#include "server-telemetry.h"
+
+#include "ggml.h"
+#include "log.h"
+
+#include <algorithm>
+#include <chrono>
+#include <cinttypes>
+#include <cmath>
+#include <cstdio>
+#include <filesystem>
+#include <fstream>
+#include <strings.h>
+
+#define TEL_INF(fmt, ...) LOG_INF("telemetry: " fmt, __VA_ARGS__)
+#define TEL_WRN(fmt, ...) LOG_WRN("telemetry: " fmt, __VA_ARGS__)
+
+namespace fs = std::filesystem;
+
+static int64_t wall_ms() {
+ return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
+}
+
+static int64_t mono_us() {
+ return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
+}
+
+static constexpr int64_t HOUR_MS = 3600 * 1000;
+
+// the first message of a rendered chat prompt: after the role line of its first <|im_start|>, up to the next special
+// token; whitespace runs become one space
+static std::string first_message(const std::string & prompt, size_t n_chars) {
+ std::string text = prompt.substr(0, 8192);
+ const size_t start = text.find("<|im_start|>");
+ if (start != std::string::npos) {
+ const size_t nl = text.find('\n', start);
+ text = nl == std::string::npos ? "" : text.substr(nl + 1);
+ }
+ const size_t end = text.find("<|");
+ if (end != std::string::npos) {
+ text.resize(end);
+ }
+ std::string res;
+ for (char c : text) {
+ const bool space = c == ' ' || c == '\n' || c == '\r' || c == '\t';
+ if (!space || (!res.empty() && res.back() != ' ')) {
+ res += space ? ' ' : c;
+ }
+ }
+ return utf8_prefix(res, n_chars);
+}
+
+server_telemetry_hint server_telemetry_hint_make(const std::map<std::string, std::string> & headers, const std::string & path,
+ const std::string & prompt) {
+ server_telemetry_hint h;
+ h.excerpt = first_message(prompt, 40);
+ h.t_admit_us = ggml_time_us();
+ h.t_admit_wall_ms = wall_ms();
+ h.endpoint = path;
+ for (const auto & [key, value] : headers) {
+ if (key.size() == 10 && strncasecmp(key.c_str(), "user-agent", 10) == 0) {
+ h.user_agent = utf8_prefix(value, 120);
+ break;
+ }
+ }
+ return h;
+}
+
+int64_t server_telemetry_parse_window(const std::string & s) {
+ if (s.empty()) {
+ return 0;
+ }
+ size_t pos = 0;
+ int64_t v = 0;
+ try {
+ v = std::stoll(s, &pos);
+ } catch (const std::exception &) {
+ return 0;
+ }
+ const std::string unit = s.substr(pos);
+ const int64_t mul = unit.empty() || unit == "s" ? 1 : unit == "m" ? 60 : unit == "h" ? 3600 : unit == "d" ? 86400 : 0;
+ return v > 0 ? v * mul : 0;
+}
+
+std::string utf8_prefix(const std::string & s, size_t n) {
+ size_t i = 0;
+ for (size_t k = 0; i < s.size() && k < n; ++k) {
+ const unsigned char c = s[i];
+ const size_t len = c < 0x80 ? 1 : (c >> 5) == 0x6 ? 2 : (c >> 4) == 0xe ? 3 : (c >> 3) == 0x1e ? 4 : 1;
+ i = std::min(s.size(), i + len);
+ }
+ return s.substr(0, i);
+}
+
+//
+// records
+//
+
+json server_telemetry_record::to_json() const {
+ json j = {
+ { "type", "req" },
+ { "t", t_wall_ms },
+ { "slot", id_slot },
+ { "id", id },
+ { "endpoint", endpoint },
+ { "client", user_agent },
+ { "excerpt", excerpt },
+ { "status", status },
+ { "reason", reason },
+ { "n_prompt", n_prompt },
+ { "n_cached", n_cached },
+ { "tier", tier },
+ { "n_scr", n_scr },
+ { "n_prefilled", n_prefilled },
+ { "scr", scr },
+ { "n_images", n_images },
+ { "n_images_encoded", n_images_encoded },
+ { "n_images_cached", n_images_cached },
+ { "image_ms", image_ms },
+ { "image_encode_ms", image_encode_ms },
+ { "queue_ms", queue_ms },
+ { "prefill_ms", prefill_ms },
+ { "prefill_tps", prefill_tps },
+ { "n_gen", n_gen },
+ { "decode_ms", decode_ms },
+ { "decode_tps", decode_tps },
+ { "n_draft", n_draft },
+ { "n_draft_accepted", n_draft_accepted },
+ { "n_cmpl", n_cmpl },
+ { "class", cache_class },
+ { "source", cache_source },
+ };
+ // missing is not zero
+ j["ttft_ms"] = ttft_ms >= 0.0 ? json(ttft_ms) : json(nullptr);
+ char fam[24];
+ snprintf(fam, sizeof(fam), "%016" PRIx64, family);
+ j["family"] = family ? std::string(fam) : std::string();
+ return j;
+}
+
+bool server_telemetry_record::from_json(const json & j, server_telemetry_record & r) {
+ try {
+ r.t_wall_ms = j.at("t").get<int64_t>();
+ r.id_slot = j.value("slot", -1);
+ r.id = j.value("id", "");
+ r.endpoint = j.value("endpoint", "");
+ r.user_agent = j.value("client", "");
+ r.excerpt = j.value("excerpt", "");
+ r.status = j.value("status", "");
+ r.reason = j.value("reason", "");
+ r.n_prompt = j.value("n_prompt", 0);
+ r.n_cached = j.value("n_cached", 0);
+ r.tier = j.value("tier", "");
+ r.n_scr = j.value("n_scr", 0);
+ r.n_prefilled = j.value("n_prefilled", 0);
+ r.scr = j.value("scr", false);
+ r.n_images = j.value("n_images", 0);
+ r.n_images_encoded = j.value("n_images_encoded", 0);
+ r.n_images_cached = j.value("n_images_cached", 0);
+ r.image_ms = j.value("image_ms", 0.0);
+ r.image_encode_ms = j.value("image_encode_ms", 0.0);
+ r.queue_ms = j.value("queue_ms", 0.0);
+ r.ttft_ms = j.contains("ttft_ms") && j.at("ttft_ms").is_number() ? j.at("ttft_ms").get<double>() : -1.0;
+ r.prefill_ms = j.value("prefill_ms", 0.0);
+ r.prefill_tps = j.value("prefill_tps", 0.0);
+ r.n_gen = j.value("n_gen", 0);
+ r.decode_ms = j.value("decode_ms", 0.0);
+ r.decode_tps = j.value("decode_tps", 0.0);
+ r.n_draft = j.value("n_draft", 0);
+ r.n_draft_accepted = j.value("n_draft_accepted", 0);
+ r.n_cmpl = j.value("n_cmpl", 1);
+ r.cache_class = j.value("class", "");
+ r.cache_source = j.value("source", "");
+ const std::string fam = j.value("family", "");
+ r.family = fam.empty() ? 0 : std::stoull(fam, nullptr, 16);
+ return true;
+ } catch (const std::exception &) {
+ return false;
+ }
+}
+
+json server_telemetry_sample::to_json() const {
+ json j = {
+ { "type", "min" },
+ { "t", t_wall_ms },
+ { "slots_busy", g.slots_busy },
+ { "slots_total", g.slots_total },
+ { "slots_busy_avg", slots_busy_avg },
+ { "ram_limit", g.ram_limit },
+ { "disk_limit", g.disk_limit },
+ };
+ j["gpu_used"] = gpu_used >= 0 ? json(gpu_used) : json(nullptr);
+ j["gpu_used_max"] = gpu_used_max >= 0 ? json(gpu_used_max) : json(nullptr);
+ j["gpu_total"] = gpu_total >= 0 ? json(gpu_total) : json(nullptr);
+ j["ram_bytes"] = g.ram_bytes >= 0 ? json(g.ram_bytes) : json(nullptr);
+ j["disk_bytes"] = g.disk_bytes >= 0 ? json(g.disk_bytes) : json(nullptr);
+ return j;
+}
+
+bool server_telemetry_sample::from_json(const json & j, server_telemetry_sample & s) {
+ const auto num = [&](const char * key) -> int64_t {
+ return j.contains(key) && j.at(key).is_number() ? j.at(key).get<int64_t>() : -1;
+ };
+ try {
+ s.t_wall_ms = j.at("t").get<int64_t>();
+ s.g.slots_busy = j.value("slots_busy", 0);
+ s.g.slots_total = j.value("slots_total", 0);
+ s.slots_busy_avg = j.value("slots_busy_avg", 0.0);
+ s.g.ram_limit = j.value("ram_limit", (int64_t) 0);
+ s.g.disk_limit = j.value("disk_limit", (int64_t) 0);
+ s.gpu_used = num("gpu_used");
+ s.gpu_used_max = num("gpu_used_max");
+ s.gpu_total = num("gpu_total");
+ s.g.ram_bytes = num("ram_bytes");
+ s.g.disk_bytes = num("disk_bytes");
+ return true;
+ } catch (const std::exception &) {
+ return false;
+ }
+}
+
+//
+// lifecycle
+//
+
+server_telemetry::~server_telemetry() {
+ stop();
+}
+
+void server_telemetry::start(const std::string & dir_, int64_t limit_mib, ggml_backend_dev_t dev_) {
+ if (dir_.empty() || running.load()) {
+ return;
+ }
+ std::error_code ec;
+ fs::create_directories(dir_, ec);
+ if (!fs::is_directory(dir_, ec)) {
+ TEL_WRN("cannot use %s, telemetry is off\n", dir_.c_str());
+ return;
+ }
+ dir = dir_;
+ limit_bytes = std::max<int64_t>(limit_mib, 1) * 1024 * 1024;
+ dev = dev_;
+ running = true;
+ thread = std::thread([this] { loop(); });
+ TEL_INF("recording to %s (limit %" PRId64 " MiB)\n", dir.c_str(), limit_bytes / (1024 * 1024));
+}
+
+void server_telemetry::stop() {
+ if (!thread.joinable()) {
+ return;
+ }
+ {
+ std::lock_guard<std::mutex> lock(qmutex);
+ quit = true;
+ }
+ qcv.notify_one();
+ thread.join();
+ running = false;
+ if (file) {
+ fclose(file);
+ file = nullptr;
+ }
+}
+
+//
+// producer side (server loop)
+//
+
+void server_telemetry::push(server_telemetry_record && rec) {
+ if (!running.load(std::memory_order_relaxed)) {
+ return;
+ }
+ {
+ std::lock_guard<std::mutex> lock(qmutex);
+ if (queue.size() >= QUEUE_MAX) {
+ n_dropped++;
+ return;
+ }
+ queue.push_back(std::move(rec));
+ }
+ qcv.notify_one();
+}
+
+void server_telemetry::set_gauges(const server_telemetry_gauges & g) {
+ if (!running.load(std::memory_order_relaxed)) {
+ return;
+ }
+ const int64_t now = mono_us();
+ std::lock_guard<std::mutex> lock(qmutex);
+ if (t_gauges > 0) {
+ busy_us += (int64_t) gauges.slots_busy * (now - t_gauges);
+ }
+ t_gauges = now;
+ gauges = g;
+}
+
+void server_telemetry::note_gpu() {
+ if (!running.load(std::memory_order_relaxed) || !dev) {
+ return;
+ }
+ // one cudaMemGetInfo, at most every 10 s
+ static thread_local int64_t t_last = 0;
+ const int64_t now = mono_us();
+ if (now - t_last < 10 * 1000 * 1000) {
+ return;
+ }
+ t_last = now;
+ size_t free = 0, total = 0;
+ ggml_backend_dev_memory(dev, &free, &total);
+ if (total == 0) {
+ return;
+ }
+ std::lock_guard<std::mutex> lock(qmutex);
+ gpu_used_last = (int64_t) (total - free);
+ gpu_total_last = (int64_t) total;
+ gpu_used_max = std::max(gpu_used_max, gpu_used_last);
+ t_gpu_us = now;
+}
+
+//
+// writer thread
+//
+
+void server_telemetry::loop() {
+ enforce_cap();
+ replay();
+ open_segment();
+
+ using clock = std::chrono::system_clock;
+ auto next_tick = clock::time_point(std::chrono::minutes(
+ std::chrono::duration_cast<std::chrono::minutes>(clock::now().time_since_epoch()).count() + 1));
+
+ std::deque<server_telemetry_record> batch;
+ while (true) {
+ bool stopping;
+ {
+ std::unique_lock<std::mutex> lock(qmutex);
+ qcv.wait_until(lock, next_tick, [this] { return quit || !queue.empty(); });
+ batch.swap(queue);
+ stopping = quit;
+ }
+
+ if (!batch.empty()) {
+ std::string lines;
+ {
+ std::lock_guard<std::mutex> lock(smutex);
+ for (const auto & r : batch) {
+ add_locked(r);
+ }
+ }
+ for (const auto & r : batch) {
+ lines += r.to_json().dump_safe();
+ lines += '\n';
+ }
+ batch.clear();
+ write_lines(lines);
+ }
+
+ if (stopping) {
+ break;
+ }
+
+ if (clock::now() >= next_tick) {
+ next_tick += std::chrono::minutes(1);
+ const auto s = take_sample(wall_ms());
+ {
+ std::lock_guard<std::mutex> lock(smutex);
+ add_sample_locked(s);
+ }
+ write_lines(s.to_json().dump() + "\n");
+ }
+ }
+}
+
+server_telemetry_sample server_telemetry::take_sample(int64_t t_wall) {
+ static int64_t t_prev = 0;
+
+ server_telemetry_sample s;
+ s.t_wall_ms = t_wall;
+
+ const int64_t now = mono_us();
+ bool read_here = false;
+ {
+ std::lock_guard<std::mutex> lock(qmutex);
+ if (t_gauges > 0) {
+ busy_us += (int64_t) gauges.slots_busy * (now - t_gauges);
+ t_gauges = now;
+ }
+ s.g = gauges;
+ if (t_prev > 0 && now > t_prev) {
+ s.slots_busy_avg = (double) busy_us / (double) (now - t_prev);
+ }
+ busy_us = 0;
+ s.gpu_used = gpu_used_last;
+ s.gpu_total = gpu_total_last;
+ s.gpu_used_max = gpu_used_max;
+ gpu_used_max = -1;
+ // the server loop reads the GPU while it is busy; an idle loop is read from here (no decode is running,
+ // and cudaMemGetInfo neither synchronizes nor conflicts with a relaxed-mode graph capture)
+ read_here = dev && s.g.slots_busy == 0 && now - t_gpu_us > 90ll * 1000 * 1000;
+ }
+ t_prev = now;
+
+ if (read_here) {
+ size_t free = 0, total = 0;
+ ggml_backend_dev_memory(dev, &free, &total);
+ if (total > 0) {
+ s.gpu_used = (int64_t) (total - free);
+ s.gpu_total = (int64_t) total;
+ s.gpu_used_max = std::max(s.gpu_used_max, s.gpu_used);
+ }
+ }
+ return s;
+}
+
+int server_telemetry::hist_bin(double v) {
+ if (!(v > 0.25)) {
+ return 0;
+ }
+ return std::min(HIST_BINS - 1, (int) std::floor(8.0 * std::log2(v / 0.25)));
+}
+
+double server_telemetry::hist_value(int bin) {
+ return 0.25 * std::exp2((bin + 0.5) / 8.0);
+}
+
+// rules for which requests a percentile counts, shared by the ring and the hourly buckets
+static bool counts_ttft(const server_telemetry_record & r) { return r.ttft_ms >= 0.0; }
+static bool counts_prefill(const server_telemetry_record & r) { return r.n_prefilled >= 32 && r.prefill_tps > 0.0; }
+static bool counts_decode(const server_telemetry_record & r) { return r.n_gen >= 8 && r.decode_tps > 0.0; }
+
+void server_telemetry::add_locked(const server_telemetry_record & r) {
+ if (t_first_ms == 0 || r.t_wall_ms < t_first_ms) {
+ t_first_ms = r.t_wall_ms;
+ }
+
+ ring.push_back(r);
+ while (ring.size() > RING_REQUESTS) {
+ ring_complete_ms = std::max(ring_complete_ms, ring.front().t_wall_ms + 1);
+ ring.pop_front();
+ }
+
+ const int64_t hour = r.t_wall_ms / HOUR_MS;
+ auto & b = hours[hour];
+ if (b.ttft.empty()) {
+ b.ttft.assign(HIST_BINS, 0);
+ b.pp.assign(HIST_BINS, 0);
+ b.tg.assign(HIST_BINS, 0);
+ }
+ b.n++;
+ b.n_cancelled += r.status == "cancelled";
+ b.n_error += r.status == "error";
+ b.prompt += r.n_prompt;
+ b.cached += r.n_cached;
+ b.scr += r.n_scr;
+ b.prefilled += r.n_prefilled;
+ b.gen += r.n_gen;
+ b.cached_slot += r.tier == "slot" ? r.n_cached : 0;
+ b.cached_ram += r.tier == "ram" ? r.n_cached : 0;
+ b.cached_disk += r.tier == "disk" ? r.n_cached : 0;
+ b.images += r.n_images;
+ b.images_cached += r.n_images_cached;
+ b.image_ms += r.image_ms;
+ b.image_encode_ms += r.image_encode_ms;
+ b.draft += r.n_draft;
+ b.draft_accepted += r.n_draft_accepted;
+ if (counts_ttft(r)) b.ttft[hist_bin(r.ttft_ms)]++;
+ if (counts_prefill(r)) b.pp[hist_bin(r.prefill_tps)]++;
+ if (counts_decode(r)) b.tg[hist_bin(r.decode_tps)]++;
+
+ while (!hours.empty() && hours.begin()->first < hour - HOURS_KEPT) {
+ hours.erase(hours.begin());
+ }
+}
+
+void server_telemetry::add_sample_locked(const server_telemetry_sample & s) {
+ samples.push_back(s);
+ while (samples.size() > RING_SAMPLES || (!samples.empty() && samples.front().t_wall_ms < s.t_wall_ms - 24 * HOUR_MS)) {
+ samples.pop_front();
+ }
+}
+
+//
+// files: telemetry-<unix ms>.jsonl segments of at most a 1/8 of the limit, the oldest go first
+//
+
+static std::vector<fs::path> list_segments(const std::string & dir) {
+ std::vector<fs::path> res;
+ std::error_code ec;
+ for (const auto & e : fs::directory_iterator(dir, ec)) {
+ const std::string name = e.path().filename().string();
+ if (e.is_regular_file(ec) && name.rfind("telemetry-", 0) == 0 && name.size() > 6 && name.substr(name.size() - 6) == ".jsonl") {
+ res.push_back(e.path());
+ }
+ }
+ std::sort(res.begin(), res.end()); // fixed-width names sort by time
+ return res;
+}
+
+void server_telemetry::replay() {
+ const int64_t t0 = mono_us();
+ const int64_t now = wall_ms();
+ std::vector<server_telemetry_record> recs;
+ std::vector<server_telemetry_sample> smps;
+ for (const auto & path : list_segments(dir)) {
+ std::ifstream in(path);
+ std::string line;
+ while (std::getline(in, line)) {
+ if (line.empty()) {
+ continue;
+ }
+ const json j = json::parse_no_throw(line);
+ if (j.is_discarded() || !j.is_object()) {
+ continue; // a torn last line after a crash
+ }
+ const std::string type = j.value("type", "");
+ if (type == "req") {
+ server_telemetry_record r;
+ if (server_telemetry_record::from_json(j, r) && r.t_wall_ms > now - (int64_t) HOURS_KEPT * HOUR_MS) {
+ recs.push_back(std::move(r));
+ }
+ } else if (type == "min") {
+ server_telemetry_sample s;
+ if (server_telemetry_sample::from_json(j, s) && s.t_wall_ms > now - 24 * HOUR_MS) {
+ smps.push_back(s);
+ }
+ }
+ }
+ }
+ std::stable_sort(recs.begin(), recs.end(), [](const auto & a, const auto & b) { return a.t_wall_ms < b.t_wall_ms; });
+ std::lock_guard<std::mutex> lock(smutex);
+ for (const auto & r : recs) {
+ add_locked(r);
+ }
+ for (const auto & s : smps) {
+ add_sample_locked(s);
+ }
+ n_replayed = recs.size();
+ if (!recs.empty()) {
+ TEL_INF("replayed %zu requests and %zu samples in %.2f s\n", recs.size(), smps.size(), (mono_us() - t0) / 1e6);
+ }
+}
+
+void server_telemetry::open_segment() {
+ if (file) {
+ fclose(file);
+ file = nullptr;
+ }
+ char name[64];
+ snprintf(name, sizeof(name), "telemetry-%013" PRId64 ".jsonl", wall_ms());
+ file_path = (fs::path(dir) / name).string();
+ file = fopen(file_path.c_str(), "ab");
+ file_size = 0;
+ if (!file) {
+ TEL_WRN("cannot open %s\n", file_path.c_str());
+ }
+ enforce_cap();
+}
+
+void server_telemetry::enforce_cap() {
+ auto segs = list_segments(dir);
+ std::error_code ec;
+ std::vector<int64_t> sizes;
+ int64_t total = 0;
+ for (const auto & p : segs) {
+ const int64_t sz = (int64_t) fs::file_size(p, ec);
+ sizes.push_back(ec ? 0 : sz);
+ total += sizes.back();
+ }
+ size_t i = 0;
+ while (total > limit_bytes && i + 1 < segs.size()) { // never the segment being written
+ fs::remove(segs[i], ec);
+ total -= sizes[i];
+ i++;
+ }
+ std::lock_guard<std::mutex> lock(smutex);
+ files_bytes = total;
+ n_files = (int32_t) (segs.size() - i);
+}
+
+void server_telemetry::write_lines(const std::string & lines) {
+ if (!file) {
+ std::lock_guard<std::mutex> lock(smutex);
+ n_write_errors++;
+ return;
+ }
+ const bool ok = fwrite(lines.data(), 1, lines.size(), file) == lines.size() && fflush(file) == 0;
+ file_size += (int64_t) lines.size();
+ {
+ std::lock_guard<std::mutex> lock(smutex);
+ n_write_errors += !ok;
+ n_written += ok;
+ files_bytes += (int64_t) lines.size();
+ }
+ if (file_size >= std::max<int64_t>(limit_bytes / 8, 64 * 1024)) {
+ open_segment();
+ }
+}
+
+//
+// readers (HTTP threads)
+//
+
+namespace {
+
+struct window_agg {
+ int64_t n = 0, n_cancelled = 0, n_error = 0;
+ int64_t prompt = 0, cached = 0, scr = 0, prefilled = 0, gen = 0;
+ int64_t cached_slot = 0, cached_ram = 0, cached_disk = 0;
+ int64_t images = 0, images_cached = 0, draft = 0, draft_accepted = 0;
+ double image_ms = 0.0, image_encode_ms = 0.0;
+};
+
+json percentiles(std::vector<double> v, std::initializer_list<double> ps) {
+ json j = json::object();
+ std::sort(v.begin(), v.end());
+ j["n"] = (int64_t) v.size();
+ for (double p : ps) {
+ const std::string key = "p" + std::to_string((int) std::lround(p * 100));
+ if (v.empty()) {
+ j[key] = nullptr;
+ } else {
+ const size_t rank = (size_t) std::ceil(p * v.size());
+ j[key] = v[std::min(v.size() - 1, rank > 0 ? rank - 1 : 0)];
+ }
+ }
+ return j;
+}
+
+json hist_percentiles(const std::vector<uint64_t> & h, std::initializer_list<double> ps, double (*value)(int)) {
+ json j = json::object();
+ uint64_t n = 0;
+ for (auto c : h) {
+ n += c;
+ }
+ j["n"] = n;
+ for (double p : ps) {
+ const std::string key = "p" + std::to_string((int) std::lround(p * 100));
+ if (n == 0) {
+ j[key] = nullptr;
+ continue;
+ }
+ const uint64_t rank = std::max<uint64_t>(1, (uint64_t) std::ceil(p * n));
+ uint64_t acc = 0;
+ for (size_t i = 0; i < h.size(); ++i) {
+ acc += h[i];
+ if (acc >= rank) {
+ j[key] = value((int) i);
+ break;
+ }
+ }
+ }
+ return j;
+}
+
+double share(int64_t a, int64_t b) {
+ return b > 0 ? (double) a / (double) b : 0.0;
+}
+
+} // namespace
+
+json server_telemetry::stats(int64_t window_s) {
+ json res = json::object();
+ res["enabled"] = enabled();
+ if (!enabled()) {
+ return res;
+ }
+
+ const int64_t now = wall_ms();
+ const int64_t since = now - window_s * 1000;
+
+ std::lock_guard<std::mutex> lock(smutex);
+
+ window_agg a;
+ json ttft, pp, tg;
+ const bool exact = ring_complete_ms <= since;
+ if (exact) {
+ std::vector<double> v_ttft, v_pp, v_tg;
+ for (const auto & r : ring) {
+ if (r.t_wall_ms < since) {
+ continue;
+ }
+ a.n++;
+ a.n_cancelled += r.status == "cancelled";
+ a.n_error += r.status == "error";
+ a.prompt += r.n_prompt; a.cached += r.n_cached; a.scr += r.n_scr; a.prefilled += r.n_prefilled; a.gen += r.n_gen;
+ a.cached_slot += r.tier == "slot" ? r.n_cached : 0;
+ a.cached_ram += r.tier == "ram" ? r.n_cached : 0;
+ a.cached_disk += r.tier == "disk" ? r.n_cached : 0;
+ a.images += r.n_images; a.images_cached += r.n_images_cached;
+ a.image_ms += r.image_ms; a.image_encode_ms += r.image_encode_ms;
+ a.draft += r.n_draft; a.draft_accepted += r.n_draft_accepted;
+ if (counts_ttft(r)) v_ttft.push_back(r.ttft_ms);
+ if (counts_prefill(r)) v_pp.push_back(r.prefill_tps);
+ if (counts_decode(r)) v_tg.push_back(r.decode_tps);
+ }
+ ttft = percentiles(v_ttft, { 0.5, 0.9 });
+ pp = percentiles(v_pp, { 0.5 });
+ tg = percentiles(v_tg, { 0.5 });
+ } else {
+ // whole hours: the window is rounded out to the hour that contains its start
+ std::vector<uint64_t> h_ttft(HIST_BINS), h_pp(HIST_BINS), h_tg(HIST_BINS);
+ for (auto it = hours.lower_bound(since / HOUR_MS); it != hours.end(); ++it) {
+ const auto & b = it->second;
+ a.n += b.n; a.n_cancelled += b.n_cancelled; a.n_error += b.n_error;
+ a.prompt += b.prompt; a.cached += b.cached; a.scr += b.scr; a.prefilled += b.prefilled; a.gen += b.gen;
+ a.cached_slot += b.cached_slot; a.cached_ram += b.cached_ram; a.cached_disk += b.cached_disk;
+ a.images += b.images; a.images_cached += b.images_cached; a.image_ms += b.image_ms; a.image_encode_ms += b.image_encode_ms;
+ a.draft += b.draft; a.draft_accepted += b.draft_accepted;
+ for (int i = 0; i < HIST_BINS; ++i) {
+ h_ttft[i] += b.ttft[i]; h_pp[i] += b.pp[i]; h_tg[i] += b.tg[i];
+ }
+ }
+ ttft = hist_percentiles(h_ttft, { 0.5, 0.9 }, hist_value);
+ pp = hist_percentiles(h_pp, { 0.5 }, hist_value);
+ tg = hist_percentiles(h_tg, { 0.5 }, hist_value);
+ }
+
+ res["window_s"] = window_s;
+ res["source"] = exact ? "requests" : "hourly";
+ res["requests"] = a.n;
+ res["completed"] = a.n - a.n_cancelled - a.n_error;
+ res["cancelled"] = a.n_cancelled;
+ res["errors"] = a.n_error;
+ res["tokens"] = {
+ { "prompt", a.prompt },
+ { "cached", a.cached },
+ { "scr", a.scr },
+ { "prefilled", a.prefilled },
+ { "generated", a.gen },
+ { "cached_by_tier", { { "slot", a.cached_slot }, { "ram", a.cached_ram }, { "disk", a.cached_disk } } },
+ };
+ res["share"] = {
+ { "cached", share(a.cached, a.prompt) },
+ { "scr", share(a.scr, a.prompt) },
+ { "prefilled", share(a.prefilled, a.prompt) },
+ };
+ res["ttft_ms"] = ttft;
+ res["prefill_tps"] = pp;
+ res["decode_tps"] = tg;
+ res["draft"] = {
+ { "proposed", a.draft },
+ { "accepted", a.draft_accepted },
+ };
+ res["draft"]["acceptance"] = a.draft > 0 ? json(share(a.draft_accepted, a.draft)) : json(nullptr);
+ res["images"] = { { "count", a.images }, { "cached", a.images_cached }, { "ms", a.image_ms }, { "encode_ms", a.image_encode_ms } };
+
+ // per hour, over at least the last 24 hours
+ json series = json::array();
+ const int64_t h_now = now / HOUR_MS;
+ const int64_t n_h = std::max<int64_t>(24, (window_s + 3599) / 3600);
+ for (int64_t h = h_now - n_h + 1; h <= h_now; ++h) {
+ const auto it = hours.find(h);
+ json p = { { "t", h * HOUR_MS } };
+ if (it == hours.end()) {
+ p["requests"] = 0; p["prompt"] = 0; p["cached"] = 0; p["scr"] = 0; p["prefilled"] = 0;
+ } else {
+ const auto & b = it->second;
+ p["requests"] = b.n; p["prompt"] = b.prompt; p["cached"] = b.cached; p["scr"] = b.scr; p["prefilled"] = b.prefilled;
+ }
+ series.push_back(p);
+ }
+ res["series"] = series;
+
+ // minute samples within the window (24 h at most), averaged down to at most 240 points
+ json smp = json::array();
+ std::vector<const server_telemetry_sample *> in;
+ for (const auto & s : samples) {
+ if (s.t_wall_ms >= since) {
+ in.push_back(&s);
+ }
+ }
+ const size_t step = std::max<size_t>(1, (in.size() + 239) / 240);
+ for (size_t i = 0; i < in.size(); i += step) {
+ const auto & s = *in[std::min(in.size() - 1, i + step - 1)];
+ int64_t gmax = -1;
+ double busy = 0.0;
+ size_t n = 0;
+ for (size_t k = i; k < std::min(in.size(), i + step); ++k, ++n) {
+ gmax = std::max(gmax, in[k]->gpu_used_max);
+ busy += in[k]->slots_busy_avg;
+ }
+ json p = s.to_json();
+ p.erase("type");
+ p["gpu_used_max"] = gmax >= 0 ? json(gmax) : json(nullptr);
+ p["slots_busy_avg"] = n ? busy / n : 0.0;
+ smp.push_back(p);
+ }
+ res["samples"] = smp;
+ if (!samples.empty()) {
+ json last = samples.back().to_json();
+ last.erase("type");
+ res["latest_sample"] = last;
+ } else {
+ res["latest_sample"] = nullptr;
+ }
+
+ res["telemetry"] = {
+ { "dir", dir },
+ { "bytes", files_bytes },
+ { "limit_bytes", limit_bytes },
+ { "files", n_files },
+ { "written", n_written },
+ { "write_errors", n_write_errors },
+ { "dropped", n_dropped.load() },
+ { "replayed", n_replayed },
+ { "ring", (int64_t) ring.size() },
+ { "since_ms", t_first_ms },
+ };
+ return res;
+}
+
+json server_telemetry::requests(int32_t limit) {
+ json res = json::object();
+ res["enabled"] = enabled();
+ json arr = json::array();
+ if (enabled()) {
+ std::lock_guard<std::mutex> lock(smutex);
+ const size_t n = std::min<size_t>(ring.size(), (size_t) std::max(limit, 0));
+ for (size_t i = 0; i < n; ++i) {
+ json j = ring[ring.size() - 1 - i].to_json(); // newest first
+ j.erase("type");
+ arr.push_back(j);
+ }
+ }
+ res["requests"] = arr;
+ res["dropped"] = n_dropped.load();
+ return res;
+}
diff --git a/tools/server/server-telemetry.h b/tools/server/server-telemetry.h
new file mode 100644
index 00000000..f064b58c
--- /dev/null
+++ b/tools/server/server-telemetry.h
@@ -0,0 +1,192 @@
+#pragma once
+
+// request telemetry: one record per finished completion request and one sample per minute, kept in memory for
+// /stats and appended to JSONL files under --telemetry-dir. the server loop only moves a record into a bounded
+// queue; a writer thread owns the aggregates and every file operation
+
+#include "ggml-backend.h"
+
+#include "json.h"
+
+#include <atomic>
+#include <condition_variable>
+#include <cstdint>
+#include <deque>
+#include <map>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+using json = common_json;
+
+// what the HTTP thread knows about a request; travels on the task
+struct server_telemetry_hint {
+ int64_t t_admit_us = 0; // ggml_time_us() after tokenization, 0 = not a telemetry request
+ int64_t t_admit_wall_ms = 0;
+ std::string endpoint;
+ std::string user_agent;
+ std::string excerpt;
+};
+
+// stamps the admission time (call after tokenization) and keeps the endpoint, the User-Agent and the start of the
+// prompt's first message (usually its system prompt)
+server_telemetry_hint server_telemetry_hint_make(const std::map<std::string, std::string> & headers, const std::string & path,
+ const std::string & prompt);
+
+// "1h", "24h", "7d", "30m" or plain seconds; 0 if invalid
+int64_t server_telemetry_parse_window(const std::string & s);
+
+// at most n code points of s
+std::string utf8_prefix(const std::string & s, size_t n);
+
+struct server_telemetry_record {
+ int64_t t_wall_ms = 0; // admission, unix ms
+ int32_t id_slot = -1;
+ std::string id; // the response id (chatcmpl-...)
+ std::string endpoint;
+ std::string user_agent;
+ std::string excerpt; // start of the prompt's first message (usually the system prompt), at most 40 characters
+
+ std::string status; // completed, cancelled or error
+ std::string reason; // finish reason (stop, word, length), the cancel phase, or the error message
+
+ int32_t n_prompt = 0;
+ int32_t n_cached = 0; // reused exactly
+ std::string tier; // where the exact reuse came from: none, slot, ram or disk
+ int32_t n_scr = 0; // reused through a splice
+ int32_t n_prefilled = 0; // evaluated, image tokens included
+ bool scr = false; // the request allowed SCR
+
+ int32_t n_images = 0;
+ int32_t n_images_encoded = 0; // run through the vision encoder
+ int32_t n_images_cached = 0; // taken from the image embedding cache
+ double image_ms = 0.0; // encode (or cache read) and the decode of the image tokens, all images
+ double image_encode_ms = 0.0; // the vision encoder alone, 0 when every image came from the cache
+
+ double queue_ms = 0.0;
+ double ttft_ms = -1.0; // admission to the first sampled token, -1 without one
+ double prefill_ms = 0.0; // the response's timings.prompt_ms
+ double prefill_tps = 0.0;
+ int32_t n_gen = 0;
+ double decode_ms = 0.0;
+ double decode_tps = 0.0; // the response's timings.predicted_per_second
+ int32_t n_draft = 0;
+ int32_t n_draft_accepted = 0;
+ int32_t n_cmpl = 1;
+
+ std::string cache_class;
+ std::string cache_source;
+ uint64_t family = 0;
+
+ json to_json() const;
+ static bool from_json(const json & j, server_telemetry_record & out);
+};
+
+// what the server loop knows; copied into every minute sample
+struct server_telemetry_gauges {
+ int32_t slots_busy = 0;
+ int32_t slots_total = 0;
+ int64_t ram_bytes = -1;
+ int64_t ram_limit = 0;
+ int64_t disk_bytes = -1; // -1 = off or not known yet
+ int64_t disk_limit = 0;
+};
+
+struct server_telemetry_sample {
+ int64_t t_wall_ms = 0;
+ int64_t gpu_used = -1; // bytes, -1 = not read
+ int64_t gpu_total = -1;
+ int64_t gpu_used_max = -1; // highest reading in the minute
+ double slots_busy_avg = 0.0;
+ server_telemetry_gauges g;
+
+ json to_json() const;
+ static bool from_json(const json & j, server_telemetry_sample & out);
+};
+
+class server_telemetry {
+public:
+ ~server_telemetry();
+
+ // starts the writer thread; an empty dir leaves telemetry off. dev is read for GPU memory (may be null)
+ void start(const std::string & dir, int64_t limit_mib, ggml_backend_dev_t dev);
+ void stop();
+
+ bool enabled() const { return running.load(); }
+
+ // server loop: never waits for disk or for readers, a full queue drops the record and counts it
+ void push(server_telemetry_record && rec);
+ void set_gauges(const server_telemetry_gauges & g);
+ // a GPU memory reading taken on the server loop while it is busy
+ void note_gpu();
+
+ // HTTP threads
+ json stats(int64_t window_s);
+ json requests(int32_t limit);
+
+private:
+ static constexpr size_t RING_REQUESTS = 2000;
+ static constexpr size_t RING_SAMPLES = 24*60;
+ static constexpr size_t QUEUE_MAX = 4096;
+ static constexpr int HOURS_KEPT = 31*24;
+ static constexpr int HIST_BINS = 192;
+
+ struct hour_bucket {
+ int64_t n = 0, n_cancelled = 0, n_error = 0;
+ int64_t prompt = 0, cached = 0, scr = 0, prefilled = 0, gen = 0;
+ int64_t cached_slot = 0, cached_ram = 0, cached_disk = 0;
+ int64_t images = 0, images_cached = 0, draft = 0, draft_accepted = 0;
+ double image_ms = 0.0, image_encode_ms = 0.0;
+ std::vector<uint32_t> ttft, pp, tg; // log histograms, HIST_BINS each
+ };
+
+ std::string dir;
+ int64_t limit_bytes = 0;
+ ggml_backend_dev_t dev = nullptr;
+
+ std::atomic<bool> running { false };
+ std::thread thread;
+
+ // producer side: the server loop holds it only to move a record or copy the gauges
+ std::mutex qmutex;
+ std::condition_variable qcv;
+ std::deque<server_telemetry_record> queue;
+ server_telemetry_gauges gauges;
+ int64_t busy_us = 0; // slot-busy integral since the last sample
+ int64_t t_gauges = 0;
+ int64_t gpu_used_last = -1, gpu_total_last = -1, gpu_used_max = -1, t_gpu_us = 0;
+ bool quit = false;
+ std::atomic<uint64_t> n_dropped { 0 };
+
+ // state side: the writer and readers
+ std::mutex smutex;
+ std::deque<server_telemetry_record> ring;
+ std::deque<server_telemetry_sample> samples;
+ std::map<int64_t, hour_bucket> hours;
+ int64_t ring_complete_ms = 0; // the ring holds every record admitted since this time
+ int64_t t_first_ms = 0; // the oldest record ever seen (also from replayed files)
+ uint64_t n_written = 0;
+ uint64_t n_write_errors = 0;
+ uint64_t n_replayed = 0;
+ int64_t files_bytes = 0;
+ int32_t n_files = 0;
+
+ // writer thread only
+ FILE * file = nullptr;
+ std::string file_path;
+ int64_t file_size = 0;
+
+ void loop();
+ void replay();
+ void add_locked(const server_telemetry_record & r);
+ void add_sample_locked(const server_telemetry_sample & s);
+ void write_lines(const std::string & lines);
+ void open_segment();
+ void rotate();
+ void enforce_cap();
+ server_telemetry_sample take_sample(int64_t t_wall_ms);
+
+ static int hist_bin(double v);
+ static double hist_value(int bin);
+};
diff --git a/tools/server/server.cpp b/tools/server/server.cpp
index dc13a51c..ab93577c 100644
--- a/tools/server/server.cpp
+++ b/tools/server/server.cpp
@@ -222,6 +222,8 @@ int llama_server(common_params & params, int argc, char ** argv) {
routes.get_cache = models_routes->proxy_get;
routes.post_cache_clear = models_routes->proxy_post;
routes.post_cache_entry = models_routes->proxy_post;
+ routes.get_stats = models_routes->proxy_get;
+ routes.get_stats_requests = models_routes->proxy_get;
// custom routes for router
routes.get_props = models_routes->get_router_props;
@@ -278,6 +280,8 @@ int llama_server(common_params & params, int argc, char ** argv) {
ctx_http.get ("/cache", ex_wrapper(routes.get_cache));
ctx_http.post("/cache/clear", ex_wrapper(routes.post_cache_clear));
ctx_http.post("/cache/entry", ex_wrapper(routes.post_cache_entry));
+ ctx_http.get ("/stats", ex_wrapper(routes.get_stats));
+ ctx_http.get ("/stats/requests", ex_wrapper(routes.get_stats_requests));
// resumable streaming: a child binds the local session factories, the router binds
// proxies that resolve the owning child, see server-stream.h