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 = /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; + // 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(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 telemetry_disk = { -1, 0 }; + std::vector slot_metas() { std::vector 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(); 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_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()); + } + // 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 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::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(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 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 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 +#include +#include +#include +#include +#include +#include +#include + +#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::system_clock::now().time_since_epoch()).count(); +} + +static int64_t mono_us() { + return std::chrono::duration_cast(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 & 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(); + 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() : -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() : -1; + }; + try { + s.t_wall_ms = j.at("t").get(); + 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(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 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 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 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 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(clock::now().time_since_epoch()).count() + 1)); + + std::deque batch; + while (true) { + bool stopping; + { + std::unique_lock 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 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 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 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-.jsonl segments of at most a 1/8 of the limit, the oldest go first +// + +static std::vector list_segments(const std::string & dir) { + std::vector 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 recs; + std::vector 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 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 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 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 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 lock(smutex); + n_write_errors += !ok; + n_written += ok; + files_bytes += (int64_t) lines.size(); + } + if (file_size >= std::max(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 v, std::initializer_list 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 & h, std::initializer_list 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(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 lock(smutex); + + window_agg a; + json ttft, pp, tg; + const bool exact = ring_complete_ms <= since; + if (exact) { + std::vector 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 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(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 in; + for (const auto & s : samples) { + if (s.t_wall_ms >= since) { + in.push_back(&s); + } + } + const size_t step = std::max(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 lock(smutex); + const size_t n = std::min(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 +#include +#include +#include +#include +#include +#include +#include +#include + +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 & 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 ttft, pp, tg; // log histograms, HIST_BINS each + }; + + std::string dir; + int64_t limit_bytes = 0; + ggml_backend_dev_t dev = nullptr; + + std::atomic 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 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 n_dropped { 0 }; + + // state side: the writer and readers + std::mutex smutex; + std::deque ring; + std::deque samples; + std::map 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