draft implementation of netdata central push server - untested

CMakeLists.txt

@@ -124,7 +124,7 @@ set(NETDATA_SOURCE_FILES
         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

src/Makefile.am

@@ -75,6 +75,7 @@ netdata_SOURCES = \
 	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 \

src/backends.c

@@ -128,11 +128,13 @@ static inline int process_opentsdb_response(BUFFER *b) {
 }
 
 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());
 
@@ -145,12 +147,15 @@ void *backends_main(void *ptr) {
     // ------------------------------------------------------------------------
     // 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");
@@ -398,26 +403,11 @@ void *backends_main(void *ptr) {
 
         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;
         }
 

src/common.h

@@ -199,12 +199,12 @@
 #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"
@@ -216,6 +216,7 @@
 #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;
@@ -225,6 +226,7 @@ extern char *netdata_configured_cache_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);

src/health.c

@@ -1,7 +1,7 @@
 #define NETDATA_HEALTH_INTERNALS
 #include "common.h"
 
-int default_localhost_health_enabled = 1;
+int default_health_enabled = 1;
 
 // ----------------------------------------------------------------------------
 // health initialization
@@ -15,9 +15,16 @@ inline char *health_config_dir(void) {
 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];

src/health.h

@@ -1,7 +1,7 @@
 #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);
 

src/main.c

@@ -2,6 +2,8 @@
 
 extern void *cgroups_main(void *ptr);
 
+char *central_netdata_to_push_data = NULL;
+
 void netdata_cleanup_and_exit(int ret) {
     netdata_exit = 1;
 
@@ -55,21 +57,39 @@ struct netdata_static_thread static_threads[] = {
     {"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);
@@ -443,7 +463,7 @@ int main(int argc, char **argv) {
                         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");
@@ -528,7 +548,7 @@ int main(int argc, char **argv) {
             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
@@ -578,7 +598,11 @@ int main(int argc, char **argv) {
     }
 
     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);
 
@@ -595,21 +619,9 @@ int main(int argc, char **argv) {
 #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];
@@ -624,11 +636,12 @@ int main(int argc, char **argv) {
         }
 
         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";
@@ -637,49 +650,102 @@ int main(int argc, char **argv) {
         }
         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.
@@ -725,7 +791,9 @@ int main(int argc, char **argv) {
         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)
@@ -739,18 +807,24 @@ int main(int argc, char **argv) {
 
         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:"");
 
@@ -758,9 +832,11 @@ int main(int argc, char **argv) {
         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();
     }
 
@@ -778,14 +854,16 @@ int main(int argc, char **argv) {
     }
 #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);
@@ -795,29 +873,35 @@ int main(int argc, char **argv) {
             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
 
@@ -842,6 +926,7 @@ int main(int argc, char **argv) {
 
     info("netdata initialization completed. Enjoy real-time performance monitoring!");
 
+
     // ------------------------------------------------------------------------
     // block signals while initializing threads.
     sigset_t sigset;

src/plugin_nfacct.c

@@ -128,8 +128,8 @@ void *nfacct_main(void *ptr) {
         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;
 
 
         // --------------------------------------------------------------------
@@ -139,17 +139,17 @@ void *nfacct_main(void *ptr) {
 
             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);
             }
 
@@ -159,17 +159,17 @@ void *nfacct_main(void *ptr) {
 
             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);
             }
 

src/plugins_d.c

@@ -83,7 +83,7 @@ static int pluginsd_split_words(char *str, char **words, int max_words) {
     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) {
@@ -92,7 +92,6 @@ inline size_t pluginsd_process(struct plugind *cd, FILE *fp, int trust_durations
     }
 
     size_t count = 0;
-    RRDHOST *host = localhost;
 
     char line[PLUGINSD_LINE_MAX + 1];
 
@@ -365,7 +364,7 @@ void *pluginsd_worker_thread(void *arg) {
 
         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);

src/plugins_d.h

@@ -34,6 +34,6 @@ struct plugind {
 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 */