diff options
Diffstat (limited to 'runtime/perctile_stats.c')
| -rw-r--r-- | runtime/perctile_stats.c | 695 |
1 files changed, 695 insertions, 0 deletions
diff --git a/runtime/perctile_stats.c b/runtime/perctile_stats.c new file mode 100644 index 0000000..55a62aa --- /dev/null +++ b/runtime/perctile_stats.c @@ -0,0 +1,695 @@ +/* + * 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 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]; + if (ctr->ref_ctr_percentile_stat) { + statsobj.DestructCounter(b->statsobj, ctr->ref_ctr_percentile_stat); + } + } + free(pstat->ctrs); + } + + if (pstat->refCtrWindowCount) { + statsobj.DestructCounter(b->statsobj, pstat->refCtrWindowCount); + } + if (pstat->refCtrWindowMin) { + statsobj.DestructCounter(b->statsobj, pstat->refCtrWindowMin); + } + if (pstat->refCtrWindowMax) { + statsobj.DestructCounter(b->statsobj, pstat->refCtrWindowMax); + } + if (pstat->refCtrWindowSum) { + statsobj.DestructCounter(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) { + 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); + statsobj.Destruct(&bkt->statsobj); + pthread_rwlock_unlock(&bkt->lock); + pthread_rwlock_destroy(&bkt->lock); + 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) { + perctile_bucket_t *head = bkts->listBuckets; + if (head) { + pthread_rwlock_wrlock(&bkts->lock); + perctile_bucket_t *pnode = head, *pnext = NULL; + while (pnode) { + pnext = pnode->next; + perctileBucketDestruct(pnode); + pnode = pnext; + } + pthread_rwlock_unlock(&bkts->lock); + } + statsobj.Destruct(&bkts->global_stats); + // 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", stat_name, stat_name_len); + 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' truncated - too long for buffer size: %d\n", stat_name, stat_name_len); + 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, const uchar* name) { + uchar *metric_name_buff, *metric_suffix; + const uchar *suffix_litteral; + int name_len; + DEFiRet; + + name_len = ustrlen(name); + CHKmalloc(metric_name_buff = malloc((name_len + PERCTILE_MAX_BUCKET_NS_METRIC_LENGTH + 1) * sizeof(uchar))); + + strcpy((char*)metric_name_buff, (char*)name); + 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) { + if (b->pOpsOverflowCtr != NULL) { + statsobj.DestructCounter(bkts->global_stats, b->pOpsOverflowCtr); + } + if (b->pNewKeyAddCtr != NULL) { + statsobj.DestructCounter(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 + 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); + + // add bucket to list of buckets + 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"); + } + + // create the statsobj for this bucket + CHKiRet(perctileInitNewBucketStats(b)); + CHKiRet(perctileAddBucketMetrics(bkts, b, name)); + } + 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); + perctiles[j] = atoi(cstr); + 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; +} |