src/web_client.h
src/web_server.c
src/web_server.h
- src/rrdhost.c src/rrdfamily.c src/rrdset.c src/rrddim.c src/health_log.c src/health_config.c src/health_json.c src/rrdcalc.c src/rrdcalctemplate.c src/rrdvar.c src/rrddimvar.c src/rrdsetvar.c)
+ src/rrdhost.c src/rrdfamily.c src/rrdset.c src/rrddim.c src/health_log.c src/health_config.c src/health_json.c src/rrdcalc.c src/rrdcalctemplate.c src/rrdvar.c src/rrddimvar.c src/rrdsetvar.c src/rrdpush.c src/rrdpush.h)
set(APPS_PLUGIN_SOURCE_FILES
src/appconfig.c
rrddimvar.c \
rrdsetvar.c \
rrd2json.c rrd2json.h \
+ rrdpush.c rrdpush.h \
storage_number.c storage_number.h \
unit_test.c unit_test.h \
url.c url.h \
}
void *backends_main(void *ptr) {
+ int default_port = 0;
+ int sock = -1;
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
BUFFER *b = buffer_create(1), *response = buffer_create(1);
- int (*backend_request_formatter)(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) = NULL;
- int (*backend_response_checker)(BUFFER *b) = NULL;
+ int (*backend_request_formatter)(BUFFER *, const char *, RRDHOST *, const char *, RRDSET *, RRDDIM *, time_t, time_t, uint32_t) = NULL;
+ int (*backend_response_checker)(BUFFER *) = NULL;
info("BACKEND thread created with task id %d", gettid());
// ------------------------------------------------------------------------
// collect configuration options
+ if(central_netdata_to_push_data) {
+ info("Backend is disabled - use the central netdata");
+ goto cleanup;
+ }
+
struct timeval timeout = {
.tv_sec = 0,
.tv_usec = 0
};
- int default_port = 0;
- int sock = -1;
uint32_t options;
int enabled = config_get_boolean("backend", "enabled", 0);
const char *source = config_get("backend", "data source", "average");
if(unlikely(sock == -1)) {
usec_t start_ut = now_monotonic_usec();
- const char *s = destination;
- while(*s) {
- const char *e = s;
-
- // skip separators, moving both s(tart) and e(nd)
- while(isspace(*e) || *e == ',') s = ++e;
+ size_t reconnects = 0;
- // move e(nd) to the first separator
- while(*e && !isspace(*e) && *e != ',') e++;
+ sock = connect_to_one_of(destination, default_port, &timeout, &reconnects);
- // is there anything?
- if(!*s || s == e) break;
-
- char buf[e - s + 1];
- strncpyz(buf, s, e - s);
- chart_backend_reconnects++;
- sock = connect_to(buf, default_port, &timeout);
- if(sock != -1) break;
- s = e;
- }
+ chart_backend_reconnects += reconnects;
chart_backend_latency += now_monotonic_usec() - start_ut;
}
#define NETDATA_OS_TYPE "linux"
#endif /* __FreeBSD__, __APPLE__*/
-#include "plugin_tc.h"
-#include "plugins_d.h"
#include "socket.h"
#include "eval.h"
#include "health.h"
#include "rrd.h"
+#include "plugin_tc.h"
+#include "plugins_d.h"
#include "rrd2json.h"
#include "web_client.h"
#include "web_server.h"
#include "backends.h"
#include "inlined.h"
#include "adaptive_resortable_list.h"
+#include "rrdpush.h"
extern char *netdata_configured_config_dir;
extern char *netdata_configured_log_dir;
extern char *netdata_configured_varlib_dir;
extern char *netdata_configured_home_dir;
extern char *netdata_configured_host_prefix;
+extern char *central_netdata_to_push_data;
extern void netdata_fix_chart_id(char *s);
extern void netdata_fix_chart_name(char *s);
#define NETDATA_HEALTH_INTERNALS
#include "common.h"
-int default_localhost_health_enabled = 1;
+int default_health_enabled = 1;
// ----------------------------------------------------------------------------
// health initialization
void health_init(void) {
debug(D_HEALTH, "Health configuration initializing");
- if(!(default_localhost_health_enabled = config_get_boolean("health", "enabled", 1))) {
- debug(D_HEALTH, "Health is disabled.");
- return;
+ if(!central_netdata_to_push_data) {
+ if(!(default_health_enabled = config_get_boolean("health", "enabled", 1))) {
+ debug(D_HEALTH, "Health is disabled.");
+ return;
+ }
+ }
+ else {
+ info("Health is disabled - setup alarms at the central netdata.");
+ config_set_boolean("health", "enabled", 0);
+ default_health_enabled = 0;
}
char pathname[FILENAME_MAX + 1];
#ifndef NETDATA_HEALTH_H
#define NETDATA_HEALTH_H
-extern int default_localhost_health_enabled;
+extern int default_health_enabled;
extern int rrdvar_compare(void *a, void *b);
extern void *cgroups_main(void *ptr);
+char *central_netdata_to_push_data = NULL;
+
void netdata_cleanup_and_exit(int ret) {
netdata_exit = 1;
{"plugins.d", NULL, NULL, 1, NULL, NULL, pluginsd_main},
{"web", NULL, NULL, 1, NULL, NULL, socket_listen_main_multi_threaded},
{"web-single-threaded", NULL, NULL, 0, NULL, NULL, socket_listen_main_single_threaded},
+ {"central-netdata-push",NULL, NULL, 0, NULL, NULL, central_netdata_push_thread},
{NULL, NULL, NULL, 0, NULL, NULL, NULL}
};
void web_server_threading_selection(void) {
- int threaded = config_get_boolean("global", "multi threaded web server", 1);
+ int multi_threaded = 0;
+ int single_threaded = 0;
+ int central_thread = 0;
+
+ if(!central_netdata_to_push_data) {
+ multi_threaded = config_get_boolean("global", "multi threaded web server", 1);
+ single_threaded = !multi_threaded;
+ }
+ else {
+ central_thread = 1;
+ info("Web servers are disabled - use the central netdata.");
+ }
int i;
for(i = 0; static_threads[i].name ; i++) {
if(static_threads[i].start_routine == socket_listen_main_multi_threaded)
- static_threads[i].enabled = threaded?1:0;
+ static_threads[i].enabled = multi_threaded;
if(static_threads[i].start_routine == socket_listen_main_single_threaded)
- static_threads[i].enabled = threaded?0:1;
+ static_threads[i].enabled = single_threaded;
+
+ if(static_threads[i].start_routine == central_netdata_push_thread)
+ static_threads[i].enabled = central_thread;
}
+ if(central_netdata_to_push_data)
+ return;
+
web_client_timeout = (int) config_get_number("global", "disconnect idle web clients after seconds", DEFAULT_DISCONNECT_IDLE_WEB_CLIENTS_AFTER_SECONDS);
respect_web_browser_do_not_track_policy = config_get_boolean("global", "respect web browser do not track policy", respect_web_browser_do_not_track_policy);
char* stacksize_string = "stacksize=";
char* debug_flags_string = "debug_flags=";
if(strcmp(optarg, "unittest") == 0) {
- default_localhost_rrd_update_every = 1;
+ default_rrd_update_every = 1;
if(run_all_mockup_tests()) exit(1);
if(unit_test_storage()) exit(1);
fprintf(stderr, "\n\nALL TESTS PASSED\n\n");
setenv("MALLOC_ARENA_MAX", pmax, 1);
#if defined(HAVE_C_MALLOPT)
- int i = config_get_number("global", "glibc malloc arena max for netdata", 1);
+ i = (int)config_get_number("global", "glibc malloc arena max for netdata", 1);
if(i > 0)
mallopt(M_ARENA_MAX, 1);
#endif
}
char *user = NULL;
+
{
+ // --------------------------------------------------------------------
+ // get the debugging flags from the configuration file
+
char *flags = config_get("global", "debug flags", "0x0000000000000000");
setenv("NETDATA_DEBUG_FLAGS", flags, 1);
#endif
}
- // --------------------------------------------------------------------
-
-#ifdef MADV_MERGEABLE
- enable_ksm = config_get_boolean("global", "memory deduplication (ksm)", enable_ksm);
-#else
-#warning "Kernel memory deduplication (KSM) is not available"
-#endif
-
- // --------------------------------------------------------------------
- get_system_HZ();
- get_system_cpus();
- get_system_pid_max();
-
// --------------------------------------------------------------------
+ // get log filenames and settings
{
char filename[FILENAME_MAX + 1];
}
error_log_throttle_period_backup =
- error_log_throttle_period = config_get_number("global", "errors flood protection period", error_log_throttle_period);
+ error_log_throttle_period = config_get_number("global", "errors flood protection period", error_log_throttle_period);
+
setenv("NETDATA_ERRORS_THROTTLE_PERIOD", config_get("global", "errors flood protection period" , ""), 1);
- error_log_errors_per_period = (unsigned long)config_get_number("global", "errors to trigger flood protection", error_log_errors_per_period);
- setenv("NETDATA_ERRORS_PER_PERIOD" , config_get("global", "errors to trigger flood protection", ""), 1);
+ error_log_errors_per_period = (unsigned long)config_get_number("global", "errors to trigger flood protection", (long long int)error_log_errors_per_period);
+ setenv("NETDATA_ERRORS_PER_PERIOD", config_get("global", "errors to trigger flood protection", ""), 1);
if(check_config) {
stdout_filename = stderr_filename = stdaccess_filename = "system";
}
error_log_limit_unlimited();
+
+ // --------------------------------------------------------------------
+ // get KSM settings
+
+#ifdef MADV_MERGEABLE
+ enable_ksm = config_get_boolean("global", "memory deduplication (ksm)", enable_ksm);
+#else
+#warning "Kernel memory deduplication (KSM) is not available"
+#endif
+
// --------------------------------------------------------------------
+ // get various system parameters
+
+ get_system_HZ();
+ get_system_cpus();
+ get_system_pid_max();
- default_localhost_rrd_memory_mode = rrd_memory_mode_id(config_get("global", "memory mode", rrd_memory_mode_name(default_localhost_rrd_memory_mode)));
// --------------------------------------------------------------------
+ // find the system hostname
{
char hostnamebuf[HOSTNAME_MAX + 1];
if(gethostname(hostnamebuf, HOSTNAME_MAX) == -1)
error("WARNING: Cannot get machine hostname.");
+
hostname = config_get("global", "hostname", hostnamebuf);
debug(D_OPTIONS, "hostname set to '%s'", hostname);
+
setenv("NETDATA_HOSTNAME", hostname, 1);
}
+
+ // --------------------------------------------------------------------
+ // find we need to send data to a central netdata
+
+ central_netdata_to_push_data = config_get("global", "central netdata to send all data", "");
+ if(central_netdata_to_push_data && !*central_netdata_to_push_data)
+ central_netdata_to_push_data = NULL;
+
+
+ // --------------------------------------------------------------------
+ // get default memory mode for the database
+
+ if(central_netdata_to_push_data) {
+ default_rrd_memory_mode = RRD_MEMORY_MODE_RAM;
+ config_set("global", "memory mode", rrd_memory_mode_name(default_rrd_memory_mode));
+ }
+ else
+ default_rrd_memory_mode = rrd_memory_mode_id(config_get("global", "memory mode", rrd_memory_mode_name(default_rrd_memory_mode)));
+
+
// --------------------------------------------------------------------
+ // get default database size
- default_localhost_rrd_history_entries = (int) config_get_number("global", "history", RRD_DEFAULT_HISTORY_ENTRIES);
- if(default_localhost_rrd_history_entries < 5 || default_localhost_rrd_history_entries > RRD_HISTORY_ENTRIES_MAX) {
- error("Invalid history entries %d given. Defaulting to %d.", default_localhost_rrd_history_entries, RRD_DEFAULT_HISTORY_ENTRIES);
- default_localhost_rrd_history_entries = RRD_DEFAULT_HISTORY_ENTRIES;
+ if(central_netdata_to_push_data) {
+ default_rrd_history_entries = 10;
+ config_set_number("global", "history", default_rrd_history_entries);
}
- else {
- debug(D_OPTIONS, "save lines set to %d.", default_localhost_rrd_history_entries);
+ else
+ default_rrd_history_entries = (int) config_get_number("global", "history", align_entries_to_pagesize(RRD_DEFAULT_HISTORY_ENTRIES));
+
+ long h = align_entries_to_pagesize(default_rrd_history_entries);
+ if(h != default_rrd_history_entries) {
+ config_set_number("global", "history", h);
+ default_rrd_history_entries = (int)h;
+ }
+
+ if(default_rrd_history_entries < 5 || default_rrd_history_entries > RRD_HISTORY_ENTRIES_MAX) {
+ error("Invalid history entries %d given. Defaulting to %d.", default_rrd_history_entries, RRD_DEFAULT_HISTORY_ENTRIES);
+ default_rrd_history_entries = RRD_DEFAULT_HISTORY_ENTRIES;
}
+ else
+ debug(D_OPTIONS, "save lines set to %d.", default_rrd_history_entries);
+
// --------------------------------------------------------------------
+ // get default database update frequency
- default_localhost_rrd_update_every = (int) config_get_number("global", "update every", UPDATE_EVERY);
- if(default_localhost_rrd_update_every < 1 || default_localhost_rrd_update_every > 600) {
- error("Invalid data collection frequency (update every) %d given. Defaulting to %d.", default_localhost_rrd_update_every, UPDATE_EVERY_MAX);
- default_localhost_rrd_update_every = UPDATE_EVERY;
+ default_rrd_update_every = (int) config_get_number("global", "update every", UPDATE_EVERY);
+ if(default_rrd_update_every < 1 || default_rrd_update_every > 600) {
+ error("Invalid data collection frequency (update every) %d given. Defaulting to %d.", default_rrd_update_every, UPDATE_EVERY_MAX);
+ default_rrd_update_every = UPDATE_EVERY;
}
- else debug(D_OPTIONS, "update timer set to %d.", default_localhost_rrd_update_every);
+ else debug(D_OPTIONS, "update timer set to %d.", default_rrd_update_every);
// let the plugins know the min update_every
{
char buf[16];
- snprintfz(buf, 15, "%d", default_localhost_rrd_update_every);
+ snprintfz(buf, 15, "%d", default_rrd_update_every);
setenv("NETDATA_UPDATE_EVERY", buf, 1);
}
+
// --------------------------------------------------------------------
+ // setup process signals
// block signals while initializing threads.
// this causes the threads to block signals.
if(sigaction(SIGUSR2, &sa, NULL) == -1)
error("Failed to change signal handler for SIGUSR2");
+
// --------------------------------------------------------------------
+ // get the required stack size of the threads of netdata
i = pthread_attr_init(&attr);
if(i != 0)
wanted_stacksize = (size_t)config_get_number("global", "pthread stack size", (long)stacksize);
+
// --------------------------------------------------------------------
+ // check which threads are enabled and initialize them
for (i = 0; static_threads[i].name != NULL ; i++) {
struct netdata_static_thread *st = &static_threads[i];
- if(st->config_name) st->enabled = config_get_boolean(st->config_section, st->config_name, st->enabled);
- if(st->enabled && st->init_routine) st->init_routine();
+ if(st->config_name)
+ st->enabled = config_get_boolean(st->config_section, st->config_name, st->enabled);
+
+ if(st->enabled && st->init_routine)
+ st->init_routine();
}
- // --------------------------------------------------------------------
+ // --------------------------------------------------------------------
// get the user we should run
+
// IMPORTANT: this is required before web_files_uid()
user = config_get("global", "run as user" , (getuid() == 0)?NETDATA_USER:"");
web_files_uid(); // IMPORTANT: web_files_uid() before web_files_gid()
web_files_gid();
+
// --------------------------------------------------------------------
+ // create the listening sockets
- if(!check_config)
+ if(!check_config && !central_netdata_to_push_data)
create_listen_sockets();
}
}
#endif /* NETDATA_INTERNAL_CHECKS */
+
// fork, switch user, create pid file, set process priority
if(become_daemon(dont_fork, user) == -1)
fatal("Cannot daemonize myself.");
info("netdata started on pid %d.", getpid());
+
// ------------------------------------------------------------------------
- // get default pthread stack size
+ // set default pthread stack size - after we have forked
if(stacksize < wanted_stacksize) {
i = pthread_attr_setstacksize(&attr, wanted_stacksize);
debug(D_SYSTEM, "Successfully set pthread stacksize to %zu bytes", wanted_stacksize);
}
+
// ------------------------------------------------------------------------
// initialize health monitoring
health_init();
+
// ------------------------------------------------------------------------
// initialize the registry
registry_init();
+
// ------------------------------------------------------------------------
// initialize rrd host
rrd_init(hostname);
+
if(check_config)
exit(1);
+
// ------------------------------------------------------------------------
// enable log flood protection
error_log_limit_reset();
+
// ------------------------------------------------------------------------
// spawn the threads
info("netdata initialization completed. Enjoy real-time performance monitoring!");
+
// ------------------------------------------------------------------------
// block signals while initializing threads.
sigset_t sigset;
usec = dt_usec(&now, &last) - susec;
debug(D_NFACCT_LOOP, "nfacct.plugin: last loop took %llu usec (worked for %llu, sleeped for %llu).", usec + susec, usec, susec);
- if(usec < (default_localhost_rrd_update_every * 1000000ULL / 2ULL)) susec = (default_localhost_rrd_update_every * 1000000ULL) - usec;
- else susec = default_localhost_rrd_update_every * 1000000ULL / 2ULL;
+ if(usec < (default_rrd_update_every * 1000000ULL / 2ULL)) susec = (default_rrd_update_every * 1000000ULL) - usec;
+ else susec = default_rrd_update_every * 1000000ULL / 2ULL;
// --------------------------------------------------------------------
st = rrdset_find_bytype("netfilter", "nfacct_packets");
if(!st) {
- st = rrdset_create("netfilter", "nfacct_packets", NULL, "nfacct", NULL, "Netfilter Accounting Packets", "packets/s", 3206, default_localhost_rrd_update_every, RRDSET_TYPE_STACKED);
+ st = rrdset_create("netfilter", "nfacct_packets", NULL, "nfacct", NULL, "Netfilter Accounting Packets", "packets/s", 3206, default_rrd_update_every, RRDSET_TYPE_STACKED);
for(i = 0; i < nfacct_list->len ; i++)
- rrddim_add(st, nfacct_list->data[i].name, NULL, 1, default_localhost_rrd_update_every, RRD_ALGORITHM_INCREMENTAL);
+ rrddim_add(st, nfacct_list->data[i].name, NULL, 1, default_rrd_update_every, RRD_ALGORITHM_INCREMENTAL);
}
else rrdset_next(st);
for(i = 0; i < nfacct_list->len ; i++) {
RRDDIM *rd = rrddim_find(st, nfacct_list->data[i].name);
- if(!rd) rd = rrddim_add(st, nfacct_list->data[i].name, NULL, 1, default_localhost_rrd_update_every, RRD_ALGORITHM_INCREMENTAL);
+ if(!rd) rd = rrddim_add(st, nfacct_list->data[i].name, NULL, 1, default_rrd_update_every, RRD_ALGORITHM_INCREMENTAL);
if(rd) rrddim_set_by_pointer(st, rd, nfacct_list->data[i].pkts);
}
st = rrdset_find_bytype("netfilter", "nfacct_bytes");
if(!st) {
- st = rrdset_create("netfilter", "nfacct_bytes", NULL, "nfacct", NULL, "Netfilter Accounting Bandwidth", "kilobytes/s", 3207, default_localhost_rrd_update_every, RRDSET_TYPE_STACKED);
+ st = rrdset_create("netfilter", "nfacct_bytes", NULL, "nfacct", NULL, "Netfilter Accounting Bandwidth", "kilobytes/s", 3207, default_rrd_update_every, RRDSET_TYPE_STACKED);
for(i = 0; i < nfacct_list->len ; i++)
- rrddim_add(st, nfacct_list->data[i].name, NULL, 1, 1000 * default_localhost_rrd_update_every, RRD_ALGORITHM_INCREMENTAL);
+ rrddim_add(st, nfacct_list->data[i].name, NULL, 1, 1000 * default_rrd_update_every, RRD_ALGORITHM_INCREMENTAL);
}
else rrdset_next(st);
for(i = 0; i < nfacct_list->len ; i++) {
RRDDIM *rd = rrddim_find(st, nfacct_list->data[i].name);
- if(!rd) rd = rrddim_add(st, nfacct_list->data[i].name, NULL, 1, 1000 * default_localhost_rrd_update_every, RRD_ALGORITHM_INCREMENTAL);
+ if(!rd) rd = rrddim_add(st, nfacct_list->data[i].name, NULL, 1, 1000 * default_rrd_update_every, RRD_ALGORITHM_INCREMENTAL);
if(rd) rrddim_set_by_pointer(st, rd, nfacct_list->data[i].bytes);
}
return i;
}
-inline size_t pluginsd_process(struct plugind *cd, FILE *fp, int trust_durations) {
+inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp, int trust_durations) {
int enabled = cd->enabled;
if(!fp || !enabled) {
}
size_t count = 0;
- RRDHOST *host = localhost;
char line[PLUGINSD_LINE_MAX + 1];
info("PLUGINSD: '%s' running on pid %d", cd->fullfilename, cd->pid);
- count = pluginsd_process(cd, fp, 0);
+ count = pluginsd_process(localhost, cd, fp, 0);
error("PLUGINSD: plugin '%s' disconnected.", cd->fullfilename);
killpid(cd->pid, SIGTERM);
extern struct plugind *pluginsd_root;
extern void *pluginsd_main(void *ptr);
-extern size_t pluginsd_process(struct plugind *cd, FILE *fp, int trust_durations);
+extern size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp, int trust_durations);
#endif /* NETDATA_PLUGINS_D_H */
char filename[FILENAME_MAX + 1];
// registry enabled?
- registry.enabled = config_get_boolean("registry", "enabled", 0);
+ if(!central_netdata_to_push_data) {
+ registry.enabled = config_get_boolean("registry", "enabled", 0);
+ }
+ else {
+ info("Registry is disabled - use the central netdata");
+ config_set_boolean("registry", "enabled", 0);
+ registry.enabled = 0;
+ }
// pathnames
snprintfz(filename, FILENAME_MAX, "%s/registry", netdata_configured_varlib_dir);
int rrd_delete_unupdated_dimensions = 0;
*/
-int default_localhost_rrd_update_every = UPDATE_EVERY;
-int default_localhost_rrd_history_entries = RRD_DEFAULT_HISTORY_ENTRIES;
-RRD_MEMORY_MODE default_localhost_rrd_memory_mode = RRD_MEMORY_MODE_SAVE;
+int default_rrd_update_every = UPDATE_EVERY;
+int default_rrd_history_entries = RRD_DEFAULT_HISTORY_ENTRIES;
+RRD_MEMORY_MODE default_rrd_memory_mode = RRD_MEMORY_MODE_SAVE;
// ----------------------------------------------------------------------------
#define RRD_DEFAULT_HISTORY_ENTRIES 3600
#define RRD_HISTORY_ENTRIES_MAX (86400*10)
-extern int default_localhost_rrd_update_every;
-extern int default_localhost_rrd_history_entries;
+extern int default_rrd_update_every;
+extern int default_rrd_history_entries;
#define RRD_ID_LENGTH_MAX 200
#define RRD_MEMORY_MODE_MAP_NAME "map"
#define RRD_MEMORY_MODE_SAVE_NAME "save"
-extern RRD_MEMORY_MODE default_localhost_rrd_memory_mode;
+extern RRD_MEMORY_MODE default_rrd_memory_mode;
extern const char *rrd_memory_mode_name(RRD_MEMORY_MODE id);
extern RRD_MEMORY_MODE rrd_memory_mode_id(const char *name);
typedef enum rrddim_flags {
RRDDIM_FLAG_HIDDEN = 1 << 0, // this dimension will not be offered to callers
RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS = 1 << 1, // do not offer RESET or OVERFLOW info to callers
- RRDDIM_FLAG_UPDATED = 1 << 31 // the dimension has been updated since the last processing
+ RRDDIM_FLAG_UPDATED = 1 << 2, // the dimension has been updated since the last processing
+ RRDDIM_FLAG_EXPOSED = 1 << 3 // when set what have sent this dimension to the central netdata
} RRDDIM_FLAGS;
#define rrddim_flag_check(rd, flag) ((rd)->flags & flag)
extern void rrdset_next_usec(RRDSET *st, usec_t microseconds);
#define rrdset_next(st) rrdset_next_usec(st, 0ULL)
-extern usec_t rrdset_done(RRDSET *st);
+extern void rrdset_done(RRDSET *st);
// get the total duration in seconds of the round robin database
#define rrdset_duration(st) ((time_t)( (((st)->counter >= ((unsigned long)(st)->entries))?(unsigned long)(st)->entries:(st)->counter) * (st)->update_every ))
extern collected_number rrddim_set_by_pointer(RRDSET *st, RRDDIM *rd, collected_number value);
extern collected_number rrddim_set(RRDSET *st, const char *id, collected_number value);
+extern long align_entries_to_pagesize(long entries);
+
// ----------------------------------------------------------------------------
// RRD internal functions
// prevent incremental calculation spikes
rd->counter = 0;
+
rrddim_flag_clear(rd, RRDDIM_FLAG_UPDATED);
+ rrddim_flag_clear(rd, RRDDIM_FLAG_EXPOSED);
+
rd->calculated_value = 0;
rd->last_calculated_value = 0;
rd->collected_value = 0;
if(!host)
host = rrdhost_create(hostname,
guid,
- default_localhost_rrd_update_every,
- default_localhost_rrd_history_entries,
- default_localhost_rrd_memory_mode,
- default_localhost_health_enabled
+ default_rrd_update_every,
+ default_rrd_history_entries,
+ default_rrd_memory_mode,
+ default_health_enabled
);
return host;
localhost = rrdhost_create(hostname,
registry_get_this_machine_guid(),
- default_localhost_rrd_update_every,
- default_localhost_rrd_history_entries,
- default_localhost_rrd_memory_mode,
- default_localhost_health_enabled
+ default_rrd_update_every,
+ default_rrd_history_entries,
+ default_rrd_memory_mode,
+ default_health_enabled
);
}
--- /dev/null
+#include "common.h"
+
+#define PIPE_READ 0
+#define PIPE_WRITE 1
+
+int rrdpush_pipe[2];
+
+static BUFFER *rrdpush_buffer = NULL;
+static pthread_mutex_t rrdpush_mutex = PTHREAD_MUTEX_INITIALIZER;
+static RRDHOST *last_host = NULL;
+
+static inline void rrdpush_lock() {
+ pthread_mutex_lock(&rrdpush_mutex);
+}
+
+static inline void rrdpush_unlock() {
+ pthread_mutex_unlock(&rrdpush_mutex);
+}
+
+static inline int need_to_send_chart_definitions(RRDSET *st) {
+ RRDDIM *rd;
+ for(rd = st->dimensions; rd ;rd = rd->next)
+ if(rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED) && !rrddim_flag_check(rd, RRDDIM_FLAG_EXPOSED))
+ return 1;
+
+ return 0;
+}
+
+static inline void send_chart_definitions(RRDSET *st) {
+ buffer_sprintf(rrdpush_buffer, "CHART '%s' '%s' '%s' '%s' '%s' '%s' '%s' %ld %d\n"
+ , st->id
+ , st->name
+ , st->title
+ , st->units
+ , st->family
+ , st->context
+ , rrdset_type_name(st->chart_type)
+ , st->priority
+ , st->update_every
+ );
+
+ RRDDIM *rd;
+ for(rd = st->dimensions; rd ;rd = rd->next) {
+ buffer_sprintf(rrdpush_buffer, "DIMENSION '%s' '%s' '%s' " COLLECTED_NUMBER_FORMAT " " COLLECTED_NUMBER_FORMAT " '%s %s'\n"
+ , rd->id
+ , rd->name
+ , rrd_algorithm_name(rd->algorithm)
+ , rd->multiplier
+ , rd->divisor
+ , rrddim_flag_check(rd, RRDDIM_FLAG_HIDDEN)?"hidden":""
+ , rrddim_flag_check(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS)?"noreset":""
+ );
+ }
+}
+
+static inline void send_chart_metrics(RRDSET *st) {
+ buffer_sprintf(rrdpush_buffer, "BEGIN %s %llu\n", st->id, st->usec_since_last_update);
+
+ RRDDIM *rd;
+ for(rd = st->dimensions; rd ;rd = rd->next) {
+ if(rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED))
+ buffer_sprintf(rrdpush_buffer, "SET %s = " COLLECTED_NUMBER_FORMAT "\n"
+ , rd->id
+ , rd->collected_value
+ );
+ }
+
+ buffer_strcat(rrdpush_buffer, "END\n");
+}
+
+static void reset_all_charts(void) {
+ rrd_rdlock();
+
+ RRDHOST *h;
+ for(h = localhost; h ;h = h->next) {
+ RRDSET *st;
+ for(st = h->rrdset_root ; st ; st = st->next) {
+ rrdset_rdlock(st);
+
+ RRDDIM *rd;
+ for(rd = st->dimensions; rd ;rd = rd->next)
+ rrddim_flag_clear(rd, RRDDIM_FLAG_EXPOSED);
+
+ rrdset_unlock(st);
+ }
+ }
+
+ last_host = NULL;
+
+ rrd_unlock();
+}
+
+void rrdset_done_push(RRDSET *st) {
+
+ if(!rrdset_flag_check(st, RRDSET_FLAG_ENABLED))
+ return;
+
+ rrdpush_lock();
+ rrdset_rdlock(st);
+
+ if(st->rrdhost != last_host)
+ buffer_sprintf(rrdpush_buffer, "HOST '%s' '%s'\n", st->rrdhost->hostname, st->rrdhost->machine_guid);
+
+ if(need_to_send_chart_definitions(st))
+ send_chart_definitions(st);
+
+ send_chart_metrics(st);
+
+ // signal the sender there are more data
+ if(write(rrdpush_pipe[PIPE_WRITE], " ", 1) == -1)
+ error("Cannot write to internal pipe");
+
+ rrdset_unlock(st);
+ rrdpush_unlock();
+}
+
+void *central_netdata_push_thread(void *ptr) {
+ struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
+
+ info("Central netdata push thread created with task id %d", gettid());
+
+ if(pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL) != 0)
+ error("Cannot set pthread cancel type to DEFERRED.");
+
+ if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
+ error("Cannot set pthread cancel state to ENABLE.");
+
+
+ rrdpush_buffer = buffer_create(1);
+
+ if(pipe(rrdpush_pipe) == -1)
+ fatal("Cannot create required pipe.");
+
+ struct timeval tv = {
+ .tv_sec = 60,
+ .tv_usec = 0
+ };
+
+ size_t begin = 0;
+ size_t max_size = 1024 * 1024;
+ size_t reconnects_counter = 0;
+ int sock = -1;
+ char buffer[1];
+
+ for(;;) {
+ if(unlikely(sock == -1)) {
+ sock = connect_to_one_of(central_netdata_to_push_data, 19999, &tv, &reconnects_counter);
+
+ if(unlikely(sock != -1)) {
+ if(fcntl(sock, F_SETFL, O_NONBLOCK) < 0)
+ error("Cannot set non-blocking mode for socket.");
+
+ buffer_sprintf(rrdpush_buffer, "GET /stream?key=%s\r\n\r\n", config_get("global", "central netdata api key", ""));
+ reset_all_charts();
+ }
+ }
+
+ if(read(rrdpush_pipe[PIPE_READ], buffer, 1) == -1) {
+ error("Cannot read from internal pipe.");
+ sleep(1);
+ }
+
+ if(likely(sock != -1)) {
+ rrdpush_lock();
+ ssize_t ret = send(sock, &rrdpush_buffer->buffer[begin], rrdpush_buffer->len, MSG_DONTWAIT);
+ if(ret == -1) {
+ error("Failed to send metrics to central netdata at %s", central_netdata_to_push_data);
+ close(sock);
+ sock = -1;
+ }
+ else {
+ begin += ret;
+ if(begin == rrdpush_buffer->len) {
+ buffer_flush(rrdpush_buffer);
+ begin = 0;
+ }
+ }
+ rrdpush_unlock();
+ }
+
+ // protection from overflow
+ if(rrdpush_buffer->len > max_size) {
+ rrdpush_lock();
+
+ error("Discarding %zu bytes of metrics data, because we cannot connect to central netdata at %s"
+ , buffer_strlen(rrdpush_buffer), central_netdata_to_push_data);
+
+ buffer_flush(rrdpush_buffer);
+
+ if(sock != -1) {
+ close(sock);
+ sock = -1;
+ }
+
+ rrdpush_unlock();
+ }
+ }
+
+cleanup:
+ debug(D_WEB_CLIENT, "Central netdata push thread exits.");
+ if(sock != -1)
+ close(sock);
+
+ static_thread->enabled = 0;
+ pthread_exit(NULL);
+ return NULL;
+}
--- /dev/null
+#ifndef NETDATA_RRDPUSH_H
+#define NETDATA_RRDPUSH_H
+
+extern void rrdset_done_push(RRDSET *st);
+extern void *central_netdata_push_thread(void *ptr);
+
+#endif //NETDATA_RRDPUSH_H
// ----------------------------------------------------------------------------
// RRDSET - helpers for rrdset_create()
-static inline long align_entries_to_pagesize(long entries) {
+inline long align_entries_to_pagesize(long entries) {
+ if(central_netdata_to_push_data)
+ return entries;
+
if(entries < 5) entries = 5;
if(entries > RRD_HISTORY_ENTRIES_MAX) entries = RRD_HISTORY_ENTRIES_MAX;
-#ifdef NETDATA_LOG_ALLOCATIONS
long page = (size_t)sysconf(_SC_PAGESIZE);
-
long size = sizeof(RRDDIM) + entries * sizeof(storage_number);
if(size % page) {
size -= (size % page);
}
return entries;
-#else
- return entries;
-#endif
}
static inline void timeval_align(struct timeval *tv, int update_every) {
// ----------------------------------------------------------------------------
// RRDSET - process the collected values for all dimensions of a chart
-usec_t rrdset_done(RRDSET *st) {
- if(unlikely(netdata_exit)) return 0;
+void rrdset_done(RRDSET *st) {
+ if(unlikely(netdata_exit)) return;
+ if(unlikely(central_netdata_to_push_data)) {
+ rrdset_done_push(st);
+ return;
+ }
debug(D_RRD_CALLS, "rrdset_done() for chart %s", st->name);
if(unlikely(pthread_setcancelstate(pthreadoldcancelstate, NULL) != 0))
error("Cannot set pthread cancel state to RESTORE (%d).", pthreadoldcancelstate);
-
- return(st->usec_since_last_update);
}
fd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
if(fd != -1) {
- if(setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, (char *)timeout, sizeof(struct timeval)) < 0)
- error("Failed to set timeout on the socket to ip '%s' port '%s'", hostBfr, servBfr);
+ if(timeout) {
+ if(setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, (char *) timeout, sizeof(struct timeval)) < 0)
+ error("Failed to set timeout on the socket to ip '%s' port '%s'", hostBfr, servBfr);
+ }
if(connect(fd, ai->ai_addr, ai->ai_addrlen) < 0) {
error("Failed to connect to '%s', port '%s'", hostBfr, servBfr);
return fd;
}
+
+int connect_to_one_of(const char *destination, int default_port, struct timeval *timeout, size_t *reconnects_counter) {
+ int sock = -1;
+
+ const char *s = destination;
+ while(*s) {
+ const char *e = s;
+
+ // skip separators, moving both s(tart) and e(nd)
+ while(isspace(*e) || *e == ',') s = ++e;
+
+ // move e(nd) to the first separator
+ while(*e && !isspace(*e) && *e != ',') e++;
+
+ // is there anything?
+ if(!*s || s == e) break;
+
+ char buf[e - s + 1];
+ strncpyz(buf, s, e - s);
+ if(reconnects_counter) *reconnects_counter += 1;
+ sock = connect_to(buf, default_port, timeout);
+ if(sock != -1) break;
+ s = e;
+ }
+
+ return sock;
+}
#define NETDATA_SOCKET_H
extern int connect_to(const char *definition, int default_port, struct timeval *timeout);
+extern int connect_to_one_of(const char *destination, int default_port, struct timeval *timeout, size_t *reconnects_counter);
#endif //NETDATA_SOCKET_H
{
fprintf(stderr, "\nRunning test '%s':\n%s\n", test->name, test->description);
- default_localhost_rrd_memory_mode = RRD_MEMORY_MODE_RAM;
- default_localhost_rrd_update_every = test->update_every;
+ default_rrd_memory_mode = RRD_MEMORY_MODE_RAM;
+ default_rrd_update_every = test->update_every;
char name[101];
snprintfz(name, 100, "unittest-%s", test->name);
snprintfz(name, 100, "unittest-%d-%ld-%ld", repeat, delay, shift);
//debug_flags = 0xffffffff;
- default_localhost_rrd_memory_mode = RRD_MEMORY_MODE_RAM;
- default_localhost_rrd_update_every = 1;
+ default_rrd_memory_mode = RRD_MEMORY_MODE_RAM;
+ default_rrd_update_every = 1;
int do_abs = 1;
int do_inc = 1;
}
int web_client_stream_request(RRDHOST *host, struct web_client *w, char *url) {
- (void)host;
-
- info("STREAM request from client '%s:%s'", w->client_ip, w->client_port);
+ info("STREAM request from client '%s:%s' for host '%s'", w->client_ip, w->client_port, host->hostname);
char *key = NULL;
struct plugind cd = {
.enabled = 1,
- .update_every = default_localhost_rrd_update_every,
+ .update_every = default_rrd_update_every,
.pid = 0,
.serial_failures = 0,
.successful_collections = 0,
}
// call the plugins.d processor to receive the metrics
- size_t count = pluginsd_process(&cd, fp, 1);
+ size_t count = pluginsd_process(host, &cd, fp, 1);
error("STREAM from '%s:%s': client disconnected.", w->client_ip, w->client_port);
// close all sockets, to let the socket worker we are done
sidebar += '<li class="" style="padding-top:15px;"><a href="https://github.com/firehol/netdata/wiki/Add-more-charts-to-netdata" target="_blank"><i class="fa fa-plus" aria-hidden="true"></i> add more charts</a></li>';
sidebar += '<li class=""><a href="https://github.com/firehol/netdata/wiki/Add-more-alarms-to-netdata" target="_blank"><i class="fa fa-plus" aria-hidden="true"></i> add more alarms</a></li>';
- sidebar += '<li class="" style="margin:20px;color:#666;"><small>netdata on <b>' + data.hostname.toString() + '</b>, collects every ' + ((data.update_every == 1)?'second':data.update_every.toString() + ' seconds') + ' <b>' + data.dimensions_count.toLocaleString() + '</b> metrics, presented as <b>' + data.charts_count.toLocaleString() + '</b> charts and monitored by <b>' + data.alarms_count.toLocaleString() + '</b> alarms, using ' + Math.round(data.rrd_memory_bytes / 1024 / 1024).toLocaleString() + ' MB of memory for ' + Math.round(data.history / (3600/data.update_every)).toLocaleString() + ' ' + ((data.history == (3600/data.update_every))?'hour':'hours').toString() + ' of real-time history.<br/> <br/><b>netdata</b><br/>v' + data.version.toString() +'</small></li>';
+ sidebar += '<li class="" style="margin:20px;color:#666;"><small>netdata on <b>' + data.hostname.toString() + '</b>, collects every ' + ((data.update_every == 1)?'second':data.update_every.toString() + ' seconds') + ' <b>' + data.dimensions_count.toLocaleString() + '</b> metrics, presented as <b>' + data.charts_count.toLocaleString() + '</b> charts and monitored by <b>' + data.alarms_count.toLocaleString() + '</b> alarms, using ' + Math.round(data.rrd_memory_bytes / 1024 / 1024).toLocaleString() + ' MB of memory for ' + seconds4human(data.update_every * data.history) + ' of real-time history.<br/> <br/><b>netdata</b><br/>v' + data.version.toString() +'</small></li>';
sidebar += '</ul>';
div.innerHTML = html;
document.getElementById('sidebar').innerHTML = sidebar;
return t.toLocaleDateString() + space + t.toLocaleTimeString();
}
- function seconds4human(seconds, options) {
- var default_options = {
- now: 'now',
- space: ' ',
- negative_suffix: 'ago',
- hour: 'hour',
- hours: 'hours',
- minute: 'minute',
- minutes: 'minutes',
- second: 'second',
- seconds: 'seconds',
- and: 'and'
- };
-
- if(typeof options !== 'object')
- options = default_options;
- else {
- var x;
- for(x in default_options) {
- if(typeof options[x] !== 'string')
- options[x] = default_options[x];
- }
- }
-
- if(typeof seconds === 'string')
- seconds = parseInt(seconds);
-
- if(seconds === 0)
- return options.now;
-
- var suffix = '';
- if(seconds < 0) {
- seconds = -seconds;
- if(options.negative_suffix !== '') suffix = options.space + options.negative_suffix;
- }
-
- var hours = Math.floor(seconds / 3600);
- seconds -= (hours * 3600);
-
- var minutes = Math.floor(seconds / 60);
- seconds -= (minutes * 60);
-
- var txt = '';
-
- if(hours > 1) txt += hours.toString() + options.space + options.hours;
- else if(hours === 1) txt += hours.toString() + options.space + options.hour;
-
- if(hours > 0 && minutes > 0 && seconds == 0)
- txt += options.space + options.and + options.space;
- else if(hours > 0 && minutes > 0 && seconds > 0)
- txt += ',' + options.space;
-
- if(minutes > 1) txt += minutes.toString() + options.space + options.minutes;
- else if(minutes === 1) txt += minutes.toString() + options.space + options.minute;
-
- if((minutes > 0 || minutes > 0) && seconds > 0)
- txt += options.space + options.and + options.space;
-
- if(seconds > 1) txt += Math.floor(seconds).toString() + options.space + options.seconds;
- else if(seconds === 1) txt += Math.floor(seconds).toString() + options.space + options.second;
-
- return txt + suffix;
- }
-
function alarm_lookup_explain(alarm, chart) {
var dimensions = ' of all values ';
});
}
+ function seconds4human(seconds, options) {
+ var default_options = {
+ now: 'now',
+ space: ' ',
+ negative_suffix: 'ago',
+ hour: 'hour',
+ hours: 'hours',
+ minute: 'minute',
+ minutes: 'minutes',
+ second: 'second',
+ seconds: 'seconds',
+ and: 'and'
+ };
+
+ if(typeof options !== 'object')
+ options = default_options;
+ else {
+ var x;
+ for(x in default_options) {
+ if(typeof options[x] !== 'string')
+ options[x] = default_options[x];
+ }
+ }
+
+ if(typeof seconds === 'string')
+ seconds = parseInt(seconds);
+
+ if(seconds === 0)
+ return options.now;
+
+ var suffix = '';
+ if(seconds < 0) {
+ seconds = -seconds;
+ if(options.negative_suffix !== '') suffix = options.space + options.negative_suffix;
+ }
+
+ var hours = Math.floor(seconds / 3600);
+ seconds -= (hours * 3600);
+
+ var minutes = Math.floor(seconds / 60);
+ seconds -= (minutes * 60);
+
+ var txt = '';
+
+ if(hours > 1) txt += hours.toString() + options.space + options.hours;
+ else if(hours === 1) txt += hours.toString() + options.space + options.hour;
+
+ if(hours > 0 && minutes > 0 && seconds == 0)
+ txt += options.space + options.and + options.space;
+ else if(hours > 0 && minutes > 0 && seconds > 0)
+ txt += ',' + options.space;
+
+ if(minutes > 1) txt += minutes.toString() + options.space + options.minutes;
+ else if(minutes === 1) txt += minutes.toString() + options.space + options.minute;
+
+ if((minutes > 0 || minutes > 0) && seconds > 0)
+ txt += options.space + options.and + options.space;
+
+ if(seconds > 1) txt += Math.floor(seconds).toString() + options.space + options.seconds;
+ else if(seconds === 1) txt += Math.floor(seconds).toString() + options.space + options.second;
+
+ return txt + suffix;
+ }
+
function alarmsCallback(data) {
var count = 0;
for(x in data.alarms) {
projects / netdata.git / commitdiff