runtime/perctile_stats.c - main - githubmirror/rsyslog

runtime/perctile_stats.c
699 строк26 KB
Cursor Agent
perctile: link buckets after initialization
27 май 2026, 00:33
27 май 2026, 00:3310f0aaf
/*
 * 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.
 */
#include "config.h"
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>
#include <assert.h>

#include "unicode-helper.h"
#include "rsyslog.h"
#include "rsconf.h"
#include "errmsg.h"
#include "perctile_stats.h"
#include "hashtable_itr.h"
#include "perctile_ringbuf.h"
#include "datetime.h"

#include <stdio.h>
#include <pthread.h>
#include <math.h>

/* Use this macro to enable debug for this module */
#ifdef PERCTILE_STATS_DEBUG
    #define _DEBUG 1
#else
    #define _DEBUG 0
#endif
#define PERCTILE_STATS_LOG(...)                   \
    do {                                          \
        if (_DEBUG) fprintf(stderr, __VA_ARGS__); \
    } while (0)

/* definitions for objects we access */
DEFobjStaticHelpers;
DEFobjCurrIf(statsobj) DEFobjCurrIf(datetime)

#define PERCTILE_CONF_PARAM_NAME "name"
#define PERCTILE_CONF_PARAM_PERCENTILES "percentiles"
#define PERCTILE_CONF_PARAM_WINDOW_SIZE "windowsize"
#define PERCTILE_CONF_PARAM_DELIM "delimiter"

#define PERCTILE_MAX_BUCKET_NS_METRIC_LENGTH 128
#define PERCTILE_METRIC_NAME_SEPARATOR '.'

    static struct cnfparamdescr modpdescr[] = {
        {PERCTILE_CONF_PARAM_NAME, eCmdHdlrString, CNFPARAM_REQUIRED},
        {PERCTILE_CONF_PARAM_DELIM, eCmdHdlrString, 0},
        {PERCTILE_CONF_PARAM_PERCENTILES, eCmdHdlrArray, 0},
        {PERCTILE_CONF_PARAM_WINDOW_SIZE, eCmdHdlrPositiveInt, 0},
};

static struct cnfparamblk modpblk = {CNFPARAMBLK_VERSION, sizeof(modpdescr) / sizeof(struct cnfparamdescr), modpdescr};

rsRetVal perctileClassInit(void) {
    DEFiRet;
    CHKiRet(objGetObjInterface(&obj));
    CHKiRet(objUse(statsobj, CORE_COMPONENT));
    CHKiRet(objUse(datetime, CORE_COMPONENT));
finalize_it:
    RETiRet;
}

static void perctileDestroyCounter(statsobj_t *stats, ctr_t **ref) {
    if (*ref != NULL) {
        if (stats != NULL) {
            statsobj.DestructCounter(stats, *ref);
        } else {
            statsobj.DestructUnlinkedCounter(*ref);
        }
        *ref = NULL;
    }
}

static uint64_t min(uint64_t a, uint64_t b) {
    return a < b ? a : b;
}

static uint64_t max(uint64_t a, uint64_t b) {
    return a > b ? a : b;
}

static void perctileStatDestruct(perctile_bucket_t *b, perctile_stat_t *pstat) {
    if (pstat) {
        if (pstat->rb_observed_stats) {
            ringbuf_del(pstat->rb_observed_stats);
        }

        if (pstat->ctrs) {
            for (size_t i = 0; i < pstat->perctile_ctrs_count; ++i) {
                perctile_ctr_t *ctr = &pstat->ctrs[i];
                perctileDestroyCounter(b->statsobj, &ctr->ref_ctr_percentile_stat);
            }
            free(pstat->ctrs);
        }

        perctileDestroyCounter(b->statsobj, &pstat->refCtrWindowCount);
        perctileDestroyCounter(b->statsobj, &pstat->refCtrWindowMin);
        perctileDestroyCounter(b->statsobj, &pstat->refCtrWindowMax);
        perctileDestroyCounter(b->statsobj, &pstat->refCtrWindowSum);
        pthread_rwlock_destroy(&pstat->stats_lock);
        free(pstat);
    }
}

static void perctileBucketDestruct(perctile_bucket_t *bkt) {
    PERCTILE_STATS_LOG("destructing perctile bucket\n");
    if (bkt) {
        /* Drop stats registry visibility before taking bucket locks and
         * freeing counter backing storage; stats scrapes hold mutStats while
         * read notifiers take the bucket locks. */
        if (bkt->statsobj != NULL) {
            statsobj.UnlinkAllCounters(bkt->statsobj);
            statsobj.Destruct(&bkt->statsobj);
        }
        pthread_rwlock_wrlock(&bkt->lock);
        // Delete all items in hashtable
        size_t count = hashtable_count(bkt->htable);
        if (count) {
            int ret = 0;
            struct hashtable_itr *itr = hashtable_iterator(bkt->htable);
            dbgprintf("%s() - All container instances, count=%zu...\n", __FUNCTION__, count);
            do {
                perctile_stat_t *pstat = hashtable_iterator_value(itr);
                perctileStatDestruct(bkt, pstat);
                ret = hashtable_iterator_advance(itr);
            } while (ret);
            free(itr);
            dbgprintf("End of container instances.\n");
        }
        hashtable_destroy(bkt->htable, 0);
        pthread_rwlock_unlock(&bkt->lock);
        pthread_rwlock_destroy(&bkt->lock);
        perctileDestroyCounter(bkt->bkts != NULL ? bkt->bkts->global_stats : NULL, &bkt->pOpsOverflowCtr);
        perctileDestroyCounter(bkt->bkts != NULL ? bkt->bkts->global_stats : NULL, &bkt->pNewKeyAddCtr);
        free(bkt->perctile_values);
        free(bkt->delim);
        free(bkt->name);
        free(bkt);
    }
}

void perctileBucketsDestruct(void) {
    perctile_buckets_t *bkts = &runConf->perctile_buckets;

    if (bkts->initialized) {
        /* Global counters point into bucket storage, so detach the object
         * before buckets start releasing that storage. */
        if (bkts->global_stats != NULL) {
            statsobj.UnlinkAllCounters(bkts->global_stats);
            statsobj.Destruct(&bkts->global_stats);
        }
        while (1) {
            perctile_bucket_t *pnode;
            pthread_rwlock_wrlock(&bkts->lock);
            pnode = bkts->listBuckets;
            if (pnode == NULL) {
                pthread_rwlock_unlock(&bkts->lock);
                break;
            }
            bkts->listBuckets = pnode->next;
            pnode->next = NULL;
            pthread_rwlock_unlock(&bkts->lock);
            perctileBucketDestruct(pnode);
        }
        // destroy any global stats we keep specifically for this.
        pthread_rwlock_destroy(&bkts->lock);
    }
}

static perctile_bucket_t *findBucket(perctile_bucket_t *head, const uchar *name) {
    perctile_bucket_t *pbkt_found = NULL;
    // walk the linked list until the name is found
    pthread_rwlock_rdlock(&head->lock);
    for (perctile_bucket_t *pnode = head; pnode != NULL; pnode = pnode->next) {
        if (ustrcmp(name, pnode->name) == 0) {
            // found.
            pbkt_found = pnode;
        }
    }
    pthread_rwlock_unlock(&head->lock);
    return pbkt_found;
}

#ifdef PERCTILE_STATS_DEBUG
static void print_perctiles(perctile_bucket_t *bkt) {
    if (hashtable_count(bkt->htable)) {
        struct hashtable_itr *itr = hashtable_iterator(bkt->htable);
        do {
            uchar *key = hashtable_iterator_key(itr);
            perctile_stat_t *perc_stat = hashtable_iterator_value(itr);
            PERCTILE_STATS_LOG("print_perctile() - key: %s, perctile stat name: %s ", key, perc_stat->name);
        } while (hashtable_iterator_advance(itr));
        PERCTILE_STATS_LOG("\n");
    }
}
#endif

// Assumes a fully created pstat and bkt, also initiliazes some values in pstat.
static rsRetVal initAndAddPerctileMetrics(perctile_stat_t *pstat, perctile_bucket_t *bkt, uchar *key) {
    char stat_name[128];
    int bytes = 0;
    int stat_name_len = sizeof(stat_name);
    DEFiRet;

    bytes = snprintf((char *)pstat->name, sizeof(pstat->name), "%s", key);
    if (bytes < 0 || bytes >= (int)sizeof(pstat->name)) {
        LogError(0, iRet, "statname '%s' truncated - too long for buffer size: %d\n", key, bytes);
        ABORT_FINALIZE(RS_RET_CONF_PARAM_INVLD);
    }

    int offset = snprintf(stat_name, stat_name_len, "%s%s", (char *)pstat->name, (char *)bkt->delim);
    if (offset < 0 || offset >= stat_name_len) {
        LogError(0, iRet, "statname '%s' delim '%s' truncated - too long for buffer size: %d\n", (char *)pstat->name,
                 (char *)bkt->delim, offset);
        ABORT_FINALIZE(RS_RET_CONF_PARAM_INVLD);
    }

    int remaining_size = stat_name_len - offset - 1;
    // initialize the counters array
    for (size_t i = 0; i < pstat->perctile_ctrs_count; ++i) {
        perctile_ctr_t *ctr = &pstat->ctrs[i];

        // bucket contains the supported percentile values.
        ctr->percentile = bkt->perctile_values[i];
        bytes = snprintf(stat_name + offset, remaining_size, "p%d", bkt->perctile_values[i]);
        if (bytes < 0 || bytes >= remaining_size) {
            LogError(0, iRet, "statname '%s' truncated - too long for buffer size: %d\n", stat_name, stat_name_len);
            ABORT_FINALIZE(RS_RET_CONF_PARAM_INVLD);
        }
        CHKiRet(statsobj.AddManagedCounter(bkt->statsobj, (uchar *)stat_name, ctrType_IntCtr, CTR_FLAG_NONE,
                                           &ctr->ctr_perctile_stat, &ctr->ref_ctr_percentile_stat, 1));
    }

    bytes = snprintf(stat_name + offset, remaining_size, "window_min");
    if (bytes < 0 || bytes >= remaining_size) {
        LogError(0, iRet, "statname '%s' truncated - too long for buffer size: %d\n", stat_name, stat_name_len);
        ABORT_FINALIZE(RS_RET_CONF_PARAM_INVLD);
    }
    CHKiRet(statsobj.AddManagedCounter(bkt->statsobj, (uchar *)stat_name, ctrType_IntCtr, CTR_FLAG_NONE,
                                       &pstat->ctrWindowMin, &pstat->refCtrWindowMin, 1));

    bytes = snprintf(stat_name + offset, remaining_size, "window_max");
    if (bytes < 0 || bytes >= remaining_size) {
        LogError(0, iRet, "statname '%s' truncated - too long for buffer size: %d\n", stat_name, stat_name_len);
        ABORT_FINALIZE(RS_RET_CONF_PARAM_INVLD);
    }
    CHKiRet(statsobj.AddManagedCounter(bkt->statsobj, (uchar *)stat_name, ctrType_IntCtr, CTR_FLAG_NONE,
                                       &pstat->ctrWindowMax, &pstat->refCtrWindowMax, 1));

    bytes = snprintf(stat_name + offset, remaining_size, "window_sum");
    if (bytes < 0 || bytes >= remaining_size) {
        LogError(0, iRet, "statname '%s' truncated - too long for buffer size: %d\n", stat_name, stat_name_len);
        ABORT_FINALIZE(RS_RET_CONF_PARAM_INVLD);
    }
    CHKiRet(statsobj.AddManagedCounter(bkt->statsobj, (uchar *)stat_name, ctrType_IntCtr, CTR_FLAG_NONE,
                                       &pstat->ctrWindowSum, &pstat->refCtrWindowSum, 1));

    bytes = snprintf(stat_name + offset, remaining_size, "window_count");
    if (bytes < 0 || bytes >= remaining_size) {
        LogError(0, iRet, "statname '%s' truncated - too long for buffer size: %d\n", stat_name, stat_name_len);
        ABORT_FINALIZE(RS_RET_CONF_PARAM_INVLD);
    }
    CHKiRet(statsobj.AddManagedCounter(bkt->statsobj, (uchar *)stat_name, ctrType_IntCtr, CTR_FLAG_NONE,
                                       &pstat->ctrWindowCount, &pstat->refCtrWindowCount, 1));

finalize_it:
    if (iRet != RS_RET_OK) {
        LogError(0, iRet, "Could not initialize percentile stats.");
    }
    RETiRet;
}

static rsRetVal perctile_observe(perctile_bucket_t *bkt, uchar *key, int64_t value) {
    uint8_t lock_initialized = 0;
    uchar *hash_key = NULL;
    DEFiRet;
    time_t now;
    datetime.GetTime(&now);

    pthread_rwlock_wrlock(&bkt->lock);
    lock_initialized = 1;
    perctile_stat_t *pstat = (perctile_stat_t *)hashtable_search(bkt->htable, key);
    if (!pstat) {
        PERCTILE_STATS_LOG("perctile_observe(): key '%s' not found - creating new pstat", key);
        // create the pstat if not found
        CHKmalloc(pstat = calloc(1, sizeof(perctile_stat_t)));
        pstat->ctrs = calloc(bkt->perctile_values_count, sizeof(perctile_ctr_t));
        if (!pstat->ctrs) {
            free(pstat);
            ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
        }
        pstat->perctile_ctrs_count = bkt->perctile_values_count;
        pstat->rb_observed_stats = ringbuf_new(bkt->window_size);
        if (!pstat->rb_observed_stats) {
            free(pstat->ctrs);
            free(pstat);
            ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
        }
        pstat->bReported = 0;
        pthread_rwlock_init(&pstat->stats_lock, NULL);

        // init all stat counters here
        pthread_rwlock_wrlock(&pstat->stats_lock);
        pstat->ctrWindowCount = pstat->ctrWindowMax = pstat->ctrWindowSum = 0;
        pstat->ctrWindowMin = value;
        pthread_rwlock_unlock(&pstat->stats_lock);

        iRet = initAndAddPerctileMetrics(pstat, bkt, key);
        if (iRet != RS_RET_OK) {
            perctileStatDestruct(bkt, pstat);
            ABORT_FINALIZE(iRet);
        }

        CHKmalloc(hash_key = ustrdup(key));
        if (!hashtable_insert(bkt->htable, hash_key, pstat)) {
            perctileStatDestruct(bkt, pstat);
            free(hash_key);
            ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
        }
        PERCTILE_STATS_LOG("perctile_observe - new pstat created - name: %s\n", pstat->name);
        STATSCOUNTER_INC(bkt->ctrNewKeyAdd, bkt->mutCtrNewKeyAdd);
    }

    // add this value into the ringbuffer
    assert(pstat->rb_observed_stats);
    if (ringbuf_append_with_overwrite(pstat->rb_observed_stats, value) != 0) {
        // ringbuffer is operating in overwrite mode, so should never see this.
        ABORT_FINALIZE(RS_RET_ERR);
    }

    // update perctile specific stats
    pthread_rwlock_wrlock(&pstat->stats_lock);
    {
        if (pstat->bReported) {
            // reset window values
            pstat->ctrWindowCount = pstat->ctrWindowSum = 0;
            pstat->ctrWindowMin = pstat->ctrWindowMax = value;
            pstat->bReported = 0;
        }
        ++(pstat->ctrWindowCount);
        pstat->ctrWindowSum += value;
        pstat->ctrWindowMin = min(pstat->ctrWindowMin, value);
        pstat->ctrWindowMax = max(pstat->ctrWindowMax, value);
    }
    pthread_rwlock_unlock(&pstat->stats_lock);

#ifdef PERCTILE_STATS_DEBUG
    PERCTILE_STATS_LOG("perctile_observe - appended value: %lld to ringbuffer\n", value);
    PERCTILE_STATS_LOG("ringbuffer contents... \n");
    for (size_t i = 0; i < pstat->rb_observed_stats->size; ++i) {
        PERCTILE_STATS_LOG("%lld ", pstat->rb_observed_stats->cb.buf[i]);
    }
    PERCTILE_STATS_LOG("\n");
    print_perctiles(bkt);
#endif
finalize_it:
    if (lock_initialized) {
        pthread_rwlock_unlock(&bkt->lock);
    }
    if (iRet != RS_RET_OK) {
        // clean up if there was an error
        if (iRet == RS_RET_OUT_OF_MEMORY) {
            STATSCOUNTER_INC(bkt->ctrOpsOverflow, bkt->mutCtrOpsOverflow);
        }
    }
    RETiRet;
}

static int cmp(const void *p1, const void *p2) {
    return (*(ITEM *)p1) - (*(ITEM *)p2);
}

static rsRetVal report_perctile_stats(perctile_bucket_t *pbkt) {
    ITEM *buf = NULL;
    struct hashtable_itr *itr = NULL;
    DEFiRet;

    pthread_rwlock_rdlock(&pbkt->lock);
    if (hashtable_count(pbkt->htable)) {
        itr = hashtable_iterator(pbkt->htable);
        CHKmalloc(buf = malloc(pbkt->window_size * sizeof(ITEM)));
        do {
            memset(buf, 0, pbkt->window_size * sizeof(ITEM));
            perctile_stat_t *perc_stat = hashtable_iterator_value(itr);
            // ringbuffer read
            size_t count = ringbuf_read_to_end(perc_stat->rb_observed_stats, buf, pbkt->window_size);
            if (!count) {
                continue;
            }
            PERCTILE_STATS_LOG("read %zu values\n", count);
            // calculate the p95 based on the
#ifdef PERCTILE_STATS_DEBUG
            PERCTILE_STATS_LOG("ringbuffer contents... \n");
            for (size_t i = 0; i < perc_stat->rb_observed_stats->size; ++i) {
                PERCTILE_STATS_LOG("%lld ", perc_stat->rb_observed_stats->cb.buf[i]);
            }
            PERCTILE_STATS_LOG("\n");

            PERCTILE_STATS_LOG("buffer contents... \n");
            for (size_t i = 0; i < perc_stat->rb_observed_stats->size; ++i) {
                PERCTILE_STATS_LOG("%lld ", buf[i]);
            }
            PERCTILE_STATS_LOG("\n");
#endif
            qsort(buf, count, sizeof(ITEM), cmp);
#ifdef PERCTILE_STATS_DEBUG
            PERCTILE_STATS_LOG("buffer contents after sort... \n");
            for (size_t i = 0; i < perc_stat->rb_observed_stats->size; ++i) {
                PERCTILE_STATS_LOG("%lld ", buf[i]);
            }
            PERCTILE_STATS_LOG("\n");
#endif
            PERCTILE_STATS_LOG("report_perctile_stats() - perctile stat has %zu counters.",
                               perc_stat->perctile_ctrs_count);
            for (size_t i = 0; i < perc_stat->perctile_ctrs_count; ++i) {
                perctile_ctr_t *pctr = &perc_stat->ctrs[i];
                // get percentile - this can be cached.
                int index = max(0, ((pctr->percentile / 100.0) * count) - 1);
                // look into if we need to lock this.
                pctr->ctr_perctile_stat = buf[index];
                PERCTILE_STATS_LOG("report_perctile_stats() - index: %d, perctile stat [%s, %d, %llu]", index,
                                   perc_stat->name, pctr->percentile, pctr->ctr_perctile_stat);
            }
            perc_stat->bReported = 1;
        } while (hashtable_iterator_advance(itr));
    }

finalize_it:
    pthread_rwlock_unlock(&pbkt->lock);
    free(itr);
    free(buf);
    RETiRet;
}

static void perctile_readCallback(statsobj_t __attribute__((unused)) * ignore, void __attribute__((unused)) * b) {
    perctile_buckets_t *bkts = &runConf->perctile_buckets;

    pthread_rwlock_rdlock(&bkts->lock);
    for (perctile_bucket_t *pbkt = bkts->listBuckets; pbkt != NULL; pbkt = pbkt->next) {
        report_perctile_stats(pbkt);
    }
    pthread_rwlock_unlock(&bkts->lock);
}

static rsRetVal perctileInitNewBucketStats(perctile_bucket_t *b) {
    DEFiRet;

    CHKiRet(statsobj.Construct(&b->statsobj));
    CHKiRet(statsobj.SetOrigin(b->statsobj, UCHAR_CONSTANT("percentile.bucket")));
    CHKiRet(statsobj.SetName(b->statsobj, b->name));
    CHKiRet(statsobj.SetReportingNamespace(b->statsobj, UCHAR_CONSTANT("values")));
    statsobj.SetReadNotifier(b->statsobj, perctile_readCallback, b);
    CHKiRet(statsobj.ConstructFinalize(b->statsobj));

finalize_it:
    RETiRet;
}

static rsRetVal perctileAddBucketMetrics(perctile_buckets_t *bkts, perctile_bucket_t *b) {
    uchar *metric_name_buff, *metric_suffix;
    const uchar *suffix_litteral;
    const size_t name_len = ustrlen(b->name);
    DEFiRet;

    CHKmalloc(metric_name_buff = malloc((name_len + PERCTILE_MAX_BUCKET_NS_METRIC_LENGTH + 1) * sizeof(uchar)));

    /* b->name is our owned duplicate created during bucket construction. */
    memcpy(metric_name_buff, b->name, name_len + 1);
    metric_suffix = metric_name_buff + name_len;
    *metric_suffix = PERCTILE_METRIC_NAME_SEPARATOR;
    metric_suffix++;

    suffix_litteral = UCHAR_CONSTANT("new_metric_add");
    ustrncpy(metric_suffix, suffix_litteral, PERCTILE_MAX_BUCKET_NS_METRIC_LENGTH);
    STATSCOUNTER_INIT(b->ctrNewKeyAdd, b->mutCtrNewKeyAdd);
    CHKiRet(statsobj.AddManagedCounter(bkts->global_stats, metric_name_buff, ctrType_IntCtr, CTR_FLAG_RESETTABLE,
                                       &(b->ctrNewKeyAdd), &b->pNewKeyAddCtr, 1));

    suffix_litteral = UCHAR_CONSTANT("ops_overflow");
    ustrncpy(metric_suffix, suffix_litteral, PERCTILE_MAX_BUCKET_NS_METRIC_LENGTH);
    STATSCOUNTER_INIT(b->ctrOpsOverflow, b->mutCtrOpsOverflow);
    CHKiRet(statsobj.AddManagedCounter(bkts->global_stats, metric_name_buff, ctrType_IntCtr, CTR_FLAG_RESETTABLE,
                                       &(b->ctrOpsOverflow), &b->pOpsOverflowCtr, 1));

finalize_it:
    free(metric_name_buff);
    if (iRet != RS_RET_OK) {
        perctileDestroyCounter(bkts->global_stats, &b->pOpsOverflowCtr);
        perctileDestroyCounter(bkts->global_stats, &b->pNewKeyAddCtr);
    }
    RETiRet;
}

/* Create new perctile bucket, and add it to our list of perctile buckets.
 */
static rsRetVal perctile_newBucket(
    const uchar *name, const uchar *delim, uint8_t *perctiles, uint32_t perctilesCount, uint32_t windowSize) {
    perctile_buckets_t *bkts;
    perctile_bucket_t *b = NULL;
    pthread_rwlockattr_t bucket_lock_attr;
    DEFiRet;

    bkts = &loadConf->perctile_buckets;

    if (bkts->initialized) {
        CHKmalloc(b = calloc(1, sizeof(perctile_bucket_t)));

        // initialize
        b->bkts = bkts;
        pthread_rwlockattr_init(&bucket_lock_attr);
        pthread_rwlock_init(&b->lock, &bucket_lock_attr);
        CHKmalloc(b->htable = create_hashtable(7, hash_from_string, key_equals_string, NULL));
        CHKmalloc(b->name = ustrdup(name));
        if (delim) {
            CHKmalloc(b->delim = ustrdup(delim));
        } else {
            CHKmalloc(b->delim = ustrdup("."));
        }

        CHKmalloc(b->perctile_values = calloc(perctilesCount, sizeof(uint8_t)));
        b->perctile_values_count = perctilesCount;
        memcpy(b->perctile_values, perctiles, perctilesCount * sizeof(uint8_t));
        b->window_size = windowSize;
        b->next = NULL;
        PERCTILE_STATS_LOG(
            "perctile_newBucket: create new bucket for %s,"
            "with windowsize: %d,  values_count: %zu\n",
            b->name, b->window_size, b->perctile_values_count);

        // create the statsobj for this bucket
        CHKiRet(perctileInitNewBucketStats(b));
        CHKiRet(perctileAddBucketMetrics(bkts, b));

        // add bucket to list of buckets
        pthread_rwlock_wrlock(&bkts->lock);
        if (!bkts->listBuckets) {
            // none yet
            bkts->listBuckets = b;
            PERCTILE_STATS_LOG("perctile_newBucket: Adding new bucket to empty list \n");
        } else {
            b->next = bkts->listBuckets;
            bkts->listBuckets = b;
            PERCTILE_STATS_LOG("perctile_newBucket: prepended new bucket list \n");
        }
        pthread_rwlock_unlock(&bkts->lock);
    } else {
        LogError(0, RS_RET_INTERNAL_ERROR,
                 "perctile: bucket creation failed, as "
                 "global-initialization of buckets was unsuccessful");
        ABORT_FINALIZE(RS_RET_INTERNAL_ERROR);
    }
finalize_it:
    if (iRet != RS_RET_OK) {
        if (b != NULL) {
            perctileBucketDestruct(b);
        }
    }
    RETiRet;
}

// Public functions
rsRetVal perctile_processCnf(struct cnfobj *o) {
    struct cnfparamvals *pvals;
    uchar *name = NULL;
    uchar *delim = NULL;
    uint8_t *perctiles = NULL;
    uint32_t perctilesCount = 0;
    uint64_t windowSize = 0;
    DEFiRet;

    pvals = nvlstGetParams(o->nvlst, &modpblk, NULL);
    if (pvals == NULL) {
        ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS);
    }

    for (short i = 0; i < modpblk.nParams; ++i) {
        if (!pvals[i].bUsed) continue;
        if (!strcmp(modpblk.descr[i].name, PERCTILE_CONF_PARAM_NAME)) {
            CHKmalloc(name = (uchar *)es_str2cstr(pvals[i].val.d.estr, NULL));
        } else if (!strcmp(modpblk.descr[i].name, PERCTILE_CONF_PARAM_DELIM)) {
            CHKmalloc(delim = (uchar *)es_str2cstr(pvals[i].val.d.estr, NULL));
        } else if (!strcmp(modpblk.descr[i].name, PERCTILE_CONF_PARAM_PERCENTILES)) {
            /* Only the first instance of this parameter will be accepted */
            if (!perctiles) {
                perctilesCount = pvals[i].val.d.ar->nmemb;
                if (perctilesCount) {
                    CHKmalloc(perctiles = calloc(perctilesCount, sizeof(uint8_t)));
                    for (int j = 0; j < pvals[i].val.d.ar->nmemb; ++j) {
                        char *cstr = es_str2cstr(pvals[i].val.d.ar->arr[j], NULL);
                        CHKmalloc(cstr);
                        char *endptr = NULL;
                        errno = 0;
                        const long val = strtol(cstr, &endptr, 10);
                        if (errno == ERANGE || endptr == cstr || *endptr != '\0' || val < 0 || val > 100) {
                            LogError(
                                0, RS_RET_PARAM_ERROR,
                                "perctile: percentile value '%s' is invalid - must be an integer between 0 and 100",
                                cstr);
                            free(cstr);
                            ABORT_FINALIZE(RS_RET_PARAM_ERROR);
                        }
                        perctiles[j] = (uint8_t)val;
                        free(cstr);
                    }
                }
            }
        } else if (!strcmp(modpblk.descr[i].name, PERCTILE_CONF_PARAM_WINDOW_SIZE)) {
            windowSize = pvals[i].val.d.n;
        } else {
            dbgprintf(
                "perctile: program error, non-handled "
                "param '%s'\n",
                modpblk.descr[i].name);
        }
    }

    if (name != NULL && perctiles != NULL) {
        CHKiRet(perctile_newBucket(name, delim, perctiles, perctilesCount, windowSize));
    }

finalize_it:
    free(name);
    free(delim);
    free(perctiles);
    cnfparamvalsDestruct(pvals, &modpblk);
    RETiRet;
}

rsRetVal perctile_initCnf(perctile_buckets_t *bkts) {
    DEFiRet;

    bkts->initialized = 0;
    bkts->listBuckets = NULL;
    CHKiRet(statsobj.Construct(&bkts->global_stats));
    CHKiRet(statsobj.SetOrigin(bkts->global_stats, UCHAR_CONSTANT("percentile")));
    CHKiRet(statsobj.SetName(bkts->global_stats, UCHAR_CONSTANT("global")));
    CHKiRet(statsobj.SetReportingNamespace(bkts->global_stats, UCHAR_CONSTANT("values")));
    CHKiRet(statsobj.ConstructFinalize(bkts->global_stats));
    pthread_rwlock_init(&bkts->lock, NULL);
    bkts->initialized = 1;

finalize_it:
    if (iRet != RS_RET_OK) {
        statsobj.Destruct(&bkts->global_stats);
    }
    RETiRet;
}

perctile_bucket_t *perctile_findBucket(const uchar *name) {
    perctile_bucket_t *b = NULL;

    perctile_buckets_t *bkts = &loadConf->perctile_buckets;
    if (bkts->initialized) {
        pthread_rwlock_rdlock(&bkts->lock);
        if (bkts->listBuckets) {
            b = findBucket(bkts->listBuckets, name);
        }
        pthread_rwlock_unlock(&bkts->lock);
    } else {
        LogError(0, RS_RET_INTERNAL_ERROR,
                 "perctile: bucket lookup failed, as global-initialization "
                 "of buckets was unsuccessful");
    }
    return b;
}

rsRetVal perctile_obs(perctile_bucket_t *perctile_bkt, uchar *key, int64_t value) {
    DEFiRet;
    if (!perctile_bkt) {
        LogError(0, RS_RET_INTERNAL_ERROR, "perctile() - perctile bkt not available");
        FINALIZE;
    }
    PERCTILE_STATS_LOG("perctile_obs() - bucket name: %s, key: %s, val: %" PRId64 "\n", perctile_bkt->name, key, value);

    CHKiRet(perctile_observe(perctile_bkt, key, value));

finalize_it:
    if (iRet != RS_RET_OK) {
        LogError(0, RS_RET_INTERNAL_ERROR, "perctile_obs(): name: %s, key: %s, val: %" PRId64 "\n", perctile_bkt->name,
                 key, value);
    }
    RETiRet;
}