You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@trafficserver.apache.org by "moonchen (via GitHub)" <gi...@apache.org> on 2026/09/29 22:20:23 UTC
[PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
moonchen opened a new pull request, #13754:
URL: https://github.com/apache/trafficserver/pull/13754
Lets stats_over_http serve its statistics from a remap rule, and makes its Prometheus output cheap to render and configurable enough to scrape each host directly. A remap rule serves one format, its ACL filters control access, and it answers methods other than GET and HEAD with a 405. The plugin renders on a task thread instead of an event thread, and for the Prometheus formats it translates each metric name once and caches the result, so a scrape only formats the values. A YAML file passed with `--config` sets how metric names become Prometheus families, labels and types, so the output can keep the names and labels that existing dashboards and alerts use. Gauges now print as signed values, where a negative gauge used to appear as a number near 2^64.
The existing Prometheus formats fix the names and labels in code, render on an event thread, and translate every name on every scrape. A site that needs other names has to run a separate process that fetches the JSON statistics and translates them, which can cost far more CPU than rendering them once in the process that owns them. Rendering thousands of metrics also holds an event thread for milliseconds, which delays the other requests that thread serves.
The first commit is #13753, which fixes a crash in the compressed responses; it should merge first.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#discussion_r4169514920
##########
plugins/stats_over_http/stats_over_http.cc:
##########
@@ -1508,6 +1289,911 @@ config_handler(TSCont cont, TSEvent /* event ATS_UNUSED */, void * /* edata ATS_
return 0;
}
+//
+// Requests for the stats. A render on a task thread answers the requests to the global plugin and to remap rules.
+//
+
+static constexpr std::string_view STATS_FORMAT_FIELD = "X-Stats-Format";
+static constexpr std::string_view ALLOWED_METHODS = "GET, HEAD";
+
+static const char REMAP_USAGE[] = "[--format=json|csv|prometheus|prometheus_v2] [--integer-counters] [--wrap-counters] "
+ "[--no-prometheus-help] [--max-age-ms=N] [--wait-timeout-ms=N] [--config=FILE] "
+ "[--on-config-error=fail|503]";
+
+static std::string_view
+format_name(output_format_t format)
+{
+ switch (format) {
+ case output_format_t::JSON_OUTPUT:
+ return "json";
+ case output_format_t::CSV_OUTPUT:
+ return "csv";
+ case output_format_t::PROMETHEUS_OUTPUT:
+ return "prometheus";
+ case output_format_t::PROMETHEUS_V2_OUTPUT:
+ return "prometheus_v2";
+ }
+ return "json";
+}
+
+static std::string_view
+format_content_type(output_format_t format)
+{
+ switch (format) {
+ case output_format_t::JSON_OUTPUT:
+ return "text/json";
+ case output_format_t::CSV_OUTPUT:
+ return "text/csv";
+ case output_format_t::PROMETHEUS_OUTPUT:
+ return "text/plain; version=0.0.4; charset=utf-8";
+ case output_format_t::PROMETHEUS_V2_OUTPUT:
+ return "text/plain; version=2.0.0; charset=utf-8";
+ }
+ return "text/json";
+}
+
+static std::string_view
+encoding_name(encoding_format_t encoding)
+{
+ switch (encoding) {
+ case encoding_format_t::DEFLATE:
+ return "deflate";
+ case encoding_format_t::GZIP:
+ return "gzip";
+ case encoding_format_t::BR:
+ return "br";
+ case encoding_format_t::NONE:
+ break;
+ }
+ return {};
+}
+
+static bool
+parse_format(std::string_view name, output_format_t &format)
+{
+ for (auto candidate : {output_format_t::JSON_OUTPUT, output_format_t::CSV_OUTPUT, output_format_t::PROMETHEUS_OUTPUT,
+ output_format_t::PROMETHEUS_V2_OUTPUT}) {
+ if (name == format_name(candidate)) {
+ format = candidate;
+ return true;
+ }
+ }
+ return false;
+}
+
+// A rendered body in one encoding. It does not change after the render publishes it, so requests on any thread share its
+// blocks.
+struct stats_snapshot {
+ stats_snapshot() : body(TSIOBufferCreate()), reader(TSIOBufferReaderAlloc(body)) {}
+ ~stats_snapshot() { TSIOBufferDestroy(body); }
+ stats_snapshot(const stats_snapshot &) = delete;
+ stats_snapshot &operator=(const stats_snapshot &) = delete;
+
+ TSIOBuffer body;
+ TSIOBufferReader reader;
+ int64_t bytes = 0;
+ encoding_format_t encoding = encoding_format_t::NONE;
+ std::chrono::steady_clock::time_point rendered;
+};
+
+using snapshot_ptr = std::shared_ptr<const stats_snapshot>;
+using snapshot_table = snapshot_ptr[FORMAT_COUNT][ENCODING_COUNT];
+
+// One request for the stats. The waiter list, the transaction hooks and the intercept share it.
+struct stats_scrape {
+ stats_scrape(TSHttpTxn txn, output_format_t fmt, encoding_format_t enc, bool rule)
+ : txnp(txn), format(fmt), encoding(enc), remap(rule)
+ {
+ }
+
+ TSHttpTxn txnp;
+ output_format_t format;
+ encoding_format_t encoding;
+ bool remap; // A remap rule serves the request, rather than the global plugin.
+ // When the request started to wait for a render.
+ std::chrono::steady_clock::time_point arrived;
+
+ // TSRemapDoRemap, scrape_wait, the render or the watchdog sets these before the transaction continues, and the intercept
+ // reads them only after that. The snapshot is null for a HEAD request to a remap rule and for a 503.
+ snapshot_ptr snapshot;
+ TSHttpStatus status = TS_HTTP_STATUS_OK;
+};
+
+struct stats_watchdog;
+
+// The stats of one remap rule, or of the global plugin. The remap rule and each continuation that works for the instance
+// hold a reference, so that a render or a watchdog can outlive the rule.
+struct stats_instance {
+ explicit stats_instance(const stats_options &opts) : options(opts) {}
+ ~stats_instance() { Dbg(dbg_ctl, "Freeing stats instance %p", this); }
+ stats_instance(const stats_instance &) = delete;
+ stats_instance &operator=(const stats_instance &) = delete;
+
+ PrometheusRenderer *
+ prometheus(output_format_t format)
+ {
+ std::unique_ptr<PrometheusRenderer> *renderer = nullptr;
+
+ if (format == output_format_t::PROMETHEUS_OUTPUT) {
+ renderer = &prometheus_v1;
+ } else if (format == output_format_t::PROMETHEUS_V2_OUTPUT) {
+ renderer = &prometheus_v2;
+ } else {
+ return nullptr;
+ }
+ if (*renderer == nullptr) {
+ *renderer = make_prometheus_renderer(format, options);
+ }
+ return renderer->get();
+ }
+
+ const stats_options options;
+ // Only the render in flight uses these, so the mutex does not guard them.
+ std::unique_ptr<PrometheusRenderer> prometheus_v1;
+ std::unique_ptr<PrometheusRenderer> prometheus_v2;
+ zlib_streams zlib;
+
+ // The mutex guards the members below it. No thread holds it while it calls an API that can run a continuation.
+ std::mutex mutex;
+ snapshot_table snapshots;
+ std::vector<std::shared_ptr<stats_scrape>> waiters;
+ bool rendering = false;
+ stats_watchdog *watchdog = nullptr; // Set while a request waits.
+};
+
+static std::shared_ptr<stats_instance>
+make_stats_instance(const stats_options &options)
+{
+ return std::make_shared<stats_instance>(options);
+}
+
+// Renders each wanted format once and compresses it for each wanted encoding.
+static void
+render_snapshots(stats_instance &instance, const bool (&wanted)[FORMAT_COUNT][ENCODING_COUNT], snapshot_table &fresh)
+{
+ auto const now = std::chrono::steady_clock::now();
+
+ for (size_t f = 0; f < FORMAT_COUNT; ++f) {
+ if (std::find(std::begin(wanted[f]), std::end(wanted[f]), true) == std::end(wanted[f])) {
+ continue;
+ }
+
+ auto const format = static_cast<output_format_t>(f);
+ auto body = std::make_shared<stats_snapshot>();
+ render_state render{body->body, &instance.options};
+
+ render_stats(format, &render, instance.prometheus(format));
+ body->bytes = TSIOBufferReaderAvail(body->reader);
+ body->rendered = now;
+
+ for (size_t e = 1; e < ENCODING_COUNT; ++e) {
+ if (!wanted[f][e]) {
+ continue;
+ }
+
+ auto const encoding = static_cast<encoding_format_t>(e);
+ auto compressed = std::make_shared<stats_snapshot>();
+
+ if (compress_body(encoding, instance.zlib, body->reader, compressed->body)) {
+ compressed->bytes = TSIOBufferReaderAvail(compressed->reader);
+ compressed->encoding = encoding;
+ compressed->rendered = now;
+ fresh[f][e] = std::move(compressed);
+ } else {
+ TSError("[%s] Cannot compress the stats, sending them uncompressed", PLUGIN_NAME);
+ fresh[f][e] = body;
+ }
+ }
+ fresh[f][static_cast<size_t>(encoding_format_t::NONE)] = std::move(body);
+ }
+}
+
+static int watchdog_handler(TSCont contp, TSEvent event, void *edata);
+
+// Answers the waiting requests with a 503 when the render takes longer than wait_timeout_ms.
+struct stats_watchdog {
+ explicit stats_watchdog(std::shared_ptr<stats_instance> inst)
+ : instance(std::move(inst)), cont(TSContCreate(watchdog_handler, TSMutexCreate()))
+ {
+ TSContDataSet(cont, this);
+ }
+
+ std::shared_ptr<stats_instance> instance;
+ TSCont cont;
+
+ // The mutex of the continuation guards these.
+ TSAction action = nullptr;
+ bool fired = false;
+};
+
+static int
+watchdog_handler(TSCont contp, TSEvent /* event ATS_UNUSED */, void * /* edata ATS_UNUSED */)
+{
+ auto *watchdog = static_cast<stats_watchdog *>(TSContDataGet(contp));
+ stats_instance &instance = *watchdog->instance;
+ std::vector<std::shared_ptr<stats_scrape>> expired;
+
+ watchdog->fired = true;
+ {
+ std::lock_guard lock{instance.mutex};
+
+ // A render took this watchdog or replaced it, and that render destroys it.
+ if (instance.watchdog != watchdog) {
+ return 0;
+ }
+ instance.watchdog = nullptr;
+ expired.swap(instance.waiters);
+ }
+
+ Dbg(dbg_ctl, "Answering %zu requests with a 503 after %" PRId64 " ms without a render of stats instance %p", expired.size(),
+ instance.options.wait_timeout_ms, &instance);
+ count(metrics().waiter_timeouts, expired.size());
+ for (auto const &scrape : expired) {
+ scrape->status = TS_HTTP_STATUS_SERVICE_UNAVAILABLE;
+ TSHttpTxnReenable(scrape->txnp, TS_EVENT_HTTP_CONTINUE);
+ }
+ TSContDestroy(contp);
+ delete watchdog;
+ return 0;
+}
+
+// Stops and destroys a watchdog that a render took from its instance.
+static void
+stop_watchdog(stats_watchdog *watchdog)
+{
+ TSMutex mutex = TSContMutexGet(watchdog->cont);
+
+ // Only the holder of its continuation's mutex can cancel an action.
+ TSMutexLock(mutex);
+ if (!watchdog->fired) {
+ TSActionCancel(watchdog->action);
+ }
+ TSMutexUnlock(mutex);
+ TSContDestroy(watchdog->cont);
+ delete watchdog;
+}
+
+// Creates a watchdog for the caller to publish under the instance mutex. The calling thread holds the mutex of the watchdog
+// until arm_watchdog sets its action, so that a render that finishes first can cancel that action.
+static stats_watchdog *
+new_watchdog(std::shared_ptr<stats_instance> instance)
+{
+ auto *watchdog = new stats_watchdog(std::move(instance));
+
+ TSMutexLock(TSContMutexGet(watchdog->cont));
+ return watchdog;
+}
+
+static void
+arm_watchdog(stats_watchdog *watchdog)
+{
+ TSMutex mutex = TSContMutexGet(watchdog->cont);
+
+ watchdog->action = TSContScheduleOnPool(watchdog->cont, watchdog->instance->options.wait_timeout_ms, TS_THREAD_POOL_NET);
+ TSMutexUnlock(mutex);
+}
+
+static int render_handler(TSCont contp, TSEvent event, void *edata);
+
+// Each render gets a new continuation, because TSContScheduleOnPool locks the mutex of the continuation on the calling
+// thread, and render_handler holds that mutex until it returns. A shared continuation could block the ET_NET thread that
+// schedules the next render.
+static void
+schedule_render(std::shared_ptr<stats_instance> instance)
+{
+ TSCont contp = TSContCreate(render_handler, TSMutexCreate());
+
+ TSContDataSet(contp, new std::shared_ptr<stats_instance>(std::move(instance)));
+ TSContScheduleOnPool(contp, 0, TS_THREAD_POOL_TASK);
+}
+
+static int
+render_handler(TSCont contp, TSEvent /* event ATS_UNUSED */, void * /* edata ATS_UNUSED */)
+{
+ auto *ref = static_cast<std::shared_ptr<stats_instance> *>(TSContDataGet(contp));
+ stats_instance &instance = **ref;
+ bool wanted[FORMAT_COUNT][ENCODING_COUNT] = {};
+ bool waiting = false;
+ snapshot_table fresh;
+
+ {
+ std::lock_guard lock{instance.mutex};
+
+ for (auto const &scrape : instance.waiters) {
+ wanted[static_cast<size_t>(scrape->format)][static_cast<size_t>(scrape->encoding)] = true;
+ waiting = true;
+ }
+ }
+ // The watchdog can answer every waiting request before the render starts.
+ if (waiting) {
+ static thread_local int64_t rest = 0;
+ int64_t const start = thread_cpu_ns();
+
+ render_snapshots(instance, wanted, fresh);
+ count(metrics().renders);
+ count_cpu_time(metrics().render_us, start, rest);
+ }
+
+ std::vector<std::shared_ptr<stats_scrape>> ready;
+ stats_watchdog *watchdog = nullptr;
+ stats_watchdog *replacement = nullptr;
+ bool again = false;
+
+ {
+ std::lock_guard lock{instance.mutex};
+
+ for (size_t f = 0; f < FORMAT_COUNT; ++f) {
+ for (size_t e = 0; e < ENCODING_COUNT; ++e) {
+ if (fresh[f][e] != nullptr) {
+ instance.snapshots[f][e] = fresh[f][e];
+ }
+ }
+ }
+
+ // As in scrape_wait, a request takes a render only if the render is younger than max_age_ms when the request arrives.
+ auto const max_age = std::chrono::milliseconds{instance.options.max_age_ms};
+ auto rendered = [&fresh, max_age](const std::shared_ptr<stats_scrape> &scrape) -> snapshot_ptr {
+ auto const &snapshot = fresh[static_cast<size_t>(scrape->format)][static_cast<size_t>(scrape->encoding)];
+
+ return snapshot != nullptr && scrape->arrived - snapshot->rendered < max_age ? snapshot : nullptr;
+ };
+ auto answered = std::stable_partition(instance.waiters.begin(), instance.waiters.end(),
+ [&rendered](const auto &scrape) { return rendered(scrape) == nullptr; });
+
+ for (auto it = answered; it != instance.waiters.end(); ++it) {
+ (*it)->snapshot = rendered(*it);
+ ready.push_back(std::move(*it));
+ }
+ instance.waiters.erase(answered, instance.waiters.end());
+
+ // Requests for another format or encoding, and requests that arrived too late for this render, wait for the next one. They
+ // get a new watchdog, because the current one can have started long before they arrived.
+ again = !instance.waiters.empty();
+ if (again) {
+ replacement = new_watchdog(*ref);
+ } else {
+ instance.rendering = false;
+ }
+ watchdog = std::exchange(instance.watchdog, replacement);
+ }
+
+ if (waiting) {
+ Dbg(dbg_ctl, "Rendered stats instance %p for %zu waiting requests", &instance, ready.size());
+ }
+ if (watchdog != nullptr) {
+ stop_watchdog(watchdog);
+ }
+ if (replacement != nullptr) {
+ arm_watchdog(replacement);
+ }
+ for (auto const &scrape : ready) {
+ TSHttpTxnReenable(scrape->txnp, TS_EVENT_HTTP_CONTINUE);
+ }
+ if (again) {
+ schedule_render(*ref);
+ }
+ delete ref;
+ TSContDestroy(contp);
+ return 0;
+}
+
+// Answers a request from a snapshot younger than max_age_ms. Otherwise the request waits for a render, which starts unless
+// one is in flight. Either way, the transaction continues once it has its answer.
+static void
+scrape_wait(const std::shared_ptr<stats_instance> &instance, std::shared_ptr<stats_scrape> scrape)
+{
+ auto const now = std::chrono::steady_clock::now();
+ bool waits = false;
+ bool render = false;
+ stats_watchdog *watchdog = nullptr;
+
+ {
+ std::lock_guard lock{instance->mutex};
+ auto const &snapshot = instance->snapshots[static_cast<size_t>(scrape->format)][static_cast<size_t>(scrape->encoding)];
+
+ if (snapshot != nullptr && now - snapshot->rendered < std::chrono::milliseconds{instance->options.max_age_ms}) {
+ scrape->snapshot = snapshot;
+ } else {
+ scrape->arrived = now;
+ instance->waiters.push_back(scrape);
+ waits = true;
+ render = !std::exchange(instance->rendering, true);
+ if (instance->watchdog == nullptr) {
+ watchdog = instance->watchdog = new_watchdog(instance);
+ }
+ }
+ }
+
+ if (!waits) {
+ TSHttpTxnReenable(scrape->txnp, TS_EVENT_HTTP_CONTINUE);
+ return;
+ }
+
+ Dbg(dbg_ctl, "Waiting for a render of stats instance %p", instance.get());
+ if (watchdog != nullptr) {
+ arm_watchdog(watchdog);
+ }
+ if (render) {
+ schedule_render(instance);
+ }
+}
+
+// The intercept holds only its scrape, because intercept events can arrive after the transaction and its remap rule are gone.
+struct scrape_intercept {
+ explicit scrape_intercept(std::shared_ptr<stats_scrape> s) : scrape(std::move(s)) {}
+ ~scrape_intercept()
+ {
+ if (net_vc != nullptr) {
+ TSVConnClose(net_vc);
+ }
+ if (req_buffer != nullptr) {
+ TSIOBufferDestroy(req_buffer);
+ }
+ if (resp_buffer != nullptr) {
+ TSIOBufferDestroy(resp_buffer);
+ }
+ }
+ scrape_intercept(const scrape_intercept &) = delete;
+ scrape_intercept &operator=(const scrape_intercept &) = delete;
+
+ std::shared_ptr<stats_scrape> scrape;
+ TSVConn net_vc = nullptr;
+ TSVIO write_vio = nullptr;
+ TSIOBuffer req_buffer = nullptr;
+ TSIOBuffer resp_buffer = nullptr;
+ TSIOBufferReader resp_reader = nullptr;
+};
+
+static std::string
+scrape_response_header(const stats_scrape &scrape)
+{
+ if (!scrape.remap) {
+ if (scrape.status != TS_HTTP_STATUS_OK) {
+ return RESP_HEADER_UNAVAILABLE;
+ }
+
+ std::string header{"HTTP/1.0 200 OK\r\nContent-Type: "};
+
+ header.append(format_content_type(scrape.format)).append("\r\n");
+ if (auto const name = encoding_name(scrape.snapshot->encoding); !name.empty()) {
+ header.append("Content-Encoding: ").append(name).append("\r\n");
+ }
+ header.append("Cache-Control: no-cache\r\n\r\n");
+ return header;
+ }
+
+ std::string header{"HTTP/1.1 "};
+
+ header.append(std::to_string(scrape.status)).append(" ").append(TSHttpHdrReasonLookup(scrape.status)).append("\r\n");
+ if (scrape.status == TS_HTTP_STATUS_OK) {
+ header.append("Content-Type: ").append(format_content_type(scrape.format)).append("\r\n");
+ }
+ header.append("Cache-Control: no-store\r\n");
+ header.append(STATS_FORMAT_FIELD).append(": ").append(format_name(scrape.format)).append("\r\n");
+ if (scrape.snapshot != nullptr || scrape.status != TS_HTTP_STATUS_OK) {
+ header.append("Content-Length: ")
+ .append(std::to_string(scrape.snapshot != nullptr ? scrape.snapshot->bytes : 0))
+ .append("\r\n");
+ }
+ header.append("\r\n");
+ return header;
+}
+
+static void
+scrape_send_response(TSCont contp, scrape_intercept *intercept)
+{
+ const stats_scrape &scrape = *intercept->scrape;
+ std::string const header = scrape_response_header(scrape);
+
+ intercept->resp_buffer = TSIOBufferCreate();
+ intercept->resp_reader = TSIOBufferReaderAlloc(intercept->resp_buffer);
+
+ int64_t bytes = TSIOBufferWrite(intercept->resp_buffer, header.data(), header.size());
+
+ if (scrape.snapshot != nullptr) {
+ // This shares the blocks of the snapshot rather than copying its data.
+ bytes += TSIOBufferCopy(intercept->resp_buffer, scrape.snapshot->reader, scrape.snapshot->bytes, 0);
+ }
+ TSVConnShutdown(intercept->net_vc, 1, 0);
+ intercept->write_vio = TSVConnWrite(intercept->net_vc, contp, intercept->resp_reader, bytes);
+ count(metrics().requests);
+ count(metrics().bytes_out, bytes);
+}
+
+static int
+scrape_intercept_handler(TSCont contp, TSEvent event, void *edata)
+{
+ static thread_local int64_t rest = 0;
+ int64_t const start = thread_cpu_ns();
+ auto *intercept = static_cast<scrape_intercept *>(TSContDataGet(contp));
+
+ switch (event) {
+ case TS_EVENT_NET_ACCEPT:
+ intercept->net_vc = static_cast<TSVConn>(edata);
+ intercept->req_buffer = TSIOBufferCreate();
+ TSVConnRead(intercept->net_vc, contp, intercept->req_buffer, INT64_MAX);
+ break;
+ case TS_EVENT_VCONN_READ_READY:
+ scrape_send_response(contp, intercept);
+ break;
+ case TS_EVENT_VCONN_WRITE_READY:
+ TSVIOReenable(intercept->write_vio);
+ break;
+ case TS_EVENT_NET_ACCEPT_FAILED:
+ case TS_EVENT_VCONN_EOS:
+ case TS_EVENT_ERROR:
+ case TS_EVENT_VCONN_INACTIVITY_TIMEOUT:
+ case TS_EVENT_VCONN_ACTIVE_TIMEOUT:
+ case TS_EVENT_VCONN_WRITE_COMPLETE:
+ Dbg(dbg_ctl, "Intercept finished on %s", TSHttpEventNameLookup(event));
+ delete intercept;
+ TSContDestroy(contp);
+ break;
+ default:
+ TSReleaseAssert(!"Unexpected Event");
+ }
+ count_cpu_time(metrics().intercept_us, start, rest);
+ return 0;
+}
+
+// The global plugin intercepts the request in its read request header hook, then waits there for the stats.
+static void
+serve_global_scrape(TSHttpTxn txnp, output_format_t format, encoding_format_t encoding)
+{
+ auto scrape = std::make_shared<stats_scrape>(txnp, format, encoding, false);
+ TSCont intercept_cont = TSContCreate(scrape_intercept_handler, TSMutexCreate());
+
+ TSContDataSet(intercept_cont, new scrape_intercept(scrape));
+ TSHttpTxnIntercept(intercept_cont, txnp);
+ scrape_wait(*global_instance, std::move(scrape));
+}
+
+//
+// Remap plugin.
+//
+
+// The transaction hooks of a GET request to a remap rule.
+struct scrape_txn {
+ std::shared_ptr<stats_instance> instance;
+ std::shared_ptr<stats_scrape> scrape;
+};
+
+static int
+scrape_txn_handler(TSCont contp, TSEvent event, void *edata)
+{
+ auto *txn = static_cast<scrape_txn *>(TSContDataGet(contp));
+
+ if (event == TS_EVENT_HTTP_CACHE_LOOKUP_COMPLETE) {
+ scrape_wait(txn->instance, txn->scrape);
+ return 0;
+ }
+ if (event == TS_EVENT_HTTP_TXN_CLOSE) {
+ delete txn;
+ TSContDestroy(contp);
+ }
+ TSHttpTxnReenable(static_cast<TSHttpTxn>(edata), TS_EVENT_HTTP_CONTINUE);
+ return 0;
+}
+
+static void
+add_allow_field(TSHttpTxn txnp)
+{
+ TSMBuffer bufp;
+ TSMLoc hdr_loc;
+
+ if (TSHttpTxnClientRespGet(txnp, &bufp, &hdr_loc) == TS_SUCCESS) {
+ TSMLoc field_loc;
+
+ // A remap ACL filter that denies the request replaces the 405 with a 403.
+ if (TSHttpHdrStatusGet(bufp, hdr_loc) == TS_HTTP_STATUS_METHOD_NOT_ALLOWED &&
+ TSMimeHdrFieldCreateNamed(bufp, hdr_loc, TS_MIME_FIELD_ALLOW, TS_MIME_LEN_ALLOW, &field_loc) == TS_SUCCESS) {
+ TSMimeHdrFieldValueStringSet(bufp, hdr_loc, field_loc, -1, ALLOWED_METHODS.data(), ALLOWED_METHODS.size());
+ TSMimeHdrFieldAppend(bufp, hdr_loc, field_loc);
+ TSHandleMLocRelease(bufp, hdr_loc, field_loc);
+ }
+ TSHandleMLocRelease(bufp, TS_NULL_MLOC, hdr_loc);
+ }
+}
+
+static int
+allow_handler(TSCont contp, TSEvent event, void *edata)
+{
+ auto txnp = static_cast<TSHttpTxn>(edata);
+
+ if (event == TS_EVENT_HTTP_SEND_RESPONSE_HDR) {
+ add_allow_field(txnp);
+ } else if (event == TS_EVENT_HTTP_TXN_CLOSE) {
+ TSContDestroy(contp);
+ }
+ TSHttpTxnReenable(txnp, TS_EVENT_HTTP_CONTINUE);
+ return 0;
+}
+
+// The settings that a remap rule or its configuration file can set.
+struct stats_settings {
+ std::optional<output_format_t> format;
+ std::optional<int64_t> max_age_ms;
+ std::optional<int64_t> wait_timeout_ms;
+ std::optional<bool> prometheus_help;
+ std::shared_ptr<const PrometheusRules> rules;
+};
+
+static std::string
+yaml_line(const YAML::Node &node)
+{
+ auto const mark = node.Mark();
+
+ return mark.is_null() ? std::string{} : "line " + std::to_string(mark.line + 1) + ": ";
+}
+
+// Reads the configuration file of a remap rule. Returns an empty string, or a description of the first error.
+static std::string
+load_settings_file(const std::string &path, stats_settings &settings)
+{
+ std::ifstream file{path};
+
+ if (!file) {
+ return std::string{"cannot open the file: "} + strerror(errno);
+ }
+ try {
+ YAML::Node const root = YAML::Load(file);
+
+ if (!root.IsMap()) {
+ return yaml_line(root) + "the file must be a map";
+ }
+ for (auto const &item : root) {
+ std::string const key = item.first.as<std::string>();
+ YAML::Node const value = item.second;
+
+ if (key == "format") {
+ output_format_t format;
+
+ if (!value.IsScalar() || !parse_format(value.Scalar(), format)) {
+ return yaml_line(value) + "format must be json, csv, prometheus or prometheus_v2";
+ }
+ settings.format = format;
+ } else if (key == "render") {
+ if (!value.IsMap()) {
+ return yaml_line(value) + "render must be a map";
+ }
+ for (auto const &setting : value) {
+ std::string const name = setting.first.as<std::string>();
+ std::string const text = setting.second.IsScalar() ? setting.second.Scalar() : std::string{};
+ int64_t number = 0;
+ int64_t min = 0;
+ bool valid = false;
+
+ if (name == "max_age_ms") {
+ valid = parse_integer(text, min, MAX_MILLISECONDS, number);
+ settings.max_age_ms = number;
+ } else if (name == "wait_timeout_ms") {
+ min = 1;
+ valid = parse_integer(text, min, MAX_MILLISECONDS, number);
+ settings.wait_timeout_ms = number;
+ } else {
+ return yaml_line(setting.first) + "unknown key render." + name;
+ }
+ if (!valid) {
+ return yaml_line(setting.second) + "render." + name + " must be an integer from " + std::to_string(min) + " to " +
+ std::to_string(MAX_MILLISECONDS) + ", not '" + text + "'";
+ }
+ }
+ } else if (key == "prometheus") {
+ auto rules = std::make_shared<PrometheusRules>();
+
+ if (std::string error = rules->load(value); !error.empty()) {
+ return error;
+ }
+ settings.prometheus_help = rules->help();
+ settings.rules = std::move(rules);
+ } else {
+ return yaml_line(item.first) + "unknown key " + key;
+ }
+ }
+ } catch (const YAML::Exception &e) {
+ return e.what();
+ }
+ return {};
+}
+
+// Registers the configuration file of a remap rule as a child of the remap configuration. After a change to the file, a
+// configuration reload loads the remap configuration again, and with it the file.
+static void
+watch_config_file(const std::string &path)
+{
+ TSMgmtString parent = nullptr;
+
+ if (TSMgmtStringGet("proxy.config.url_remap.filename", &parent) == TS_SUCCESS) {
+ TSMgmtConfigFileAdd(parent, path.c_str());
+ } else {
+ TSWarning("[%s] Cannot read proxy.config.url_remap.filename, so a configuration reload does not detect a change to %s",
+ PLUGIN_NAME, path.c_str());
+ }
+ TSfree(parent);
+}
+
+TSReturnCode
+TSRemapInit(TSRemapInterface * /* api_info ATS_UNUSED */, char * /* errbuf ATS_UNUSED */, int /* errbuf_size ATS_UNUSED */)
+{
+ metrics();
+ return TS_SUCCESS;
+}
+
+TSReturnCode
+TSRemapNewInstance(int argc, char *argv[], void **ih, char *errbuf, int errbuf_size)
+{
+ static const struct option longopts[] = {
+ {"format", required_argument, nullptr, 'f'},
+ {"integer-counters", no_argument, nullptr, 'i'},
+ {"wrap-counters", no_argument, nullptr, 'w'},
+ {"no-prometheus-help", no_argument, nullptr, 'n'},
+ {"max-age-ms", required_argument, nullptr, 'a'},
+ {"wait-timeout-ms", required_argument, nullptr, 't'},
+ {"config", required_argument, nullptr, 'c'},
+ {"on-config-error", required_argument, nullptr, 'e'},
+ {nullptr, 0, nullptr, 0 }
+ };
+ stats_options options;
+ stats_settings rule;
+ std::string config_path;
+ bool answer_errors = false;
+ int64_t number = 0;
+
+ options.prometheus_epoch = false;
+
+ // argv[0] is the "from" URL. Skip it so that the "to" URL poses as the program name.
+ --argc;
+ ++argv;
+ for (int opt, option_index = 0; (opt = getopt_long(argc, argv, "", longopts, &option_index)) != -1;) {
Review Comment:
`getopt_long()` retains process-global parser state, but this entry point runs once per remap rule. After the first instance (or `TSPluginInit`) leaves `optind` at the end of its argv, later rules can skip options such as `--format` and `--config` or parse from the wrong position. Reset getopt before every instance parse, as the ESI and statichit remap plugins do.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "moonchen (via GitHub)" <gi...@apache.org>.
moonchen commented on code in PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#discussion_r4189264104
##########
plugins/stats_over_http/stats_over_http.cc:
##########
@@ -1508,6 +1289,911 @@ config_handler(TSCont cont, TSEvent /* event ATS_UNUSED */, void * /* edata ATS_
return 0;
}
+//
+// Requests for the stats. A render on a task thread answers the requests to the global plugin and to remap rules.
+//
+
+static constexpr std::string_view STATS_FORMAT_FIELD = "X-Stats-Format";
+static constexpr std::string_view ALLOWED_METHODS = "GET, HEAD";
+
+static const char REMAP_USAGE[] = "[--format=json|csv|prometheus|prometheus_v2] [--integer-counters] [--wrap-counters] "
+ "[--no-prometheus-help] [--max-age-ms=N] [--wait-timeout-ms=N] [--config=FILE] "
+ "[--on-config-error=fail|503]";
+
+static std::string_view
+format_name(output_format_t format)
+{
+ switch (format) {
+ case output_format_t::JSON_OUTPUT:
+ return "json";
+ case output_format_t::CSV_OUTPUT:
+ return "csv";
+ case output_format_t::PROMETHEUS_OUTPUT:
+ return "prometheus";
+ case output_format_t::PROMETHEUS_V2_OUTPUT:
+ return "prometheus_v2";
+ }
+ return "json";
+}
+
+static std::string_view
+format_content_type(output_format_t format)
+{
+ switch (format) {
+ case output_format_t::JSON_OUTPUT:
+ return "text/json";
+ case output_format_t::CSV_OUTPUT:
+ return "text/csv";
+ case output_format_t::PROMETHEUS_OUTPUT:
+ return "text/plain; version=0.0.4; charset=utf-8";
+ case output_format_t::PROMETHEUS_V2_OUTPUT:
+ return "text/plain; version=2.0.0; charset=utf-8";
+ }
+ return "text/json";
+}
+
+static std::string_view
+encoding_name(encoding_format_t encoding)
+{
+ switch (encoding) {
+ case encoding_format_t::DEFLATE:
+ return "deflate";
+ case encoding_format_t::GZIP:
+ return "gzip";
+ case encoding_format_t::BR:
+ return "br";
+ case encoding_format_t::NONE:
+ break;
+ }
+ return {};
+}
+
+static bool
+parse_format(std::string_view name, output_format_t &format)
+{
+ for (auto candidate : {output_format_t::JSON_OUTPUT, output_format_t::CSV_OUTPUT, output_format_t::PROMETHEUS_OUTPUT,
+ output_format_t::PROMETHEUS_V2_OUTPUT}) {
+ if (name == format_name(candidate)) {
+ format = candidate;
+ return true;
+ }
+ }
+ return false;
+}
+
+// A rendered body in one encoding. It does not change after the render publishes it, so requests on any thread share its
+// blocks.
+struct stats_snapshot {
+ stats_snapshot() : body(TSIOBufferCreate()), reader(TSIOBufferReaderAlloc(body)) {}
+ ~stats_snapshot() { TSIOBufferDestroy(body); }
+ stats_snapshot(const stats_snapshot &) = delete;
+ stats_snapshot &operator=(const stats_snapshot &) = delete;
+
+ TSIOBuffer body;
+ TSIOBufferReader reader;
+ int64_t bytes = 0;
+ encoding_format_t encoding = encoding_format_t::NONE;
+ std::chrono::steady_clock::time_point rendered;
+};
+
+using snapshot_ptr = std::shared_ptr<const stats_snapshot>;
+using snapshot_table = snapshot_ptr[FORMAT_COUNT][ENCODING_COUNT];
+
+// One request for the stats. The waiter list, the transaction hooks and the intercept share it.
+struct stats_scrape {
+ stats_scrape(TSHttpTxn txn, output_format_t fmt, encoding_format_t enc, bool rule)
+ : txnp(txn), format(fmt), encoding(enc), remap(rule)
+ {
+ }
+
+ TSHttpTxn txnp;
+ output_format_t format;
+ encoding_format_t encoding;
+ bool remap; // A remap rule serves the request, rather than the global plugin.
+ // When the request started to wait for a render.
+ std::chrono::steady_clock::time_point arrived;
+
+ // TSRemapDoRemap, scrape_wait, the render or the watchdog sets these before the transaction continues, and the intercept
+ // reads them only after that. The snapshot is null for a HEAD request to a remap rule and for a 503.
+ snapshot_ptr snapshot;
+ TSHttpStatus status = TS_HTTP_STATUS_OK;
+};
+
+struct stats_watchdog;
+
+// The stats of one remap rule, or of the global plugin. The remap rule and each continuation that works for the instance
+// hold a reference, so that a render or a watchdog can outlive the rule.
+struct stats_instance {
+ explicit stats_instance(const stats_options &opts) : options(opts) {}
+ ~stats_instance() { Dbg(dbg_ctl, "Freeing stats instance %p", this); }
+ stats_instance(const stats_instance &) = delete;
+ stats_instance &operator=(const stats_instance &) = delete;
+
+ PrometheusRenderer *
+ prometheus(output_format_t format)
+ {
+ std::unique_ptr<PrometheusRenderer> *renderer = nullptr;
+
+ if (format == output_format_t::PROMETHEUS_OUTPUT) {
+ renderer = &prometheus_v1;
+ } else if (format == output_format_t::PROMETHEUS_V2_OUTPUT) {
+ renderer = &prometheus_v2;
+ } else {
+ return nullptr;
+ }
+ if (*renderer == nullptr) {
+ *renderer = make_prometheus_renderer(format, options);
+ }
+ return renderer->get();
+ }
+
+ const stats_options options;
+ // Only the render in flight uses these, so the mutex does not guard them.
+ std::unique_ptr<PrometheusRenderer> prometheus_v1;
+ std::unique_ptr<PrometheusRenderer> prometheus_v2;
+ zlib_streams zlib;
+
+ // The mutex guards the members below it. No thread holds it while it calls an API that can run a continuation.
+ std::mutex mutex;
+ snapshot_table snapshots;
+ std::vector<std::shared_ptr<stats_scrape>> waiters;
+ bool rendering = false;
+ stats_watchdog *watchdog = nullptr; // Set while a request waits.
+};
+
+static std::shared_ptr<stats_instance>
+make_stats_instance(const stats_options &options)
+{
+ return std::make_shared<stats_instance>(options);
+}
+
+// Renders each wanted format once and compresses it for each wanted encoding.
+static void
+render_snapshots(stats_instance &instance, const bool (&wanted)[FORMAT_COUNT][ENCODING_COUNT], snapshot_table &fresh)
+{
+ auto const now = std::chrono::steady_clock::now();
+
+ for (size_t f = 0; f < FORMAT_COUNT; ++f) {
+ if (std::find(std::begin(wanted[f]), std::end(wanted[f]), true) == std::end(wanted[f])) {
+ continue;
+ }
+
+ auto const format = static_cast<output_format_t>(f);
+ auto body = std::make_shared<stats_snapshot>();
+ render_state render{body->body, &instance.options};
+
+ render_stats(format, &render, instance.prometheus(format));
+ body->bytes = TSIOBufferReaderAvail(body->reader);
+ body->rendered = now;
+
+ for (size_t e = 1; e < ENCODING_COUNT; ++e) {
+ if (!wanted[f][e]) {
+ continue;
+ }
+
+ auto const encoding = static_cast<encoding_format_t>(e);
+ auto compressed = std::make_shared<stats_snapshot>();
+
+ if (compress_body(encoding, instance.zlib, body->reader, compressed->body)) {
+ compressed->bytes = TSIOBufferReaderAvail(compressed->reader);
+ compressed->encoding = encoding;
+ compressed->rendered = now;
+ fresh[f][e] = std::move(compressed);
+ } else {
+ TSError("[%s] Cannot compress the stats, sending them uncompressed", PLUGIN_NAME);
+ fresh[f][e] = body;
+ }
+ }
+ fresh[f][static_cast<size_t>(encoding_format_t::NONE)] = std::move(body);
+ }
+}
+
+static int watchdog_handler(TSCont contp, TSEvent event, void *edata);
+
+// Answers the waiting requests with a 503 when the render takes longer than wait_timeout_ms.
+struct stats_watchdog {
+ explicit stats_watchdog(std::shared_ptr<stats_instance> inst)
+ : instance(std::move(inst)), cont(TSContCreate(watchdog_handler, TSMutexCreate()))
+ {
+ TSContDataSet(cont, this);
+ }
+
+ std::shared_ptr<stats_instance> instance;
+ TSCont cont;
+
+ // The mutex of the continuation guards these.
+ TSAction action = nullptr;
+ bool fired = false;
+};
+
+static int
+watchdog_handler(TSCont contp, TSEvent /* event ATS_UNUSED */, void * /* edata ATS_UNUSED */)
+{
+ auto *watchdog = static_cast<stats_watchdog *>(TSContDataGet(contp));
+ stats_instance &instance = *watchdog->instance;
+ std::vector<std::shared_ptr<stats_scrape>> expired;
+
+ watchdog->fired = true;
+ {
+ std::lock_guard lock{instance.mutex};
+
+ // A render took this watchdog or replaced it, and that render destroys it.
+ if (instance.watchdog != watchdog) {
+ return 0;
+ }
+ instance.watchdog = nullptr;
+ expired.swap(instance.waiters);
+ }
+
+ Dbg(dbg_ctl, "Answering %zu requests with a 503 after %" PRId64 " ms without a render of stats instance %p", expired.size(),
+ instance.options.wait_timeout_ms, &instance);
+ count(metrics().waiter_timeouts, expired.size());
+ for (auto const &scrape : expired) {
+ scrape->status = TS_HTTP_STATUS_SERVICE_UNAVAILABLE;
+ TSHttpTxnReenable(scrape->txnp, TS_EVENT_HTTP_CONTINUE);
+ }
+ TSContDestroy(contp);
+ delete watchdog;
+ return 0;
+}
+
+// Stops and destroys a watchdog that a render took from its instance.
+static void
+stop_watchdog(stats_watchdog *watchdog)
+{
+ TSMutex mutex = TSContMutexGet(watchdog->cont);
+
+ // Only the holder of its continuation's mutex can cancel an action.
+ TSMutexLock(mutex);
+ if (!watchdog->fired) {
+ TSActionCancel(watchdog->action);
+ }
+ TSMutexUnlock(mutex);
+ TSContDestroy(watchdog->cont);
+ delete watchdog;
+}
+
+// Creates a watchdog for the caller to publish under the instance mutex. The calling thread holds the mutex of the watchdog
+// until arm_watchdog sets its action, so that a render that finishes first can cancel that action.
+static stats_watchdog *
+new_watchdog(std::shared_ptr<stats_instance> instance)
+{
+ auto *watchdog = new stats_watchdog(std::move(instance));
+
+ TSMutexLock(TSContMutexGet(watchdog->cont));
+ return watchdog;
+}
+
+static void
+arm_watchdog(stats_watchdog *watchdog)
+{
+ TSMutex mutex = TSContMutexGet(watchdog->cont);
+
+ watchdog->action = TSContScheduleOnPool(watchdog->cont, watchdog->instance->options.wait_timeout_ms, TS_THREAD_POOL_NET);
+ TSMutexUnlock(mutex);
+}
+
+static int render_handler(TSCont contp, TSEvent event, void *edata);
+
+// Each render gets a new continuation, because TSContScheduleOnPool locks the mutex of the continuation on the calling
+// thread, and render_handler holds that mutex until it returns. A shared continuation could block the ET_NET thread that
+// schedules the next render.
+static void
+schedule_render(std::shared_ptr<stats_instance> instance)
+{
+ TSCont contp = TSContCreate(render_handler, TSMutexCreate());
+
+ TSContDataSet(contp, new std::shared_ptr<stats_instance>(std::move(instance)));
+ TSContScheduleOnPool(contp, 0, TS_THREAD_POOL_TASK);
+}
+
+static int
+render_handler(TSCont contp, TSEvent /* event ATS_UNUSED */, void * /* edata ATS_UNUSED */)
+{
+ auto *ref = static_cast<std::shared_ptr<stats_instance> *>(TSContDataGet(contp));
+ stats_instance &instance = **ref;
+ bool wanted[FORMAT_COUNT][ENCODING_COUNT] = {};
+ bool waiting = false;
+ snapshot_table fresh;
+
+ {
+ std::lock_guard lock{instance.mutex};
+
+ for (auto const &scrape : instance.waiters) {
+ wanted[static_cast<size_t>(scrape->format)][static_cast<size_t>(scrape->encoding)] = true;
+ waiting = true;
+ }
+ }
+ // The watchdog can answer every waiting request before the render starts.
+ if (waiting) {
+ static thread_local int64_t rest = 0;
+ int64_t const start = thread_cpu_ns();
+
+ render_snapshots(instance, wanted, fresh);
+ count(metrics().renders);
+ count_cpu_time(metrics().render_us, start, rest);
+ }
+
+ std::vector<std::shared_ptr<stats_scrape>> ready;
+ stats_watchdog *watchdog = nullptr;
+ stats_watchdog *replacement = nullptr;
+ bool again = false;
+
+ {
+ std::lock_guard lock{instance.mutex};
+
+ for (size_t f = 0; f < FORMAT_COUNT; ++f) {
+ for (size_t e = 0; e < ENCODING_COUNT; ++e) {
+ if (fresh[f][e] != nullptr) {
+ instance.snapshots[f][e] = fresh[f][e];
+ }
+ }
+ }
+
+ // As in scrape_wait, a request takes a render only if the render is younger than max_age_ms when the request arrives.
+ auto const max_age = std::chrono::milliseconds{instance.options.max_age_ms};
+ auto rendered = [&fresh, max_age](const std::shared_ptr<stats_scrape> &scrape) -> snapshot_ptr {
+ auto const &snapshot = fresh[static_cast<size_t>(scrape->format)][static_cast<size_t>(scrape->encoding)];
+
+ return snapshot != nullptr && scrape->arrived - snapshot->rendered < max_age ? snapshot : nullptr;
+ };
+ auto answered = std::stable_partition(instance.waiters.begin(), instance.waiters.end(),
+ [&rendered](const auto &scrape) { return rendered(scrape) == nullptr; });
+
+ for (auto it = answered; it != instance.waiters.end(); ++it) {
+ (*it)->snapshot = rendered(*it);
+ ready.push_back(std::move(*it));
+ }
+ instance.waiters.erase(answered, instance.waiters.end());
+
+ // Requests for another format or encoding, and requests that arrived too late for this render, wait for the next one. They
+ // get a new watchdog, because the current one can have started long before they arrived.
+ again = !instance.waiters.empty();
+ if (again) {
+ replacement = new_watchdog(*ref);
+ } else {
+ instance.rendering = false;
+ }
+ watchdog = std::exchange(instance.watchdog, replacement);
+ }
+
+ if (waiting) {
+ Dbg(dbg_ctl, "Rendered stats instance %p for %zu waiting requests", &instance, ready.size());
+ }
+ if (watchdog != nullptr) {
+ stop_watchdog(watchdog);
+ }
+ if (replacement != nullptr) {
+ arm_watchdog(replacement);
+ }
+ for (auto const &scrape : ready) {
+ TSHttpTxnReenable(scrape->txnp, TS_EVENT_HTTP_CONTINUE);
+ }
+ if (again) {
+ schedule_render(*ref);
+ }
+ delete ref;
+ TSContDestroy(contp);
+ return 0;
+}
+
+// Answers a request from a snapshot younger than max_age_ms. Otherwise the request waits for a render, which starts unless
+// one is in flight. Either way, the transaction continues once it has its answer.
+static void
+scrape_wait(const std::shared_ptr<stats_instance> &instance, std::shared_ptr<stats_scrape> scrape)
+{
+ auto const now = std::chrono::steady_clock::now();
+ bool waits = false;
+ bool render = false;
+ stats_watchdog *watchdog = nullptr;
+
+ {
+ std::lock_guard lock{instance->mutex};
+ auto const &snapshot = instance->snapshots[static_cast<size_t>(scrape->format)][static_cast<size_t>(scrape->encoding)];
+
+ if (snapshot != nullptr && now - snapshot->rendered < std::chrono::milliseconds{instance->options.max_age_ms}) {
+ scrape->snapshot = snapshot;
+ } else {
+ scrape->arrived = now;
+ instance->waiters.push_back(scrape);
+ waits = true;
+ render = !std::exchange(instance->rendering, true);
+ if (instance->watchdog == nullptr) {
+ watchdog = instance->watchdog = new_watchdog(instance);
+ }
+ }
+ }
+
+ if (!waits) {
+ TSHttpTxnReenable(scrape->txnp, TS_EVENT_HTTP_CONTINUE);
+ return;
+ }
+
+ Dbg(dbg_ctl, "Waiting for a render of stats instance %p", instance.get());
+ if (watchdog != nullptr) {
+ arm_watchdog(watchdog);
+ }
+ if (render) {
+ schedule_render(instance);
+ }
+}
+
+// The intercept holds only its scrape, because intercept events can arrive after the transaction and its remap rule are gone.
+struct scrape_intercept {
+ explicit scrape_intercept(std::shared_ptr<stats_scrape> s) : scrape(std::move(s)) {}
+ ~scrape_intercept()
+ {
+ if (net_vc != nullptr) {
+ TSVConnClose(net_vc);
+ }
+ if (req_buffer != nullptr) {
+ TSIOBufferDestroy(req_buffer);
+ }
+ if (resp_buffer != nullptr) {
+ TSIOBufferDestroy(resp_buffer);
+ }
+ }
+ scrape_intercept(const scrape_intercept &) = delete;
+ scrape_intercept &operator=(const scrape_intercept &) = delete;
+
+ std::shared_ptr<stats_scrape> scrape;
+ TSVConn net_vc = nullptr;
+ TSVIO write_vio = nullptr;
+ TSIOBuffer req_buffer = nullptr;
+ TSIOBuffer resp_buffer = nullptr;
+ TSIOBufferReader resp_reader = nullptr;
+};
+
+static std::string
+scrape_response_header(const stats_scrape &scrape)
+{
+ if (!scrape.remap) {
+ if (scrape.status != TS_HTTP_STATUS_OK) {
+ return RESP_HEADER_UNAVAILABLE;
+ }
+
+ std::string header{"HTTP/1.0 200 OK\r\nContent-Type: "};
+
+ header.append(format_content_type(scrape.format)).append("\r\n");
+ if (auto const name = encoding_name(scrape.snapshot->encoding); !name.empty()) {
+ header.append("Content-Encoding: ").append(name).append("\r\n");
+ }
+ header.append("Cache-Control: no-cache\r\n\r\n");
+ return header;
+ }
+
+ std::string header{"HTTP/1.1 "};
+
+ header.append(std::to_string(scrape.status)).append(" ").append(TSHttpHdrReasonLookup(scrape.status)).append("\r\n");
+ if (scrape.status == TS_HTTP_STATUS_OK) {
+ header.append("Content-Type: ").append(format_content_type(scrape.format)).append("\r\n");
+ }
+ header.append("Cache-Control: no-store\r\n");
+ header.append(STATS_FORMAT_FIELD).append(": ").append(format_name(scrape.format)).append("\r\n");
+ if (scrape.snapshot != nullptr || scrape.status != TS_HTTP_STATUS_OK) {
+ header.append("Content-Length: ")
+ .append(std::to_string(scrape.snapshot != nullptr ? scrape.snapshot->bytes : 0))
+ .append("\r\n");
+ }
+ header.append("\r\n");
+ return header;
+}
+
+static void
+scrape_send_response(TSCont contp, scrape_intercept *intercept)
+{
+ const stats_scrape &scrape = *intercept->scrape;
+ std::string const header = scrape_response_header(scrape);
+
+ intercept->resp_buffer = TSIOBufferCreate();
+ intercept->resp_reader = TSIOBufferReaderAlloc(intercept->resp_buffer);
+
+ int64_t bytes = TSIOBufferWrite(intercept->resp_buffer, header.data(), header.size());
+
+ if (scrape.snapshot != nullptr) {
+ // This shares the blocks of the snapshot rather than copying its data.
+ bytes += TSIOBufferCopy(intercept->resp_buffer, scrape.snapshot->reader, scrape.snapshot->bytes, 0);
+ }
+ TSVConnShutdown(intercept->net_vc, 1, 0);
+ intercept->write_vio = TSVConnWrite(intercept->net_vc, contp, intercept->resp_reader, bytes);
+ count(metrics().requests);
+ count(metrics().bytes_out, bytes);
+}
+
+static int
+scrape_intercept_handler(TSCont contp, TSEvent event, void *edata)
+{
+ static thread_local int64_t rest = 0;
+ int64_t const start = thread_cpu_ns();
+ auto *intercept = static_cast<scrape_intercept *>(TSContDataGet(contp));
+
+ switch (event) {
+ case TS_EVENT_NET_ACCEPT:
+ intercept->net_vc = static_cast<TSVConn>(edata);
+ intercept->req_buffer = TSIOBufferCreate();
+ TSVConnRead(intercept->net_vc, contp, intercept->req_buffer, INT64_MAX);
+ break;
+ case TS_EVENT_VCONN_READ_READY:
+ scrape_send_response(contp, intercept);
+ break;
+ case TS_EVENT_VCONN_WRITE_READY:
+ TSVIOReenable(intercept->write_vio);
+ break;
+ case TS_EVENT_NET_ACCEPT_FAILED:
+ case TS_EVENT_VCONN_EOS:
+ case TS_EVENT_ERROR:
+ case TS_EVENT_VCONN_INACTIVITY_TIMEOUT:
+ case TS_EVENT_VCONN_ACTIVE_TIMEOUT:
+ case TS_EVENT_VCONN_WRITE_COMPLETE:
+ Dbg(dbg_ctl, "Intercept finished on %s", TSHttpEventNameLookup(event));
+ delete intercept;
+ TSContDestroy(contp);
+ break;
+ default:
+ TSReleaseAssert(!"Unexpected Event");
+ }
+ count_cpu_time(metrics().intercept_us, start, rest);
+ return 0;
+}
+
+// The global plugin intercepts the request in its read request header hook, then waits there for the stats.
+static void
+serve_global_scrape(TSHttpTxn txnp, output_format_t format, encoding_format_t encoding)
+{
+ auto scrape = std::make_shared<stats_scrape>(txnp, format, encoding, false);
+ TSCont intercept_cont = TSContCreate(scrape_intercept_handler, TSMutexCreate());
+
+ TSContDataSet(intercept_cont, new scrape_intercept(scrape));
+ TSHttpTxnIntercept(intercept_cont, txnp);
+ scrape_wait(*global_instance, std::move(scrape));
+}
+
+//
+// Remap plugin.
+//
+
+// The transaction hooks of a GET request to a remap rule.
+struct scrape_txn {
+ std::shared_ptr<stats_instance> instance;
+ std::shared_ptr<stats_scrape> scrape;
+};
+
+static int
+scrape_txn_handler(TSCont contp, TSEvent event, void *edata)
+{
+ auto *txn = static_cast<scrape_txn *>(TSContDataGet(contp));
+
+ if (event == TS_EVENT_HTTP_CACHE_LOOKUP_COMPLETE) {
+ scrape_wait(txn->instance, txn->scrape);
+ return 0;
+ }
+ if (event == TS_EVENT_HTTP_TXN_CLOSE) {
+ delete txn;
+ TSContDestroy(contp);
+ }
+ TSHttpTxnReenable(static_cast<TSHttpTxn>(edata), TS_EVENT_HTTP_CONTINUE);
+ return 0;
+}
+
+static void
+add_allow_field(TSHttpTxn txnp)
+{
+ TSMBuffer bufp;
+ TSMLoc hdr_loc;
+
+ if (TSHttpTxnClientRespGet(txnp, &bufp, &hdr_loc) == TS_SUCCESS) {
+ TSMLoc field_loc;
+
+ // A remap ACL filter that denies the request replaces the 405 with a 403.
+ if (TSHttpHdrStatusGet(bufp, hdr_loc) == TS_HTTP_STATUS_METHOD_NOT_ALLOWED &&
+ TSMimeHdrFieldCreateNamed(bufp, hdr_loc, TS_MIME_FIELD_ALLOW, TS_MIME_LEN_ALLOW, &field_loc) == TS_SUCCESS) {
+ TSMimeHdrFieldValueStringSet(bufp, hdr_loc, field_loc, -1, ALLOWED_METHODS.data(), ALLOWED_METHODS.size());
+ TSMimeHdrFieldAppend(bufp, hdr_loc, field_loc);
+ TSHandleMLocRelease(bufp, hdr_loc, field_loc);
+ }
+ TSHandleMLocRelease(bufp, TS_NULL_MLOC, hdr_loc);
+ }
+}
+
+static int
+allow_handler(TSCont contp, TSEvent event, void *edata)
+{
+ auto txnp = static_cast<TSHttpTxn>(edata);
+
+ if (event == TS_EVENT_HTTP_SEND_RESPONSE_HDR) {
+ add_allow_field(txnp);
+ } else if (event == TS_EVENT_HTTP_TXN_CLOSE) {
+ TSContDestroy(contp);
+ }
+ TSHttpTxnReenable(txnp, TS_EVENT_HTTP_CONTINUE);
+ return 0;
+}
+
+// The settings that a remap rule or its configuration file can set.
+struct stats_settings {
+ std::optional<output_format_t> format;
+ std::optional<int64_t> max_age_ms;
+ std::optional<int64_t> wait_timeout_ms;
+ std::optional<bool> prometheus_help;
+ std::shared_ptr<const PrometheusRules> rules;
+};
+
+static std::string
+yaml_line(const YAML::Node &node)
+{
+ auto const mark = node.Mark();
+
+ return mark.is_null() ? std::string{} : "line " + std::to_string(mark.line + 1) + ": ";
+}
+
+// Reads the configuration file of a remap rule. Returns an empty string, or a description of the first error.
+static std::string
+load_settings_file(const std::string &path, stats_settings &settings)
+{
+ std::ifstream file{path};
+
+ if (!file) {
+ return std::string{"cannot open the file: "} + strerror(errno);
+ }
+ try {
+ YAML::Node const root = YAML::Load(file);
+
+ if (!root.IsMap()) {
+ return yaml_line(root) + "the file must be a map";
+ }
+ for (auto const &item : root) {
+ std::string const key = item.first.as<std::string>();
+ YAML::Node const value = item.second;
+
+ if (key == "format") {
+ output_format_t format;
+
+ if (!value.IsScalar() || !parse_format(value.Scalar(), format)) {
+ return yaml_line(value) + "format must be json, csv, prometheus or prometheus_v2";
+ }
+ settings.format = format;
+ } else if (key == "render") {
+ if (!value.IsMap()) {
+ return yaml_line(value) + "render must be a map";
+ }
+ for (auto const &setting : value) {
+ std::string const name = setting.first.as<std::string>();
+ std::string const text = setting.second.IsScalar() ? setting.second.Scalar() : std::string{};
+ int64_t number = 0;
+ int64_t min = 0;
+ bool valid = false;
+
+ if (name == "max_age_ms") {
+ valid = parse_integer(text, min, MAX_MILLISECONDS, number);
+ settings.max_age_ms = number;
+ } else if (name == "wait_timeout_ms") {
+ min = 1;
+ valid = parse_integer(text, min, MAX_MILLISECONDS, number);
+ settings.wait_timeout_ms = number;
+ } else {
+ return yaml_line(setting.first) + "unknown key render." + name;
+ }
+ if (!valid) {
+ return yaml_line(setting.second) + "render." + name + " must be an integer from " + std::to_string(min) + " to " +
+ std::to_string(MAX_MILLISECONDS) + ", not '" + text + "'";
+ }
+ }
+ } else if (key == "prometheus") {
+ auto rules = std::make_shared<PrometheusRules>();
+
+ if (std::string error = rules->load(value); !error.empty()) {
+ return error;
+ }
+ settings.prometheus_help = rules->help();
+ settings.rules = std::move(rules);
+ } else {
+ return yaml_line(item.first) + "unknown key " + key;
+ }
+ }
+ } catch (const YAML::Exception &e) {
+ return e.what();
+ }
+ return {};
+}
+
+// Registers the configuration file of a remap rule as a child of the remap configuration. After a change to the file, a
+// configuration reload loads the remap configuration again, and with it the file.
+static void
+watch_config_file(const std::string &path)
+{
+ TSMgmtString parent = nullptr;
+
+ if (TSMgmtStringGet("proxy.config.url_remap.filename", &parent) == TS_SUCCESS) {
+ TSMgmtConfigFileAdd(parent, path.c_str());
+ } else {
+ TSWarning("[%s] Cannot read proxy.config.url_remap.filename, so a configuration reload does not detect a change to %s",
+ PLUGIN_NAME, path.c_str());
+ }
+ TSfree(parent);
+}
+
+TSReturnCode
+TSRemapInit(TSRemapInterface * /* api_info ATS_UNUSED */, char * /* errbuf ATS_UNUSED */, int /* errbuf_size ATS_UNUSED */)
+{
+ metrics();
+ return TS_SUCCESS;
+}
+
+TSReturnCode
+TSRemapNewInstance(int argc, char *argv[], void **ih, char *errbuf, int errbuf_size)
+{
+ static const struct option longopts[] = {
+ {"format", required_argument, nullptr, 'f'},
+ {"integer-counters", no_argument, nullptr, 'i'},
+ {"wrap-counters", no_argument, nullptr, 'w'},
+ {"no-prometheus-help", no_argument, nullptr, 'n'},
+ {"max-age-ms", required_argument, nullptr, 'a'},
+ {"wait-timeout-ms", required_argument, nullptr, 't'},
+ {"config", required_argument, nullptr, 'c'},
+ {"on-config-error", required_argument, nullptr, 'e'},
+ {nullptr, 0, nullptr, 0 }
+ };
+ stats_options options;
+ stats_settings rule;
+ std::string config_path;
+ bool answer_errors = false;
+ int64_t number = 0;
+
+ options.prometheus_epoch = false;
+
+ // argv[0] is the "from" URL. Skip it so that the "to" URL poses as the program name.
+ --argc;
+ ++argv;
+ for (int opt, option_index = 0; (opt = getopt_long(argc, argv, "", longopts, &option_index)) != -1;) {
Review Comment:
`RemapPluginInfo::initInstance()` resets `optind`, `optreset`, `opterr` and `optarg` before every `new_instance` call (`src/proxy/http/remap/RemapPluginInfo.cc`), so each rule starts with fresh getopt state. The plugin then shifts `argv` by one, so the "to" URL stands in for the program name and parsing starts at the first plugin parameter.
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,739 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+PrometheusRenderer::begin()
+{
+ ++_renders;
+ _position = 0;
+ _stats = {};
+}
+
+void
+PrometheusRenderer::add(const char *name, TSRecordDataType data_type, const TSRecordData &datum)
+{
+ uint32_t const position = _position++;
+ uint32_t index;
+
+ if (position < _slots.size() && _slots[position].name == name && _records[_slots[position].record].data_type == data_type) {
+ index = _slots[position].record;
+ } else {
+ index = find(name, data_type);
+ if (position < _slots.size()) {
+ _slots[position] = {name, index};
+ } else {
+ _slots.push_back({name, index});
+ }
+ }
+
+ Record &record = _records[index];
+ record.last_render = _renders;
+ record.datum = datum;
+ if (record.value == Value::TEXT_LABEL || record.value == Value::TEXT_NUMBER) {
+ // The string is valid only while TSRecordDump calls back.
+ _info[index].text.assign(datum.rec_string != nullptr ? datum.rec_string : "");
+ } else if (record.value == Value::NONE && _info[index].invalid) {
+ ++_stats.dropped;
+ }
+}
+
+uint32_t
+PrometheusRenderer::find(const char *name, TSRecordDataType data_type)
+{
+ if (auto it = _by_pointer.find(name); it != _by_pointer.end() && _records[it->second].data_type == data_type) {
+ return it->second;
+ }
+
+ auto [it, added] = _by_name.try_emplace(name, 0);
+
+ if (added || _records[it->second].data_type != data_type) {
+ it->second = translate(it->first, data_type);
+ }
+ _by_pointer[name] = it->second;
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::add_record(std::string_view name, TSRecordDataType data_type)
+{
+ auto const index = static_cast<uint32_t>(_records.size());
+
+ _records.emplace_back().data_type = data_type;
+ _info.emplace_back().name = name;
+ _writers.push_back({index, 0});
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::add_family(const std::string &name, std::string_view record_name, PrometheusType type, uint32_t record)
+{
+ auto [it, added] = _family_index.try_emplace(name, static_cast<uint32_t>(_families.size()));
+
+ if (added) {
+ Family &family = _families.emplace_back();
+
+ _family_info.push_back({name, record});
+ _reorder = true;
+ family.type = type;
+ if (_options.help) {
+ family.header.append("# HELP ").append(name).append(" ").append(record_name).append("\n");
+ }
+ if (type != PrometheusType::UNTYPED) {
+ family.header.append("# TYPE ").append(name).append(" ").append(prometheus_type_name(type)).append("\n");
+ }
+ }
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::translate(std::string_view name, TSRecordDataType data_type)
+{
+ if (_options.rules != nullptr) {
+ return translate_rules(name, data_type);
+ }
+
+ uint32_t const index = add_record(name, data_type);
+
+ if (!is_numeric(data_type)) {
+ Dbg(dbg_ctl, "Prometheus supports only numeric values, skipping: %.*s", static_cast<int>(name.size()), name.data());
+ } else if (PrometheusName translated = _options.namer(name, data_type); !translated.family.empty()) {
+ Record &record = _records[index];
+
+ record.value = Value::NUMBER;
+ record.type = translated.type;
+ _info[index].family = add_family(translated.family, name, translated.type, index);
+
+ Family &family = _families[_info[index].family];
+
+ if (family.type != translated.type) {
+ Dbg(dbg_ctl, "Inconsistent types for metric family %s: %.*s is %s, the family is %s", translated.family.c_str(),
+ static_cast<int>(name.size()), name.data(), prometheus_type_name(translated.type).data(),
+ prometheus_type_name(family.type).data());
+ }
+ record.prefix = std::move(translated.family);
+ if (!translated.labels.empty()) {
+ record.prefix.append("{").append(translated.labels).append("}");
+ }
+ record.prefix += ' ';
+ family.members.push_back(index);
+ }
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::translate_rules(std::string_view name, TSRecordDataType data_type)
+{
+ uint32_t const index = add_record(name, data_type);
+ PrometheusRuleMatch const match = _options.rules->translate(name, data_type);
+ Record &record = _records[index];
+ RecordInfo &info = _info[index];
+
+ if (match.result == PrometheusRuleMatch::Result::INVALID_NAME) {
+ info.invalid = true;
+ if (!_warned_invalid && _options.warn != nullptr) {
+ _warned_invalid = true;
+ _options.warn("Leaving out " + std::string{name} + ", because the rules name it " + match.family +
+ ", which is not a valid metric name. Other such records are left out without a message");
+ }
+ return index;
+ }
+ if (match.result == PrometheusRuleMatch::Result::EXCLUDED) {
+ return index;
+ }
+
+ if (match.text_label) {
+ record.value = Value::TEXT_LABEL;
+ } else {
+ record.value = data_type == TS_RECORDDATATYPE_STRING ? Value::TEXT_NUMBER : Value::NUMBER;
+ }
+ record.type = match.type;
+ info.rule_index = match.rule_index;
+ info.label_names = match.label_names;
+ info.family = add_family(match.family, name, match.type, index);
+
+ Family &family = _families[info.family];
+ FamilyInfo &family_info = _family_info[info.family];
+
+ family.members.push_back(index);
+ mark_unsorted(info.family);
+ if (family_info.definer != index && defines_over(index, family_info.definer)) {
+ family_info.definer = index;
+ regroup(info.family);
+ } else {
+ place(index, match.label_values);
+ }
+ return index;
+}
+
+void
+PrometheusRenderer::place(uint32_t index, const std::vector<std::string> &label_values)
+{
+ Record &record = _records[index];
+ RecordInfo &info = _info[index];
+ FamilyInfo const &family = _family_info[info.family];
+ const std::vector<std::string> &own = *info.label_names;
+ const std::vector<std::string> &defined = *_info[family.definer].label_names;
+ bool const same = std::is_permutation(own.begin(), own.end(), defined.begin(), defined.end());
+ // The same label names in another order keep their values. Other names take the names of the family by position.
+ const std::vector<std::string> &names = same ? own : defined;
+
+ _writers[index].first = index;
+ record.dropped = own.size() != defined.size() || other_kind(index);
+ record.relabeled = !record.dropped && !same;
+ record.prefix.clear();
+ if (record.dropped) {
+ return;
+ }
+
+ struct Label {
+ std::string_view name;
+ std::string_view value;
+ bool text;
+ };
+
+ std::vector<Label> labels;
+ // The string value takes the last position.
+ size_t const text = record.value == Value::TEXT_LABEL ? names.size() - 1 : names.size();
+
+ for (size_t i = 0; i < names.size(); ++i) {
+ labels.push_back({names[i], i == text ? std::string_view{} : std::string_view{label_values[i]}, i == text});
+ }
+ for (auto const &[name, value] : _options.rules->const_labels()) {
+ labels.push_back({name, value, false});
+ }
+ std::sort(labels.begin(), labels.end(), [](const Label &a, const Label &b) { return a.name < b.name; });
+
+ record.prefix = family.name;
+ if (!labels.empty()) {
+ record.prefix += '{';
+ for (size_t i = 0; i < labels.size(); ++i) {
+ if (i > 0) {
+ record.prefix += ',';
+ }
+ record.prefix.append(labels[i].name).append("=\"");
+ if (labels[i].text) {
+ info.split = static_cast<uint32_t>(record.prefix.size());
+ } else {
+ prometheus_escape_label_value(record.prefix, labels[i].value);
+ }
+ record.prefix += '"';
+ }
+ record.prefix += '}';
+ }
+ record.prefix += ' ';
+
+ size_t const hash = std::hash<std::string>{}(record.prefix);
+ auto const [first, last] = _series.equal_range(hash);
+
+ for (auto it = first; it != last; ++it) {
+ if (_records[it->second].prefix == record.prefix) {
+ _writers[index].first = it->second;
+ return;
+ }
+ }
+ _series.emplace(hash, index);
+}
+
+void
+PrometheusRenderer::forget_series(uint32_t index)
+{
+ Record const &record = _records[index];
+
+ if (record.dropped || _writers[index].first != index) {
+ return;
+ }
+
+ auto const [first, last] = _series.equal_range(std::hash<std::string>{}(record.prefix));
+
+ for (auto it = first; it != last; ++it) {
+ if (it->second == index) {
+ _series.erase(it);
+ return;
+ }
+ }
+}
+
+// The label names of the family changed, so place each member again.
+void
+PrometheusRenderer::regroup(uint32_t family)
+{
+ auto &members = _families[family].members;
+
+ for (uint32_t member : members) {
+ forget_series(member);
+ }
+ std::sort(members.begin(), members.end());
+ for (uint32_t member : members) {
+ place(member, _options.rules->translate(_info[member].name, _records[member].data_type).label_values);
Review Comment:
This work is bounded. `defines_over()` lets the definer change only to a record whose rule index is higher, or within one rule from a text-label record to a numeric one, so a family regroups at most twice per rule that feeds it, and only when records are first translated. The cost is the family's members times its rules, not quadratic in records. Caching the label values would keep them in memory for every record to speed up that path, so I have left it as is.
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,739 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+PrometheusRenderer::begin()
+{
+ ++_renders;
+ _position = 0;
+ _stats = {};
+}
+
+void
+PrometheusRenderer::add(const char *name, TSRecordDataType data_type, const TSRecordData &datum)
+{
+ uint32_t const position = _position++;
+ uint32_t index;
+
+ if (position < _slots.size() && _slots[position].name == name && _records[_slots[position].record].data_type == data_type) {
+ index = _slots[position].record;
+ } else {
+ index = find(name, data_type);
+ if (position < _slots.size()) {
+ _slots[position] = {name, index};
+ } else {
+ _slots.push_back({name, index});
+ }
+ }
+
+ Record &record = _records[index];
+ record.last_render = _renders;
+ record.datum = datum;
+ if (record.value == Value::TEXT_LABEL || record.value == Value::TEXT_NUMBER) {
+ // The string is valid only while TSRecordDump calls back.
+ _info[index].text.assign(datum.rec_string != nullptr ? datum.rec_string : "");
+ } else if (record.value == Value::NONE && _info[index].invalid) {
+ ++_stats.dropped;
+ }
+}
+
+uint32_t
+PrometheusRenderer::find(const char *name, TSRecordDataType data_type)
+{
+ if (auto it = _by_pointer.find(name); it != _by_pointer.end() && _records[it->second].data_type == data_type) {
+ return it->second;
+ }
+
+ auto [it, added] = _by_name.try_emplace(name, 0);
+
+ if (added || _records[it->second].data_type != data_type) {
+ it->second = translate(it->first, data_type);
+ }
+ _by_pointer[name] = it->second;
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::add_record(std::string_view name, TSRecordDataType data_type)
+{
+ auto const index = static_cast<uint32_t>(_records.size());
+
+ _records.emplace_back().data_type = data_type;
+ _info.emplace_back().name = name;
+ _writers.push_back({index, 0});
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::add_family(const std::string &name, std::string_view record_name, PrometheusType type, uint32_t record)
+{
+ auto [it, added] = _family_index.try_emplace(name, static_cast<uint32_t>(_families.size()));
+
+ if (added) {
+ Family &family = _families.emplace_back();
+
+ _family_info.push_back({name, record});
+ _reorder = true;
+ family.type = type;
+ if (_options.help) {
+ family.header.append("# HELP ").append(name).append(" ").append(record_name).append("\n");
Review Comment:
Fixed in 28b6c1d6f9: HELP text now escapes a backslash and a line feed. The renderer on master writes the raw name the same way, so this also fixes existing output.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#discussion_r4138964694
##########
tests/gold_tests/pluginTest/stats_over_http/task_render.py:
##########
@@ -0,0 +1,421 @@
+'''
+Drive stats_over_http requests that wait for a render on a task thread, and check the render in traffic.out.
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+import http.client
+import json
+import os
+import re
+import socket
+import struct
+import subprocess
+import sys
+import threading
+import time
+
+from fetch_stats import decode, request, test_metric_values
+
+WAITING = r'Waiting for a render of stats instance (?P<instance>0x[0-9a-f]+)'
+
+
+def rendered(instance: str = r'(?P<instance>0x[0-9a-f]+)', count: str = r'(?P<count>\d+)') -> str:
+ '''The log line of a render, which runs on a task thread.'''
+ return rf'\[ET_TASK \d+\] .*Rendered stats instance {instance} for {count} waiting requests'
+
+
+RENDERED = rendered()
+
+
+class Failure(Exception):
+ pass
+
+
+class Log:
+ '''Reads traffic.out from an offset, so that each check sees only the lines of its own scenario.'''
+
+ def __init__(self, path: str):
+ self.path = path
+ self.offset = os.path.getsize(path)
+
+ def text(self) -> str:
+ with open(self.path, 'rb') as f:
+ f.seek(self.offset)
+ return f.read().decode(errors='replace')
+
+ def wait_for(self, pattern: str, timeout: float = 30) -> re.Match:
+ deadline = time.monotonic() + timeout
+ while True:
+ match = re.search(pattern, self.text())
+ if match:
+ return match
+ if time.monotonic() > deadline:
+ raise Failure(f'traffic.out has no match for {pattern!r} after {timeout} s')
+ time.sleep(0.05)
+
+ def all(self, pattern: str) -> list[re.Match]:
+ return list(re.finditer(pattern, self.text()))
+
+
+class Client:
+
+ def __init__(self, args: argparse.Namespace):
+ self.args = args
+
+ def get(self, path: str, encoding: str | None = None, timeout: float = 60) -> tuple[int, http.client.HTTPMessage, bytes]:
+ response, body = request(self.args.port, path, encoding, timeout=timeout)
+ return response.status, response.headers, body
+
+ def verify(self, path: str, stats_format: str, encoding: str | None = None) -> None:
+ status, headers, body = self.get(path, encoding)
+ if status != 200:
+ raise Failure(f'{path} returned status {status}')
+ if headers.get('content-encoding') != encoding:
+ raise Failure(f'{path} returned Content-Encoding {headers.get("content-encoding")}, expected {encoding}')
+ values = test_metric_values(stats_format, decode(encoding, body).decode('utf-8'))
+ if values != list(range(self.args.count)):
+ raise Failure(f'{path} with {encoding} has {len(values)} test metrics, expected {self.args.count}')
+
+ def render_time(self, path: str) -> int:
+ '''GET the JSON stats and return their current_time_epoch_ms, which each render sets.'''
+ status, _, body = self.get(path)
+ if status != 200:
+ raise Failure(f'{path} returned status {status}')
+ return int(json.loads(body)['global']['current_time_epoch_ms'])
+
+ def stall(self, ms: int, log: Log | None = None) -> Log:
+ '''Hold up the record dumps of the renders for ms milliseconds, and return the log from before the stall.'''
+ log = log or Log(self.args.traffic_out)
+ subprocess.run(['traffic_ctl', 'plugin', 'msg', 'test_metrics.stall', str(ms)], check=True, capture_output=True)
+ log.wait_for(f'Stalling record dumps for {ms} ms')
+ return log
+
+ def hold_task(self, ms: int, log: Log) -> re.Match:
+ '''Queue a task that keeps a task thread busy for ms milliseconds, and return its log line.'''
+ subprocess.run(['traffic_ctl', 'plugin', 'msg', 'test_metrics.hold_task', str(ms)], check=True, capture_output=True)
+ return log.wait_for(f'Queued a task that holds a task thread for {ms} ms')
+
+ def metric(self, name: str) -> int:
+ out = subprocess.run(['traffic_ctl', 'metric', 'get', name], check=True, capture_output=True, text=True).stdout
+ return int(out.split()[-1])
+
+ def reload_remap(self) -> None:
+ # Traffic Server reloads a file only when its modification time changes.
+ os.utime(self.args.remap_config)
+ subprocess.run(
+ ['traffic_ctl', 'config', 'reload', '--monitor', '--initial-wait=0.1', '--refresh-int=0.1', '--timeout=60s'],
+ check=True,
+ capture_output=True)
+
+
+def concurrent(client: Client) -> None:
+ '''Requests that arrive during a render all wait for it, and one render answers them all.'''
+ before = client.metric('plugin.stats_over_http.renders')
+ log = client.stall(3000)
+ with ThreadPoolExecutor(client.args.requests) as pool:
+ list(pool.map(lambda _: client.verify(client.args.path, 'prometheus'), range(client.args.requests)))
+ renders = log.all(RENDERED)
+ if [int(render['count']) for render in renders] != [client.args.requests]:
+ raise Failure(f'Expected one render for {client.args.requests} requests: {[render[0] for render in renders]}')
+ counted = client.metric('plugin.stats_over_http.renders') - before
+ if counted != 1:
+ raise Failure(f'plugin.stats_over_http.renders counted {counted} renders, expected 1')
+ print(f'One render answered {client.args.requests} concurrent requests')
+
+
+def reuse(client: Client) -> None:
+ '''A render answers later requests until it is older than --max-age-ms. The next request then renders again.'''
+ max_age = client.args.max_age_ms / 1000
+ log = Log(client.args.traffic_out)
+ start = time.monotonic()
+ first = client.render_time(client.args.path)
+ instance = log.wait_for(WAITING)['instance']
+ log.wait_for(rendered(instance, '1'))
+
+ # The first render happened after start, so a request answered before start + max_age finds it younger than max_age.
+ second = client.render_time(client.args.path)
+ if time.monotonic() - start >= max_age:
+ raise Failure(f'The second request took longer than {max_age} s, so it cannot show the reuse of the render')
+ if second != first:
+ raise Failure(f'A request within --max-age-ms got a new render: current_time_epoch_ms {first}, then {second}')
+
+ deadline = time.monotonic() + max_age + 30
+ while (later := client.render_time(client.args.path)) == first:
+ if time.monotonic() > deadline:
+ raise Failure(f'No new render {max_age + 30} s after the first one')
+ time.sleep(0.1)
+ elapsed = time.monotonic() - start
+ if elapsed < max_age:
+ raise Failure(f'A new render answered a request {elapsed:.2f} s after the first request, within --max-age-ms')
+
+ counts = [render['count'] for render in log.all(rendered(instance))]
+ if counts != ['1', '1']:
+ raise Failure(f'Expected one render for the first request and one after --max-age-ms: {counts}')
+ print(f'The render at {first} answered the second request, and the render at {later} answered a request after {elapsed:.2f} s')
+
+
+def abort(client: Client, reset: bool) -> None:
+ '''A client that goes away while its request waits for a render leaves nothing behind.'''
+ log = client.stall(3000)
+ sock = socket.create_connection(('127.0.0.1', client.args.port))
+ sock.sendall(f'GET {client.args.path} HTTP/1.1\r\nHost: 127.0.0.1:{client.args.port}\r\n\r\n'.encode())
+ instance = log.wait_for(WAITING)['instance']
+ if reset:
+ sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+ sock.close()
+
+ render = log.wait_for(rendered(instance, '1'))
+ if reset:
+ # The rule turns off half-open connections, so Traffic Server releases the intercept (NET_ACCEPT_FAILED) when
+ # the client closes the connection while the transaction waits. The transaction closes after the render
+ # reenables it.
+ closed = log.wait_for(r'Intercept finished on (?P<event>TS_EVENT_NET_ACCEPT_FAILED)')
+ if closed.start() > render.start():
+ raise Failure('The intercept of the reset request finished after the render')
+ else:
+ # The transaction keeps waiting for the render, then writes to a client that has gone.
+ closed = log.wait_for(r'Intercept finished on (?P<event>\S+)')
+ if closed.start() < render.start():
+ raise Failure('The intercept of the closed request finished before the render')
+ print(f'The render answered the {"reset" if reset else "closed"} request, whose intercept finished on {closed["event"]}')
+ client.verify(client.args.path, 'json')
+
+
+def timeout_then_reload(client: Client) -> None:
+ '''The watchdog answers a request that waits too long, and a remap reload deletes the rule while the render runs.
+
+ The render then holds the last reference to the instance, and frees it when it finishes. The reload and the render run
+ on task threads, so the reload can queue behind the render on the same thread. Try again when that happens.
+ '''
+ for attempt in range(1, 6):
+ log = client.stall(6000)
+ start = time.monotonic()
+ status, headers, body = client.get(client.args.path)
+ elapsed = time.monotonic() - start
+ if status != 503 or headers.get('x-stats-format') != 'prometheus' or headers.get('content-length') != '0' or body:
+ raise Failure(f'Expected an empty 503 from the watchdog, got {status} {headers} {body[:100]!r}')
+ if elapsed < client.args.wait_timeout_ms / 1000 - 0.1:
+ raise Failure(f'The 503 arrived after {elapsed:.2f} s, before the wait timeout')
+ instance = log.wait_for(WAITING)['instance']
+ log.wait_for(
+ rf'Answering 1 requests with a 503 after {client.args.wait_timeout_ms} ms without a render of stats instance {instance}'
+ )
+
+ client.reload_remap()
+ released = log.wait_for(rf'Releasing remap instance {instance}')
+ render = log.wait_for(rendered(instance, '0'))
+ freed = log.wait_for(rf'Freeing stats instance {instance}')
+ if released.start() < render.start() < freed.start():
+ print(f'Attempt {attempt}: the render freed stats instance {instance} after the reload released it')
+ log.wait_for('Stopped stalling record dumps')
+ return
+ print(f'Attempt {attempt}: the reload released stats instance {instance} after the render, trying again')
+ log.wait_for('Stopped stalling record dumps')
+ raise Failure('The reload never released the instance while the render ran')
+
+
+def reload_loop(client: Client) -> None:
+ '''Remap reloads replace the instance while requests keep it rendering.'''
+ log = Log(client.args.traffic_out)
+ stop = threading.Event()
+ statuses: list[int] = []
+ errors: list[str] = []
+
+ def scrape() -> None:
+ while not stop.is_set():
+ try:
+ client.verify(client.args.path, 'prometheus')
+ statuses.append(200)
+ except Exception as e:
+ errors.append(str(e))
+
+ scrapers = [threading.Thread(target=scrape) for _ in range(4)]
+ for scraper in scrapers:
+ scraper.start()
+ for _ in range(client.args.reloads):
+ client.reload_remap()
+ stop.set()
+ for scraper in scrapers:
+ scraper.join()
+
+ instances = {render['instance'] for render in log.all(RENDERED)}
+ released = log.all(r'Releasing remap instance')
+ if errors:
+ raise Failure(f'{len(errors)} of {len(errors) + len(statuses)} requests failed, the first: {errors[0]}')
+ if len(instances) < 2 or not released:
+ raise Failure(
+ f'The reloads should replace the rendering instance: {len(instances)} instances rendered, '
+ f'{len(released)} released')
+ print(f'{len(statuses)} requests during {client.args.reloads} reloads, {len(instances)} instances rendered')
+
+
+def global_mix(client: Client) -> None:
+ '''One render answers the global plugin's requests for each format and encoding that arrive while it runs.'''
+ encodings = ['gzip', 'deflate'] + (['br'] if client.args.brotli else [])
+ later = [
+ ('csv', None),
+ ('prometheus', 'gzip'),
+ ('prometheus_v2', None),
+ ('json', encodings[-1]),
+ ('json', 'deflate'),
+ ('csv', 'gzip'),
+ ('prometheus', None),
+ ]
+ log = client.stall(3000)
+ with ThreadPoolExecutor(len(later) + 1) as pool:
+ first = pool.submit(client.verify, '/_stats/json', 'json')
+ log.wait_for(WAITING)
+ rest = [pool.submit(client.verify, f'/_stats/{fmt}', fmt, encoding) for fmt, encoding in later]
+ for future in [first, *rest]:
+ future.result()
+ counts = [int(render['count']) for render in log.all(RENDERED)]
+ if counts != [1, len(later)]:
+ raise Failure(f'Expected a render for the first request, then one for the other {len(later)}: {counts}')
+ print(f'Two renders answered {len(later) + 1} requests for {len(later) + 1} formats and encodings')
+
+
+def global_timeout(client: Client) -> None:
+ '''The watchdog answers a request to the global plugin that waits too long with an empty 503.'''
+ timeout = client.args.wait_timeout_ms / 1000
+ log = client.stall(client.args.wait_timeout_ms + 1000)
+ start = time.monotonic()
+ status, headers, body = client.get('/_stats/json')
+ elapsed = time.monotonic() - start
+ if status != 503 or body or 'content-type' in headers or 'x-stats-format' in headers:
+ raise Failure(f'Expected an empty 503 without stats headers from the watchdog, got {status} {headers} {body[:100]!r}')
+ if elapsed < timeout - 0.1:
+ raise Failure(f'The 503 arrived after {elapsed:.2f} s, before the wait timeout')
+ instance = log.wait_for(WAITING)['instance']
+ log.wait_for(
+ rf'Answering 1 requests with a 503 after {client.args.wait_timeout_ms} ms without a render of stats instance {instance}')
+ log.wait_for(rendered(instance, '0'))
+ client.verify('/_stats/json', 'json')
+ print(f'The watchdog answered after {elapsed:.2f} s, and the next request got the stats')
+
+
+def global_handover(client: Client) -> None:
+ '''A request that waits for the render after the one in flight gets a wait timeout that starts when that render ends.
+
+ A render schedules the next render on its own task thread, ahead of the work that other threads queue after the
+ render starts. With one task thread and a wait timeout W:
+
+ 0 A held task keeps the thread busy until W/2. The first request waits, and its render queues behind the
+ held task. A second held task of 5/8 W queues behind the render.
+ W/2 The render starts, and a stall holds it until 3/4 W. A request for another format arrives and waits.
+ 3/4 W The render answers the first request and schedules the next render behind the second held task. The
+ wait timeout of the second request starts now and ends at 7/4 W.
+ 11/8 W The second held task ends, and the next render answers the second request. This is after W, when the
+ wait timeout of the first request would end, and before 7/4 W.
+ '''
+ timeout = client.args.wait_timeout_ms
+ log = Log(client.args.traffic_out)
+ client.hold_task(timeout // 2, log)
+ log.wait_for(f'Holding a task thread for {timeout // 2} ms')
+ client.stall(timeout * 3 // 4, log)
+ with ThreadPoolExecutor(2) as pool:
+ first = pool.submit(client.verify, '/_stats/json', 'json')
+ log.wait_for(WAITING)
+ queued = client.hold_task(timeout * 5 // 8, log)
+ released = log.wait_for('Released the task thread')
+ if queued.start() > released.start():
+ raise Failure('The second held task was queued after the first one ended, so the host was too slow for this scenario')
+ second = pool.submit(client.verify, '/_stats/csv', 'csv')
+ log.wait_for(r'Waiting for a render[^\n]*\n[\s\S]*Waiting for a render')
+ for future in [first, second]:
+ future.result()
+ held = log.wait_for(rf'Holding a task thread for {timeout * 5 // 8} ms')
+
+ renders = log.all(RENDERED)
+ if [int(render['count']) for render in renders] != [1, 1]:
+ raise Failure(f'Expected a render for each request: {[render[0] for render in renders]}')
+ if not renders[0].start() < held.start() < renders[1].start():
+ raise Failure('The second held task did not run between the renders')
+ if timeouts := log.all(r'Answering \d+ requests with a 503'):
+ raise Failure(f'The watchdog answered a request: {timeouts[0][0]}')
+ log.wait_for(r'Released the task thread[\s\S]*Released the task thread')
+ print('The second render answered its request after the wait timeout of the first request')
+
+
+def global_fresh(client: Client) -> None:
+ '''With --max-age-ms=0, a request that arrives during a render waits for the next render.'''
+ log = client.stall(client.args.wait_timeout_ms // 2)
+ with ThreadPoolExecutor(2) as pool:
+ first = pool.submit(client.render_time, '/_stats/json')
+ log.wait_for(WAITING)
+ second = pool.submit(client.render_time, '/_stats/json')
+ first_time, second_time = first.result(), second.result()
+ counts = [int(render['count']) for render in log.all(RENDERED)]
+ if counts != [1, 1]:
+ raise Failure(f'Expected a render for each request: {counts}')
+ if second_time <= first_time:
+ raise Failure(
+ f'The second request got a render from before it arrived: current_time_epoch_ms {first_time}, then {second_time}')
Review Comment:
Separate renders are not guaranteed to produce distinct `current_time_epoch_ms` values: the timestamp uses wall-clock milliseconds, so two consecutive renders can be equal (or the clock can move backward). The `[1, 1]` render-count assertion immediately above already proves that the second request used another render; this comparison can make the AuTest fail despite correct behavior.
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,707 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+PrometheusRenderer::begin()
+{
+ ++_renders;
+ _position = 0;
+ _stats = {};
+}
+
+void
+PrometheusRenderer::add(const char *name, TSRecordDataType data_type, const TSRecordData &datum)
+{
+ uint32_t const position = _position++;
+ uint32_t index;
+
+ if (position < _slots.size() && _slots[position].name == name && _records[_slots[position].record].data_type == data_type) {
+ index = _slots[position].record;
+ } else {
+ index = find(name, data_type);
+ if (position < _slots.size()) {
+ _slots[position] = {name, index};
+ } else {
+ _slots.push_back({name, index});
+ }
+ }
+
+ Record &record = _records[index];
+ record.last_render = _renders;
+ record.datum = datum;
+ if (record.value == Value::TEXT_LABEL || record.value == Value::TEXT_NUMBER) {
+ // The string is valid only while TSRecordDump calls back.
+ _info[index].text.assign(datum.rec_string != nullptr ? datum.rec_string : "");
+ } else if (record.value == Value::NONE && _info[index].invalid) {
+ ++_stats.dropped;
+ }
+}
+
+uint32_t
+PrometheusRenderer::find(const char *name, TSRecordDataType data_type)
+{
+ if (auto it = _by_pointer.find(name); it != _by_pointer.end() && _records[it->second].data_type == data_type) {
+ return it->second;
+ }
+
+ auto [it, added] = _by_name.try_emplace(name, 0);
+
+ if (added || _records[it->second].data_type != data_type) {
+ it->second = translate(it->first, data_type);
+ }
+ _by_pointer[name] = it->second;
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::add_record(std::string_view name, TSRecordDataType data_type)
+{
+ auto const index = static_cast<uint32_t>(_records.size());
+
+ _records.emplace_back().data_type = data_type;
+ _info.emplace_back().name = name;
+ _writers.push_back({index, 0});
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::add_family(const std::string &name, std::string_view record_name, PrometheusType type, uint32_t record)
+{
+ auto [it, added] = _family_index.try_emplace(name, static_cast<uint32_t>(_families.size()));
+
+ if (added) {
+ Family &family = _families.emplace_back();
+
+ _family_info.push_back({name, record});
+ _reorder = true;
+ family.type = type;
+ if (_options.help) {
+ family.header.append("# HELP ").append(name).append(" ").append(record_name).append("\n");
+ }
+ if (type != PrometheusType::UNTYPED) {
+ family.header.append("# TYPE ").append(name).append(" ").append(prometheus_type_name(type)).append("\n");
+ }
+ }
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::translate(std::string_view name, TSRecordDataType data_type)
+{
+ if (_options.rules != nullptr) {
+ return translate_rules(name, data_type);
+ }
+
+ uint32_t const index = add_record(name, data_type);
+
+ if (!is_numeric(data_type)) {
+ Dbg(dbg_ctl, "Prometheus supports only numeric values, skipping: %.*s", static_cast<int>(name.size()), name.data());
+ } else if (PrometheusName translated = _options.namer(name, data_type); !translated.family.empty()) {
+ Record &record = _records[index];
+
+ record.value = Value::NUMBER;
+ record.type = translated.type;
+ _info[index].family = add_family(translated.family, name, translated.type, index);
+
+ Family &family = _families[_info[index].family];
+
+ if (family.type != translated.type) {
+ Dbg(dbg_ctl, "Inconsistent types for metric family %s: %.*s is %s, the family is %s", translated.family.c_str(),
+ static_cast<int>(name.size()), name.data(), prometheus_type_name(translated.type).data(),
+ prometheus_type_name(family.type).data());
+ }
+ record.prefix = std::move(translated.family);
+ if (!translated.labels.empty()) {
+ record.prefix.append("{").append(translated.labels).append("}");
+ }
+ record.prefix += ' ';
+ family.members.push_back(index);
+ }
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::translate_rules(std::string_view name, TSRecordDataType data_type)
+{
+ uint32_t const index = add_record(name, data_type);
+ PrometheusRuleMatch const match = _options.rules->translate(name, data_type);
+ Record &record = _records[index];
+ RecordInfo &info = _info[index];
+
+ if (match.result == PrometheusRuleMatch::Result::INVALID_NAME) {
+ info.invalid = true;
+ if (!_warned_invalid && _options.warn != nullptr) {
+ _warned_invalid = true;
+ _options.warn("Leaving out " + std::string{name} + ", because the rules name it " + match.family +
+ ", which is not a valid metric name. Other such records are left out without a message");
+ }
+ return index;
+ }
+ if (match.result == PrometheusRuleMatch::Result::EXCLUDED) {
+ return index;
+ }
+
+ if (match.text_label) {
+ record.value = Value::TEXT_LABEL;
+ } else {
+ record.value = data_type == TS_RECORDDATATYPE_STRING ? Value::TEXT_NUMBER : Value::NUMBER;
+ }
+ record.type = match.type;
+ info.rule_index = match.rule_index;
+ info.label_names = match.label_names;
+ info.family = add_family(match.family, name, match.type, index);
+
+ Family &family = _families[info.family];
+ FamilyInfo &family_info = _family_info[info.family];
+
+ family.members.push_back(index);
+ mark_unsorted(info.family);
+ if (record.type != family.type) {
+ report(index, WARN_TYPE);
+ }
+ if (family_info.definer != index && info.rule_index > _info[family_info.definer].rule_index) {
+ family_info.definer = index;
+ regroup(info.family);
+ } else {
+ place(index, match.label_values);
+ }
+ return index;
+}
+
+void
+PrometheusRenderer::place(uint32_t index, const std::vector<std::string> &label_values)
+{
+ Record &record = _records[index];
+ RecordInfo &info = _info[index];
+ FamilyInfo const &family = _family_info[info.family];
+ const std::vector<std::string> &own = *info.label_names;
+ const std::vector<std::string> &defined = *_info[family.definer].label_names;
+ bool const same = std::is_permutation(own.begin(), own.end(), defined.begin(), defined.end());
+ // The same label names in another order keep their values. Other names take the names of the family by position.
+ const std::vector<std::string> &names = same ? own : defined;
+
+ _writers[index].first = index;
+ record.dropped = own.size() != defined.size();
+ record.relabeled = !record.dropped && !same;
+ record.prefix.clear();
+ if (record.dropped) {
+ return;
+ }
+
+ struct Label {
+ std::string_view name;
+ std::string_view value;
+ bool text;
+ };
+
+ std::vector<Label> labels;
+ // The string value takes the last position.
+ size_t const text = record.value == Value::TEXT_LABEL ? names.size() - 1 : names.size();
+
+ for (size_t i = 0; i < names.size(); ++i) {
+ labels.push_back({names[i], i == text ? std::string_view{} : std::string_view{label_values[i]}, i == text});
+ }
+ for (auto const &[name, value] : _options.rules->const_labels()) {
+ labels.push_back({name, value, false});
+ }
+ std::sort(labels.begin(), labels.end(), [](const Label &a, const Label &b) { return a.name < b.name; });
+
+ record.prefix = family.name;
+ if (!labels.empty()) {
+ record.prefix += '{';
+ for (size_t i = 0; i < labels.size(); ++i) {
+ if (i > 0) {
+ record.prefix += ',';
+ }
+ record.prefix.append(labels[i].name).append("=\"");
+ if (labels[i].text) {
+ info.split = static_cast<uint32_t>(record.prefix.size());
+ } else {
+ prometheus_escape_label_value(record.prefix, labels[i].value);
+ }
+ record.prefix += '"';
+ }
+ record.prefix += '}';
+ }
+ record.prefix += ' ';
+
+ size_t const hash = std::hash<std::string>{}(record.prefix);
+ auto const [first, last] = _series.equal_range(hash);
+
+ for (auto it = first; it != last; ++it) {
+ if (_records[it->second].prefix == record.prefix) {
+ _writers[index].first = it->second;
Review Comment:
The cached series key omits the runtime value of `TEXT_LABEL` records because `record.prefix` contains only an empty insertion point. If two configured string metrics map to the same family and static labels but currently have different string values, this loop permanently marks the second as a duplicate even though Prometheus considers the differing `value` labels distinct series; changing either value later does not rebuild that relationship. Include the escaped text-label value in per-render series identity (and ordering) before duplicate suppression.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#discussion_r4161707674
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,739 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+PrometheusRenderer::begin()
+{
+ ++_renders;
+ _position = 0;
+ _stats = {};
+}
+
+void
+PrometheusRenderer::add(const char *name, TSRecordDataType data_type, const TSRecordData &datum)
+{
+ uint32_t const position = _position++;
+ uint32_t index;
+
+ if (position < _slots.size() && _slots[position].name == name && _records[_slots[position].record].data_type == data_type) {
+ index = _slots[position].record;
+ } else {
+ index = find(name, data_type);
+ if (position < _slots.size()) {
+ _slots[position] = {name, index};
+ } else {
+ _slots.push_back({name, index});
+ }
+ }
+
+ Record &record = _records[index];
+ record.last_render = _renders;
+ record.datum = datum;
+ if (record.value == Value::TEXT_LABEL || record.value == Value::TEXT_NUMBER) {
+ // The string is valid only while TSRecordDump calls back.
+ _info[index].text.assign(datum.rec_string != nullptr ? datum.rec_string : "");
+ } else if (record.value == Value::NONE && _info[index].invalid) {
+ ++_stats.dropped;
+ }
+}
+
+uint32_t
+PrometheusRenderer::find(const char *name, TSRecordDataType data_type)
+{
+ if (auto it = _by_pointer.find(name); it != _by_pointer.end() && _records[it->second].data_type == data_type) {
+ return it->second;
+ }
+
+ auto [it, added] = _by_name.try_emplace(name, 0);
+
+ if (added || _records[it->second].data_type != data_type) {
+ it->second = translate(it->first, data_type);
+ }
+ _by_pointer[name] = it->second;
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::add_record(std::string_view name, TSRecordDataType data_type)
+{
+ auto const index = static_cast<uint32_t>(_records.size());
+
+ _records.emplace_back().data_type = data_type;
+ _info.emplace_back().name = name;
+ _writers.push_back({index, 0});
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::add_family(const std::string &name, std::string_view record_name, PrometheusType type, uint32_t record)
+{
+ auto [it, added] = _family_index.try_emplace(name, static_cast<uint32_t>(_families.size()));
+
+ if (added) {
+ Family &family = _families.emplace_back();
+
+ _family_info.push_back({name, record});
+ _reorder = true;
+ family.type = type;
+ if (_options.help) {
+ family.header.append("# HELP ").append(name).append(" ").append(record_name).append("\n");
Review Comment:
The original record name is inserted into a Prometheus `HELP` line without escaping. ATS metric creation does not restrict names, so a name containing `\` or a newline produces invalid exposition (a newline also injects an extra line). Escape backslashes and newlines in HELP text as required by the Prometheus text format.
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,739 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+PrometheusRenderer::begin()
+{
+ ++_renders;
+ _position = 0;
+ _stats = {};
+}
+
+void
+PrometheusRenderer::add(const char *name, TSRecordDataType data_type, const TSRecordData &datum)
+{
+ uint32_t const position = _position++;
+ uint32_t index;
+
+ if (position < _slots.size() && _slots[position].name == name && _records[_slots[position].record].data_type == data_type) {
+ index = _slots[position].record;
+ } else {
+ index = find(name, data_type);
+ if (position < _slots.size()) {
+ _slots[position] = {name, index};
+ } else {
+ _slots.push_back({name, index});
+ }
+ }
+
+ Record &record = _records[index];
+ record.last_render = _renders;
+ record.datum = datum;
+ if (record.value == Value::TEXT_LABEL || record.value == Value::TEXT_NUMBER) {
+ // The string is valid only while TSRecordDump calls back.
+ _info[index].text.assign(datum.rec_string != nullptr ? datum.rec_string : "");
+ } else if (record.value == Value::NONE && _info[index].invalid) {
+ ++_stats.dropped;
+ }
+}
+
+uint32_t
+PrometheusRenderer::find(const char *name, TSRecordDataType data_type)
+{
+ if (auto it = _by_pointer.find(name); it != _by_pointer.end() && _records[it->second].data_type == data_type) {
+ return it->second;
+ }
+
+ auto [it, added] = _by_name.try_emplace(name, 0);
+
+ if (added || _records[it->second].data_type != data_type) {
+ it->second = translate(it->first, data_type);
+ }
+ _by_pointer[name] = it->second;
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::add_record(std::string_view name, TSRecordDataType data_type)
+{
+ auto const index = static_cast<uint32_t>(_records.size());
+
+ _records.emplace_back().data_type = data_type;
+ _info.emplace_back().name = name;
+ _writers.push_back({index, 0});
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::add_family(const std::string &name, std::string_view record_name, PrometheusType type, uint32_t record)
+{
+ auto [it, added] = _family_index.try_emplace(name, static_cast<uint32_t>(_families.size()));
+
+ if (added) {
+ Family &family = _families.emplace_back();
+
+ _family_info.push_back({name, record});
+ _reorder = true;
+ family.type = type;
+ if (_options.help) {
+ family.header.append("# HELP ").append(name).append(" ").append(record_name).append("\n");
+ }
+ if (type != PrometheusType::UNTYPED) {
+ family.header.append("# TYPE ").append(name).append(" ").append(prometheus_type_name(type)).append("\n");
+ }
+ }
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::translate(std::string_view name, TSRecordDataType data_type)
+{
+ if (_options.rules != nullptr) {
+ return translate_rules(name, data_type);
+ }
+
+ uint32_t const index = add_record(name, data_type);
+
+ if (!is_numeric(data_type)) {
+ Dbg(dbg_ctl, "Prometheus supports only numeric values, skipping: %.*s", static_cast<int>(name.size()), name.data());
+ } else if (PrometheusName translated = _options.namer(name, data_type); !translated.family.empty()) {
+ Record &record = _records[index];
+
+ record.value = Value::NUMBER;
+ record.type = translated.type;
+ _info[index].family = add_family(translated.family, name, translated.type, index);
+
+ Family &family = _families[_info[index].family];
+
+ if (family.type != translated.type) {
+ Dbg(dbg_ctl, "Inconsistent types for metric family %s: %.*s is %s, the family is %s", translated.family.c_str(),
+ static_cast<int>(name.size()), name.data(), prometheus_type_name(translated.type).data(),
+ prometheus_type_name(family.type).data());
+ }
+ record.prefix = std::move(translated.family);
+ if (!translated.labels.empty()) {
+ record.prefix.append("{").append(translated.labels).append("}");
+ }
+ record.prefix += ' ';
+ family.members.push_back(index);
+ }
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::translate_rules(std::string_view name, TSRecordDataType data_type)
+{
+ uint32_t const index = add_record(name, data_type);
+ PrometheusRuleMatch const match = _options.rules->translate(name, data_type);
+ Record &record = _records[index];
+ RecordInfo &info = _info[index];
+
+ if (match.result == PrometheusRuleMatch::Result::INVALID_NAME) {
+ info.invalid = true;
+ if (!_warned_invalid && _options.warn != nullptr) {
+ _warned_invalid = true;
+ _options.warn("Leaving out " + std::string{name} + ", because the rules name it " + match.family +
+ ", which is not a valid metric name. Other such records are left out without a message");
+ }
+ return index;
+ }
+ if (match.result == PrometheusRuleMatch::Result::EXCLUDED) {
+ return index;
+ }
+
+ if (match.text_label) {
+ record.value = Value::TEXT_LABEL;
+ } else {
+ record.value = data_type == TS_RECORDDATATYPE_STRING ? Value::TEXT_NUMBER : Value::NUMBER;
+ }
+ record.type = match.type;
+ info.rule_index = match.rule_index;
+ info.label_names = match.label_names;
+ info.family = add_family(match.family, name, match.type, index);
+
+ Family &family = _families[info.family];
+ FamilyInfo &family_info = _family_info[info.family];
+
+ family.members.push_back(index);
+ mark_unsorted(info.family);
+ if (family_info.definer != index && defines_over(index, family_info.definer)) {
+ family_info.definer = index;
+ regroup(info.family);
+ } else {
+ place(index, match.label_values);
+ }
+ return index;
+}
+
+void
+PrometheusRenderer::place(uint32_t index, const std::vector<std::string> &label_values)
+{
+ Record &record = _records[index];
+ RecordInfo &info = _info[index];
+ FamilyInfo const &family = _family_info[info.family];
+ const std::vector<std::string> &own = *info.label_names;
+ const std::vector<std::string> &defined = *_info[family.definer].label_names;
+ bool const same = std::is_permutation(own.begin(), own.end(), defined.begin(), defined.end());
+ // The same label names in another order keep their values. Other names take the names of the family by position.
+ const std::vector<std::string> &names = same ? own : defined;
+
+ _writers[index].first = index;
+ record.dropped = own.size() != defined.size() || other_kind(index);
+ record.relabeled = !record.dropped && !same;
+ record.prefix.clear();
+ if (record.dropped) {
+ return;
+ }
+
+ struct Label {
+ std::string_view name;
+ std::string_view value;
+ bool text;
+ };
+
+ std::vector<Label> labels;
+ // The string value takes the last position.
+ size_t const text = record.value == Value::TEXT_LABEL ? names.size() - 1 : names.size();
+
+ for (size_t i = 0; i < names.size(); ++i) {
+ labels.push_back({names[i], i == text ? std::string_view{} : std::string_view{label_values[i]}, i == text});
+ }
+ for (auto const &[name, value] : _options.rules->const_labels()) {
+ labels.push_back({name, value, false});
+ }
+ std::sort(labels.begin(), labels.end(), [](const Label &a, const Label &b) { return a.name < b.name; });
+
+ record.prefix = family.name;
+ if (!labels.empty()) {
+ record.prefix += '{';
+ for (size_t i = 0; i < labels.size(); ++i) {
+ if (i > 0) {
+ record.prefix += ',';
+ }
+ record.prefix.append(labels[i].name).append("=\"");
+ if (labels[i].text) {
+ info.split = static_cast<uint32_t>(record.prefix.size());
+ } else {
+ prometheus_escape_label_value(record.prefix, labels[i].value);
+ }
+ record.prefix += '"';
+ }
+ record.prefix += '}';
+ }
+ record.prefix += ' ';
+
+ size_t const hash = std::hash<std::string>{}(record.prefix);
+ auto const [first, last] = _series.equal_range(hash);
+
+ for (auto it = first; it != last; ++it) {
+ if (_records[it->second].prefix == record.prefix) {
+ _writers[index].first = it->second;
+ return;
+ }
+ }
+ _series.emplace(hash, index);
+}
+
+void
+PrometheusRenderer::forget_series(uint32_t index)
+{
+ Record const &record = _records[index];
+
+ if (record.dropped || _writers[index].first != index) {
+ return;
+ }
+
+ auto const [first, last] = _series.equal_range(std::hash<std::string>{}(record.prefix));
+
+ for (auto it = first; it != last; ++it) {
+ if (it->second == index) {
+ _series.erase(it);
+ return;
+ }
+ }
+}
+
+// The label names of the family changed, so place each member again.
+void
+PrometheusRenderer::regroup(uint32_t family)
+{
+ auto &members = _families[family].members;
+
+ for (uint32_t member : members) {
+ forget_series(member);
+ }
+ std::sort(members.begin(), members.end());
+ for (uint32_t member : members) {
+ place(member, _options.rules->translate(_info[member].name, _records[member].data_type).label_values);
Review Comment:
`regroup()` retranslates every existing member whenever a later rule becomes the family definer. A configuration with many rules feeding one family can therefore run the regex/rule pipeline repeatedly (quadratic work during the first render or after reload), contrary to the cached-translation design and potentially causing scrape timeouts. Cache each match's `label_values` in `RecordInfo` when the record is first translated and reuse them here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "moonchen (via GitHub)" <gi...@apache.org>.
moonchen commented on PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#issuecomment-6348360594
Thanks, Bryan.
- NUL in `records[]`: `prometheus_record_filter()` now drops a decoded name that contains a NUL, because no record name has one. A query whose only names are dropped still counts as a `records[]` request, so it gets an empty body, not the full output. The unit tests cover a `%00` in the middle and at the end of a name.
- String metrics: the docs now say that a string metric, such as `proxy.process.version.server.short`, cannot be found by its name, so a response has it only after the plugin has seen it in a walk of every record. The `record_exists()` comment says the same.
I also rebased onto master after #13545 and moved this PR's tests to Uranium. The AuTests became pytest modules or `.test.yaml` replays, and the two tests that #13545 converted took this PR's changes. Each commit builds and passes its own tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "moonchen (via GitHub)" <gi...@apache.org>.
moonchen commented on code in PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#discussion_r4233721951
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,761 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+prometheus_escape_help_text(std::string &out, std::string_view text)
+{
+ for (char c : text) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+PrometheusRenderer::begin()
+{
+ ++_renders;
+ _position = 0;
+ _stats = {};
+}
+
+void
+PrometheusRenderer::add(const char *name, TSRecordDataType data_type, const TSRecordData &datum)
+{
+ uint32_t const position = _position++;
+ uint32_t index;
+
+ if (position < _slots.size() && _slots[position].name == name && _records[_slots[position].record].data_type == data_type) {
+ index = _slots[position].record;
+ } else {
+ index = find(name, data_type);
+ if (position < _slots.size()) {
+ _slots[position] = {name, index};
+ } else {
+ _slots.push_back({name, index});
+ }
+ }
+
+ Record &record = _records[index];
+ record.last_render = _renders;
+ record.datum = datum;
+ if (record.value == Value::TEXT_LABEL || record.value == Value::TEXT_NUMBER) {
+ // The string is valid only while TSRecordDump calls back.
+ _info[index].text.assign(datum.rec_string != nullptr ? datum.rec_string : "");
+ } else if (record.value == Value::NONE && _info[index].invalid) {
+ ++_stats.dropped;
+ }
+}
+
+uint32_t
+PrometheusRenderer::find(const char *name, TSRecordDataType data_type)
+{
+ if (auto it = _by_pointer.find(name); it != _by_pointer.end() && _records[it->second].data_type == data_type) {
+ return it->second;
+ }
+
+ auto [it, added] = _by_name.try_emplace(name, 0);
+
+ if (added || _records[it->second].data_type != data_type) {
+ it->second = translate(it->first, data_type);
+ }
+ _by_pointer[name] = it->second;
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::add_record(std::string_view name, TSRecordDataType data_type)
+{
+ auto const index = static_cast<uint32_t>(_records.size());
+
+ _records.emplace_back().data_type = data_type;
+ _info.emplace_back().name = name;
+ _writers.push_back({index, 0});
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::add_family(const std::string &name, std::string_view record_name, PrometheusType type, uint32_t record)
+{
+ auto [it, added] = _family_index.try_emplace(name, static_cast<uint32_t>(_families.size()));
+
+ if (added) {
+ Family &family = _families.emplace_back();
+
+ _family_info.push_back({name, record});
+ _reorder = true;
+ family.type = type;
+ if (_options.help) {
+ family.header.append("# HELP ").append(name).append(" ");
+ prometheus_escape_help_text(family.header, record_name);
+ family.header.append("\n");
+ }
+ if (type != PrometheusType::UNTYPED) {
+ family.header.append("# TYPE ").append(name).append(" ").append(prometheus_type_name(type)).append("\n");
+ }
+ }
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::translate(std::string_view name, TSRecordDataType data_type)
+{
+ if (_options.rules != nullptr) {
+ return translate_rules(name, data_type);
+ }
+
+ uint32_t const index = add_record(name, data_type);
+
+ if (!is_numeric(data_type)) {
+ Dbg(dbg_ctl, "Prometheus supports only numeric values, skipping: %.*s", static_cast<int>(name.size()), name.data());
+ } else if (PrometheusName translated = _options.namer(name, data_type); !translated.family.empty()) {
+ Record &record = _records[index];
+
+ record.value = Value::NUMBER;
+ record.type = translated.type;
+ _info[index].family = add_family(translated.family, name, translated.type, index);
+
+ Family &family = _families[_info[index].family];
+
+ if (family.type != translated.type) {
+ Dbg(dbg_ctl, "Inconsistent types for metric family %s: %.*s is %s, the family is %s", translated.family.c_str(),
+ static_cast<int>(name.size()), name.data(), prometheus_type_name(translated.type).data(),
+ prometheus_type_name(family.type).data());
+ }
+ record.prefix = std::move(translated.family);
+ if (!translated.labels.empty()) {
+ record.prefix.append("{").append(translated.labels).append("}");
+ }
+ record.prefix += ' ';
+ family.members.push_back(index);
Review Comment:
This is the existing behavior on master: `prometheus_out_stat()` writes a separate HELP, TYPE and sample block for each record as `TSRecordDump` visits it, so two names that sanitize to the same string already produce the same series twice. This PR keeps that output for the formats without rules; it only groups those samples under one TYPE line. Deduplicating the flat formats changes their output for existing users, so it belongs in a separate change with an upgrade note.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#discussion_r4189331821
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,761 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+prometheus_escape_help_text(std::string &out, std::string_view text)
+{
+ for (char c : text) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+PrometheusRenderer::begin()
+{
+ ++_renders;
+ _position = 0;
+ _stats = {};
+}
+
+void
+PrometheusRenderer::add(const char *name, TSRecordDataType data_type, const TSRecordData &datum)
+{
+ uint32_t const position = _position++;
+ uint32_t index;
+
+ if (position < _slots.size() && _slots[position].name == name && _records[_slots[position].record].data_type == data_type) {
+ index = _slots[position].record;
+ } else {
+ index = find(name, data_type);
+ if (position < _slots.size()) {
+ _slots[position] = {name, index};
+ } else {
+ _slots.push_back({name, index});
+ }
+ }
+
+ Record &record = _records[index];
+ record.last_render = _renders;
+ record.datum = datum;
+ if (record.value == Value::TEXT_LABEL || record.value == Value::TEXT_NUMBER) {
+ // The string is valid only while TSRecordDump calls back.
+ _info[index].text.assign(datum.rec_string != nullptr ? datum.rec_string : "");
+ } else if (record.value == Value::NONE && _info[index].invalid) {
+ ++_stats.dropped;
+ }
+}
+
+uint32_t
+PrometheusRenderer::find(const char *name, TSRecordDataType data_type)
+{
+ if (auto it = _by_pointer.find(name); it != _by_pointer.end() && _records[it->second].data_type == data_type) {
+ return it->second;
+ }
+
+ auto [it, added] = _by_name.try_emplace(name, 0);
+
+ if (added || _records[it->second].data_type != data_type) {
+ it->second = translate(it->first, data_type);
+ }
+ _by_pointer[name] = it->second;
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::add_record(std::string_view name, TSRecordDataType data_type)
+{
+ auto const index = static_cast<uint32_t>(_records.size());
+
+ _records.emplace_back().data_type = data_type;
+ _info.emplace_back().name = name;
+ _writers.push_back({index, 0});
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::add_family(const std::string &name, std::string_view record_name, PrometheusType type, uint32_t record)
+{
+ auto [it, added] = _family_index.try_emplace(name, static_cast<uint32_t>(_families.size()));
+
+ if (added) {
+ Family &family = _families.emplace_back();
+
+ _family_info.push_back({name, record});
+ _reorder = true;
+ family.type = type;
+ if (_options.help) {
+ family.header.append("# HELP ").append(name).append(" ");
+ prometheus_escape_help_text(family.header, record_name);
+ family.header.append("\n");
+ }
+ if (type != PrometheusType::UNTYPED) {
+ family.header.append("# TYPE ").append(name).append(" ").append(prometheus_type_name(type)).append("\n");
+ }
+ }
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::translate(std::string_view name, TSRecordDataType data_type)
+{
+ if (_options.rules != nullptr) {
+ return translate_rules(name, data_type);
+ }
+
+ uint32_t const index = add_record(name, data_type);
+
+ if (!is_numeric(data_type)) {
+ Dbg(dbg_ctl, "Prometheus supports only numeric values, skipping: %.*s", static_cast<int>(name.size()), name.data());
+ } else if (PrometheusName translated = _options.namer(name, data_type); !translated.family.empty()) {
+ Record &record = _records[index];
+
+ record.value = Value::NUMBER;
+ record.type = translated.type;
+ _info[index].family = add_family(translated.family, name, translated.type, index);
+
+ Family &family = _families[_info[index].family];
+
+ if (family.type != translated.type) {
+ Dbg(dbg_ctl, "Inconsistent types for metric family %s: %.*s is %s, the family is %s", translated.family.c_str(),
+ static_cast<int>(name.size()), name.data(), prometheus_type_name(translated.type).data(),
+ prometheus_type_name(family.type).data());
+ }
+ record.prefix = std::move(translated.family);
+ if (!translated.labels.empty()) {
+ record.prefix.append("{").append(translated.labels).append("}");
+ }
+ record.prefix += ' ';
+ family.members.push_back(index);
Review Comment:
The no-rules renderer does not deduplicate series. The production v1 namer sanitizes metric names many-to-one (for example, both `a.b` and `a_b` become the unlabeled series `a_b`), so adding both records here makes one scrape emit `a_b` twice. Prometheus requires each label set to be unique within a scrape and may reject the scrape as containing duplicate samples. Track the generated prefix in this path and suppress later records with the same family and labels, as the rules path already does.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "moonchen (via GitHub)" <gi...@apache.org>.
moonchen commented on PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#issuecomment-6505718851
Rebased onto master (only `upgrading.en.rst` needed merging) and made a few more correctness fixes on top of the approved revision. Re-requesting review for the delta.
- Type rules: `PrometheusRules::translate` stripped a label value from the name before matching the type rules by searching for the value, which could hit an earlier, unrelated copy — e.g. an `origin` label value `ur` matched the `ur` inside `current`, so `current_connection` lost its type-rule match and the sample was typed a counter instead of a gauge. It now erases each captured span by its offset. A unit case covers the collision.
- `records[]` and unlisted metrics: a full scrape skips a metric that ATS has unlisted (for example a per-server connection metric after a mode change), but the by-name path did not, so `records[]` could serve a metric the full output omits. `record_exists()` and `read_record()` now check `ts::Metrics::listed()`, so the two agree. Covered by a test that unlists a metric and checks `records[]` drops it.
- String metrics by name: a `records[]` request for a string metric such as `proxy.process.version.server.short` walked every record on every request, because the plugin could not read it by name. It now reads string metrics from `ts::Metrics::StaticString`, so a request walks only when it first names a record the plugin has not seen. The docs and the comment match.
- Rules file errors: a bad scalar for `help`, `rules.labels.<name>` or `limits.max_series` returned a raw yaml-cpp "bad conversion" message with no key. It now names the setting. A repeated key in a map (for example two `rules:` keys) now fails to load rather than silently using the first.
- Tests and docs: the compression test again asserts no `ERROR:`, `FATAL:` or unrecognized-value line after shutdown; the upgrade note gives both `--max-age-ms` defaults (0 for the global plugin, 1000 for a remap rule).
- Spelled out the concrete types for several `auto` locals where the type was not on the line.
Each change is in the commit it belongs to.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "moonchen (via GitHub)" <gi...@apache.org>.
moonchen commented on code in PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#discussion_r4161673628
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,707 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+PrometheusRenderer::begin()
+{
+ ++_renders;
+ _position = 0;
+ _stats = {};
+}
+
+void
+PrometheusRenderer::add(const char *name, TSRecordDataType data_type, const TSRecordData &datum)
+{
+ uint32_t const position = _position++;
+ uint32_t index;
+
+ if (position < _slots.size() && _slots[position].name == name && _records[_slots[position].record].data_type == data_type) {
+ index = _slots[position].record;
+ } else {
+ index = find(name, data_type);
+ if (position < _slots.size()) {
+ _slots[position] = {name, index};
+ } else {
+ _slots.push_back({name, index});
+ }
+ }
+
+ Record &record = _records[index];
+ record.last_render = _renders;
+ record.datum = datum;
+ if (record.value == Value::TEXT_LABEL || record.value == Value::TEXT_NUMBER) {
+ // The string is valid only while TSRecordDump calls back.
+ _info[index].text.assign(datum.rec_string != nullptr ? datum.rec_string : "");
+ } else if (record.value == Value::NONE && _info[index].invalid) {
+ ++_stats.dropped;
+ }
+}
+
+uint32_t
+PrometheusRenderer::find(const char *name, TSRecordDataType data_type)
+{
+ if (auto it = _by_pointer.find(name); it != _by_pointer.end() && _records[it->second].data_type == data_type) {
+ return it->second;
+ }
+
+ auto [it, added] = _by_name.try_emplace(name, 0);
+
+ if (added || _records[it->second].data_type != data_type) {
+ it->second = translate(it->first, data_type);
+ }
+ _by_pointer[name] = it->second;
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::add_record(std::string_view name, TSRecordDataType data_type)
+{
+ auto const index = static_cast<uint32_t>(_records.size());
+
+ _records.emplace_back().data_type = data_type;
+ _info.emplace_back().name = name;
+ _writers.push_back({index, 0});
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::add_family(const std::string &name, std::string_view record_name, PrometheusType type, uint32_t record)
+{
+ auto [it, added] = _family_index.try_emplace(name, static_cast<uint32_t>(_families.size()));
+
+ if (added) {
+ Family &family = _families.emplace_back();
+
+ _family_info.push_back({name, record});
+ _reorder = true;
+ family.type = type;
+ if (_options.help) {
+ family.header.append("# HELP ").append(name).append(" ").append(record_name).append("\n");
+ }
+ if (type != PrometheusType::UNTYPED) {
+ family.header.append("# TYPE ").append(name).append(" ").append(prometheus_type_name(type)).append("\n");
+ }
+ }
+ return it->second;
+}
+
+uint32_t
+PrometheusRenderer::translate(std::string_view name, TSRecordDataType data_type)
+{
+ if (_options.rules != nullptr) {
+ return translate_rules(name, data_type);
+ }
+
+ uint32_t const index = add_record(name, data_type);
+
+ if (!is_numeric(data_type)) {
+ Dbg(dbg_ctl, "Prometheus supports only numeric values, skipping: %.*s", static_cast<int>(name.size()), name.data());
+ } else if (PrometheusName translated = _options.namer(name, data_type); !translated.family.empty()) {
+ Record &record = _records[index];
+
+ record.value = Value::NUMBER;
+ record.type = translated.type;
+ _info[index].family = add_family(translated.family, name, translated.type, index);
+
+ Family &family = _families[_info[index].family];
+
+ if (family.type != translated.type) {
+ Dbg(dbg_ctl, "Inconsistent types for metric family %s: %.*s is %s, the family is %s", translated.family.c_str(),
+ static_cast<int>(name.size()), name.data(), prometheus_type_name(translated.type).data(),
+ prometheus_type_name(family.type).data());
+ }
+ record.prefix = std::move(translated.family);
+ if (!translated.labels.empty()) {
+ record.prefix.append("{").append(translated.labels).append("}");
+ }
+ record.prefix += ' ';
+ family.members.push_back(index);
+ }
+ return index;
+}
+
+uint32_t
+PrometheusRenderer::translate_rules(std::string_view name, TSRecordDataType data_type)
+{
+ uint32_t const index = add_record(name, data_type);
+ PrometheusRuleMatch const match = _options.rules->translate(name, data_type);
+ Record &record = _records[index];
+ RecordInfo &info = _info[index];
+
+ if (match.result == PrometheusRuleMatch::Result::INVALID_NAME) {
+ info.invalid = true;
+ if (!_warned_invalid && _options.warn != nullptr) {
+ _warned_invalid = true;
+ _options.warn("Leaving out " + std::string{name} + ", because the rules name it " + match.family +
+ ", which is not a valid metric name. Other such records are left out without a message");
+ }
+ return index;
+ }
+ if (match.result == PrometheusRuleMatch::Result::EXCLUDED) {
+ return index;
+ }
+
+ if (match.text_label) {
+ record.value = Value::TEXT_LABEL;
+ } else {
+ record.value = data_type == TS_RECORDDATATYPE_STRING ? Value::TEXT_NUMBER : Value::NUMBER;
+ }
+ record.type = match.type;
+ info.rule_index = match.rule_index;
+ info.label_names = match.label_names;
+ info.family = add_family(match.family, name, match.type, index);
+
+ Family &family = _families[info.family];
+ FamilyInfo &family_info = _family_info[info.family];
+
+ family.members.push_back(index);
+ mark_unsorted(info.family);
+ if (record.type != family.type) {
+ report(index, WARN_TYPE);
+ }
+ if (family_info.definer != index && info.rule_index > _info[family_info.definer].rule_index) {
+ family_info.definer = index;
+ regroup(info.family);
+ } else {
+ place(index, match.label_values);
+ }
+ return index;
+}
+
+void
+PrometheusRenderer::place(uint32_t index, const std::vector<std::string> &label_values)
+{
+ Record &record = _records[index];
+ RecordInfo &info = _info[index];
+ FamilyInfo const &family = _family_info[info.family];
+ const std::vector<std::string> &own = *info.label_names;
+ const std::vector<std::string> &defined = *_info[family.definer].label_names;
+ bool const same = std::is_permutation(own.begin(), own.end(), defined.begin(), defined.end());
+ // The same label names in another order keep their values. Other names take the names of the family by position.
+ const std::vector<std::string> &names = same ? own : defined;
+
+ _writers[index].first = index;
+ record.dropped = own.size() != defined.size();
+ record.relabeled = !record.dropped && !same;
+ record.prefix.clear();
+ if (record.dropped) {
+ return;
+ }
+
+ struct Label {
+ std::string_view name;
+ std::string_view value;
+ bool text;
+ };
+
+ std::vector<Label> labels;
+ // The string value takes the last position.
+ size_t const text = record.value == Value::TEXT_LABEL ? names.size() - 1 : names.size();
+
+ for (size_t i = 0; i < names.size(); ++i) {
+ labels.push_back({names[i], i == text ? std::string_view{} : std::string_view{label_values[i]}, i == text});
+ }
+ for (auto const &[name, value] : _options.rules->const_labels()) {
+ labels.push_back({name, value, false});
+ }
+ std::sort(labels.begin(), labels.end(), [](const Label &a, const Label &b) { return a.name < b.name; });
+
+ record.prefix = family.name;
+ if (!labels.empty()) {
+ record.prefix += '{';
+ for (size_t i = 0; i < labels.size(); ++i) {
+ if (i > 0) {
+ record.prefix += ',';
+ }
+ record.prefix.append(labels[i].name).append("=\"");
+ if (labels[i].text) {
+ info.split = static_cast<uint32_t>(record.prefix.size());
+ } else {
+ prometheus_escape_label_value(record.prefix, labels[i].value);
+ }
+ record.prefix += '"';
+ }
+ record.prefix += '}';
+ }
+ record.prefix += ' ';
+
+ size_t const hash = std::hash<std::string>{}(record.prefix);
+ auto const [first, last] = _series.equal_range(hash);
+
+ for (auto it = first; it != last; ++it) {
+ if (_records[it->second].prefix == record.prefix) {
+ _writers[index].first = it->second;
Review Comment:
Handled differently. A listed string metric has one value at a time, so its string is the value of its sample, not part of its identity. Two such metrics that the rules give the same family and other labels are a naming collision, whatever their strings: the plugin writes the first, counts the other in `series_duplicates`, and logs a warning that names both metrics. A family also holds either listed string metrics or other metrics, never both, so every duplicate follows from the names and is decided once, when the names are translated. The docs and the unit tests cover both rules.
##########
tests/gold_tests/pluginTest/stats_over_http/task_render.py:
##########
@@ -0,0 +1,421 @@
+'''
+Drive stats_over_http requests that wait for a render on a task thread, and check the render in traffic.out.
+'''
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+import http.client
+import json
+import os
+import re
+import socket
+import struct
+import subprocess
+import sys
+import threading
+import time
+
+from fetch_stats import decode, request, test_metric_values
+
+WAITING = r'Waiting for a render of stats instance (?P<instance>0x[0-9a-f]+)'
+
+
+def rendered(instance: str = r'(?P<instance>0x[0-9a-f]+)', count: str = r'(?P<count>\d+)') -> str:
+ '''The log line of a render, which runs on a task thread.'''
+ return rf'\[ET_TASK \d+\] .*Rendered stats instance {instance} for {count} waiting requests'
+
+
+RENDERED = rendered()
+
+
+class Failure(Exception):
+ pass
+
+
+class Log:
+ '''Reads traffic.out from an offset, so that each check sees only the lines of its own scenario.'''
+
+ def __init__(self, path: str):
+ self.path = path
+ self.offset = os.path.getsize(path)
+
+ def text(self) -> str:
+ with open(self.path, 'rb') as f:
+ f.seek(self.offset)
+ return f.read().decode(errors='replace')
+
+ def wait_for(self, pattern: str, timeout: float = 30) -> re.Match:
+ deadline = time.monotonic() + timeout
+ while True:
+ match = re.search(pattern, self.text())
+ if match:
+ return match
+ if time.monotonic() > deadline:
+ raise Failure(f'traffic.out has no match for {pattern!r} after {timeout} s')
+ time.sleep(0.05)
+
+ def all(self, pattern: str) -> list[re.Match]:
+ return list(re.finditer(pattern, self.text()))
+
+
+class Client:
+
+ def __init__(self, args: argparse.Namespace):
+ self.args = args
+
+ def get(self, path: str, encoding: str | None = None, timeout: float = 60) -> tuple[int, http.client.HTTPMessage, bytes]:
+ response, body = request(self.args.port, path, encoding, timeout=timeout)
+ return response.status, response.headers, body
+
+ def verify(self, path: str, stats_format: str, encoding: str | None = None) -> None:
+ status, headers, body = self.get(path, encoding)
+ if status != 200:
+ raise Failure(f'{path} returned status {status}')
+ if headers.get('content-encoding') != encoding:
+ raise Failure(f'{path} returned Content-Encoding {headers.get("content-encoding")}, expected {encoding}')
+ values = test_metric_values(stats_format, decode(encoding, body).decode('utf-8'))
+ if values != list(range(self.args.count)):
+ raise Failure(f'{path} with {encoding} has {len(values)} test metrics, expected {self.args.count}')
+
+ def render_time(self, path: str) -> int:
+ '''GET the JSON stats and return their current_time_epoch_ms, which each render sets.'''
+ status, _, body = self.get(path)
+ if status != 200:
+ raise Failure(f'{path} returned status {status}')
+ return int(json.loads(body)['global']['current_time_epoch_ms'])
+
+ def stall(self, ms: int, log: Log | None = None) -> Log:
+ '''Hold up the record dumps of the renders for ms milliseconds, and return the log from before the stall.'''
+ log = log or Log(self.args.traffic_out)
+ subprocess.run(['traffic_ctl', 'plugin', 'msg', 'test_metrics.stall', str(ms)], check=True, capture_output=True)
+ log.wait_for(f'Stalling record dumps for {ms} ms')
+ return log
+
+ def hold_task(self, ms: int, log: Log) -> re.Match:
+ '''Queue a task that keeps a task thread busy for ms milliseconds, and return its log line.'''
+ subprocess.run(['traffic_ctl', 'plugin', 'msg', 'test_metrics.hold_task', str(ms)], check=True, capture_output=True)
+ return log.wait_for(f'Queued a task that holds a task thread for {ms} ms')
+
+ def metric(self, name: str) -> int:
+ out = subprocess.run(['traffic_ctl', 'metric', 'get', name], check=True, capture_output=True, text=True).stdout
+ return int(out.split()[-1])
+
+ def reload_remap(self) -> None:
+ # Traffic Server reloads a file only when its modification time changes.
+ os.utime(self.args.remap_config)
+ subprocess.run(
+ ['traffic_ctl', 'config', 'reload', '--monitor', '--initial-wait=0.1', '--refresh-int=0.1', '--timeout=60s'],
+ check=True,
+ capture_output=True)
+
+
+def concurrent(client: Client) -> None:
+ '''Requests that arrive during a render all wait for it, and one render answers them all.'''
+ before = client.metric('plugin.stats_over_http.renders')
+ log = client.stall(3000)
+ with ThreadPoolExecutor(client.args.requests) as pool:
+ list(pool.map(lambda _: client.verify(client.args.path, 'prometheus'), range(client.args.requests)))
+ renders = log.all(RENDERED)
+ if [int(render['count']) for render in renders] != [client.args.requests]:
+ raise Failure(f'Expected one render for {client.args.requests} requests: {[render[0] for render in renders]}')
+ counted = client.metric('plugin.stats_over_http.renders') - before
+ if counted != 1:
+ raise Failure(f'plugin.stats_over_http.renders counted {counted} renders, expected 1')
+ print(f'One render answered {client.args.requests} concurrent requests')
+
+
+def reuse(client: Client) -> None:
+ '''A render answers later requests until it is older than --max-age-ms. The next request then renders again.'''
+ max_age = client.args.max_age_ms / 1000
+ log = Log(client.args.traffic_out)
+ start = time.monotonic()
+ first = client.render_time(client.args.path)
+ instance = log.wait_for(WAITING)['instance']
+ log.wait_for(rendered(instance, '1'))
+
+ # The first render happened after start, so a request answered before start + max_age finds it younger than max_age.
+ second = client.render_time(client.args.path)
+ if time.monotonic() - start >= max_age:
+ raise Failure(f'The second request took longer than {max_age} s, so it cannot show the reuse of the render')
+ if second != first:
+ raise Failure(f'A request within --max-age-ms got a new render: current_time_epoch_ms {first}, then {second}')
+
+ deadline = time.monotonic() + max_age + 30
+ while (later := client.render_time(client.args.path)) == first:
+ if time.monotonic() > deadline:
+ raise Failure(f'No new render {max_age + 30} s after the first one')
+ time.sleep(0.1)
+ elapsed = time.monotonic() - start
+ if elapsed < max_age:
+ raise Failure(f'A new render answered a request {elapsed:.2f} s after the first request, within --max-age-ms')
+
+ counts = [render['count'] for render in log.all(rendered(instance))]
+ if counts != ['1', '1']:
+ raise Failure(f'Expected one render for the first request and one after --max-age-ms: {counts}')
+ print(f'The render at {first} answered the second request, and the render at {later} answered a request after {elapsed:.2f} s')
+
+
+def abort(client: Client, reset: bool) -> None:
+ '''A client that goes away while its request waits for a render leaves nothing behind.'''
+ log = client.stall(3000)
+ sock = socket.create_connection(('127.0.0.1', client.args.port))
+ sock.sendall(f'GET {client.args.path} HTTP/1.1\r\nHost: 127.0.0.1:{client.args.port}\r\n\r\n'.encode())
+ instance = log.wait_for(WAITING)['instance']
+ if reset:
+ sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+ sock.close()
+
+ render = log.wait_for(rendered(instance, '1'))
+ if reset:
+ # The rule turns off half-open connections, so Traffic Server releases the intercept (NET_ACCEPT_FAILED) when
+ # the client closes the connection while the transaction waits. The transaction closes after the render
+ # reenables it.
+ closed = log.wait_for(r'Intercept finished on (?P<event>TS_EVENT_NET_ACCEPT_FAILED)')
+ if closed.start() > render.start():
+ raise Failure('The intercept of the reset request finished after the render')
+ else:
+ # The transaction keeps waiting for the render, then writes to a client that has gone.
+ closed = log.wait_for(r'Intercept finished on (?P<event>\S+)')
+ if closed.start() < render.start():
+ raise Failure('The intercept of the closed request finished before the render')
+ print(f'The render answered the {"reset" if reset else "closed"} request, whose intercept finished on {closed["event"]}')
+ client.verify(client.args.path, 'json')
+
+
+def timeout_then_reload(client: Client) -> None:
+ '''The watchdog answers a request that waits too long, and a remap reload deletes the rule while the render runs.
+
+ The render then holds the last reference to the instance, and frees it when it finishes. The reload and the render run
+ on task threads, so the reload can queue behind the render on the same thread. Try again when that happens.
+ '''
+ for attempt in range(1, 6):
+ log = client.stall(6000)
+ start = time.monotonic()
+ status, headers, body = client.get(client.args.path)
+ elapsed = time.monotonic() - start
+ if status != 503 or headers.get('x-stats-format') != 'prometheus' or headers.get('content-length') != '0' or body:
+ raise Failure(f'Expected an empty 503 from the watchdog, got {status} {headers} {body[:100]!r}')
+ if elapsed < client.args.wait_timeout_ms / 1000 - 0.1:
+ raise Failure(f'The 503 arrived after {elapsed:.2f} s, before the wait timeout')
+ instance = log.wait_for(WAITING)['instance']
+ log.wait_for(
+ rf'Answering 1 requests with a 503 after {client.args.wait_timeout_ms} ms without a render of stats instance {instance}'
+ )
+
+ client.reload_remap()
+ released = log.wait_for(rf'Releasing remap instance {instance}')
+ render = log.wait_for(rendered(instance, '0'))
+ freed = log.wait_for(rf'Freeing stats instance {instance}')
+ if released.start() < render.start() < freed.start():
+ print(f'Attempt {attempt}: the render freed stats instance {instance} after the reload released it')
+ log.wait_for('Stopped stalling record dumps')
+ return
+ print(f'Attempt {attempt}: the reload released stats instance {instance} after the render, trying again')
+ log.wait_for('Stopped stalling record dumps')
+ raise Failure('The reload never released the instance while the render ran')
+
+
+def reload_loop(client: Client) -> None:
+ '''Remap reloads replace the instance while requests keep it rendering.'''
+ log = Log(client.args.traffic_out)
+ stop = threading.Event()
+ statuses: list[int] = []
+ errors: list[str] = []
+
+ def scrape() -> None:
+ while not stop.is_set():
+ try:
+ client.verify(client.args.path, 'prometheus')
+ statuses.append(200)
+ except Exception as e:
+ errors.append(str(e))
+
+ scrapers = [threading.Thread(target=scrape) for _ in range(4)]
+ for scraper in scrapers:
+ scraper.start()
+ for _ in range(client.args.reloads):
+ client.reload_remap()
+ stop.set()
+ for scraper in scrapers:
+ scraper.join()
+
+ instances = {render['instance'] for render in log.all(RENDERED)}
+ released = log.all(r'Releasing remap instance')
+ if errors:
+ raise Failure(f'{len(errors)} of {len(errors) + len(statuses)} requests failed, the first: {errors[0]}')
+ if len(instances) < 2 or not released:
+ raise Failure(
+ f'The reloads should replace the rendering instance: {len(instances)} instances rendered, '
+ f'{len(released)} released')
+ print(f'{len(statuses)} requests during {client.args.reloads} reloads, {len(instances)} instances rendered')
+
+
+def global_mix(client: Client) -> None:
+ '''One render answers the global plugin's requests for each format and encoding that arrive while it runs.'''
+ encodings = ['gzip', 'deflate'] + (['br'] if client.args.brotli else [])
+ later = [
+ ('csv', None),
+ ('prometheus', 'gzip'),
+ ('prometheus_v2', None),
+ ('json', encodings[-1]),
+ ('json', 'deflate'),
+ ('csv', 'gzip'),
+ ('prometheus', None),
+ ]
+ log = client.stall(3000)
+ with ThreadPoolExecutor(len(later) + 1) as pool:
+ first = pool.submit(client.verify, '/_stats/json', 'json')
+ log.wait_for(WAITING)
+ rest = [pool.submit(client.verify, f'/_stats/{fmt}', fmt, encoding) for fmt, encoding in later]
+ for future in [first, *rest]:
+ future.result()
+ counts = [int(render['count']) for render in log.all(RENDERED)]
+ if counts != [1, len(later)]:
+ raise Failure(f'Expected a render for the first request, then one for the other {len(later)}: {counts}')
+ print(f'Two renders answered {len(later) + 1} requests for {len(later) + 1} formats and encodings')
+
+
+def global_timeout(client: Client) -> None:
+ '''The watchdog answers a request to the global plugin that waits too long with an empty 503.'''
+ timeout = client.args.wait_timeout_ms / 1000
+ log = client.stall(client.args.wait_timeout_ms + 1000)
+ start = time.monotonic()
+ status, headers, body = client.get('/_stats/json')
+ elapsed = time.monotonic() - start
+ if status != 503 or body or 'content-type' in headers or 'x-stats-format' in headers:
+ raise Failure(f'Expected an empty 503 without stats headers from the watchdog, got {status} {headers} {body[:100]!r}')
+ if elapsed < timeout - 0.1:
+ raise Failure(f'The 503 arrived after {elapsed:.2f} s, before the wait timeout')
+ instance = log.wait_for(WAITING)['instance']
+ log.wait_for(
+ rf'Answering 1 requests with a 503 after {client.args.wait_timeout_ms} ms without a render of stats instance {instance}')
+ log.wait_for(rendered(instance, '0'))
+ client.verify('/_stats/json', 'json')
+ print(f'The watchdog answered after {elapsed:.2f} s, and the next request got the stats')
+
+
+def global_handover(client: Client) -> None:
+ '''A request that waits for the render after the one in flight gets a wait timeout that starts when that render ends.
+
+ A render schedules the next render on its own task thread, ahead of the work that other threads queue after the
+ render starts. With one task thread and a wait timeout W:
+
+ 0 A held task keeps the thread busy until W/2. The first request waits, and its render queues behind the
+ held task. A second held task of 5/8 W queues behind the render.
+ W/2 The render starts, and a stall holds it until 3/4 W. A request for another format arrives and waits.
+ 3/4 W The render answers the first request and schedules the next render behind the second held task. The
+ wait timeout of the second request starts now and ends at 7/4 W.
+ 11/8 W The second held task ends, and the next render answers the second request. This is after W, when the
+ wait timeout of the first request would end, and before 7/4 W.
+ '''
+ timeout = client.args.wait_timeout_ms
+ log = Log(client.args.traffic_out)
+ client.hold_task(timeout // 2, log)
+ log.wait_for(f'Holding a task thread for {timeout // 2} ms')
+ client.stall(timeout * 3 // 4, log)
+ with ThreadPoolExecutor(2) as pool:
+ first = pool.submit(client.verify, '/_stats/json', 'json')
+ log.wait_for(WAITING)
+ queued = client.hold_task(timeout * 5 // 8, log)
+ released = log.wait_for('Released the task thread')
+ if queued.start() > released.start():
+ raise Failure('The second held task was queued after the first one ended, so the host was too slow for this scenario')
+ second = pool.submit(client.verify, '/_stats/csv', 'csv')
+ log.wait_for(r'Waiting for a render[^\n]*\n[\s\S]*Waiting for a render')
+ for future in [first, second]:
+ future.result()
+ held = log.wait_for(rf'Holding a task thread for {timeout * 5 // 8} ms')
+
+ renders = log.all(RENDERED)
+ if [int(render['count']) for render in renders] != [1, 1]:
+ raise Failure(f'Expected a render for each request: {[render[0] for render in renders]}')
+ if not renders[0].start() < held.start() < renders[1].start():
+ raise Failure('The second held task did not run between the renders')
+ if timeouts := log.all(r'Answering \d+ requests with a 503'):
+ raise Failure(f'The watchdog answered a request: {timeouts[0][0]}')
+ log.wait_for(r'Released the task thread[\s\S]*Released the task thread')
+ print('The second render answered its request after the wait timeout of the first request')
+
+
+def global_fresh(client: Client) -> None:
+ '''With --max-age-ms=0, a request that arrives during a render waits for the next render.'''
+ log = client.stall(client.args.wait_timeout_ms // 2)
+ with ThreadPoolExecutor(2) as pool:
+ first = pool.submit(client.render_time, '/_stats/json')
+ log.wait_for(WAITING)
+ second = pool.submit(client.render_time, '/_stats/json')
+ first_time, second_time = first.result(), second.result()
+ counts = [int(render['count']) for render in log.all(RENDERED)]
+ if counts != [1, 1]:
+ raise Failure(f'Expected a render for each request: {counts}')
+ if second_time <= first_time:
+ raise Failure(
+ f'The second request got a render from before it arrived: current_time_epoch_ms {first_time}, then {second_time}')
Review Comment:
Fixed. `global_fresh` no longer compares `current_time_epoch_ms`. It keeps the `[1, 1]` render-count check and requires the second request's "Waiting for a render" log line to come before the first render finishes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] stats_over_http: Remap mode and Prometheus rules [trafficserver]
Posted by "bryancall (via GitHub)" <gi...@apache.org>.
bryancall commented on code in PR #13754:
URL: https://github.com/apache/trafficserver/pull/13754#discussion_r4332851886
##########
plugins/stats_over_http/prometheus_render.cc:
##########
@@ -0,0 +1,934 @@
+/** @file
+
+ Render records in the Prometheus text format from a cache of translated names.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#include "prometheus_render.h"
+
+#include <algorithm>
+#include <cerrno>
+#include <charconv>
+#include <cmath>
+#include <cstdio>
+#include <cstdlib>
+#include <functional>
+#include <limits>
+#include <numeric>
+
+#include "prometheus_rules.h"
+#include "tsutil/DbgCtl.h"
+#include "tsutil/StringCompare.h"
+
+namespace
+{
+DbgCtl dbg_ctl{"stats_over_http"};
+
+bool
+is_numeric(TSRecordDataType data_type)
+{
+ return data_type == TS_RECORDDATATYPE_COUNTER || data_type == TS_RECORDDATATYPE_INT || data_type == TS_RECORDDATATYPE_FLOAT;
+}
+
+bool
+is_digit(char c)
+{
+ return c >= '0' && c <= '9';
+}
+
+// Accepts a decimal number with an optional sign and exponent, Inf or Infinity with an optional sign, or NaN. The case of
+// the letters does not matter. Rejects a number too large for a double.
+bool
+parse_number(const std::string &text, double &value)
+{
+ std::string_view body{text};
+
+ if (ts::iequals(body, "nan")) {
+ value = NAN;
+ return true;
+ }
+ if (!body.empty() && (body[0] == '+' || body[0] == '-')) {
+ body.remove_prefix(1);
+ }
+ if (!ts::iequals(body, "inf") && !ts::iequals(body, "infinity")) {
+ size_t i = 0;
+ size_t digits = 0;
+
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ if (i < body.size() && body[i] == '.') {
+ for (++i; i < body.size() && is_digit(body[i]); ++i) {
+ ++digits;
+ }
+ }
+ if (digits == 0) {
+ return false;
+ }
+ if (i < body.size() && (body[i] == 'e' || body[i] == 'E')) {
+ size_t exponent = 0;
+
+ if (++i < body.size() && (body[i] == '+' || body[i] == '-')) {
+ ++i;
+ }
+ for (; i < body.size() && is_digit(body[i]); ++i) {
+ ++exponent;
+ }
+ if (exponent == 0) {
+ return false;
+ }
+ }
+ if (i != body.size()) {
+ return false;
+ }
+ }
+ errno = 0;
+ value = std::strtod(text.c_str(), nullptr);
+ return !(errno == ERANGE && std::isinf(value));
+}
+
+// Appends @a value with the fewest significant digits, from @a min to @a max, that parse back to the same value. With
+// @a single, the digits must parse back to the same float. NaN and the infinities get their Prometheus spellings.
+void
+append_number(std::string &out, double value, int min, int max, bool single)
+{
+ if (std::isnan(value)) {
+ out += "NaN";
+ return;
+ }
+ if (std::isinf(value)) {
+ out += value > 0 ? "+Inf" : "-Inf";
+ return;
+ }
+
+ char buffer[32];
+ int length = 0;
+
+ for (int precision = min; precision <= max; ++precision) {
+ length = snprintf(buffer, sizeof(buffer), "%.*g", precision, value);
+ if (single ? std::strtof(buffer, nullptr) == static_cast<float>(value) : std::strtod(buffer, nullptr) == value) {
+ break;
+ }
+ }
+ out.append(buffer, length);
+}
+
+std::string
+join(const std::vector<std::string> &names)
+{
+ std::string result;
+
+ for (const std::string &name : names) {
+ if (!result.empty()) {
+ result += ", ";
+ }
+ result += name;
+ }
+ return result;
+}
+
+int
+hex_value(char c)
+{
+ if (c >= '0' && c <= '9') {
+ return c - '0';
+ }
+ if (c >= 'a' && c <= 'f') {
+ return c - 'a' + 10;
+ }
+ if (c >= 'A' && c <= 'F') {
+ return c - 'A' + 10;
+ }
+ return -1;
+}
+
+// Decodes the percent escapes and the plus signs of a query string component. A malformed escape stays as it is.
+std::string
+query_decode(std::string_view text)
+{
+ std::string out;
+
+ out.reserve(text.size());
+ for (size_t i = 0; i < text.size(); ++i) {
+ char const c = text[i];
+
+ if (c == '+') {
+ out += ' ';
+ continue;
+ }
+ if (c == '%' && i + 2 < text.size()) {
+ int const high = hex_value(text[i + 1]);
+ int const low = hex_value(text[i + 2]);
+
+ if (high >= 0 && low >= 0) {
+ out += static_cast<char>(high * 16 + low);
+ i += 2;
+ continue;
+ }
+ }
+ out += c;
+ }
+ return out;
+}
+} // namespace
+
+std::string_view
+prometheus_type_name(PrometheusType type)
+{
+ switch (type) {
+ case PrometheusType::COUNTER:
+ return "counter";
+ case PrometheusType::GAUGE:
+ return "gauge";
+ case PrometheusType::UNTYPED:
+ break;
+ }
+ return "untyped";
+}
+
+void
+prometheus_escape_label_value(std::string &out, std::string_view value)
+{
+ for (char c : value) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '"':
+ out += "\\\"";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+void
+prometheus_escape_help_text(std::string &out, std::string_view text)
+{
+ for (char c : text) {
+ switch (c) {
+ case '\\':
+ out += "\\\\";
+ break;
+ case '\n':
+ out += "\\n";
+ break;
+ default:
+ out += c;
+ break;
+ }
+ }
+}
+
+std::optional<std::vector<std::string>>
+prometheus_record_filter(std::string_view query)
+{
+ static constexpr std::string_view KEY = "records[]";
+
+ std::optional<std::vector<std::string>> names;
+
+ while (!query.empty()) {
+ size_t const amp = query.find('&');
+ std::string_view const param = query.substr(0, amp);
+
+ query.remove_prefix(amp == std::string_view::npos ? query.size() : amp + 1);
+
+ size_t const equal = param.find('=');
+ std::string_view const key = param.substr(0, equal);
+
+ if (key != KEY && (key.find_first_of("%+") == std::string_view::npos || query_decode(key) != KEY)) {
+ continue;
+ }
+ if (!names) {
+ names.emplace();
+ }
+ names->push_back(equal == std::string_view::npos ? std::string{} : query_decode(param.substr(equal + 1)));
Review Comment:
A decoded name containing a NUL could be dropped here, before it reaches `record_exists()`.
##########
plugins/stats_over_http/stats_over_http.cc:
##########
@@ -1001,114 +805,135 @@ csv_out_stats(stats_state *my_state)
}
static void
-prometheus_out_stats(stats_state *my_state)
+count_render_stats(const PrometheusRenderStats &stats)
{
- TSRecordDump((TSRecordType)(TS_RECORDTYPE_PLUGIN | TS_RECORDTYPE_NODE | TS_RECORDTYPE_PROCESS), prometheus_out_stat, my_state);
- APPEND_STAT_PROMETHEUS_NUMERIC("current_time_epoch_ms", "%" PRIu64, ms_since_epoch());
- // No version printed, since string stats are not supported by Prometheus.
+ count(metrics().series, stats.series);
+ count(metrics().series_dropped, stats.dropped);
+ count(metrics().series_relabeled, stats.relabeled);
+ count(metrics().series_duplicates, stats.duplicates);
+ count(metrics().series_type_conflicts, stats.type_conflicts);
}
+// Passes every record to the renderer, in the order of TSRecordDump.
static void
-prometheus_v2_out_stats(stats_state *my_state)
+walk_records(PrometheusRenderer &renderer)
{
- TSRecordDump((TSRecordType)(TS_RECORDTYPE_PLUGIN | TS_RECORDTYPE_NODE | TS_RECORDTYPE_PROCESS), prometheus_v2_out_stat, my_state);
-
- for (const auto &sanitized_name : my_state->prometheus_v2_family_order) {
- const auto &family = my_state->prometheus_v2_families.at(sanitized_name);
+ renderer.begin();
+ TSRecordDump((TSRecordType)(TS_RECORDTYPE_PLUGIN | TS_RECORDTYPE_NODE | TS_RECORDTYPE_PROCESS), prometheus_add_stat, &renderer);
+}
- APPEND("# HELP ");
- APPEND(sanitized_name.c_str());
- APPEND(" ");
- APPEND(family.help.c_str());
- APPEND("\n");
+static void
+prometheus_out_stats(output_format_t format, PrometheusRenderer &renderer, render_state *my_state)
+{
+ walk_records(renderer);
- const char *type_str = (family.data_type == TS_RECORDDATATYPE_COUNTER) ? "counter" : "gauge";
- APPEND("# TYPE ");
- APPEND(sanitized_name.c_str());
- APPEND(" ");
- APPEND(type_str);
- APPEND("\n");
+ std::string &body = renderer.render();
- for (const auto &sample : family.samples) {
- APPEND(sample.c_str());
+ if (my_state->options->prometheus_epoch) {
+ if (format == output_format_t::PROMETHEUS_V2_OUTPUT) {
+ if (renderer.options().help) {
+ body.append("# HELP current_time_epoch_ms Current time in milliseconds since epoch.\n");
+ }
+ body.append("# TYPE current_time_epoch_ms gauge\n");
}
+ body.append("current_time_epoch_ms ").append(std::to_string(ms_since_epoch())).append("\n");
}
+ TSIOBufferWrite(my_state->resp_buffer, body.data(), body.size());
+ count_render_stats(renderer.stats());
+}
+
+// Whether ATS has a record named @a name. TSMgmtDataTypeGet finds a record of librecords, and TSStatFindName a metric. Neither
+// finds a string metric, so a request finds one only after a render without records[] has seen it.
+static bool
+record_exists(const std::string &name)
+{
+ TSRecordDataType data_type;
+ int id;
+
+ return TSMgmtDataTypeGet(name.c_str(), &data_type) == TS_SUCCESS || TSStatFindName(name.c_str(), &id) == TS_SUCCESS;
Review Comment:
This still checks only the part of `name` before an embedded NUL, so `...completed_requests%00a` is treated as a real record and lands in `undumped` with its suffix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@trafficserver.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org