archive/netdata_netdata
0
0
Fork 0
mirror of https://github.com/netdata/netdata.git synced 2025-04-13 01:08:11 +00:00
Code Issues Projects Releases Wiki Activity
netdata_netdata/database/rrdfunctions.c
Costa Tsaousis 3e508c8f95
New logging layer () 
* cleanup of logging - wip

* first working iteration

* add errno annotator

* replace old logging functions with netdata_logger()

* cleanup

* update error_limit

* fix remanining error_limit references

* work on fatal()

* started working on structured logs

* full cleanup

* default logging to files; fix all plugins initialization

* fix formatting of numbers

* cleanup and reorg

* fix coverity issues

* cleanup obsolete code

* fix formatting of numbers

* fix log rotation

* fix for older systems

* add detection of systemd journal via stderr

* finished on access.log

* remove left-over transport

* do not add empty fields to the logs

* journal get compact uuids; X-Transaction-ID header is added in web responses

* allow compiling on systems without memfd sealing

* added libnetdata/uuid directory

* move datetime formatters to libnetdata

* add missing files

* link the makefiles in libnetdata

* added uuid_parse_flexi() to parse UUIDs with and without hyphens; the web server now read X-Transaction-ID and uses it for functions and web responses

* added stream receiver, sender, proc plugin and pluginsd log stack

* iso8601 advanced usage; line_splitter module in libnetdata; code cleanup

* add message ids to streaming inbound and outbound connections

* cleanup line_splitter between lines to avoid logging garbage; when killing children, kill them with SIGABRT if internal checks is enabled

* send SIGABRT to external plugins only if we are not shutting down

* fix cross cleanup in pluginsd parser

* fatal when there is a stack error in logs

* compile netdata with -fexceptions

* do not kill external plugins with SIGABRT

* metasync info logs to debug level

* added severity to logs

* added json output; added options per log output; added documentation; fixed issues mentioned

* allow memfd only on linux

* moved journal low level functions to journal.c/h

* move health logs to daemon.log with proper priorities

* fixed a couple of bugs; health log in journal

* updated docs

* systemd-cat-native command to push structured logs to journal from the command line

* fix makefiles

* restored NETDATA_LOG_SEVERITY_LEVEL

* fix makefiles

* systemd-cat-native can also work as the logger of Netdata scripts

* do not require a socket to systemd-journal to log-as-netdata

* alarm notify logs in native format

* properly compare log ids

* fatals log alerts; alarm-notify.sh working

* fix overflow warning

* alarm-notify.sh now logs the request (command line)

* anotate external plugins logs with the function cmd they run

* added context, component and type to alarm-notify.sh; shell sanitization removes control character and characters that may be expanded by bash

* reformatted alarm-notify logs

* unify cgroup-network-helper.sh

* added quotes around params

* charts.d.plugin switched logging to journal native

* quotes for logfmt

* unify the status codes of streaming receivers and senders

* alarm-notify: dont log anything, if there is nothing to do

* all external plugins log to stderr when running outside netdata; alarm-notify now shows an error when notifications menthod are needed but are not available

* migrate cgroup-name.sh to new logging

* systemd-cat-native now supports messages with newlines

* socket.c logs use priority

* cleanup log field types

* inherit the systemd set INVOCATION_ID if found

* allow systemd-cat-native to send messages to a systemd-journal-remote URL

* log2journal command that can convert structured logs to journal export format

* various fixes and documentation of log2journal

* updated log2journal docs

* updated log2journal docs

* updated documentation of fields

* allow compiling without libcurl

* do not use socket as format string

* added version information to newly added tools

* updated documentation and help messages

* fix the namespace socket path

* print errno with error

* do not timeout

* updated docs

* updated docs

* updated docs

* log2journal updated docs and params

* when talking to a remote journal, systemd-cat-native batches the messages

* enable lz4 compression for systemd-cat-native when sending messages to a systemd-journal-remote

* Revert "enable lz4 compression for systemd-cat-native when sending messages to a systemd-journal-remote"

This reverts commit b079d53c11.

* note about uncompressed traffic

* log2journal: code reorg and cleanup to make modular

* finished rewriting log2journal

* more comments

* rewriting rules support

* increased limits

* updated docs

* updated docs

* fix old log call

* use journal only when stderr is connected to journal

* update netdata.spec for libcurl, libpcre2 and log2journal

* pcre2-devel

* do not require pcre2 in centos < 8, amazonlinux < 2023, open suse

* log2journal only on systems pcre2 is available

* ignore log2journal in .gitignore

* avoid log2journal on centos 7, amazonlinux 2 and opensuse

* add pcre2-8 to static build

* undo last commit

* Bundle to static

Signed-off-by: Tasos Katsoulas <tasos@netdata.cloud>

* Add build deps for deb packages

Signed-off-by: Tasos Katsoulas <tasos@netdata.cloud>

* Add dependencies; build from source

Signed-off-by: Tasos Katsoulas <tasos@netdata.cloud>

* Test build for amazon linux and centos expect to fail for suse

Signed-off-by: Tasos Katsoulas <tasos@netdata.cloud>

* fix minor oversight

Signed-off-by: Tasos Katsoulas <tasos@netdata.cloud>

* Reorg code

* Add the install from source (deps) as a TODO
* Not enable the build on suse ecosystem

Signed-off-by: Tasos Katsoulas <tasos@netdata.cloud>

---------

Signed-off-by: Tasos Katsoulas <tasos@netdata.cloud>
Co-authored-by: Tasos Katsoulas <tasos@netdata.cloud>
2023-11-22 10:27:25 +02:00

1821 lines
76 KiB
C
Raw Blame History

// 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;
}
Reference in a new issue View git blame Copy permalink