ipng/nginx-ipng-stats-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

PRE-RELEASE v0.8.2

Browse Source
This commit is contained in:
Pim van Pelt
2026-04-19 22:34:30 +02:00
parent 7ed77f5b22
commit 4b1b412fb7
7 changed files with 475 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

353
src/ngx_http_ipng_stats_module.c
View File

@@ -159,6 +159,34 @@ typedef struct ngx_http_ipng_stats_slot_s {
} ngx_http_ipng_stats_slot_t;
/* In-flight gauge node: one per (source_id, vip_id) pair, tracking the
* count of requests currently in each lifecycle phase. Inserted under
* the slab mutex the first time a pair is observed, then cached in the
* per-request ctx so transitions touch only the atomic lanes — no
* locking on the hot path after the first hit for a given pair.
*
* Never evicted: one node per distinct (source, vip) pair is small in
* practice (tens) and the node is cheap. */
typedef struct {
ngx_rbtree_node_t rbnode; /* key = hash of (source_id, vip_id) */
ngx_queue_t lru; /* for iteration at render time */
ngx_uint_t source_id;
ngx_uint_t vip_id;
ngx_atomic_uint_t active;
ngx_atomic_uint_t reading;
ngx_atomic_uint_t writing;
} ngx_http_ipng_stats_gauge_t;
/* Per-request state carried through the lifecycle hooks. Allocated at
* POST_READ, transitioned by the header filter, decremented by the
* pool cleanup. `state` is 1 = reading, 2 = writing, 0 = not tracked. */
typedef struct {
ngx_http_ipng_stats_gauge_t *gauge;
ngx_uint_t state;
} ngx_http_ipng_stats_ctx_t;
/* String interning tables live at the head of the shared-memory zone.
* They're flat arrays of ngx_str_t whose data pointers reference memory
* allocated from the zone's slab pool. Workers look up strings by
@@ -179,6 +207,13 @@ typedef struct {
ngx_http_ipng_stats_intern_t sources;
ngx_http_ipng_stats_intern_t vips;
/* In-flight gauges, keyed by (source_id, vip_id). Atomically
* updated from the request lifecycle hooks; snapshotted by the
* scrape handler under the slab mutex. */
ngx_rbtree_t gauge_rbtree;
ngx_rbtree_node_t gauge_sentinel;
ngx_queue_t gauge_lru;
/* Meta-counters for the plugin itself (FR-6 observability of
* the plugin in the design doc). */
ngx_atomic_uint_t zone_full_events;
@@ -328,6 +363,12 @@ static void ngx_http_ipng_stats_exit_worker(ngx_cycle_t *cycle);
static void ngx_http_ipng_stats_rescan_timer(ngx_event_t *ev);
static ngx_int_t ngx_http_ipng_stats_log_handler(ngx_http_request_t *r);
static ngx_int_t ngx_http_ipng_stats_post_read_handler(ngx_http_request_t *r);
static ngx_int_t ngx_http_ipng_stats_header_filter(ngx_http_request_t *r);
static void ngx_http_ipng_stats_ctx_cleanup(void *data);
static ngx_http_ipng_stats_gauge_t *ngx_http_ipng_stats_gauge_get(
ngx_http_ipng_stats_shctx_t *sh, ngx_slab_pool_t *slab,
ngx_uint_t source_id, ngx_uint_t vip_id);
static ngx_int_t ngx_http_ipng_stats_content_handler(ngx_http_request_t *r);
static void ngx_http_ipng_stats_flush_timer(ngx_event_t *ev);
@@ -365,6 +406,8 @@ static char *(*ngx_http_core_listen_orig)(ngx_conf_t *cf,
static ngx_http_ipng_stats_worker_t ngx_http_ipng_stats_worker;
static ngx_http_output_header_filter_pt ngx_http_ipng_stats_next_header_filter;
extern ngx_module_t ngx_http_core_module;
@@ -1200,7 +1243,7 @@ ngx_http_ipng_stats_logtail(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
/* ----------------------------------------------------------------- */
/* Postconfig: install log-phase handler */
/* Postconfig: install phase handlers and header filter */
/* ----------------------------------------------------------------- */
static ngx_int_t
@@ -1240,6 +1283,22 @@ ngx_http_ipng_stats_postconfig(ngx_conf_t *cf)
}
*h = ngx_http_ipng_stats_log_handler;
/* POST_READ is the earliest phase at which the request is parsed
* enough to resolve source and vip; we register the in-flight
* gauge there so `reading` covers rewrite/access/content. */
h = ngx_array_push(&cmcf->phases[NGX_HTTP_POST_READ_PHASE].handlers);
if (h == NULL) {
return NGX_ERROR;
}
*h = ngx_http_ipng_stats_post_read_handler;
/* Header filter: transitions reading -> writing when nginx starts
* sending the response. Inserted at the top of the chain so the
* transition is observed before any downstream filter mutates the
* response. */
ngx_http_ipng_stats_next_header_filter = ngx_http_top_header_filter;
ngx_http_top_header_filter = ngx_http_ipng_stats_header_filter;
return NGX_OK;
}
@@ -1313,6 +1372,10 @@ ngx_http_ipng_stats_init_zone(ngx_shm_zone_t *shm_zone, void *data)
ngx_http_ipng_stats_rbtree_insert);
ngx_queue_init(&sh->lru);
ngx_rbtree_init(&sh->gauge_rbtree, &sh->gauge_sentinel,
ngx_http_ipng_stats_rbtree_insert);
ngx_queue_init(&sh->gauge_lru);
sh->sources.nalloc = 16;
sh->sources.entries = ngx_slab_alloc(slab,
sh->sources.nalloc * sizeof(ngx_str_t));
@@ -2116,6 +2179,141 @@ ngx_http_ipng_stats_log_handler(ngx_http_request_t *r)
}
/* ----------------------------------------------------------------- */
/* In-flight gauge lifecycle hooks */
/* ----------------------------------------------------------------- */
/* POST_READ phase handler: first point at which the request is fully
* parsed enough to resolve source and vip. Finds or creates the
* gauge node, increments active + reading, and registers a pool
* cleanup that decrements on request finalization. Skips subrequests
* and internal redirects (the gauge is already ticking on r->main). */
static ngx_int_t
ngx_http_ipng_stats_post_read_handler(ngx_http_request_t *r)
{
ngx_http_ipng_stats_main_conf_t *imcf;
ngx_http_ipng_stats_loc_conf_t *ilcf;
ngx_http_ipng_stats_shctx_t *sh;
ngx_slab_pool_t *slab;
ngx_http_ipng_stats_ctx_t *ctx;
ngx_http_ipng_stats_gauge_t *g;
ngx_pool_cleanup_t *cln;
ngx_str_t source, vip;
u_char vipbuf[NGX_SOCKADDR_STRLEN];
ngx_uint_t source_id, vip_id;
if (r != r->main || r->internal) {
return NGX_DECLINED;
}
imcf = ngx_http_get_module_main_conf(r, ngx_http_ipng_stats_module);
if (imcf == NULL || imcf->shm_zone == NULL || !imcf->enabled) {
return NGX_DECLINED;
}
ilcf = ngx_http_get_module_loc_conf(r, ngx_http_ipng_stats_module);
if (ilcf == NULL || !ilcf->enabled) {
return NGX_DECLINED;
}
if (ngx_http_ipng_stats_resolve_source(r, imcf, &source) != NGX_OK) {
return NGX_DECLINED;
}
if (ngx_connection_local_sockaddr(r->connection, NULL, 0) != NGX_OK) {
return NGX_DECLINED;
}
if (ngx_http_ipng_stats_canonical_vip(r, vipbuf, sizeof(vipbuf), &vip)
!= NGX_OK)
{
return NGX_DECLINED;
}
slab = (ngx_slab_pool_t *) imcf->shm_zone->shm.addr;
sh = imcf->shm_zone->data;
ngx_shmtx_lock(&slab->mutex);
if (ngx_http_ipng_stats_intern_shared(sh, slab, &sh->sources, &source,
&source_id) != NGX_OK
|| ngx_http_ipng_stats_intern_shared(sh, slab, &sh->vips, &vip,
&vip_id) != NGX_OK)
{
ngx_shmtx_unlock(&slab->mutex);
return NGX_DECLINED;
}
ngx_shmtx_unlock(&slab->mutex);
g = ngx_http_ipng_stats_gauge_get(sh, slab, source_id, vip_id);
if (g == NULL) {
return NGX_DECLINED;
}
ctx = ngx_pcalloc(r->pool, sizeof(*ctx));
cln = ngx_pool_cleanup_add(r->pool, 0);
if (ctx == NULL || cln == NULL) {
return NGX_DECLINED;
}
ctx->gauge = g;
ctx->state = 1; /* reading */
cln->handler = ngx_http_ipng_stats_ctx_cleanup;
cln->data = ctx;
ngx_http_set_ctx(r, ctx, ngx_http_ipng_stats_module);
(void) ngx_atomic_fetch_add(&g->active, 1);
(void) ngx_atomic_fetch_add(&g->reading, 1);
return NGX_DECLINED;
}
/* Header filter: first call for the main request transitions reading
* -> writing. Subrequests have their own header-filter invocations;
* we ignore those so a subrequest doesn't prematurely flip the main
* request's gauge state. Internal redirects re-enter the filter
* chain; the state check prevents double transitions. */
static ngx_int_t
ngx_http_ipng_stats_header_filter(ngx_http_request_t *r)
{
ngx_http_ipng_stats_ctx_t *ctx;
if (r == r->main) {
ctx = ngx_http_get_module_ctx(r, ngx_http_ipng_stats_module);
if (ctx != NULL && ctx->gauge != NULL && ctx->state == 1) {
(void) ngx_atomic_fetch_add(&ctx->gauge->reading,
(ngx_atomic_uint_t) -1);
(void) ngx_atomic_fetch_add(&ctx->gauge->writing, 1);
ctx->state = 2;
}
}
return ngx_http_ipng_stats_next_header_filter(r);
}
static void
ngx_http_ipng_stats_ctx_cleanup(void *data)
{
ngx_http_ipng_stats_ctx_t *ctx = data;
if (ctx->gauge == NULL) {
return;
}
if (ctx->state == 1) {
(void) ngx_atomic_fetch_add(&ctx->gauge->reading,
(ngx_atomic_uint_t) -1);
} else if (ctx->state == 2) {
(void) ngx_atomic_fetch_add(&ctx->gauge->writing,
(ngx_atomic_uint_t) -1);
}
(void) ngx_atomic_fetch_add(&ctx->gauge->active,
(ngx_atomic_uint_t) -1);
ctx->gauge = NULL;
ctx->state = 0;
}
/* ----------------------------------------------------------------- */
/* String interning (called under slab mutex) */
/* ----------------------------------------------------------------- */
@@ -2170,6 +2368,66 @@ ngx_http_ipng_stats_intern_shared(ngx_http_ipng_stats_shctx_t *sh,
}
/* ----------------------------------------------------------------- */
/* In-flight gauges */
/* ----------------------------------------------------------------- */
/* Hash of (source_id, vip_id). Separate constants from the counter
* rbtree's hash so collisions don't line up between the two trees. */
static ngx_inline ngx_uint_t
ngx_http_ipng_stats_gauge_hash(ngx_uint_t source_id, ngx_uint_t vip_id)
{
return (source_id * 2246822519u) ^ (vip_id * 3266489917u);
}
/* Find or create the gauge node for (source_id, vip_id). Takes the
* slab mutex only when a new node must be inserted; the caller holds
* the returned pointer for the rest of the request and does lock-free
* atomic inc/dec on the lanes. Returns NULL on slab exhaustion. */
static ngx_http_ipng_stats_gauge_t *
ngx_http_ipng_stats_gauge_get(ngx_http_ipng_stats_shctx_t *sh,
ngx_slab_pool_t *slab, ngx_uint_t source_id, ngx_uint_t vip_id)
{
ngx_uint_t hash;
ngx_rbtree_node_t *rb;
ngx_http_ipng_stats_gauge_t *g = NULL;
hash = ngx_http_ipng_stats_gauge_hash(source_id, vip_id);
ngx_shmtx_lock(&slab->mutex);
rb = sh->gauge_rbtree.root;
while (rb != &sh->gauge_sentinel) {
if (hash < rb->key) { rb = rb->left; continue; }
if (hash > rb->key) { rb = rb->right; continue; }
g = (ngx_http_ipng_stats_gauge_t *) rb;
if (g->source_id == source_id && g->vip_id == vip_id) {
break;
}
rb = rb->right;
g = NULL;
}
if (g == NULL) {
g = ngx_slab_calloc_locked(slab, sizeof(*g));
if (g == NULL) {
(void) ngx_atomic_fetch_add(&sh->zone_full_events, 1);
ngx_shmtx_unlock(&slab->mutex);
return NULL;
}
g->rbnode.key = hash;
g->source_id = source_id;
g->vip_id = vip_id;
ngx_rbtree_insert(&sh->gauge_rbtree, &g->rbnode);
ngx_queue_insert_tail(&sh->gauge_lru, &g->lru);
}
ngx_shmtx_unlock(&slab->mutex);
return g;
}
/* ----------------------------------------------------------------- */
/* Global logtail: write + flush */
/* ----------------------------------------------------------------- */
@@ -2461,6 +2719,9 @@ typedef struct {
uint64_t bytes_out_sum;
uint64_t req_total; /* total requests for this (source, vip) */
uint64_t up_total; /* upstream observations */
uint64_t active; /* in-flight gauges (request lifecycle) */
uint64_t reading;
uint64_t writing;
uint64_t *dhist; /* nbuckets+1 */
uint64_t *uhist;
uint64_t *bin_hist; /* nbytebuckets+1 */
@@ -2737,6 +2998,60 @@ ngx_http_ipng_stats_snapshot_nodes(ngx_http_ipng_stats_shctx_t *sh,
}
/* Walk the gauge rbtree under the slab mutex and fold in-flight gauge
* values into the aggregation table. Creates new agg entries for
* (source, vip) pairs that have in-flight requests but no completed
* ones yet, so scrapes reflect load even before the first log-phase
* update for a pair. Respects the same filter semantics as the
* counter walk. */
static void
ngx_http_ipng_stats_snapshot_gauges(ngx_http_ipng_stats_shctx_t *sh,
ngx_str_t *filter_src, ngx_str_t *filter_vip,
ngx_str_t *src_tbl, ngx_uint_t n_src,
ngx_str_t *vip_tbl, ngx_uint_t n_vip,
ngx_http_ipng_stats_agg_t *aggs, ngx_uint_t naggs_alloc,
ngx_uint_t *naggs_io)
{
ngx_queue_t *q;
ngx_http_ipng_stats_gauge_t *g;
ngx_str_t *src_entry, *vip_entry;
ngx_http_ipng_stats_agg_t *a;
for (q = ngx_queue_head(&sh->gauge_lru);
q != ngx_queue_sentinel(&sh->gauge_lru);
q = ngx_queue_next(q))
{
g = ngx_queue_data(q, ngx_http_ipng_stats_gauge_t, lru);
if (g->source_id >= n_src || g->vip_id >= n_vip) continue;
src_entry = &src_tbl[g->source_id];
vip_entry = &vip_tbl[g->vip_id];
if (filter_src->len > 0
&& (src_entry->len != filter_src->len
|| ngx_memcmp(src_entry->data, filter_src->data,
filter_src->len) != 0))
{
continue;
}
if (filter_vip->len > 0
&& (vip_entry->len != filter_vip->len
|| ngx_memcmp(vip_entry->data, filter_vip->data,
filter_vip->len) != 0))
{
continue;
}
a = ngx_http_ipng_stats_agg_get(aggs, naggs_io, naggs_alloc,
g->source_id, g->vip_id);
if (a == NULL) continue;
a->active = g->active;
a->reading = g->reading;
a->writing = g->writing;
}
}
/* -- Prometheus ---------------------------------------------------- */
@@ -2817,6 +3132,13 @@ ngx_http_ipng_stats_render_prom(ngx_http_request_t *r,
"# TYPE nginx_ipng_bytes_in histogram\n"
"# HELP nginx_ipng_bytes_out Request size histogram in bytes.\n"
"# TYPE nginx_ipng_bytes_out histogram\n"
"# HELP nginx_ipng_active Requests currently in flight.\n"
"# TYPE nginx_ipng_active gauge\n"
"# HELP nginx_ipng_reading In-flight requests in the pre-response "
"phases (rewrite/access/content).\n"
"# TYPE nginx_ipng_reading gauge\n"
"# HELP nginx_ipng_writing In-flight requests past header send.\n"
"# TYPE nginx_ipng_writing gauge\n"
"# HELP nginx_ipng_ifindex_misses_total Connections whose ingress "
"ifindex did not match any configured device= binding.\n"
"# TYPE nginx_ipng_ifindex_misses_total counter\n"
@@ -2891,6 +3213,10 @@ ngx_http_ipng_stats_render_prom(ngx_http_request_t *r,
snaps, nsnaps_alloc, &nsnaps,
aggs, naggs_alloc, &naggs);
ngx_http_ipng_stats_snapshot_gauges(sh, filter_source, filter_vip,
src_tbl, n_src, vip_tbl, n_vip,
aggs, naggs_alloc, &naggs);
ngx_shmtx_unlock(&slab->mutex);
/* Per-node counters. */
@@ -2915,9 +3241,11 @@ ngx_http_ipng_stats_render_prom(ngx_http_request_t *r,
}
}
/* One chain link per (source, vip) for the four aggregated histograms.
* Size: per-bucket line ~96B, + sum/count/+Inf per metric ~96B each. */
hist_sz = 256 + 96 * (2 * (nb + 1) + 2 * (nbb + 1)) + 4 * 200;
/* One chain link per (source, vip) for the four aggregated
* histograms plus the three in-flight gauges. Size: per-bucket
* line ~96B, sum/count/+Inf per metric ~96B each, three gauge
* lines ~80B each. */
hist_sz = 512 + 96 * (2 * (nb + 1) + 2 * (nbb + 1)) + 4 * 200;
for (i = 0; i < naggs; i++) {
ngx_http_ipng_stats_agg_t *a = &aggs[i];
@@ -2944,6 +3272,13 @@ ngx_http_ipng_stats_render_prom(ngx_http_request_t *r,
"nginx_ipng_bytes_out", src, vip,
imcf->byte_bucket_bounds, nbb, a->bout_hist,
(double) a->bytes_out_sum, 0);
p = ngx_sprintf(p,
"nginx_ipng_active{source_tag=\"%V\",vip=\"%V\"} %uL\n"
"nginx_ipng_reading{source_tag=\"%V\",vip=\"%V\"} %uL\n"
"nginx_ipng_writing{source_tag=\"%V\",vip=\"%V\"} %uL\n",
src, vip, a->active,
src, vip, a->reading,
src, vip, a->writing);
cl->buf->last = p;
if (ngx_http_ipng_stats_append(&last, cl) != NGX_OK) {
return NGX_HTTP_INTERNAL_SERVER_ERROR;
@@ -3045,11 +3380,15 @@ ngx_http_ipng_stats_render_json(ngx_http_request_t *r,
snaps, nsnaps_alloc, &nsnaps,
aggs, naggs_alloc, &naggs);
ngx_http_ipng_stats_snapshot_gauges(sh, filter_source, filter_vip,
src_tbl, n_src, vip_tbl, n_vip,
aggs, naggs_alloc, &naggs);
ngx_shmtx_unlock(&slab->mutex);
/* One JSON record per aggregated (source, vip). Size upper-bound
* accounts for: fixed overhead, up to NCLASSES class entries, 4
* histograms. */
* histograms, 3 gauges. */
rec_sz = 512
+ 160 * NGX_HTTP_IPNG_STATS_NCLASSES
+ 48 * (2 * (nb + 1) + 2 * (nbb + 1))
@@ -3131,7 +3470,9 @@ ngx_http_ipng_stats_render_json(ngx_http_request_t *r,
imcf->byte_bucket_bounds[j], a->bout_hist[j]);
}
}
p = ngx_sprintf(p, "}}}");
p = ngx_sprintf(p,
"}},\"gauges\":{\"active\":%uL,\"reading\":%uL,\"writing\":%uL}}",
a->active, a->reading, a->writing);
cl->buf->last = p;
if (ngx_http_ipng_stats_append(&last, cl) != NGX_OK) {