123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821 |
- // SPDX-License-Identifier: GPL-3.0-or-later
- #define NETDATA_RRD_INTERNALS
- #include "rrd.h"
- #define MAX_FUNCTION_LENGTH (PLUGINSD_LINE_MAX - 512) // we need some space for the rest of the line
- static unsigned char functions_allowed_chars[256] = {
- [0] = '\0', //
- [1] = '_', //
- [2] = '_', //
- [3] = '_', //
- [4] = '_', //
- [5] = '_', //
- [6] = '_', //
- [7] = '_', //
- [8] = '_', //
- [9] = ' ', // Horizontal Tab
- [10] = ' ', // Line Feed
- [11] = ' ', // Vertical Tab
- [12] = ' ', // Form Feed
- [13] = ' ', // Carriage Return
- [14] = '_', //
- [15] = '_', //
- [16] = '_', //
- [17] = '_', //
- [18] = '_', //
- [19] = '_', //
- [20] = '_', //
- [21] = '_', //
- [22] = '_', //
- [23] = '_', //
- [24] = '_', //
- [25] = '_', //
- [26] = '_', //
- [27] = '_', //
- [28] = '_', //
- [29] = '_', //
- [30] = '_', //
- [31] = '_', //
- [32] = ' ', // SPACE keep
- [33] = '!', // ! keep
- [34] = '"', // " keep
- [35] = '#', // # keep
- [36] = '$', // $ keep
- [37] = '%', // % keep
- [38] = '&', // & keep
- [39] = '\'', // ' keep
- [40] = '(', // ( keep
- [41] = ')', // ) keep
- [42] = '*', // * keep
- [43] = '+', // + keep
- [44] = ',', // , keep
- [45] = '-', // - keep
- [46] = '.', // . keep
- [47] = '/', // / keep
- [48] = '0', // 0 keep
- [49] = '1', // 1 keep
- [50] = '2', // 2 keep
- [51] = '3', // 3 keep
- [52] = '4', // 4 keep
- [53] = '5', // 5 keep
- [54] = '6', // 6 keep
- [55] = '7', // 7 keep
- [56] = '8', // 8 keep
- [57] = '9', // 9 keep
- [58] = ':', // : keep
- [59] = ';', // ; keep
- [60] = '<', // < keep
- [61] = '=', // = keep
- [62] = '>', // > keep
- [63] = '?', // ? keep
- [64] = '@', // @ keep
- [65] = 'A', // A keep
- [66] = 'B', // B keep
- [67] = 'C', // C keep
- [68] = 'D', // D keep
- [69] = 'E', // E keep
- [70] = 'F', // F keep
- [71] = 'G', // G keep
- [72] = 'H', // H keep
- [73] = 'I', // I keep
- [74] = 'J', // J keep
- [75] = 'K', // K keep
- [76] = 'L', // L keep
- [77] = 'M', // M keep
- [78] = 'N', // N keep
- [79] = 'O', // O keep
- [80] = 'P', // P keep
- [81] = 'Q', // Q keep
- [82] = 'R', // R keep
- [83] = 'S', // S keep
- [84] = 'T', // T keep
- [85] = 'U', // U keep
- [86] = 'V', // V keep
- [87] = 'W', // W keep
- [88] = 'X', // X keep
- [89] = 'Y', // Y keep
- [90] = 'Z', // Z keep
- [91] = '[', // [ keep
- [92] = '\\', // backslash keep
- [93] = ']', // ] keep
- [94] = '^', // ^ keep
- [95] = '_', // _ keep
- [96] = '`', // ` keep
- [97] = 'a', // a keep
- [98] = 'b', // b keep
- [99] = 'c', // c keep
- [100] = 'd', // d keep
- [101] = 'e', // e keep
- [102] = 'f', // f keep
- [103] = 'g', // g keep
- [104] = 'h', // h keep
- [105] = 'i', // i keep
- [106] = 'j', // j keep
- [107] = 'k', // k keep
- [108] = 'l', // l keep
- [109] = 'm', // m keep
- [110] = 'n', // n keep
- [111] = 'o', // o keep
- [112] = 'p', // p keep
- [113] = 'q', // q keep
- [114] = 'r', // r keep
- [115] = 's', // s keep
- [116] = 't', // t keep
- [117] = 'u', // u keep
- [118] = 'v', // v keep
- [119] = 'w', // w keep
- [120] = 'x', // x keep
- [121] = 'y', // y keep
- [122] = 'z', // z keep
- [123] = '{', // { keep
- [124] = '|', // | keep
- [125] = '}', // } keep
- [126] = '~', // ~ keep
- [127] = '_', //
- [128] = '_', //
- [129] = '_', //
- [130] = '_', //
- [131] = '_', //
- [132] = '_', //
- [133] = '_', //
- [134] = '_', //
- [135] = '_', //
- [136] = '_', //
- [137] = '_', //
- [138] = '_', //
- [139] = '_', //
- [140] = '_', //
- [141] = '_', //
- [142] = '_', //
- [143] = '_', //
- [144] = '_', //
- [145] = '_', //
- [146] = '_', //
- [147] = '_', //
- [148] = '_', //
- [149] = '_', //
- [150] = '_', //
- [151] = '_', //
- [152] = '_', //
- [153] = '_', //
- [154] = '_', //
- [155] = '_', //
- [156] = '_', //
- [157] = '_', //
- [158] = '_', //
- [159] = '_', //
- [160] = '_', //
- [161] = '_', //
- [162] = '_', //
- [163] = '_', //
- [164] = '_', //
- [165] = '_', //
- [166] = '_', //
- [167] = '_', //
- [168] = '_', //
- [169] = '_', //
- [170] = '_', //
- [171] = '_', //
- [172] = '_', //
- [173] = '_', //
- [174] = '_', //
- [175] = '_', //
- [176] = '_', //
- [177] = '_', //
- [178] = '_', //
- [179] = '_', //
- [180] = '_', //
- [181] = '_', //
- [182] = '_', //
- [183] = '_', //
- [184] = '_', //
- [185] = '_', //
- [186] = '_', //
- [187] = '_', //
- [188] = '_', //
- [189] = '_', //
- [190] = '_', //
- [191] = '_', //
- [192] = '_', //
- [193] = '_', //
- [194] = '_', //
- [195] = '_', //
- [196] = '_', //
- [197] = '_', //
- [198] = '_', //
- [199] = '_', //
- [200] = '_', //
- [201] = '_', //
- [202] = '_', //
- [203] = '_', //
- [204] = '_', //
- [205] = '_', //
- [206] = '_', //
- [207] = '_', //
- [208] = '_', //
- [209] = '_', //
- [210] = '_', //
- [211] = '_', //
- [212] = '_', //
- [213] = '_', //
- [214] = '_', //
- [215] = '_', //
- [216] = '_', //
- [217] = '_', //
- [218] = '_', //
- [219] = '_', //
- [220] = '_', //
- [221] = '_', //
- [222] = '_', //
- [223] = '_', //
- [224] = '_', //
- [225] = '_', //
- [226] = '_', //
- [227] = '_', //
- [228] = '_', //
- [229] = '_', //
- [230] = '_', //
- [231] = '_', //
- [232] = '_', //
- [233] = '_', //
- [234] = '_', //
- [235] = '_', //
- [236] = '_', //
- [237] = '_', //
- [238] = '_', //
- [239] = '_', //
- [240] = '_', //
- [241] = '_', //
- [242] = '_', //
- [243] = '_', //
- [244] = '_', //
- [245] = '_', //
- [246] = '_', //
- [247] = '_', //
- [248] = '_', //
- [249] = '_', //
- [250] = '_', //
- [251] = '_', //
- [252] = '_', //
- [253] = '_', //
- [254] = '_', //
- [255] = '_' //
- };
- static inline size_t sanitize_function_text(char *dst, const char *src, size_t dst_len) {
- return text_sanitize((unsigned char *)dst, (const unsigned char *)src, dst_len,
- functions_allowed_chars, true, "", NULL);
- }
- // we keep a dictionary per RRDSET with these functions
- // the dictionary is created on demand (only when a function is added to an RRDSET)
- typedef enum __attribute__((packed)) {
- RRD_FUNCTION_LOCAL = (1 << 0),
- RRD_FUNCTION_GLOBAL = (1 << 1),
- // this is 8-bit
- } RRD_FUNCTION_OPTIONS;
- struct rrd_host_function {
- bool sync; // when true, the function is called synchronously
- RRD_FUNCTION_OPTIONS options; // RRD_FUNCTION_OPTIONS
- STRING *help;
- int timeout; // the default timeout of the function
- rrd_function_execute_cb_t execute_cb;
- void *execute_cb_data;
- struct rrd_collector *collector;
- };
- // Each function points to this collector structure
- // so that when the collector exits, all of them will
- // be invalidated (running == false)
- // The last function that is using this collector
- // frees the structure too (or when the collector calls
- // rrdset_collector_finished()).
- struct rrd_collector {
- int32_t refcount;
- int32_t refcount_canceller;
- pid_t tid;
- bool running;
- };
- // Each thread that adds RRDSET functions, has to call
- // rrdset_collector_started() and rrdset_collector_finished()
- // to create the collector structure.
- static __thread struct rrd_collector *thread_rrd_collector = NULL;
- static void rrd_collector_free(struct rrd_collector *rdc) {
- if(rdc->running)
- return;
- int32_t expected = 0;
- if(!__atomic_compare_exchange_n(&rdc->refcount, &expected, -1, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED)) {
- // the collector is still referenced by charts.
- // leave it hanging there, the last chart will actually free it.
- return;
- }
- // we can free it now
- freez(rdc);
- }
- // called once per collector
- void rrd_collector_started(void) {
- if(!thread_rrd_collector)
- thread_rrd_collector = callocz(1, sizeof(struct rrd_collector));
- thread_rrd_collector->tid = gettid();
- __atomic_store_n(&thread_rrd_collector->running, true, __ATOMIC_RELAXED);
- }
- // called once per collector
- void rrd_collector_finished(void) {
- if(!thread_rrd_collector)
- return;
- __atomic_store_n(&thread_rrd_collector->running, false, __ATOMIC_RELAXED);
- int32_t expected = 0;
- while(!__atomic_compare_exchange_n(&thread_rrd_collector->refcount_canceller, &expected, -1, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED)) {
- expected = 0;
- sleep_usec(1 * USEC_PER_MS);
- }
- rrd_collector_free(thread_rrd_collector);
- thread_rrd_collector = NULL;
- }
- #define rrd_collector_running(c) __atomic_load_n(&(c)->running, __ATOMIC_RELAXED)
- static struct rrd_collector *rrd_collector_acquire(void) {
- rrd_collector_started();
- int32_t expected = __atomic_load_n(&thread_rrd_collector->refcount, __ATOMIC_RELAXED), wanted = 0;
- do {
- if(expected < 0 || !rrd_collector_running(thread_rrd_collector)) {
- internal_fatal(true, "FUNCTIONS: Trying to acquire a collector that is exiting.");
- return thread_rrd_collector;
- }
- wanted = expected + 1;
- } while(!__atomic_compare_exchange_n(&thread_rrd_collector->refcount, &expected, wanted, false, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED));
- return thread_rrd_collector;
- }
- static void rrd_collector_release(struct rrd_collector *rdc) {
- if(unlikely(!rdc)) return;
- int32_t expected = __atomic_load_n(&rdc->refcount, __ATOMIC_RELAXED), wanted = 0;
- do {
- if(expected < 0) {
- internal_fatal(true, "FUNCTIONS: Trying to release a collector that is exiting.");
- return;
- }
- if(expected == 0) {
- internal_fatal(true, "FUNCTIONS: Trying to release a collector that is not acquired.");
- return;
- }
- wanted = expected - 1;
- } while(!__atomic_compare_exchange_n(&rdc->refcount, &expected, wanted, false, __ATOMIC_RELEASE, __ATOMIC_RELAXED));
- if(wanted == 0)
- rrd_collector_free(rdc);
- }
- static void rrd_functions_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func, void *rrdhost) {
- RRDHOST *host = rrdhost; (void)host;
- struct rrd_host_function *rdcf = func;
- rrd_collector_started();
- rdcf->collector = rrd_collector_acquire();
- // internal_error(true, "FUNCTIONS: adding function '%s' on host '%s', collection tid %d, %s",
- // dictionary_acquired_item_name(item), rrdhost_hostname(host),
- // rdcf->collector->tid, rdcf->collector->running ? "running" : "NOT running");
- }
- static void rrd_functions_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func,
- void *rrdhost __maybe_unused) {
- struct rrd_host_function *rdcf = func;
- rrd_collector_release(rdcf->collector);
- }
- static bool rrd_functions_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func,
- void *new_func, void *rrdhost) {
- RRDHOST *host = rrdhost; (void)host;
- struct rrd_host_function *rdcf = func;
- struct rrd_host_function *new_rdcf = new_func;
- rrd_collector_started();
- bool changed = false;
- if(rdcf->collector != thread_rrd_collector) {
- netdata_log_info("FUNCTIONS: function '%s' of host '%s' changed collector from %d to %d",
- dictionary_acquired_item_name(item), rrdhost_hostname(host), rdcf->collector->tid, thread_rrd_collector->tid);
- struct rrd_collector *old_rdc = rdcf->collector;
- rdcf->collector = rrd_collector_acquire();
- rrd_collector_release(old_rdc);
- changed = true;
- }
- if(rdcf->execute_cb != new_rdcf->execute_cb) {
- netdata_log_info("FUNCTIONS: function '%s' of host '%s' changed execute callback",
- dictionary_acquired_item_name(item), rrdhost_hostname(host));
- rdcf->execute_cb = new_rdcf->execute_cb;
- changed = true;
- }
- if(rdcf->help != new_rdcf->help) {
- netdata_log_info("FUNCTIONS: function '%s' of host '%s' changed help text",
- dictionary_acquired_item_name(item), rrdhost_hostname(host));
- STRING *old = rdcf->help;
- rdcf->help = new_rdcf->help;
- string_freez(old);
- changed = true;
- }
- else
- string_freez(new_rdcf->help);
- if(rdcf->timeout != new_rdcf->timeout) {
- netdata_log_info("FUNCTIONS: function '%s' of host '%s' changed timeout",
- dictionary_acquired_item_name(item), rrdhost_hostname(host));
- rdcf->timeout = new_rdcf->timeout;
- changed = true;
- }
- if(rdcf->sync != new_rdcf->sync) {
- netdata_log_info("FUNCTIONS: function '%s' of host '%s' changed sync/async mode",
- dictionary_acquired_item_name(item), rrdhost_hostname(host));
- rdcf->sync = new_rdcf->sync;
- changed = true;
- }
- if(rdcf->execute_cb_data != new_rdcf->execute_cb_data) {
- netdata_log_info("FUNCTIONS: function '%s' of host '%s' changed execute callback data",
- dictionary_acquired_item_name(item), rrdhost_hostname(host));
- rdcf->execute_cb_data = new_rdcf->execute_cb_data;
- changed = true;
- }
- // internal_error(true, "FUNCTIONS: adding function '%s' on host '%s', collection tid %d, %s",
- // dictionary_acquired_item_name(item), rrdhost_hostname(host),
- // rdcf->collector->tid, rdcf->collector->running ? "running" : "NOT running");
- return changed;
- }
- void rrdfunctions_host_init(RRDHOST *host) {
- if(host->functions) return;
- host->functions = dictionary_create_advanced(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE,
- &dictionary_stats_category_functions, sizeof(struct rrd_host_function));
- dictionary_register_insert_callback(host->functions, rrd_functions_insert_callback, host);
- dictionary_register_delete_callback(host->functions, rrd_functions_delete_callback, host);
- dictionary_register_conflict_callback(host->functions, rrd_functions_conflict_callback, host);
- }
- void rrdfunctions_host_destroy(RRDHOST *host) {
- dictionary_destroy(host->functions);
- }
- void rrd_function_add(RRDHOST *host, RRDSET *st, const char *name, int timeout, const char *help,
- bool sync, rrd_function_execute_cb_t execute_cb, void *execute_cb_data) {
- // RRDSET *st may be NULL in this function
- // to create a GLOBAL function
- if(st && !st->functions_view)
- st->functions_view = dictionary_create_view(host->functions);
- char key[PLUGINSD_LINE_MAX + 1];
- sanitize_function_text(key, name, PLUGINSD_LINE_MAX);
- struct rrd_host_function tmp = {
- .sync = sync,
- .timeout = timeout,
- .options = (st)?RRD_FUNCTION_LOCAL:RRD_FUNCTION_GLOBAL,
- .execute_cb = execute_cb,
- .execute_cb_data = execute_cb_data,
- .help = string_strdupz(help),
- };
- const DICTIONARY_ITEM *item = dictionary_set_and_acquire_item(host->functions, key, &tmp, sizeof(tmp));
- if(st)
- dictionary_view_set(st->functions_view, key, item);
- else
- rrdhost_flag_set(host, RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED);
- dictionary_acquired_item_release(host->functions, item);
- }
- void rrd_functions_expose_rrdpush(RRDSET *st, BUFFER *wb) {
- if(!st->functions_view)
- return;
- struct rrd_host_function *tmp;
- dfe_start_read(st->functions_view, tmp) {
- buffer_sprintf(wb
- , PLUGINSD_KEYWORD_FUNCTION " \"%s\" %d \"%s\"\n"
- , tmp_dfe.name
- , tmp->timeout
- , string2str(tmp->help)
- );
- }
- dfe_done(tmp);
- }
- void rrd_functions_expose_global_rrdpush(RRDHOST *host, BUFFER *wb) {
- rrdhost_flag_clear(host, RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED);
- struct rrd_host_function *tmp;
- dfe_start_read(host->functions, tmp) {
- if(!(tmp->options & RRD_FUNCTION_GLOBAL))
- continue;
- buffer_sprintf(wb
- , PLUGINSD_KEYWORD_FUNCTION " GLOBAL \"%s\" %d \"%s\"\n"
- , tmp_dfe.name
- , tmp->timeout
- , string2str(tmp->help)
- );
- }
- dfe_done(tmp);
- }
- struct {
- const char *format;
- HTTP_CONTENT_TYPE content_type;
- } function_formats[] = {
- { .format = "application/json", CT_APPLICATION_JSON },
- { .format = "text/plain", CT_TEXT_PLAIN },
- { .format = "application/xml", CT_APPLICATION_XML },
- { .format = "prometheus", CT_PROMETHEUS },
- { .format = "text", CT_TEXT_PLAIN },
- { .format = "txt", CT_TEXT_PLAIN },
- { .format = "json", CT_APPLICATION_JSON },
- { .format = "html", CT_TEXT_HTML },
- { .format = "text/html", CT_TEXT_HTML },
- { .format = "xml", CT_APPLICATION_XML },
- // terminator
- { .format = NULL, CT_TEXT_PLAIN },
- };
- uint8_t functions_format_to_content_type(const char *format) {
- if(format && *format) {
- for (int i = 0; function_formats[i].format; i++)
- if (strcmp(function_formats[i].format, format) == 0)
- return function_formats[i].content_type;
- }
- return CT_TEXT_PLAIN;
- }
- const char *functions_content_type_to_format(HTTP_CONTENT_TYPE content_type) {
- for (int i = 0; function_formats[i].format; i++)
- if (function_formats[i].content_type == content_type)
- return function_formats[i].format;
- return "text/plain";
- }
- int rrd_call_function_error(BUFFER *wb, const char *msg, int code) {
- char buffer[PLUGINSD_LINE_MAX];
- json_escape_string(buffer, msg, PLUGINSD_LINE_MAX);
- buffer_flush(wb);
- buffer_sprintf(wb, "{\"status\":%d,\"error_message\":\"%s\"}", code, buffer);
- wb->content_type = CT_APPLICATION_JSON;
- buffer_no_cacheable(wb);
- return code;
- }
- static int rrd_call_function_find(RRDHOST *host, BUFFER *wb, const char *name, size_t key_length, const DICTIONARY_ITEM **item) {
- char buffer[MAX_FUNCTION_LENGTH + 1];
- strncpyz(buffer, name, MAX_FUNCTION_LENGTH);
- char *s = NULL;
- bool found = false;
- *item = NULL;
- if(host->functions) {
- while (buffer[0]) {
- if((*item = dictionary_get_and_acquire_item(host->functions, buffer))) {
- found = true;
- struct rrd_host_function *rdcf = dictionary_acquired_item_value(*item);
- if(rrd_collector_running(rdcf->collector)) {
- break;
- }
- else {
- dictionary_acquired_item_release(host->functions, *item);
- *item = NULL;
- }
- }
- // if s == NULL, set it to the end of the buffer
- // this should happen only the first time
- if (unlikely(!s))
- s = &buffer[key_length - 1];
- // skip a word from the end
- while (s >= buffer && !isspace(*s)) *s-- = '\0';
- // skip all spaces
- while (s >= buffer && isspace(*s)) *s-- = '\0';
- }
- }
- buffer_flush(wb);
- if(!(*item)) {
- if(found)
- return rrd_call_function_error(wb,
- "The collector that registered this function, is not currently running.",
- HTTP_RESP_SERVICE_UNAVAILABLE);
- else
- return rrd_call_function_error(wb,
- "No collector is supplying this function on this host at this time.",
- HTTP_RESP_NOT_FOUND);
- }
- return HTTP_RESP_OK;
- }
- // ----------------------------------------------------------------------------
- struct rrd_function_inflight {
- bool used;
- RRDHOST *host;
- const char *transaction;
- const char *cmd;
- const char *sanitized_cmd;
- size_t sanitized_cmd_length;
- int timeout;
- bool cancelled;
- const DICTIONARY_ITEM *host_function_acquired;
- // the collector
- // we acquire this structure at the beginning,
- // and we release it at the end
- struct rrd_host_function *rdcf;
- struct {
- BUFFER *wb;
- // in async mode,
- // the function to call to send the result back
- rrd_function_result_callback_t cb;
- void *data;
- } result;
- struct {
- // to be called in sync mode
- // while the function is running
- // to check if the function has been cancelled
- rrd_function_is_cancelled_cb_t cb;
- void *data;
- } is_cancelled;
- struct {
- // to be registered by the function itself
- // used to signal the function to cancel
- rrd_function_canceller_cb_t cb;
- void *data;
- } canceller;
- };
- static DICTIONARY *rrd_functions_inflight_requests = NULL;
- static void rrd_functions_inflight_delete_cb(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *data __maybe_unused) {
- struct rrd_function_inflight *r = value;
- // internal_error(true, "FUNCTIONS: transaction '%s' finished", r->transaction);
- freez((void *)r->transaction);
- freez((void *)r->cmd);
- freez((void *)r->sanitized_cmd);
- dictionary_acquired_item_release(r->host->functions, r->host_function_acquired);
- }
- void rrd_functions_inflight_init(void) {
- if(rrd_functions_inflight_requests)
- return;
- rrd_functions_inflight_requests = dictionary_create_advanced(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct rrd_function_inflight));
- dictionary_register_delete_callback(rrd_functions_inflight_requests, rrd_functions_inflight_delete_cb, NULL);
- }
- void rrd_functions_inflight_destroy(void) {
- if(!rrd_functions_inflight_requests)
- return;
- dictionary_destroy(rrd_functions_inflight_requests);
- rrd_functions_inflight_requests = NULL;
- }
- static void rrd_inflight_async_function_register_canceller_cb(void *register_canceller_cb_data, rrd_function_canceller_cb_t canceller_cb, void *canceller_cb_data) {
- struct rrd_function_inflight *r = register_canceller_cb_data;
- r->canceller.cb = canceller_cb;
- r->canceller.data = canceller_cb_data;
- }
- // ----------------------------------------------------------------------------
- // waiting for async function completion
- struct rrd_function_call_wait {
- RRDHOST *host;
- const DICTIONARY_ITEM *host_function_acquired;
- char *transaction;
- bool free_with_signal;
- bool data_are_ready;
- netdata_mutex_t mutex;
- pthread_cond_t cond;
- int code;
- };
- static void rrd_inflight_function_cleanup(RRDHOST *host __maybe_unused,
- const DICTIONARY_ITEM *host_function_acquired __maybe_unused,
- const char *transaction) {
- dictionary_del(rrd_functions_inflight_requests, transaction);
- dictionary_garbage_collect(rrd_functions_inflight_requests);
- }
- static void rrd_function_call_wait_free(struct rrd_function_call_wait *tmp) {
- rrd_inflight_function_cleanup(tmp->host, tmp->host_function_acquired, tmp->transaction);
- freez(tmp->transaction);
- pthread_cond_destroy(&tmp->cond);
- netdata_mutex_destroy(&tmp->mutex);
- freez(tmp);
- }
- static void rrd_async_function_signal_when_ready(BUFFER *temp_wb __maybe_unused, int code, void *callback_data) {
- struct rrd_function_call_wait *tmp = callback_data;
- bool we_should_free = false;
- netdata_mutex_lock(&tmp->mutex);
- // since we got the mutex,
- // the waiting thread is either in pthread_cond_timedwait()
- // or gave up and left.
- tmp->code = code;
- tmp->data_are_ready = true;
- if(tmp->free_with_signal)
- we_should_free = true;
- pthread_cond_signal(&tmp->cond);
- netdata_mutex_unlock(&tmp->mutex);
- if(we_should_free) {
- buffer_free(temp_wb);
- rrd_function_call_wait_free(tmp);
- }
- }
- static void rrd_inflight_async_function_nowait_finished(BUFFER *wb, int code, void *data) {
- struct rrd_function_inflight *r = data;
- if(r->result.cb)
- r->result.cb(wb, code, r->result.data);
- rrd_inflight_function_cleanup(r->host, r->host_function_acquired, r->transaction);
- }
- static bool rrd_inflight_async_function_is_cancelled(void *data) {
- struct rrd_function_inflight *r = data;
- return __atomic_load_n(&r->cancelled, __ATOMIC_RELAXED);
- }
- static inline int rrd_call_function_async_and_dont_wait(struct rrd_function_inflight *r) {
- int code = r->rdcf->execute_cb(r->result.wb, r->timeout, r->sanitized_cmd, r->rdcf->execute_cb_data,
- rrd_inflight_async_function_nowait_finished, r,
- rrd_inflight_async_function_is_cancelled, r,
- rrd_inflight_async_function_register_canceller_cb, r);
- if(code != HTTP_RESP_OK) {
- if (!buffer_strlen(r->result.wb))
- rrd_call_function_error(r->result.wb, "Failed to send request to the collector.", code);
- rrd_inflight_function_cleanup(r->host, r->host_function_acquired, r->transaction);
- }
- return code;
- }
- static int rrd_call_function_async_and_wait(struct rrd_function_inflight *r) {
- struct timespec tp;
- clock_gettime(CLOCK_REALTIME, &tp);
- usec_t now_ut = tp.tv_sec * USEC_PER_SEC + tp.tv_nsec / NSEC_PER_USEC;
- usec_t end_ut = now_ut + r->timeout * USEC_PER_SEC + RRDFUNCTIONS_TIMEOUT_EXTENSION_UT;
- struct rrd_function_call_wait *tmp = mallocz(sizeof(struct rrd_function_call_wait));
- tmp->free_with_signal = false;
- tmp->data_are_ready = false;
- tmp->host = r->host;
- tmp->host_function_acquired = r->host_function_acquired;
- tmp->transaction = strdupz(r->transaction);
- netdata_mutex_init(&tmp->mutex);
- pthread_cond_init(&tmp->cond, NULL);
- // we need a temporary BUFFER, because we may time out and the caller supplied one may vanish
- // so, we create a new one we guarantee will survive until the collector finishes...
- bool we_should_free = true;
- BUFFER *temp_wb = buffer_create(PLUGINSD_LINE_MAX + 1, &netdata_buffers_statistics.buffers_functions); // we need it because we may give up on it
- temp_wb->content_type = r->result.wb->content_type;
- int code = r->rdcf->execute_cb(temp_wb, r->timeout, r->sanitized_cmd, r->rdcf->execute_cb_data,
- // we overwrite the result callbacks,
- // so that we can clean up the allocations made
- rrd_async_function_signal_when_ready, tmp,
- rrd_inflight_async_function_is_cancelled, r,
- rrd_inflight_async_function_register_canceller_cb, r);
- if (code == HTTP_RESP_OK) {
- netdata_mutex_lock(&tmp->mutex);
- bool cancelled = false;
- int rc = 0;
- while (rc == 0 && !cancelled && !tmp->data_are_ready) {
- clock_gettime(CLOCK_REALTIME, &tp);
- now_ut = tp.tv_sec * USEC_PER_SEC + tp.tv_nsec / NSEC_PER_USEC;
- if(now_ut >= end_ut) {
- rc = ETIMEDOUT;
- break;
- }
- tp.tv_nsec += 10 * NSEC_PER_MSEC;
- if(tp.tv_nsec > (long)(1 * NSEC_PER_SEC)) {
- tp.tv_sec++;
- tp.tv_nsec -= 1 * NSEC_PER_SEC;
- }
- // the mutex is unlocked within pthread_cond_timedwait()
- rc = pthread_cond_timedwait(&tmp->cond, &tmp->mutex, &tp);
- // the mutex is again ours
- if(rc == ETIMEDOUT) {
- rc = 0;
- if (!tmp->data_are_ready && r->is_cancelled.cb &&
- r->is_cancelled.cb(r->is_cancelled.data)) {
- // internal_error(true, "FUNCTIONS: transaction '%s' is cancelled while waiting for response",
- // r->transaction);
- rc = 0;
- cancelled = true;
- rrd_function_cancel(r->transaction);
- break;
- }
- }
- }
- if (tmp->data_are_ready) {
- // we have a response
- buffer_fast_strcat(r->result.wb, buffer_tostring(temp_wb), buffer_strlen(temp_wb));
- r->result.wb->content_type = temp_wb->content_type;
- r->result.wb->expires = temp_wb->expires;
- if(r->result.wb->expires)
- buffer_cacheable(r->result.wb);
- else
- buffer_no_cacheable(r->result.wb);
- code = tmp->code;
- }
- else if (rc == ETIMEDOUT || cancelled) {
- // timeout
- // we will go away and let the callback free the structure
- tmp->free_with_signal = true;
- we_should_free = false;
- if(cancelled)
- code = rrd_call_function_error(r->result.wb,
- "Request cancelled",
- HTTP_RESP_CLIENT_CLOSED_REQUEST);
- else
- code = rrd_call_function_error(r->result.wb,
- "Timeout while waiting for a response from the collector.",
- HTTP_RESP_GATEWAY_TIMEOUT);
- }
- else
- code = rrd_call_function_error(r->result.wb,
- "Internal error while communicating with the collector",
- HTTP_RESP_INTERNAL_SERVER_ERROR);
- netdata_mutex_unlock(&tmp->mutex);
- }
- else {
- if(!buffer_strlen(r->result.wb))
- rrd_call_function_error(r->result.wb, "The collector returned an error.", code);
- }
- if (we_should_free) {
- rrd_function_call_wait_free(tmp);
- buffer_free(temp_wb);
- }
- return code;
- }
- static inline int rrd_call_function_async(struct rrd_function_inflight *r, bool wait) {
- if(wait)
- return rrd_call_function_async_and_wait(r);
- else
- return rrd_call_function_async_and_dont_wait(r);
- }
- void call_virtual_function_async(BUFFER *wb, RRDHOST *host, const char *name, const char *payload, rrd_function_result_callback_t callback, void *callback_data);
- // ----------------------------------------------------------------------------
- int rrd_function_run(RRDHOST *host, BUFFER *result_wb, int timeout, const char *cmd,
- bool wait, const char *transaction,
- rrd_function_result_callback_t result_cb, void *result_cb_data,
- rrd_function_is_cancelled_cb_t is_cancelled_cb, void *is_cancelled_cb_data, const char *payload) {
- int code;
- char sanitized_cmd[PLUGINSD_LINE_MAX + 1];
- const DICTIONARY_ITEM *host_function_acquired = NULL;
- // ------------------------------------------------------------------------
- // find the function
- size_t sanitized_cmd_length = sanitize_function_text(sanitized_cmd, cmd, PLUGINSD_LINE_MAX);
- if (is_dyncfg_function(sanitized_cmd, DYNCFG_FUNCTION_TYPE_ALL)) {
- call_virtual_function_async(result_wb, host, sanitized_cmd, payload, result_cb, result_cb_data);
- return HTTP_RESP_OK;
- }
- code = rrd_call_function_find(host, result_wb, sanitized_cmd, sanitized_cmd_length, &host_function_acquired);
- if(code != HTTP_RESP_OK)
- return code;
- struct rrd_host_function *rdcf = dictionary_acquired_item_value(host_function_acquired);
- if(timeout <= 0)
- timeout = rdcf->timeout;
- // ------------------------------------------------------------------------
- // the function can only be executed in sync mode
- if(rdcf->sync) {
- // the caller has to wait
- code = rdcf->execute_cb(result_wb, timeout, sanitized_cmd, rdcf->execute_cb_data,
- result_cb, result_cb_data,
- is_cancelled_cb, is_cancelled_cb_data, // it is ok to pass these, we block the caller
- NULL, NULL); // no need to pass, we will wait
- if (code != HTTP_RESP_OK && !buffer_strlen(result_wb))
- rrd_call_function_error(result_wb, "Collector reported error.", code);
- dictionary_acquired_item_release(host->functions, host_function_acquired);
- return code;
- }
- // ------------------------------------------------------------------------
- // the function can only be executed in async mode
- // put the function into the inflight requests
- char uuid_str[UUID_COMPACT_STR_LEN];
- if(!transaction) {
- uuid_t uuid;
- uuid_generate_random(uuid);
- uuid_unparse_lower_compact(uuid, uuid_str);
- transaction = uuid_str;
- }
- // put the request into the inflight requests
- struct rrd_function_inflight t = {
- .used = false,
- .host = host,
- .cmd = strdupz(cmd),
- .sanitized_cmd = strdupz(sanitized_cmd),
- .sanitized_cmd_length = sanitized_cmd_length,
- .transaction = strdupz(transaction),
- .timeout = timeout,
- .cancelled = false,
- .host_function_acquired = host_function_acquired,
- .rdcf = rdcf,
- .result = {
- .wb = result_wb,
- .cb = result_cb,
- .data = result_cb_data,
- },
- .is_cancelled = {
- .cb = is_cancelled_cb,
- .data = is_cancelled_cb_data,
- }
- };
- struct rrd_function_inflight *r = dictionary_set(rrd_functions_inflight_requests, transaction, &t, sizeof(t));
- if(r->used) {
- netdata_log_info("FUNCTIONS: duplicate transaction '%s', function: '%s'", t.transaction, t.cmd);
- code = rrd_call_function_error(result_wb, "duplicate transaction", HTTP_RESP_BAD_REQUEST);
- freez((void *)t.transaction);
- freez((void *)t.cmd);
- freez((void *)t.sanitized_cmd);
- dictionary_acquired_item_release(r->host->functions, t.host_function_acquired);
- return code;
- }
- r->used = true;
- // internal_error(true, "FUNCTIONS: transaction '%s' started", r->transaction);
- return rrd_call_function_async(r, wait);
- }
- void rrd_function_cancel(const char *transaction) {
- // internal_error(true, "FUNCTIONS: request to cancel transaction '%s'", transaction);
- const DICTIONARY_ITEM *item = dictionary_get_and_acquire_item(rrd_functions_inflight_requests, transaction);
- if(!item) {
- netdata_log_info("FUNCTIONS: received a cancel request for transaction '%s', but the transaction is not running.",
- transaction);
- return;
- }
- struct rrd_function_inflight *r = dictionary_acquired_item_value(item);
- bool cancelled = __atomic_load_n(&r->cancelled, __ATOMIC_RELAXED);
- if(cancelled) {
- netdata_log_info("FUNCTIONS: received a cancel request for transaction '%s', but it is already cancelled.",
- transaction);
- goto cleanup;
- }
- __atomic_store_n(&r->cancelled, true, __ATOMIC_RELAXED);
- int32_t expected = __atomic_load_n(&r->rdcf->collector->refcount_canceller, __ATOMIC_RELAXED);
- int32_t wanted;
- do {
- if(expected < 0) {
- netdata_log_info("FUNCTIONS: received a cancel request for transaction '%s', but the collector is not running.",
- transaction);
- goto cleanup;
- }
- wanted = expected + 1;
- } while(!__atomic_compare_exchange_n(&r->rdcf->collector->refcount_canceller, &expected, wanted, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
- if(r->canceller.cb)
- r->canceller.cb(r->canceller.data);
- __atomic_sub_fetch(&r->rdcf->collector->refcount_canceller, 1, __ATOMIC_RELAXED);
- cleanup:
- dictionary_acquired_item_release(rrd_functions_inflight_requests, item);
- }
- // ----------------------------------------------------------------------------
- static void functions2json(DICTIONARY *functions, BUFFER *wb)
- {
- struct rrd_host_function *t;
- dfe_start_read(functions, t)
- {
- if (!rrd_collector_running(t->collector))
- continue;
- buffer_json_member_add_object(wb, t_dfe.name);
- buffer_json_member_add_string_or_empty(wb, "help", string2str(t->help));
- buffer_json_member_add_int64(wb, "timeout", (int64_t)t->timeout);
- char options[65];
- snprintfz(
- options,
- 64,
- "%s%s",
- (t->options & RRD_FUNCTION_LOCAL) ? "LOCAL " : "",
- (t->options & RRD_FUNCTION_GLOBAL) ? "GLOBAL" : "");
- buffer_json_member_add_string_or_empty(wb, "options", options);
- buffer_json_object_close(wb);
- }
- dfe_done(t);
- }
- void chart_functions2json(RRDSET *st, BUFFER *wb) {
- if(!st || !st->functions_view) return;
- functions2json(st->functions_view, wb);
- }
- void host_functions2json(RRDHOST *host, BUFFER *wb) {
- if(!host || !host->functions) return;
- buffer_json_member_add_object(wb, "functions");
- struct rrd_host_function *t;
- dfe_start_read(host->functions, t) {
- if(!rrd_collector_running(t->collector)) continue;
- buffer_json_member_add_object(wb, t_dfe.name);
- buffer_json_member_add_string(wb, "help", string2str(t->help));
- buffer_json_member_add_int64(wb, "timeout", t->timeout);
- buffer_json_member_add_array(wb, "options");
- if(t->options & RRD_FUNCTION_GLOBAL)
- buffer_json_add_array_item_string(wb, "GLOBAL");
- if(t->options & RRD_FUNCTION_LOCAL)
- buffer_json_add_array_item_string(wb, "LOCAL");
- buffer_json_array_close(wb);
- buffer_json_object_close(wb);
- }
- dfe_done(t);
- buffer_json_object_close(wb);
- }
- void chart_functions_to_dict(DICTIONARY *rrdset_functions_view, DICTIONARY *dst, void *value, size_t value_size) {
- if(!rrdset_functions_view || !dst) return;
- struct rrd_host_function *t;
- dfe_start_read(rrdset_functions_view, t) {
- if(!rrd_collector_running(t->collector)) continue;
- dictionary_set(dst, t_dfe.name, value, value_size);
- }
- dfe_done(t);
- }
- void host_functions_to_dict(RRDHOST *host, DICTIONARY *dst, void *value, size_t value_size, STRING **help) {
- if(!host || !host->functions || !dictionary_entries(host->functions) || !dst) return;
- struct rrd_host_function *t;
- dfe_start_read(host->functions, t) {
- if(!rrd_collector_running(t->collector)) continue;
- if(help)
- *help = t->help;
- dictionary_set(dst, t_dfe.name, value, value_size);
- }
- dfe_done(t);
- }
- // ----------------------------------------------------------------------------
- int rrdhost_function_streaming(BUFFER *wb, int timeout __maybe_unused, const char *function __maybe_unused,
- void *collector_data __maybe_unused,
- rrd_function_result_callback_t result_cb, void *result_cb_data,
- rrd_function_is_cancelled_cb_t is_cancelled_cb, void *is_cancelled_cb_data,
- rrd_function_register_canceller_cb_t register_canceller_cb __maybe_unused,
- void *register_canceller_cb_data __maybe_unused) {
- time_t now = now_realtime_sec();
- buffer_flush(wb);
- wb->content_type = CT_APPLICATION_JSON;
- buffer_json_initialize(wb, "\"", "\"", 0, true, BUFFER_JSON_OPTIONS_DEFAULT);
- buffer_json_member_add_string(wb, "hostname", rrdhost_hostname(localhost));
- buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
- buffer_json_member_add_string(wb, "type", "table");
- buffer_json_member_add_time_t(wb, "update_every", 1);
- buffer_json_member_add_string(wb, "help", RRDFUNCTIONS_STREAMING_HELP);
- buffer_json_member_add_array(wb, "data");
- size_t max_sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_MAX];
- size_t max_db_metrics = 0, max_db_instances = 0, max_db_contexts = 0;
- size_t max_collection_replication_instances = 0, max_streaming_replication_instances = 0;
- size_t max_ml_anomalous = 0, max_ml_normal = 0, max_ml_trained = 0, max_ml_pending = 0, max_ml_silenced = 0;
- {
- RRDHOST *host;
- dfe_start_read(rrdhost_root_index, host) {
- RRDHOST_STATUS s;
- rrdhost_status(host, now, &s);
- buffer_json_add_array_item_array(wb);
- if(s.db.metrics > max_db_metrics)
- max_db_metrics = s.db.metrics;
- if(s.db.instances > max_db_instances)
- max_db_instances = s.db.instances;
- if(s.db.contexts > max_db_contexts)
- max_db_contexts = s.db.contexts;
- if(s.ingest.replication.instances > max_collection_replication_instances)
- max_collection_replication_instances = s.ingest.replication.instances;
- if(s.stream.replication.instances > max_streaming_replication_instances)
- max_streaming_replication_instances = s.stream.replication.instances;
- for(int i = 0; i < STREAM_TRAFFIC_TYPE_MAX ;i++) {
- if (s.stream.sent_bytes_on_this_connection_per_type[i] >
- max_sent_bytes_on_this_connection_per_type[i])
- max_sent_bytes_on_this_connection_per_type[i] =
- s.stream.sent_bytes_on_this_connection_per_type[i];
- }
- // retention
- buffer_json_add_array_item_string(wb, rrdhost_hostname(s.host)); // Node
- buffer_json_add_array_item_uint64(wb, s.db.first_time_s * MSEC_PER_SEC); // dbFrom
- buffer_json_add_array_item_uint64(wb, s.db.last_time_s * MSEC_PER_SEC); // dbTo
- if(s.db.first_time_s && s.db.last_time_s && s.db.last_time_s > s.db.first_time_s)
- buffer_json_add_array_item_uint64(wb, s.db.last_time_s - s.db.first_time_s); // dbDuration
- else
- buffer_json_add_array_item_string(wb, NULL); // dbDuration
- buffer_json_add_array_item_uint64(wb, s.db.metrics); // dbMetrics
- buffer_json_add_array_item_uint64(wb, s.db.instances); // dbInstances
- buffer_json_add_array_item_uint64(wb, s.db.contexts); // dbContexts
- // statuses
- buffer_json_add_array_item_string(wb, rrdhost_ingest_status_to_string(s.ingest.status)); // InStatus
- buffer_json_add_array_item_string(wb, rrdhost_streaming_status_to_string(s.stream.status)); // OutStatus
- buffer_json_add_array_item_string(wb, rrdhost_ml_status_to_string(s.ml.status)); // MLStatus
- // collection
- if(s.ingest.since) {
- buffer_json_add_array_item_uint64(wb, s.ingest.since * MSEC_PER_SEC); // InSince
- buffer_json_add_array_item_time_t(wb, s.now - s.ingest.since); // InAge
- }
- else {
- buffer_json_add_array_item_string(wb, NULL); // InSince
- buffer_json_add_array_item_string(wb, NULL); // InAge
- }
- buffer_json_add_array_item_string(wb, stream_handshake_error_to_string(s.ingest.reason)); // InReason
- buffer_json_add_array_item_uint64(wb, s.ingest.hops); // InHops
- buffer_json_add_array_item_double(wb, s.ingest.replication.completion); // InReplCompletion
- buffer_json_add_array_item_uint64(wb, s.ingest.replication.instances); // InReplInstances
- buffer_json_add_array_item_string(wb, s.ingest.peers.local.ip); // InLocalIP
- buffer_json_add_array_item_uint64(wb, s.ingest.peers.local.port); // InLocalPort
- buffer_json_add_array_item_string(wb, s.ingest.peers.peer.ip); // InRemoteIP
- buffer_json_add_array_item_uint64(wb, s.ingest.peers.peer.port); // InRemotePort
- buffer_json_add_array_item_string(wb, s.ingest.ssl ? "SSL" : "PLAIN"); // InSSL
- stream_capabilities_to_json_array(wb, s.ingest.capabilities, NULL); // InCapabilities
- // streaming
- if(s.stream.since) {
- buffer_json_add_array_item_uint64(wb, s.stream.since * MSEC_PER_SEC); // OutSince
- buffer_json_add_array_item_time_t(wb, s.now - s.stream.since); // OutAge
- }
- else {
- buffer_json_add_array_item_string(wb, NULL); // OutSince
- buffer_json_add_array_item_string(wb, NULL); // OutAge
- }
- buffer_json_add_array_item_string(wb, stream_handshake_error_to_string(s.stream.reason)); // OutReason
- buffer_json_add_array_item_uint64(wb, s.stream.hops); // OutHops
- buffer_json_add_array_item_double(wb, s.stream.replication.completion); // OutReplCompletion
- buffer_json_add_array_item_uint64(wb, s.stream.replication.instances); // OutReplInstances
- buffer_json_add_array_item_string(wb, s.stream.peers.local.ip); // OutLocalIP
- buffer_json_add_array_item_uint64(wb, s.stream.peers.local.port); // OutLocalPort
- buffer_json_add_array_item_string(wb, s.stream.peers.peer.ip); // OutRemoteIP
- buffer_json_add_array_item_uint64(wb, s.stream.peers.peer.port); // OutRemotePort
- buffer_json_add_array_item_string(wb, s.stream.ssl ? "SSL" : "PLAIN"); // OutSSL
- buffer_json_add_array_item_string(wb, s.stream.compression ? "COMPRESSED" : "UNCOMPRESSED"); // OutCompression
- stream_capabilities_to_json_array(wb, s.stream.capabilities, NULL); // OutCapabilities
- buffer_json_add_array_item_uint64(wb, s.stream.sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_DATA]);
- buffer_json_add_array_item_uint64(wb, s.stream.sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_METADATA]);
- buffer_json_add_array_item_uint64(wb, s.stream.sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_REPLICATION]);
- buffer_json_add_array_item_uint64(wb, s.stream.sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_FUNCTIONS]);
- buffer_json_add_array_item_array(wb); // OutAttemptHandshake
- time_t last_attempt = 0;
- for(struct rrdpush_destinations *d = host->destinations; d ; d = d->next) {
- if(d->since > last_attempt)
- last_attempt = d->since;
- buffer_json_add_array_item_string(wb, stream_handshake_error_to_string(d->reason));
- }
- buffer_json_array_close(wb); // // OutAttemptHandshake
- if(!last_attempt) {
- buffer_json_add_array_item_string(wb, NULL); // OutAttemptSince
- buffer_json_add_array_item_string(wb, NULL); // OutAttemptAge
- }
- else {
- buffer_json_add_array_item_uint64(wb, last_attempt * 1000); // OutAttemptSince
- buffer_json_add_array_item_time_t(wb, s.now - last_attempt); // OutAttemptAge
- }
- // ML
- if(s.ml.status == RRDHOST_ML_STATUS_RUNNING) {
- buffer_json_add_array_item_uint64(wb, s.ml.metrics.anomalous); // MlAnomalous
- buffer_json_add_array_item_uint64(wb, s.ml.metrics.normal); // MlNormal
- buffer_json_add_array_item_uint64(wb, s.ml.metrics.trained); // MlTrained
- buffer_json_add_array_item_uint64(wb, s.ml.metrics.pending); // MlPending
- buffer_json_add_array_item_uint64(wb, s.ml.metrics.silenced); // MlSilenced
- if(s.ml.metrics.anomalous > max_ml_anomalous)
- max_ml_anomalous = s.ml.metrics.anomalous;
- if(s.ml.metrics.normal > max_ml_normal)
- max_ml_normal = s.ml.metrics.normal;
- if(s.ml.metrics.trained > max_ml_trained)
- max_ml_trained = s.ml.metrics.trained;
- if(s.ml.metrics.pending > max_ml_pending)
- max_ml_pending = s.ml.metrics.pending;
- if(s.ml.metrics.silenced > max_ml_silenced)
- max_ml_silenced = s.ml.metrics.silenced;
- }
- else {
- buffer_json_add_array_item_string(wb, NULL); // MlAnomalous
- buffer_json_add_array_item_string(wb, NULL); // MlNormal
- buffer_json_add_array_item_string(wb, NULL); // MlTrained
- buffer_json_add_array_item_string(wb, NULL); // MlPending
- buffer_json_add_array_item_string(wb, NULL); // MlSilenced
- }
- // close
- buffer_json_array_close(wb);
- }
- dfe_done(host);
- }
- buffer_json_array_close(wb); // data
- buffer_json_member_add_object(wb, "columns");
- {
- size_t field_id = 0;
- // Node
- buffer_rrdf_table_add_field(wb, field_id++, "Node", "Node's Hostname",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_VISIBLE | RRDF_FIELD_OPTS_UNIQUE_KEY | RRDF_FIELD_OPTS_STICKY,
- NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "dbFrom", "DB Data Retention From",
- RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "dbTo", "DB Data Retention To",
- RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "dbDuration", "DB Data Retention Duration",
- RRDF_FIELD_TYPE_DURATION, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DURATION_S,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "dbMetrics", "Time-series Metrics in the DB",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, NULL, max_db_metrics, RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "dbInstances", "Instances in the DB",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, NULL, max_db_instances, RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "dbContexts", "Contexts in the DB",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, NULL, max_db_contexts, RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- // --- statuses ---
- buffer_rrdf_table_add_field(wb, field_id++, "InStatus", "Data Collection Online Status",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutStatus", "Streaming Online Status",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "MlStatus", "ML Status",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- // --- collection ---
- buffer_rrdf_table_add_field(wb, field_id++, "InSince", "Last Data Collection Status Change",
- RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
- 0, NULL, NAN, RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InAge", "Last Data Collection Online Status Change Age",
- RRDF_FIELD_TYPE_DURATION, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DURATION_S,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InReason", "Data Collection Online Status Reason",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InHops", "Data Collection Distance Hops from Origin Node",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InReplCompletion", "Inbound Replication Completion",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_BAR, RRDF_FIELD_TRANSFORM_NUMBER,
- 1, "%", 100.0, RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InReplInstances", "Inbound Replicating Instances",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "instances", max_collection_replication_instances, RRDF_FIELD_SORT_DESCENDING,
- NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InLocalIP", "Inbound Local IP",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InLocalPort", "Inbound Local Port",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InRemoteIP", "Inbound Remote IP",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InRemotePort", "Inbound Remote Port",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InSSL", "Inbound SSL Connection",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "InCapabilities", "Inbound Connection Capabilities",
- RRDF_FIELD_TYPE_ARRAY, RRDF_FIELD_VISUAL_PILL, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- // --- streaming ---
- buffer_rrdf_table_add_field(wb, field_id++, "OutSince", "Last Streaming Status Change",
- RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
- 0, NULL, NAN, RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutAge", "Last Streaming Status Change Age",
- RRDF_FIELD_TYPE_DURATION, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DURATION_S,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutReason", "Streaming Status Reason",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutHops", "Streaming Distance Hops from Origin Node",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutReplCompletion", "Outbound Replication Completion",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_BAR, RRDF_FIELD_TRANSFORM_NUMBER,
- 1, "%", 100.0, RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutReplInstances", "Outbound Replicating Instances",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "instances", max_streaming_replication_instances, RRDF_FIELD_SORT_DESCENDING,
- NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutLocalIP", "Outbound Local IP",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutLocalPort", "Outbound Local Port",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutRemoteIP", "Outbound Remote IP",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutRemotePort", "Outbound Remote Port",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutSSL", "Outbound SSL Connection",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutCompression", "Outbound Compressed Connection",
- RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutCapabilities", "Outbound Connection Capabilities",
- RRDF_FIELD_TYPE_ARRAY, RRDF_FIELD_VISUAL_PILL, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutTrafficData", "Outbound Metric Data Traffic",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "bytes", max_sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_DATA],
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutTrafficMetadata", "Outbound Metric Metadata Traffic",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "bytes",
- max_sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_METADATA],
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutTrafficReplication", "Outbound Metric Replication Traffic",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "bytes",
- max_sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_REPLICATION],
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutTrafficFunctions", "Outbound Metric Functions Traffic",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "bytes",
- max_sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_FUNCTIONS],
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutAttemptHandshake",
- "Outbound Connection Attempt Handshake Status",
- RRDF_FIELD_TYPE_ARRAY, RRDF_FIELD_VISUAL_PILL, RRDF_FIELD_TRANSFORM_NONE,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutAttemptSince",
- "Last Outbound Connection Attempt Status Change Time",
- RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
- 0, NULL, NAN, RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_MAX, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "OutAttemptAge",
- "Last Outbound Connection Attempt Status Change Age",
- RRDF_FIELD_TYPE_DURATION, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DURATION_S,
- 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
- RRDF_FIELD_SUMMARY_MIN, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_VISIBLE, NULL);
- // --- ML ---
- buffer_rrdf_table_add_field(wb, field_id++, "MlAnomalous", "Number of Anomalous Metrics",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "metrics",
- max_ml_anomalous,
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "MlNormal", "Number of Not Anomalous Metrics",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "metrics",
- max_ml_normal,
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "MlTrained", "Number of Trained Metrics",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "metrics",
- max_ml_trained,
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "MlPending", "Number of Pending Metrics",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "metrics",
- max_ml_pending,
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- buffer_rrdf_table_add_field(wb, field_id++, "MlSilenced", "Number of Silenced Metrics",
- RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NUMBER,
- 0, "metrics",
- max_ml_silenced,
- RRDF_FIELD_SORT_DESCENDING, NULL,
- RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
- RRDF_FIELD_OPTS_NONE, NULL);
- }
- buffer_json_object_close(wb); // columns
- buffer_json_member_add_string(wb, "default_sort_column", "Node");
- buffer_json_member_add_object(wb, "charts");
- {
- // Data Collection Age chart
- buffer_json_member_add_object(wb, "InAge");
- {
- buffer_json_member_add_string(wb, "name", "Data Collection Age");
- buffer_json_member_add_string(wb, "type", "stacked-bar");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "InAge");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- // Streaming Age chart
- buffer_json_member_add_object(wb, "OutAge");
- {
- buffer_json_member_add_string(wb, "name", "Streaming Age");
- buffer_json_member_add_string(wb, "type", "stacked-bar");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "OutAge");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- // DB Duration
- buffer_json_member_add_object(wb, "dbDuration");
- {
- buffer_json_member_add_string(wb, "name", "Retention Duration");
- buffer_json_member_add_string(wb, "type", "stacked-bar");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "dbDuration");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- }
- buffer_json_object_close(wb); // charts
- buffer_json_member_add_array(wb, "default_charts");
- {
- buffer_json_add_array_item_array(wb);
- buffer_json_add_array_item_string(wb, "InAge");
- buffer_json_add_array_item_string(wb, "Node");
- buffer_json_array_close(wb);
- buffer_json_add_array_item_array(wb);
- buffer_json_add_array_item_string(wb, "OutAge");
- buffer_json_add_array_item_string(wb, "Node");
- buffer_json_array_close(wb);
- }
- buffer_json_array_close(wb);
- buffer_json_member_add_object(wb, "group_by");
- {
- buffer_json_member_add_object(wb, "Node");
- {
- buffer_json_member_add_string(wb, "name", "Node");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "Node");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- buffer_json_member_add_object(wb, "InStatus");
- {
- buffer_json_member_add_string(wb, "name", "Nodes by Collection Status");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "InStatus");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- buffer_json_member_add_object(wb, "OutStatus");
- {
- buffer_json_member_add_string(wb, "name", "Nodes by Streaming Status");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "OutStatus");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- buffer_json_member_add_object(wb, "MlStatus");
- {
- buffer_json_member_add_string(wb, "name", "Nodes by ML Status");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "MlStatus");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- buffer_json_member_add_object(wb, "InRemoteIP");
- {
- buffer_json_member_add_string(wb, "name", "Nodes by Inbound IP");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "InRemoteIP");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- buffer_json_member_add_object(wb, "OutRemoteIP");
- {
- buffer_json_member_add_string(wb, "name", "Nodes by Outbound IP");
- buffer_json_member_add_array(wb, "columns");
- {
- buffer_json_add_array_item_string(wb, "OutRemoteIP");
- }
- buffer_json_array_close(wb);
- }
- buffer_json_object_close(wb);
- }
- buffer_json_object_close(wb); // group_by
- buffer_json_member_add_time_t(wb, "expires", now_realtime_sec() + 1);
- buffer_json_finalize(wb);
- int response = HTTP_RESP_OK;
- if(is_cancelled_cb && is_cancelled_cb(is_cancelled_cb_data)) {
- buffer_flush(wb);
- response = HTTP_RESP_CLIENT_CLOSED_REQUEST;
- }
- if(result_cb)
- result_cb(wb, response, result_cb_data);
- return response;
- }
|