runtime/statsobj.c - main - githubmirror/rsyslog

runtime/statsobj.c
1 170 строк38 KB
Rainer Gerhards
statsobj: guard Prometheus escape writes
20 июл 2026, 13:06
20 июл 2026, 13:0664c8fd6
/* The statsobj object.
 *
 * This object provides a statistics-gathering facility inside rsyslog. This
 * functionality will be pragmatically implemented and extended.
 *
 * Copyright 2010-2021 Adiscon GmbH.
 *
 * This file is part of the rsyslog runtime library.
 *
 * Licensed 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
 *       -or-
 *       see COPYING.ASL20 in the source distribution
 *
 * 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.
 */
/**
 * @file statsobj.c
 * @brief Implementation of the rsyslog statistics object
 *
 * Each statsobj_t instance maintains a name, origin and a set of counters.
 * All instances are linked together so that GetAllStatsLines() can iterate
 * through them and emit their values. Counters may be 64 bit integers or
 * plain ints and are modified with atomic helpers when required.
 *
 * The module can output collected counters in several formats:
 *  - legacy key=value lines
 *  - JSON or JSON-ES (Elasticsearch compatible)
 *  - CEE / lumberjack records
 *  - Prometheus text exposition
 *
 * Statistics gathering is controlled by the ::GatherStats flag and can be
 * enabled via EnableStats().
 */

#include "config.h"
#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <inttypes.h>
#include <stdint.h>
#include <pthread.h>
#include <errno.h>
#include <time.h>
#include <assert.h>
#include <json.h>

#include "rsyslog.h"
#include "unicode-helper.h"
#include "obj.h"
#include "statsobj.h"
#include "srUtils.h"
#include "stringbuf.h"
#include "errmsg.h"
#include "hashtable.h"
#include "hashtable_itr.h"
#include "rsconf.h"


/* externally-visiable data (see statsobj.h for explanation) */
int GatherStats = 0;

/* static data */
DEFobjStaticHelpers;

/* doubly linked list of stats objects. Object is automatically linked to it
 * upon construction. Enqueue always happens at the front (simplifies logic).
 */
static statsobj_t *objRoot = NULL;
static statsobj_t *objLast = NULL;

static pthread_mutex_t mutStats;
static pthread_mutex_t mutSenders;

static struct hashtable *stats_senders = NULL;

/* ------------------------------ statsobj linked list maintenance  ------------------------------ */

static rsRetVal statsobjLock(pthread_mutex_t *mutex, const char *lockName) {
    const int ret = pthread_mutex_lock(mutex);
    if (ret != 0) {
        LogError(ret, RS_RET_CONC_CTRL_ERR, "statsobj: error locking %s mutex", lockName);
        return RS_RET_CONC_CTRL_ERR;
    }
    return RS_RET_OK;
}

static void statsobjUnlock(pthread_mutex_t *mutex, const char *lockName) {
    const int ret = pthread_mutex_unlock(mutex);
    if (ret != 0) {
        LogError(ret, RS_RET_CONC_CTRL_ERR, "statsobj: error unlocking %s mutex", lockName);
        abort();
    }
}

static rsRetVal addToObjList(statsobj_t *pThis) {
    DEFiRet;
    CHKiRet(statsobjLock(&mutStats, "stats object list"));
    if (pThis->flags & STATSOBJ_FLAG_DO_PREPEND) {
        pThis->next = objRoot;
        if (objRoot != NULL) {
            objRoot->prev = pThis;
        }
        objRoot = pThis;
        if (objLast == NULL) objLast = pThis;
    } else {
        pThis->prev = objLast;
        if (objLast != NULL) objLast->next = pThis;
        objLast = pThis;
        if (objRoot == NULL) objRoot = pThis;
    }
    /* Unlock failures happen after shared state was already updated.  Log them,
     * but do not make callers roll back objects that are now visible.
     */
    statsobjUnlock(&mutStats, "stats object list");
finalize_it:
    RETiRet;
}


static rsRetVal removeFromObjList(statsobj_t *pThis) {
    DEFiRet;
    CHKiRet(statsobjLock(&mutStats, "stats object list"));
    if (pThis->prev != NULL) pThis->prev->next = pThis->next;
    if (pThis->next != NULL) pThis->next->prev = pThis->prev;
    if (objLast == pThis) objLast = pThis->prev;
    if (objRoot == pThis) objRoot = pThis->next;
    statsobjUnlock(&mutStats, "stats object list");
finalize_it:
    RETiRet;
}


static rsRetVal addCtrToList(statsobj_t *pThis, ctr_t *pCtr) {
    DEFiRet;
    CHKiRet(statsobjLock(&pThis->mutCtr, "counter list"));
    pCtr->prev = pThis->ctrLast;
    if (pThis->ctrLast != NULL) pThis->ctrLast->next = pCtr;
    pThis->ctrLast = pCtr;
    if (pThis->ctrRoot == NULL) pThis->ctrRoot = pCtr;
    /* See addToObjList(): rollback after successful list mutation is unsafe. */
    statsobjUnlock(&pThis->mutCtr, "counter list");
finalize_it:
    RETiRet;
}

/* ------------------------------ methods ------------------------------ */


/* Standard-Constructor
 */
BEGINobjConstruct(statsobj) /* be sure to specify the object type also in END macro! */
    CHKiConcCtrl(pthread_mutex_init(&pThis->mutCtr, NULL));
    pThis->ctrLast = NULL;
    pThis->ctrRoot = NULL;
    pThis->read_notifier = NULL;
    pThis->flags = 0;
finalize_it:
ENDobjConstruct(statsobj)


/* ConstructionFinalizer
 */
static rsRetVal statsobjConstructFinalize(statsobj_t *pThis) {
    DEFiRet;
    ISOBJ_TYPE_assert(pThis, statsobj);
    CHKiRet(addToObjList(pThis));
finalize_it:
    RETiRet;
}

/* set read_notifier (a function which is invoked after stats are read).
 */
static rsRetVal setReadNotifier(statsobj_t *pThis, statsobj_read_notifier_t notifier, void *ctx) {
    DEFiRet;
    pThis->read_notifier = notifier;
    pThis->read_notifier_ctx = ctx;
    RETiRet;
}


/* set origin (module name, etc).
 * Note that we make our own copy of the memory, caller is
 * responsible to free up name it passes in (if required).
 */
static rsRetVal setOrigin(statsobj_t *pThis, uchar *origin) {
    DEFiRet;
    CHKmalloc(pThis->origin = ustrdup(origin));
finalize_it:
    RETiRet;
}


/* set name. Note that we make our own copy of the memory, caller is
 * responsible to free up name it passes in (if required).
 */
static rsRetVal setName(statsobj_t *pThis, uchar *name) {
    DEFiRet;
    CHKmalloc(pThis->name = ustrdup(name));
finalize_it:
    RETiRet;
}

static void setStatsObjFlags(statsobj_t *pThis, int flags) {
    pThis->flags = flags;
}

static rsRetVal setReportingNamespace(statsobj_t *pThis, uchar *ns) {
    DEFiRet;
    CHKmalloc(pThis->reporting_ns = ustrdup(ns));
finalize_it:
    RETiRet;
}

/* add a counter to an object
 * ctrName is duplicated, caller must free it if requried
 * NOTE: The counter is READ-ONLY and MUST NOT be modified (most
 * importantly, it must not be initialized, so the caller must
 * ensure the counter is properly initialized before AddCounter()
 * is called.
 */
static rsRetVal addManagedCounter(statsobj_t *pThis,
                                  const uchar *ctrName,
                                  statsCtrType_t ctrType,
                                  int8_t flags,
                                  void *pCtr,
                                  ctr_t **entryRef,
                                  int8_t linked) {
    ctr_t *ctr;
    DEFiRet;

    *entryRef = NULL;

    CHKmalloc(ctr = calloc(1, sizeof(ctr_t)));
    ctr->next = NULL;
    ctr->prev = NULL;
    if ((ctr->name = ustrdup(ctrName)) == NULL) {
        DBGPRINTF("addCounter: OOM in strdup()\n");
        ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
    }
    ctr->flags = flags;
    ctr->ctrType = ctrType;
    switch (ctrType) {
        case ctrType_IntCtr:
            ctr->val.pIntCtr = (intctr_t *)pCtr;
            break;
        case ctrType_Int:
            ctr->val.pInt = (int *)pCtr;
            break;
        default:
            // No action needed for other cases
            break;
    }
    if (linked) {
        CHKiRet(addCtrToList(pThis, ctr));
    }
    *entryRef = ctr;

finalize_it:
    if (iRet != RS_RET_OK) {
        if (ctr != NULL) {
            free(ctr->name);
            free(ctr);
        }
    }
    RETiRet;
}

static rsRetVal addPreCreatedCounter(statsobj_t *pThis, ctr_t *pCtr) {
    pCtr->next = NULL;
    pCtr->prev = NULL;
    return addCtrToList(pThis, pCtr);
}

static rsRetVal addCounter(statsobj_t *pThis, const uchar *ctrName, statsCtrType_t ctrType, int8_t flags, void *pCtr) {
    ctr_t *ctr;
    DEFiRet;
    iRet = addManagedCounter(pThis, ctrName, ctrType, flags, pCtr, &ctr, 1);
    RETiRet;
}

static void destructUnlinkedCounter(ctr_t *ctr) {
    free(ctr->name);
    free(ctr);
}

static void destructCounter(statsobj_t *pThis, ctr_t *pCtr) {
    pthread_mutex_lock(&pThis->mutCtr);
    if (pCtr->prev != NULL) {
        pCtr->prev->next = pCtr->next;
    }
    if (pCtr->next != NULL) {
        pCtr->next->prev = pCtr->prev;
    }
    if (pThis->ctrLast == pCtr) {
        pThis->ctrLast = pCtr->prev;
    }
    if (pThis->ctrRoot == pCtr) {
        pThis->ctrRoot = pCtr->next;
    }
    pthread_mutex_unlock(&pThis->mutCtr);
    destructUnlinkedCounter(pCtr);
}

static intctr_t getIntCtrValue(const intctr_t *const ctr) {
    return PREFER_LOAD_uint64(ctr);
}


static void resetIntCtrValue(intctr_t *const ctr) {
    PREFER_STORE_uint64(ctr, 0);
}


static int getIntValue(const int *const ctr) {
    return PREFER_LOAD_INT(ctr);
}


static void resetIntValue(int *const ctr) {
    PREFER_STORE_INT(ctr, 0);
}


static void resetResettableCtr(ctr_t *pCtr, int8_t bResetCtrs) {
    if ((bResetCtrs && (pCtr->flags & CTR_FLAG_RESETTABLE)) || (pCtr->flags & CTR_FLAG_MUST_RESET)) {
        switch (pCtr->ctrType) {
            case ctrType_IntCtr:
                resetIntCtrValue(pCtr->val.pIntCtr);
                break;
            case ctrType_Int:
                resetIntValue(pCtr->val.pInt);
                break;
            default:
                // No action needed for other cases
                break;
        }
    }
}

static rsRetVal addCtrForReporting(json_object *to, const uchar *field_name, intctr_t value) {
    json_object *v;
    DEFiRet;

    /*We should migrate libfastjson to support uint64_t in addition to int64_t.
      Although no counter is likely to grow to int64 max-value, this is theoritically
      incorrect (as intctr_t is uint64)*/
    CHKmalloc(v = json_object_new_int64((int64_t)value));

    json_object_object_add(to, (const char *)field_name, v);
finalize_it:
    /* v cannot be NULL in error case, as this would only happen during malloc fail,
     * which itself sets it to NULL -- so not doing cleanup here.
     */
    RETiRet;
}

static rsRetVal addContextForReporting(json_object *to, const uchar *field_name, const uchar *value) {
    json_object *v;
    DEFiRet;

    CHKmalloc(v = json_object_new_string((const char *)value));

    json_object_object_add(to, (const char *)field_name, v);
finalize_it:
    RETiRet;
}

static intctr_t accumulatedValue(ctr_t *pCtr) {
    switch (pCtr->ctrType) {
        case ctrType_IntCtr:
            return getIntCtrValue(pCtr->val.pIntCtr);
        case ctrType_Int:
            return (intctr_t)getIntValue(pCtr->val.pInt);
        default:
            // No action needed for other cases
            break;
    }
    return -1;
}


/* get all the object's countes together as CEE. */
static rsRetVal getStatsLineCEE(statsobj_t *pThis, cstr_t **ppcstr, const statsFmtType_t fmt, const int8_t bResetCtrs) {
    cstr_t *pcstr = NULL;
    ctr_t *pCtr;
    json_object *root, *values;
    int locked = 0;
    DEFiRet;

    root = values = NULL;

    CHKiRet(cstrConstruct(&pcstr));

    if (fmt == statsFmt_CEE)
        CHKiRet(rsCStrAppendStrWithLen(pcstr, UCHAR_CONSTANT(CONST_CEE_COOKIE " "), CONST_LEN_CEE_COOKIE + 1));

    CHKmalloc(root = json_object_new_object());

    CHKiRet(addContextForReporting(root, UCHAR_CONSTANT("name"), pThis->name));

    if (pThis->origin != NULL) {
        CHKiRet(addContextForReporting(root, UCHAR_CONSTANT("origin"), pThis->origin));
    }

    if (pThis->reporting_ns == NULL) {
        values = json_object_get(root);
    } else {
        CHKmalloc(values = json_object_new_object());
        json_object_object_add(root, (const char *)pThis->reporting_ns, json_object_get(values));
    }

    /* now add all counters to this line */
    pthread_mutex_lock(&pThis->mutCtr);
    locked = 1;
    for (pCtr = pThis->ctrRoot; pCtr != NULL; pCtr = pCtr->next) {
        if (fmt == statsFmt_JSON_ES) {
            /* work-around for broken Elasticsearch JSON implementation:
             * we need to replace dots by a different char, we use bang.
             * Note: ES 2.0 does not longer accept dot in name
             */
            uchar esbuf[256];
            const size_t ctr_name_len = strlen((char *)pCtr->name);
            const size_t esbuf_len = ctr_name_len < sizeof(esbuf) - 1 ? ctr_name_len : sizeof(esbuf) - 1;
            memcpy(esbuf, pCtr->name, esbuf_len);
            esbuf[esbuf_len] = '\0';
            for (uchar *c = esbuf; *c; ++c) {
                if (*c == '.') *c = '!';
            }
            CHKiRet(addCtrForReporting(values, esbuf, accumulatedValue(pCtr)));
        } else {
            CHKiRet(addCtrForReporting(values, pCtr->name, accumulatedValue(pCtr)));
        }
        resetResettableCtr(pCtr, bResetCtrs);
    }
    pthread_mutex_unlock(&pThis->mutCtr);
    locked = 0;
    CHKiRet(rsCStrAppendStr(pcstr, (const uchar *)json_object_to_json_string(root)));

    cstrFinalize(pcstr);
    *ppcstr = pcstr;
    pcstr = NULL;

finalize_it:
    if (locked) {
        pthread_mutex_unlock(&pThis->mutCtr);
    }

    if (pcstr != NULL) {
        cstrDestruct(&pcstr);
    }
    if (root != NULL) {
        json_object_put(root);
    }
    if (values != NULL) {
        json_object_put(values);
    }

    RETiRet;
}

/* get all the object's countes together with object name as one line.
 */
static rsRetVal getStatsLine(statsobj_t *pThis, cstr_t **ppcstr, int8_t bResetCtrs) {
    cstr_t *pcstr;
    ctr_t *pCtr;
    DEFiRet;

    CHKiRet(cstrConstruct(&pcstr));
    rsCStrAppendStr(pcstr, pThis->name);
    rsCStrAppendStrWithLen(pcstr, UCHAR_CONSTANT(": "), 2);

    if (pThis->origin != NULL) {
        rsCStrAppendStrWithLen(pcstr, UCHAR_CONSTANT("origin="), 7);
        rsCStrAppendStr(pcstr, pThis->origin);
        cstrAppendChar(pcstr, ' ');
    }

    /* now add all counters to this line */
    pthread_mutex_lock(&pThis->mutCtr);
    for (pCtr = pThis->ctrRoot; pCtr != NULL; pCtr = pCtr->next) {
        rsCStrAppendStr(pcstr, pCtr->name);
        cstrAppendChar(pcstr, '=');
        switch (pCtr->ctrType) {
            case ctrType_IntCtr:
                rsCStrAppendInt(pcstr, getIntCtrValue(pCtr->val.pIntCtr));
                break;
            case ctrType_Int:
                rsCStrAppendInt(pcstr, getIntValue(pCtr->val.pInt));
                break;
            default:
                // No action needed for other cases
                break;
        }
        cstrAppendChar(pcstr, ' ');
        resetResettableCtr(pCtr, bResetCtrs);
    }
    pthread_mutex_unlock(&pThis->mutCtr);

    cstrFinalize(pcstr);
    *ppcstr = pcstr;

finalize_it:
    RETiRet;
}


/* this function obtains all sender stats. hlper to getAllStatsLines()
 * We need to keep this looked to avoid resizing of the hash table
 * (what could otherwise cause a segfault).
 */
static void getSenderStats(rsRetVal (*cb)(void *, const char *),
                           void *usrptr,
                           statsFmtType_t fmt,
                           const int8_t bResetCtrs) {
    struct hashtable_itr *itr = NULL;
    struct sender_stats *stat;
    char fmtbuf[2048];

    pthread_mutex_lock(&mutSenders);

    /* Iterator constructor only returns a valid iterator if
     * the hashtable is not empty
     */
    if (hashtable_count(stats_senders) > 0) {
        itr = hashtable_iterator(stats_senders);
        do {
            stat = (struct sender_stats *)hashtable_iterator_value(itr);
            if (fmt == statsFmt_Legacy) {
                snprintf(fmtbuf, sizeof(fmtbuf), "_sender_stat: sender=%s messages=%" PRIu64, stat->sender,
                         stat->nMsgs);
            } else {
                snprintf(fmtbuf, sizeof(fmtbuf),
                         "{ \"name\":\"_sender_stat\", "
                         "\"origin\":\"impstats\", "
                         "\"sender\":\"%s\", \"messages\":%" PRIu64 "}",
                         stat->sender, stat->nMsgs);
            }
            fmtbuf[sizeof(fmtbuf) - 1] = '\0';
            cb(usrptr, fmtbuf);
            if (bResetCtrs) stat->nMsgs = 0;
        } while (hashtable_iterator_advance(itr));
    }

    free(itr);
    pthread_mutex_unlock(&mutSenders);
}


/*
 * Prometheus values escaping reserves the U__ prefix. Ordinary legacy-safe
 * names remain unchanged; all other names use a reversible U__ encoding.
 */
static int prometheusLegacySafeName(const uchar *name) {
    const uchar *p;

    if (name == NULL || name[0] == '\0') return 0;
    if (!((name[0] >= 'a' && name[0] <= 'z') || (name[0] >= 'A' && name[0] <= 'Z') || name[0] == '_' ||
          name[0] == ':')) {
        return 0;
    }
    if (name[0] == 'U' && strncmp((const char *)name, "U__", 3) == 0) return 0;
    for (p = name + 1; *p != '\0'; ++p) {
        if (!((*p >= 'a' && *p <= 'z') || (*p >= 'A' && *p <= 'Z') || (*p >= '0' && *p <= '9') || *p == '_' ||
              *p == ':')) {
            return 0;
        }
    }
    return 1;
}

static int decodeUtf8Codepoint(const uchar *input, size_t len, size_t *consumed, uint32_t *codepoint) {
    const uint32_t b0 = input[0];

    if (b0 < 0x80) {
        *consumed = 1;
        *codepoint = b0;
        return 1;
    }
    if (b0 >= 0xc2 && b0 <= 0xdf && len >= 2 && (input[1] & 0xc0) == 0x80) {
        *consumed = 2;
        *codepoint = ((b0 & 0x1f) << 6) | (input[1] & 0x3f);
        return 1;
    }
    if (b0 >= 0xe0 && b0 <= 0xef && len >= 3 && (input[1] & 0xc0) == 0x80 && (input[2] & 0xc0) == 0x80) {
        *codepoint = ((b0 & 0x0f) << 12) | ((input[1] & 0x3f) << 6) | (input[2] & 0x3f);
        if (*codepoint >= 0x800 && !(*codepoint >= 0xd800 && *codepoint <= 0xdfff)) {
            *consumed = 3;
            return 1;
        }
    }
    if (b0 >= 0xf0 && b0 <= 0xf4 && len >= 4 && (input[1] & 0xc0) == 0x80 && (input[2] & 0xc0) == 0x80 &&
        (input[3] & 0xc0) == 0x80) {
        *codepoint = ((b0 & 0x07) << 18) | ((input[1] & 0x3f) << 12) | ((input[2] & 0x3f) << 6) | (input[3] & 0x3f);
        if (*codepoint >= 0x10000 && *codepoint <= 0x10ffff) {
            *consumed = 4;
            return 1;
        }
    }
    *consumed = 1;
    *codepoint = b0;
    return 0;
}

static rsRetVal encodePrometheusMetricName(const uchar *raw_name, char **encoded_name) {
    const size_t raw_len = raw_name == NULL ? 0 : strlen((const char *)raw_name);
    size_t offset = 0;
    size_t out_len = 0;
    size_t remaining;
    int written;
    char *out = NULL;
    DEFiRet;

    if (encoded_name == NULL || raw_name == NULL) ABORT_FINALIZE(RS_RET_PARAM_ERROR);
    *encoded_name = NULL;
    if (prometheusLegacySafeName(raw_name)) {
        CHKmalloc(out = strdup((const char *)raw_name));
        *encoded_name = out;
        out = NULL;
        FINALIZE;
    }
    if (raw_len > (SIZE_MAX - 4) / 10) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
    CHKmalloc(out = malloc(raw_len * 10 + 4));
    memcpy(out, "U__", 3);
    out_len = 3;
    while (offset < raw_len) {
        size_t consumed;
        uint32_t codepoint;
        const int valid = decodeUtf8Codepoint(raw_name + offset, raw_len - offset, &consumed, &codepoint);
        if (valid && ((codepoint >= 'a' && codepoint <= 'z') || (codepoint >= 'A' && codepoint <= 'Z') ||
                      (codepoint >= '0' && codepoint <= '9') || codepoint == ':')) {
            out[out_len++] = (char)codepoint;
        } else if (valid && codepoint == '_') {
            out[out_len++] = '_';
            out[out_len++] = '_';
        } else if (valid) {
            remaining = raw_len * 10 + 4 - out_len;
            written = snprintf(out + out_len, remaining, "_%X_", codepoint);
            if (written < 0 || (size_t)written >= remaining) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
            out_len += (size_t)written;
        } else {
            remaining = raw_len * 10 + 4 - out_len;
            written = snprintf(out + out_len, remaining, "_x%02X_", codepoint);
            if (written < 0 || (size_t)written >= remaining) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
            out_len += (size_t)written;
        }
        offset += consumed;
    }
    out[out_len] = '\0';
    *encoded_name = out;
    out = NULL;
finalize_it:
    free(out);
    RETiRet;
}

static rsRetVal escapePrometheusHelp(const char *input, const char **escaped, char **allocated) {
    size_t i;
    size_t len;
    size_t out_len = 0;
    char *out = NULL;
    DEFiRet;

    if (input == NULL || escaped == NULL || allocated == NULL) ABORT_FINALIZE(RS_RET_PARAM_ERROR);
    *escaped = input;
    *allocated = NULL;
    len = strlen(input);
    for (i = 0; i < len; ++i) {
        if (input[i] == '\\' || input[i] == '"' || input[i] == '\n') break;
    }
    if (i == len) FINALIZE;
    if (len > (SIZE_MAX - 1) / 2) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
    CHKmalloc(out = malloc(len * 2 + 1));
    for (i = 0; i < len; ++i) {
        if (input[i] == '\\' || input[i] == '"') {
            out[out_len++] = '\\';
            out[out_len++] = input[i];
        } else if (input[i] == '\n') {
            out[out_len++] = '\\';
            out[out_len++] = 'n';
        } else {
            out[out_len++] = input[i];
        }
    }
    out[out_len] = '\0';
    *escaped = out;
    *allocated = out;
    out = NULL;
finalize_it:
    free(out);
    RETiRet;
}

/**
 * Helper: For a single statsobj_t (named o->name), iterate its counters
 * and emit Prometheus lines via cb.  We generate, for each counter:
 *   # HELP <obj>_<ctr> Generic help: "<origin> object, counter <ctr>"
 *   # TYPE <obj>_<ctr> counter
 *   <obj>_<ctr> <value>
 *
 * If bResetCtrs=TRUE and the counter has CTR_FLAG_RESETTABLE, zero it after reading.
 *
 * Note: by rsyslog stats subsystem design decision, read and write counters is racy
 *       because we need the performance. It is OK that the counters in question are
 *       not 100% precise.
 */
static ATTR_NO_SANITIZE_THREAD rsRetVal emitPrometheusForObject(statsobj_t *o,
                                                                rsRetVal (*cb)(void *, const char *),
                                                                void *usrptr,
                                                                int8_t bResetCtrs) {
    ctr_t *pCtr;
    char *raw_name = NULL;
    char *metric_name = NULL;
    const char *escaped_origin;
    const char *escaped_object;
    const char *escaped_counter;
    char *escaped_origin_alloc = NULL;
    char *escaped_object_alloc = NULL;
    char *escaped_counter_alloc = NULL;
    char *line = NULL;
    rsRetVal iRet;
    uint64_t value;
    const char *objName = (const char *)o->name;
    const char *origin = o->origin ? (const char *)o->origin : "";

    /* Iterate each counter in o->ctrRoot.  Lock while walking the linked list. */
    pthread_mutex_lock(&o->mutCtr);
    for (pCtr = o->ctrRoot; pCtr != NULL; pCtr = pCtr->next) {
        /* 1) Read the current accumulated value.  Might be IntCtr or Int. */
        switch (pCtr->ctrType) {
            case ctrType_IntCtr:
                value = getIntCtrValue(pCtr->val.pIntCtr);
                break;
            case ctrType_Int:
                value = (uint64_t)getIntValue(pCtr->val.pInt);
                break;
            default:
                value = 0;
                break;
        }

        /* 2) Optionally reset if requested and allowed. */
        if ((bResetCtrs && (pCtr->flags & CTR_FLAG_RESETTABLE)) || (pCtr->flags & CTR_FLAG_MUST_RESET)) {
            switch (pCtr->ctrType) {
                case ctrType_IntCtr:
                    resetIntCtrValue(pCtr->val.pIntCtr);
                    break;
                case ctrType_Int:
                    resetIntValue(pCtr->val.pInt);
                    break;
                default:
                    break;
            }
        }
        pthread_mutex_unlock(&o->mutCtr);

        /* 3) Build the metric name: "<object>_<counter>_total". */
        if (asprintf(&raw_name, "%s_%s_total", objName, pCtr->name) < 0) {
            raw_name = NULL;
            return RS_RET_OUT_OF_MEMORY;
        }
        if (raw_name == NULL || encodePrometheusMetricName((const uchar *)raw_name, &metric_name) != RS_RET_OK ||
            escapePrometheusHelp(origin, &escaped_origin, &escaped_origin_alloc) != RS_RET_OK ||
            escapePrometheusHelp(objName, &escaped_object, &escaped_object_alloc) != RS_RET_OK ||
            escapePrometheusHelp((const char *)pCtr->name, &escaped_counter, &escaped_counter_alloc) != RS_RET_OK) {
            free(raw_name);
            free(metric_name);
            free(escaped_origin_alloc);
            free(escaped_object_alloc);
            free(escaped_counter_alloc);
            return RS_RET_OUT_OF_MEMORY;
        }
        if (asprintf(&line,
                     "# HELP %s rsyslog stats: origin=\"%s\" object=\"%s\", counter=\"%s\"\n"
                     "# TYPE %s counter\n"
                     "%s %llu\n",
                     metric_name, escaped_origin, escaped_object, escaped_counter, metric_name, metric_name,
                     (unsigned long long)value) < 0) {
            line = NULL;
            free(raw_name);
            free(metric_name);
            free(escaped_origin_alloc);
            free(escaped_object_alloc);
            free(escaped_counter_alloc);
            return RS_RET_OUT_OF_MEMORY;
        }
        iRet = cb(usrptr, line);
        free(raw_name);
        free(metric_name);
        free(escaped_origin_alloc);
        free(escaped_object_alloc);
        free(escaped_counter_alloc);
        free(line);
        raw_name = metric_name = escaped_origin_alloc = escaped_object_alloc = escaped_counter_alloc = line = NULL;
        if (iRet != RS_RET_OK) return iRet;

        /* Acquire the lock again before advancing to the next counter */
        pthread_mutex_lock(&o->mutCtr);
    }
    pthread_mutex_unlock(&o->mutCtr);
    return RS_RET_OK;
}


static rsRetVal generatePrometheusStats(rsRetVal (*cb)(void *, const char *), void *usrptr, int8_t bResetCtrs) {
    statsobj_t *o;
    int listLocked = 0;
    DEFiRet;

    /* For each statsobj in our linked list, emit Prometheus lines. */
    pthread_mutex_lock(&mutStats);
    listLocked = 1;
    for (o = objRoot; o != NULL; o = o->next) {
        CHKiRet(emitPrometheusForObject(o, cb, usrptr, bResetCtrs));
        /* If the object has a read_notifier, call it now */
        if (o->read_notifier != NULL) {
            o->read_notifier(o, o->read_notifier_ctx);
        }
    }
    pthread_mutex_unlock(&mutStats);
    listLocked = 0;
    /* Optionally, handle sender stats as additional metrics:
     * e.g. emit "rsyslog_sender_<sender> <nMsgs>" lines.
     * For simplicity, we skip this, or you can extend similarly. */
finalize_it:
    if (listLocked) {
        pthread_mutex_unlock(&mutStats);
    }
    RETiRet;
}

/* this function can be used to obtain all stats lines. In this case,
 * a callback must be provided. This module than iterates over all objects and
 * submits each stats line to the callback. The callback has two parameters:
 * the first one is a caller-provided void*, the second one the cstr_t with the
 * line. If the callback reports an error, processing is stopped.
 */
static rsRetVal getAllStatsLines(rsRetVal (*cb)(void *, const char *),
                                 void *const usrptr,
                                 statsFmtType_t fmt,
                                 const int8_t bResetCtrs) {
    statsobj_t *o;
    cstr_t *cstr = NULL;
    int listLocked = 0;
    DEFiRet;


    if (fmt == statsFmt_Prometheus) {
        CHKiRet(generatePrometheusStats(cb, usrptr, bResetCtrs));
        FINALIZE;
    }

    pthread_mutex_lock(&mutStats);
    listLocked = 1;
    for (o = objRoot; o != NULL; o = o->next) {
        switch (fmt) {
            case statsFmt_Legacy:
                CHKiRet(getStatsLine(o, &cstr, bResetCtrs));
                break;
            case statsFmt_CEE:
            case statsFmt_JSON:
            case statsFmt_JSON_ES:
                CHKiRet(getStatsLineCEE(o, &cstr, fmt, bResetCtrs));
                break;
            case statsFmt_Prometheus:
                /* already handled above */
                break;
            default:
                // No action needed for other cases
                break;
        }
        CHKiRet(cb(usrptr, (const char *)cstrGetSzStrNoNULL(cstr)));
        rsCStrDestruct(&cstr);
        if (o->read_notifier != NULL) {
            o->read_notifier(o, o->read_notifier_ctx);
        }
    }
    pthread_mutex_unlock(&mutStats);
    listLocked = 0;

    getSenderStats(cb, usrptr, fmt, bResetCtrs);

finalize_it:
    if (listLocked) {
        pthread_mutex_unlock(&mutStats);
    }
    if (cstr != NULL) {
        rsCStrDestruct(&cstr);
    }
    RETiRet;
}

/**
 * getAllCounters() - Native counter iteration API
 *
 * Iterates through all statsobj instances and their counters, invoking the callback
 * with raw counter values. This eliminates text serialization/parsing overhead.
 *
 * @param cb  Callback function invoked for each counter
 * @param ctx User context passed to callback
 * @return RS_RET_OK on success, error code on failure
 */
static rsRetVal getAllCounters(statsobj_counter_cb_t cb, void *ctx) {
    DEFiRet;
    statsobj_t *o;
    ctr_t *ctr;
    int listLocked = 0;

    if (cb == NULL) {
        ABORT_FINALIZE(RS_RET_PARAM_ERROR);
    }

    /* Iterate through all statsobj instances */
    pthread_mutex_lock(&mutStats);
    listLocked = 1;
    for (o = objRoot; o != NULL; o = o->next) {
        /* Iterate through all counters in this object */
        pthread_mutex_lock(&o->mutCtr);
        for (ctr = o->ctrRoot; ctr != NULL; ctr = ctr->next) {
            uint64_t value;

            /* Read counter value based on type */
            switch (ctr->ctrType) {
                case ctrType_IntCtr:
                    value = getIntCtrValue(ctr->val.pIntCtr);
                    break;

                case ctrType_Int:
                    /* Plain int - read without lock. This matches GetAllStatsLines()
                     * behavior. Value may be stale but acceptable for monitoring.
                     * Most ctrType_Int counters are gauges (queue size, open files)
                     * protected by application-level mutexes. */
                    value = (uint64_t)getIntValue(ctr->val.pInt);
                    break;

                default:
                    value = 0;
                    break;
            }

            /* Invoke callback with counter metadata and value.
             * Keep mutCtr locked to prevent list modification during iteration.
             * Callback must not call back into statsobj or deadlock may occur. */
            rsRetVal localRet = cb(ctx, o->name, o->origin, ctr->name, ctr->ctrType, value, ctr->flags);

            if (localRet != RS_RET_OK) {
                pthread_mutex_unlock(&o->mutCtr);
                ABORT_FINALIZE(localRet);
            }
        }
        pthread_mutex_unlock(&o->mutCtr);

        /* Call read notifier after counters are read (e.g., for percentile stats) */
        if (o->read_notifier != NULL) {
            o->read_notifier(o, o->read_notifier_ctx);
        }
    }
    pthread_mutex_unlock(&mutStats);
    listLocked = 0;

finalize_it:
    if (listLocked) {
        pthread_mutex_unlock(&mutStats);
    }
    RETiRet;
}

/* Enable statistics gathering. currently there is no function to disable it
 * again, as this is right now not needed.
 */
static rsRetVal enableStats(void) {
    PREFER_STORE_1_TO_INT(&GatherStats);
    return RS_RET_OK;
}


rsRetVal statsRecordSender(const uchar *sender, unsigned nMsgs, time_t lastSeen) {
    struct sender_stats *stat;
    int mustUnlock = 0;
    DEFiRet;

    if (stats_senders == NULL) FINALIZE; /* unlikely: we could not init our hash table */

    pthread_mutex_lock(&mutSenders);
    mustUnlock = 1;
    stat = hashtable_search(stats_senders, (void *)sender);
    if (stat == NULL) {
        DBGPRINTF("statsRecordSender: sender '%s' not found, adding\n", sender);
        CHKmalloc(stat = calloc(1, sizeof(struct sender_stats)));
        stat->sender = (const uchar *)strdup((const char *)sender);
        stat->nMsgs = 0;
        if (runConf->globals.reportNewSenders) {
            LogMsg(0, RS_RET_SENDER_APPEARED, LOG_INFO, "new sender '%s'", stat->sender);
        }
        if (hashtable_insert(stats_senders, (void *)stat->sender, (void *)stat) == 0) {
            LogError(errno, RS_RET_INTERNAL_ERROR,
                     "error inserting sender '%s' into sender "
                     "hash table",
                     sender);
            ABORT_FINALIZE(RS_RET_INTERNAL_ERROR);
        }
    }

    stat->nMsgs += nMsgs;
    stat->lastSeen = lastSeen;
    DBGPRINTF("DDDDD: statsRecordSender: '%s', nmsgs %u [%llu], lastSeen %llu\n", sender, nMsgs,
              (long long unsigned)stat->nMsgs, (long long unsigned)lastSeen);

finalize_it:
    if (mustUnlock) pthread_mutex_unlock(&mutSenders);
    RETiRet;
}

static ctr_t *unlinkAllCounters(statsobj_t *pThis) {
    ctr_t *ctr;
    pthread_mutex_lock(&pThis->mutCtr);
    ctr = pThis->ctrRoot;
    pThis->ctrLast = NULL;
    pThis->ctrRoot = NULL;
    pthread_mutex_unlock(&pThis->mutCtr);
    return ctr;
}

static void destructUnlinkedCounters(ctr_t *ctr) {
    ctr_t *ctrToDel;

    while (ctr != NULL) {
        ctrToDel = ctr;
        ctr = ctr->next;
        destructUnlinkedCounter(ctrToDel);
    }
}

/* check if a sender has not sent info to us for an extended period
 * of time.
 */
void checkGoneAwaySenders(const time_t tCurr) {
    struct hashtable_itr *itr = NULL;
    struct sender_stats *stat;
    const time_t rqdLast = tCurr - runConf->globals.senderStatsTimeout;
    struct tm tm;

    pthread_mutex_lock(&mutSenders);

    /* Iterator constructor only returns a valid iterator if
     * the hashtable is not empty
     */
    if (hashtable_count(stats_senders) > 0) {
        itr = hashtable_iterator(stats_senders);
        do {
            stat = (struct sender_stats *)hashtable_iterator_value(itr);
            if (stat->lastSeen < rqdLast) {
                if (runConf->globals.reportGoneAwaySenders) {
                    localtime_r(&stat->lastSeen, &tm);
                    LogMsg(0, RS_RET_SENDER_GONE_AWAY, LOG_WARNING,
                           "removing sender '%s' from connection "
                           "table, last seen at "
                           "%4.4d-%2.2d-%2.2d %2.2d:%2.2d:%2.2d",
                           stat->sender, tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday, tm.tm_hour, tm.tm_min,
                           tm.tm_sec);
                }
                hashtable_remove(stats_senders, (void *)stat->sender);
            }
        } while (hashtable_iterator_advance(itr));
    }

    pthread_mutex_unlock(&mutSenders);
    free(itr);
}

/* destructor for the statsobj object */
BEGINobjDestruct(statsobj) /* be sure to specify the object type also in END and CODESTART macros! */
    CODESTARTobjDestruct(statsobj);
    iRet = removeFromObjList(pThis);
    /* Do not enter ENDobjDestruct on unlink failure: that macro frees pThis
     * and would leave objRoot/objLast pointing at released memory. There is no
     * safe recovery path once we cannot lock the global list for destruction.
     */
    if (iRet != RS_RET_OK) abort();

    /* destruct counters */
    destructUnlinkedCounters(unlinkAllCounters(pThis));

    pthread_mutex_destroy(&pThis->mutCtr);
    free(pThis->name);
    free(pThis->origin);
    free(pThis->reporting_ns);
ENDobjDestruct(statsobj)


/* debugprint for the statsobj object */
BEGINobjDebugPrint(statsobj) /* be sure to specify the object type also in END and CODESTART macros! */
    CODESTARTobjDebugPrint(statsobj);
    dbgoprint((obj_t *)pThis, "statsobj object, currently no state info available\n");
ENDobjDebugPrint(statsobj)


/* queryInterface function
 */
BEGINobjQueryInterface(statsobj)
    CODESTARTobjQueryInterface(statsobj);
    if (pIf->ifVersion != statsobjCURR_IF_VERSION) { /* check for current version, increment on each change */
        ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED);
    }

    /* ok, we have the right interface, so let's fill it
     * Please note that we may also do some backwards-compatibility
     * work here (if we can support an older interface version - that,
     * of course, also affects the "if" above).
     */
    pIf->Construct = statsobjConstruct;
    pIf->ConstructFinalize = statsobjConstructFinalize;
    pIf->Destruct = statsobjDestruct;
    pIf->DebugPrint = statsobjDebugPrint;
    pIf->SetName = setName;
    pIf->SetOrigin = setOrigin;
    pIf->SetReadNotifier = setReadNotifier;
    pIf->SetReportingNamespace = setReportingNamespace;
    pIf->SetStatsObjFlags = setStatsObjFlags;
    pIf->GetAllStatsLines = getAllStatsLines;
    pIf->GetAllCounters = getAllCounters;
    pIf->EncodePrometheusMetricName = encodePrometheusMetricName;
    pIf->AddCounter = addCounter;
    pIf->AddManagedCounter = addManagedCounter;
    pIf->AddPreCreatedCtr = addPreCreatedCounter;
    pIf->DestructCounter = destructCounter;
    pIf->DestructUnlinkedCounter = destructUnlinkedCounter;
    pIf->UnlinkAllCounters = unlinkAllCounters;
    pIf->EnableStats = enableStats;
finalize_it:
ENDobjQueryInterface(statsobj)


/* Initialize the statsobj class. Must be called as the very first method
 * before anything else is called inside this class.
 */
BEGINAbstractObjClassInit(statsobj, 1, OBJ_IS_CORE_MODULE) /* class, version */
    /* request objects we use */

    /* set our own handlers */
    OBJSetMethodHandler(objMethod_DEBUGPRINT, statsobjDebugPrint);
    OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, statsobjConstructFinalize);

    /* init other data items */
    CHKiConcCtrl(pthread_mutex_init(&mutStats, NULL));
    CHKiConcCtrl(pthread_mutex_init(&mutSenders, NULL));

    if ((stats_senders = create_hashtable(100, hash_from_string, key_equals_string, NULL)) == NULL) {
        LogError(0, RS_RET_INTERNAL_ERROR,
                 "error trying to initialize hash-table "
                 "for sender table. Sender statistics and warnings are disabled.");
        ABORT_FINALIZE(RS_RET_INTERNAL_ERROR);
    }
ENDObjClassInit(statsobj)

/* Exit the class.
 */
BEGINObjClassExit(statsobj, OBJ_IS_CORE_MODULE) /* class, version */
    /* release objects we no longer need */
    pthread_mutex_destroy(&mutStats);
    pthread_mutex_destroy(&mutSenders);
    hashtable_destroy(stats_senders, 1);
ENDObjClassExit(statsobj)