]> arthur.barton.de Git - netdata.git/commitdiff
added /stream/ URL that redirects socket to plugins.d for processing - api key valida...
authorCosta Tsaousis (ktsaou) <costa@tsaousis.gr>
Sun, 19 Feb 2017 19:07:17 +0000 (21:07 +0200)
committerCosta Tsaousis (ktsaou) <costa@tsaousis.gr>
Tue, 21 Feb 2017 23:00:25 +0000 (01:00 +0200)
src/plugins_d.c
src/plugins_d.h
src/web_client.c

index a4065089757158a5655ed83e46d56507cfa4caa7..0c73408b28904e62f1af75feb82365b4f83621ea 100644 (file)
@@ -83,17 +83,18 @@ 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) {
+    int enabled = cd->enabled;
 
-void *pluginsd_worker_thread(void *arg) {
-    struct plugind *cd = (struct plugind *)arg;
-    cd->obsolete = 0;
+    if(!fp || !enabled) {
+        cd->enabled = 0;
+        return 0;
+    }
 
-    char line[PLUGINSD_LINE_MAX + 1];
+    size_t count = 0;
+    RRDHOST *host = localhost;
 
-#ifdef DETACH_PLUGINS_FROM_NETDATA
-    usec_t usec = 0, susec = 0;
-    struct timeval last = {0, 0} , now = {0, 0};
-#endif
+    char line[PLUGINSD_LINE_MAX + 1];
 
     char *words[MAX_WORDS] = { NULL };
     uint32_t HOST_HASH = simple_hash("HOST");
@@ -103,305 +104,277 @@ void *pluginsd_worker_thread(void *arg) {
     uint32_t CHART_HASH = simple_hash("CHART");
     uint32_t DIMENSION_HASH = simple_hash("DIMENSION");
     uint32_t DISABLE_HASH = simple_hash("DISABLE");
-#ifdef DETACH_PLUGINS_FROM_NETDATA
-    uint32_t MYPID_HASH = simple_hash("MYPID");
-    uint32_t STOPPING_WAKE_ME_UP_PLEASE_HASH = simple_hash("STOPPING_WAKE_ME_UP_PLEASE");
-#endif
 
-    size_t count = 0;
-    RRDHOST *host = localhost;
+    RRDSET *st = NULL;
+    uint32_t hash;
 
-    for(;;) {
+    while(likely(fgets(line, PLUGINSD_LINE_MAX, fp) != NULL)) {
         if(unlikely(netdata_exit)) break;
 
-        FILE *fp = mypopen(cd->cmd, &cd->pid);
-        if(unlikely(!fp)) {
-            error("Cannot popen(\"%s\", \"r\").", cd->cmd);
-            break;
-        }
-
-        info("PLUGINSD: '%s' running on pid %d", cd->fullfilename, cd->pid);
+        line[PLUGINSD_LINE_MAX] = '\0';
 
-        RRDSET *st = NULL;
-        uint32_t hash;
+        // debug(D_PLUGINSD, "PLUGINSD: %s: %s", cd->filename, line);
 
-        while(likely(fgets(line, PLUGINSD_LINE_MAX, fp) != NULL)) {
-            if(unlikely(netdata_exit)) break;
+        int w = pluginsd_split_words(line, words, MAX_WORDS);
+        char *s = words[0];
+        if(unlikely(!s || !*s || !w)) {
+            // debug(D_PLUGINSD, "PLUGINSD: empty line");
+            continue;
+        }
 
-            line[PLUGINSD_LINE_MAX] = '\0';
+        // debug(D_PLUGINSD, "PLUGINSD: words 0='%s' 1='%s' 2='%s' 3='%s' 4='%s' 5='%s' 6='%s' 7='%s' 8='%s' 9='%s'", words[0], words[1], words[2], words[3], words[4], words[5], words[6], words[7], words[8], words[9]);
 
-            // debug(D_PLUGINSD, "PLUGINSD: %s: %s", cd->filename, line);
+        if(likely(!simple_hash_strcmp(s, "SET", &hash))) {
+            char *dimension = words[1];
+            char *value = words[2];
 
-            int w = pluginsd_split_words(line, words, MAX_WORDS);
-            char *s = words[0];
-            if(unlikely(!s || !*s || !w)) {
-                // debug(D_PLUGINSD, "PLUGINSD: empty line");
-                continue;
+            if(unlikely(!dimension || !*dimension)) {
+                error("PLUGINSD: '%s' is requesting a SET on chart '%s', without a dimension. Disabling it.", cd->fullfilename, st->id);
+                enabled = 0;
+                break;
             }
 
-            // debug(D_PLUGINSD, "PLUGINSD: words 0='%s' 1='%s' 2='%s' 3='%s' 4='%s' 5='%s' 6='%s' 7='%s' 8='%s' 9='%s'", words[0], words[1], words[2], words[3], words[4], words[5], words[6], words[7], words[8], words[9]);
-
-            if(likely(!simple_hash_strcmp(s, "SET", &hash))) {
-                char *dimension = words[1];
-                char *value = words[2];
-
-                if(unlikely(!dimension || !*dimension)) {
-                    error("PLUGINSD: '%s' is requesting a SET on chart '%s', without a dimension. Disabling it.", cd->fullfilename, st->id);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
+            if(unlikely(!value || !*value)) value = NULL;
 
-                if(unlikely(!value || !*value)) value = NULL;
+            if(unlikely(!st)) {
+                error("PLUGINSD: '%s' is requesting a SET on dimension %s with value %s, without a BEGIN. Disabling it.", cd->fullfilename, dimension, value?value:"<nothing>");
+                enabled = 0;
+                break;
+            }
 
-                if(unlikely(!st)) {
-                    error("PLUGINSD: '%s' is requesting a SET on dimension %s with value %s, without a BEGIN. Disabling it.", cd->fullfilename, dimension, value?value:"<nothing>");
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
+            if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) debug(D_PLUGINSD, "PLUGINSD: '%s' is setting dimension %s/%s to %s", cd->fullfilename, st->id, dimension, value?value:"<nothing>");
 
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) debug(D_PLUGINSD, "PLUGINSD: '%s' is setting dimension %s/%s to %s", cd->fullfilename, st->id, dimension, value?value:"<nothing>");
+            if(value) rrddim_set(st, dimension, strtoll(value, NULL, 0));
+        }
+        else if(likely(hash == BEGIN_HASH && !strcmp(s, "BEGIN"))) {
+            char *id = words[1];
+            char *microseconds_txt = words[2];
 
-                if(value) rrddim_set(st, dimension, strtoll(value, NULL, 0));
+            if(unlikely(!id)) {
+                error("PLUGINSD: '%s' is requesting a BEGIN without a chart id. Disabling it.", cd->fullfilename);
+                enabled = 0;
+                break;
             }
-            else if(likely(hash == BEGIN_HASH && !strcmp(s, "BEGIN"))) {
-                char *id = words[1];
-                char *microseconds_txt = words[2];
 
-                if(unlikely(!id)) {
-                    error("PLUGINSD: '%s' is requesting a BEGIN without a chart id. Disabling it.", cd->fullfilename);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
+            st = rrdset_find(host, id);
+            if(unlikely(!st)) {
+                error("PLUGINSD: '%s' is requesting a BEGIN on chart '%s', which does not exist. Disabling it.", cd->fullfilename, id);
+                enabled = 0;
+                break;
+            }
 
-                st = rrdset_find(host, id);
-                if(unlikely(!st)) {
-                    error("PLUGINSD: '%s' is requesting a BEGIN on chart '%s', which does not exist. Disabling it.", cd->fullfilename, id);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
+            if(likely(st->counter_done)) {
+                usec_t microseconds = 0;
+                if(microseconds_txt && *microseconds_txt) microseconds = str2ull(microseconds_txt);
 
-                if(likely(st->counter_done)) {
-                    usec_t microseconds = 0;
-                    if(microseconds_txt && *microseconds_txt) microseconds = str2ull(microseconds_txt);
-                    if(microseconds) rrdset_next_usec(st, microseconds);
-                    else rrdset_next(st);
+                if(likely(microseconds)) {
+                    if(trust_durations)
+                        rrdset_next_usec_unfiltered(st, microseconds);
+                    else
+                        rrdset_next_usec(st, microseconds);
                 }
+                else rrdset_next(st);
+            }
+        }
+        else if(likely(hash == END_HASH && !strcmp(s, "END"))) {
+            if(unlikely(!st)) {
+                error("PLUGINSD: '%s' is requesting an END, without a BEGIN. Disabling it.", cd->fullfilename);
+                enabled = 0;
+                break;
             }
-            else if(likely(hash == END_HASH && !strcmp(s, "END"))) {
-                if(unlikely(!st)) {
-                    error("PLUGINSD: '%s' is requesting an END, without a BEGIN. Disabling it.", cd->fullfilename);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
 
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) debug(D_PLUGINSD, "PLUGINSD: '%s' is requesting an END on chart %s", cd->fullfilename, st->id);
+            if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) debug(D_PLUGINSD, "PLUGINSD: '%s' is requesting an END on chart %s", cd->fullfilename, st->id);
 
-                rrdset_done(st);
-                st = NULL;
+            rrdset_done(st);
+            st = NULL;
 
-                count++;
-            }
-            else if(likely(hash == HOST_HASH && !strcmp(s, "HOST"))) {
-                char *guid = words[1];
-                char *hostname = words[2];
-
-                if(unlikely(!guid || !*guid)) {
-                    error("PLUGINSD: '%s' is requesting a HOST, without a guid. Disabling it.", cd->fullfilename);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
-                if(unlikely(!hostname || !*hostname)) {
-                    error("PLUGINSD: '%s' is requesting a HOST, without a hostname. Disabling it.", cd->fullfilename);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
+            count++;
+        }
+        else if(likely(hash == HOST_HASH && !strcmp(s, "HOST"))) {
+            char *guid = words[1];
+            char *hostname = words[2];
 
-                host = rrdhost_find_or_create(hostname, guid);
+            if(unlikely(!guid || !*guid)) {
+                error("PLUGINSD: '%s' is requesting a HOST, without a guid. Disabling it.", cd->fullfilename);
+                enabled = 0;
+                break;
             }
-            else if(likely(hash == FLUSH_HASH && !strcmp(s, "FLUSH"))) {
-                debug(D_PLUGINSD, "PLUGINSD: '%s' is requesting a FLUSH", cd->fullfilename);
-                st = NULL;
+            if(unlikely(!hostname || !*hostname)) {
+                error("PLUGINSD: '%s' is requesting a HOST, without a hostname. Disabling it.", cd->fullfilename);
+                enabled = 0;
+                break;
             }
-            else if(likely(hash == CHART_HASH && !strcmp(s, "CHART"))) {
-                int noname = 0;
-                st = NULL;
-
-                if((words[1]) != NULL && (words[2]) != NULL && strcmp(words[1], words[2]) == 0)
-                    noname = 1;
-
-                char *type = words[1];
-                char *id = NULL;
-                if(likely(type)) {
-                    id = strchr(type, '.');
-                    if(likely(id)) { *id = '\0'; id++; }
-                }
-                char *name = words[2];
-                char *title = words[3];
-                char *units = words[4];
-                char *family = words[5];
-                char *context = words[6];
-                char *chart = words[7];
-                char *priority_s = words[8];
-                char *update_every_s = words[9];
-
-                if(unlikely(!type || !*type || !id || !*id)) {
-                    error("PLUGINSD: '%s' is requesting a CHART, without a type.id. Disabling it.", cd->fullfilename);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
 
-                int priority = 1000;
-                if(likely(priority_s)) priority = str2i(priority_s);
-
-                int update_every = cd->update_every;
-                if(likely(update_every_s)) update_every = str2i(update_every_s);
-                if(unlikely(!update_every)) update_every = cd->update_every;
-
-                RRDSET_TYPE chart_type = RRDSET_TYPE_LINE;
-                if(unlikely(chart)) chart_type = rrdset_type_id(chart);
-
-                if(unlikely(noname || !name || !*name || strcasecmp(name, "NULL") == 0 || strcasecmp(name, "(NULL)") == 0)) name = NULL;
-                if(unlikely(!family || !*family)) family = NULL;
-                if(unlikely(!context || !*context)) context = NULL;
-
-                st = rrdset_find_bytype(host, type, id);
-                if(unlikely(!st)) {
-                    debug(D_PLUGINSD, "PLUGINSD: Creating chart type='%s', id='%s', name='%s', family='%s', context='%s', chart='%s', priority=%d, update_every=%d"
-                        , type, id
-                        , name?name:""
-                        , family?family:""
-                        , context?context:""
-                        , rrdset_type_name(chart_type)
-                        , priority
-                        , update_every
-                        );
-
-                    st = rrdset_create(host, type, id, name, family, context, title, units, priority, update_every
-                                                 , chart_type);
-                    cd->update_every = update_every;
-                }
-                else debug(D_PLUGINSD, "PLUGINSD: Chart '%s' already exists. Not adding it again.", st->id);
+            host = rrdhost_find_or_create(hostname, guid);
+        }
+        else if(likely(hash == FLUSH_HASH && !strcmp(s, "FLUSH"))) {
+            debug(D_PLUGINSD, "PLUGINSD: '%s' is requesting a FLUSH", cd->fullfilename);
+            st = NULL;
+        }
+        else if(likely(hash == CHART_HASH && !strcmp(s, "CHART"))) {
+            int noname = 0;
+            st = NULL;
+
+            if((words[1]) != NULL && (words[2]) != NULL && strcmp(words[1], words[2]) == 0)
+                noname = 1;
+
+            char *type = words[1];
+            char *id = NULL;
+            if(likely(type)) {
+                id = strchr(type, '.');
+                if(likely(id)) { *id = '\0'; id++; }
+            }
+            char *name = words[2];
+            char *title = words[3];
+            char *units = words[4];
+            char *family = words[5];
+            char *context = words[6];
+            char *chart = words[7];
+            char *priority_s = words[8];
+            char *update_every_s = words[9];
+
+            if(unlikely(!type || !*type || !id || !*id)) {
+                error("PLUGINSD: '%s' is requesting a CHART, without a type.id. Disabling it.", cd->fullfilename);
+                enabled = 0;
+                break;
             }
-            else if(likely(hash == DIMENSION_HASH && !strcmp(s, "DIMENSION"))) {
-                char *id = words[1];
-                char *name = words[2];
-                char *algorithm = words[3];
-                char *multiplier_s = words[4];
-                char *divisor_s = words[5];
-                char *options = words[6];
-
-                if(unlikely(!id || !*id)) {
-                    error("PLUGINSD: '%s' is requesting a DIMENSION, without an id. Disabling it.", cd->fullfilename);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
-
-                if(unlikely(!st)) {
-                    error("PLUGINSD: '%s' is requesting a DIMENSION, without a CHART. Disabling it.", cd->fullfilename);
-                    cd->enabled = 0;
-                    killpid(cd->pid, SIGTERM);
-                    break;
-                }
 
-                long multiplier = 1;
-                if(multiplier_s && *multiplier_s) multiplier = strtol(multiplier_s, NULL, 0);
-                if(unlikely(!multiplier)) multiplier = 1;
-
-                long divisor = 1;
-                if(likely(divisor_s && *divisor_s)) divisor = strtol(divisor_s, NULL, 0);
-                if(unlikely(!divisor)) divisor = 1;
-
-                if(unlikely(!algorithm || !*algorithm)) algorithm = "absolute";
-
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_PLUGINSD, "PLUGINSD: Creating dimension in chart %s, id='%s', name='%s', algorithm='%s', multiplier=%ld, divisor=%ld, hidden='%s'"
-                        , st->id
-                        , id
-                        , name?name:""
-                        , rrd_algorithm_name(rrd_algorithm_id(algorithm))
-                        , multiplier
-                        , divisor
-                        , options?options:""
-                        );
-
-                RRDDIM *rd = rrddim_find(st, id);
-                if(unlikely(!rd)) {
-                    rd = rrddim_add(st, id, name, multiplier, divisor, rrd_algorithm_id(algorithm));
-                    rd->flags = 0x00000000;
-                    if(options && *options) {
-                        if(strstr(options, "hidden") != NULL) rrddim_flag_set(rd, RRDDIM_FLAG_HIDDEN);
-                        if(strstr(options, "noreset") != NULL) rrddim_flag_set(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS);
-                        if(strstr(options, "nooverflow") != NULL) rrddim_flag_set(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS);
-                    }
-                }
-                else if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_PLUGINSD, "PLUGINSD: dimension %s/%s already exists. Not adding it again.", st->id, id);
+            int priority = 1000;
+            if(likely(priority_s)) priority = str2i(priority_s);
+
+            int update_every = cd->update_every;
+            if(likely(update_every_s)) update_every = str2i(update_every_s);
+            if(unlikely(!update_every)) update_every = cd->update_every;
+
+            RRDSET_TYPE chart_type = RRDSET_TYPE_LINE;
+            if(unlikely(chart)) chart_type = rrdset_type_id(chart);
+
+            if(unlikely(noname || !name || !*name || strcasecmp(name, "NULL") == 0 || strcasecmp(name, "(NULL)") == 0)) name = NULL;
+            if(unlikely(!family || !*family)) family = NULL;
+            if(unlikely(!context || !*context)) context = NULL;
+
+            st = rrdset_find_bytype(host, type, id);
+            if(unlikely(!st)) {
+                debug(D_PLUGINSD, "PLUGINSD: Creating chart type='%s', id='%s', name='%s', family='%s', context='%s', chart='%s', priority=%d, update_every=%d"
+                      , type, id
+                      , name?name:""
+                      , family?family:""
+                      , context?context:""
+                      , rrdset_type_name(chart_type)
+                      , priority
+                      , update_every
+                );
+
+                st = rrdset_create(host, type, id, name, family, context, title, units, priority, update_every, chart_type);
+                cd->update_every = update_every;
             }
-            else if(unlikely(hash == DISABLE_HASH && !strcmp(s, "DISABLE"))) {
-                info("PLUGINSD: '%s' called DISABLE. Disabling it.", cd->fullfilename);
-                cd->enabled = 0;
-                killpid(cd->pid, SIGTERM);
+            else debug(D_PLUGINSD, "PLUGINSD: Chart '%s' already exists. Not adding it again.", st->id);
+        }
+        else if(likely(hash == DIMENSION_HASH && !strcmp(s, "DIMENSION"))) {
+            char *id = words[1];
+            char *name = words[2];
+            char *algorithm = words[3];
+            char *multiplier_s = words[4];
+            char *divisor_s = words[5];
+            char *options = words[6];
+
+            if(unlikely(!id || !*id)) {
+                error("PLUGINSD: '%s' is requesting a DIMENSION, without an id. Disabling it.", cd->fullfilename);
+                enabled = 0;
                 break;
             }
-#ifdef DETACH_PLUGINS_FROM_NETDATA
-            else if(likely(hash == MYPID_HASH && !strcmp(s, "MYPID"))) {
-                char *pid_s = words[1];
-                pid_t pid = strtod(pid_s, NULL, 0);
 
-                if(likely(pid)) cd->pid = pid;
-                debug(D_PLUGINSD, "PLUGINSD: %s is on pid %d", cd->id, cd->pid);
+            if(unlikely(!st)) {
+                error("PLUGINSD: '%s' is requesting a DIMENSION, without a CHART. Disabling it.", cd->fullfilename);
+                enabled = 0;
+                break;
             }
-            else if(likely(hash == STOPPING_WAKE_ME_UP_PLEASE_HASH && !strcmp(s, "STOPPING_WAKE_ME_UP_PLEASE"))) {
-                error("PLUGINSD: '%s' (pid %d) called STOPPING_WAKE_ME_UP_PLEASE.", cd->fullfilename, cd->pid);
 
-                now_realtime_timeval(&now);
-                if(unlikely(!usec && !susec)) {
-                    // our first run
-                    susec = cd->rrd_update_every * USEC_PER_SEC;
-                }
-                else {
-                    // second+ run
-                    usec = dt_usec(&now, &last) - susec;
-                    error("PLUGINSD: %s last loop took %llu usec (worked for %llu, sleeped for %llu).\n", cd->fullfilename, usec + susec, usec, susec);
-                    if(unlikely(usec < (localhost->rrd_update_every * USEC_PER_SEC / 2ULL))) susec = (localhost->rrd_update_every * USEC_PER_SEC) - usec;
-                    else susec = localhost->rrd_update_every * USEC_PER_SEC / 2ULL;
+            long multiplier = 1;
+            if(multiplier_s && *multiplier_s) multiplier = strtol(multiplier_s, NULL, 0);
+            if(unlikely(!multiplier)) multiplier = 1;
+
+            long divisor = 1;
+            if(likely(divisor_s && *divisor_s)) divisor = strtol(divisor_s, NULL, 0);
+            if(unlikely(!divisor)) divisor = 1;
+
+            if(unlikely(!algorithm || !*algorithm)) algorithm = "absolute";
+
+            if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
+                debug(D_PLUGINSD, "PLUGINSD: Creating dimension in chart %s, id='%s', name='%s', algorithm='%s', multiplier=%ld, divisor=%ld, hidden='%s'"
+                      , st->id
+                      , id
+                      , name?name:""
+                      , rrd_algorithm_name(rrd_algorithm_id(algorithm))
+                      , multiplier
+                      , divisor
+                      , options?options:""
+                );
+
+            RRDDIM *rd = rrddim_find(st, id);
+            if(unlikely(!rd)) {
+                rd = rrddim_add(st, id, name, multiplier, divisor, rrd_algorithm_id(algorithm));
+                rd->flags = 0x00000000;
+                if(options && *options) {
+                    if(strstr(options, "hidden") != NULL) rrddim_flag_set(rd, RRDDIM_FLAG_HIDDEN);
+                    if(strstr(options, "noreset") != NULL) rrddim_flag_set(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS);
+                    if(strstr(options, "nooverflow") != NULL) rrddim_flag_set(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS);
                 }
-
-                error("PLUGINSD: %s sleeping for %llu. Will kill with SIGCONT pid %d to wake it up.\n", cd->fullfilename, susec, cd->pid);
-                usleep(susec);
-                killpid(cd->pid, SIGCONT);
-                memmove(&last, &now, sizeof(struct timeval));
-                break;
-            }
-#endif
-            else {
-                error("PLUGINSD: '%s' is sending command '%s' which is not known by netdata. Disabling it.", cd->fullfilename, s);
-                cd->enabled = 0;
-                killpid(cd->pid, SIGTERM);
-                break;
             }
+            else if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
+                debug(D_PLUGINSD, "PLUGINSD: dimension %s/%s already exists. Not adding it again.", st->id, id);
+        }
+        else if(unlikely(hash == DISABLE_HASH && !strcmp(s, "DISABLE"))) {
+            info("PLUGINSD: '%s' called DISABLE. Disabling it.", cd->fullfilename);
+            enabled = 0;
+            break;
+        }
+        else {
+            error("PLUGINSD: '%s' is sending command '%s' which is not known by netdata. Disabling it.", cd->fullfilename, s);
+            enabled = 0;
+            break;
         }
-        if(likely(count)) {
-            cd->successful_collections += count;
-            cd->serial_failures = 0;
+    }
+
+    cd->enabled = enabled;
+
+    if(likely(count)) {
+        cd->successful_collections += count;
+        cd->serial_failures = 0;
+    }
+    else
+        cd->serial_failures++;
+
+    return count;
+}
+
+void *pluginsd_worker_thread(void *arg) {
+    struct plugind *cd = (struct plugind *)arg;
+    cd->obsolete = 0;
+
+    size_t count = 0;
+
+    for(;;) {
+        if(unlikely(netdata_exit)) break;
+
+        FILE *fp = mypopen(cd->cmd, &cd->pid);
+        if(unlikely(!fp)) {
+            error("Cannot popen(\"%s\", \"r\").", cd->cmd);
+            break;
         }
-        else
-            cd->serial_failures++;
+
+        info("PLUGINSD: '%s' running on pid %d", cd->fullfilename, cd->pid);
+
+        count = pluginsd_process(cd, fp, 0);
+        error("PLUGINSD: plugin '%s' disconnected.", cd->fullfilename);
+
+        killpid(cd->pid, SIGTERM);
 
         info("PLUGINSD: '%s' on pid %d stopped after %zu successful data collections (ENDs).", cd->fullfilename, cd->pid, count);
 
         // get the return code
         int code = mypclose(fp, cd->pid);
-        
+
         if(unlikely(netdata_exit)) break;
         else if(code != 0) {
             // the plugin reports failure
index 91c5c14b5d3fd2e8f6311cbbc819d68c5ce2d0f4..e6ad8c1e895217e2dd932b23e0f8bd09a5aab030 100644 (file)
@@ -34,5 +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);
 
 #endif /* NETDATA_PLUGINS_D_H */
index 1b89d69a43589f3da68c0fdbf203707bafc030e0..db9eeee1af12c9621bc1051cfaf264901207ff79 100644 (file)
@@ -1667,6 +1667,92 @@ int web_client_api_old_data_request(RRDHOST *host, struct web_client *w, char *u
     return 200;
 }
 
+int validate_stream_api_key(const char *key) {
+    return 0;
+}
+
+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);
+
+    char *key = NULL;
+
+    while(url) {
+        char *value = mystrsep(&url, "?&");
+        if(!value || !*value) continue;
+
+        char *name = mystrsep(&value, "=");
+        if(!name || !*name) continue;
+        if(!value || !*value) continue;
+
+        if(!strcmp(name, "key"))
+            key = value;
+    }
+
+    if(!key || !*key) {
+        buffer_flush(w->response.data);
+        buffer_sprintf(w->response.data, "You need an API key for this request.");
+        error("STREAM request from client '%s:%s', without an API key. Forbidding access.", w->client_ip, w->client_port);
+        return 401;
+    }
+
+    if(!validate_stream_api_key(key)) {
+        buffer_flush(w->response.data);
+        buffer_sprintf(w->response.data, "Your API key is not permitted access.");
+        error("STREAM request from client '%s:%s': API key '%s' is not allowed. Forbidding access.", w->client_ip, w->client_port, key);
+        return 401;
+    }
+
+    struct plugind cd = {
+            .enabled = 1,
+            .update_every = default_localhost_rrd_update_every,
+            .pid = 0,
+            .serial_failures = 0,
+            .successful_collections = 0,
+            .obsolete = 0,
+            .started_t = now_realtime_sec(),
+            .next = NULL,
+    };
+
+    // put the client IP and port into the buffers used by plugins.d
+    snprintfz(cd.id,           CONFIG_MAX_NAME,  "%s:%s", w->client_ip, w->client_port);
+    snprintfz(cd.filename,     FILENAME_MAX,     "%s:%s", w->client_ip, w->client_port);
+    snprintfz(cd.fullfilename, FILENAME_MAX,     "%s:%s", w->client_ip, w->client_port);
+    snprintfz(cd.cmd,          PLUGINSD_CMD_MAX, "%s:%s", w->client_ip, w->client_port);
+
+    // remove the non-blocking flag from the socket
+    if(fcntl(w->ifd, F_SETFL, fcntl(w->ifd, F_GETFL, 0) & ~O_NONBLOCK) == -1)
+        error("STREAM from '%s:%s': cannot remove the non-blocking flag from socket %d", w->client_ip, w->client_port, w->ifd);
+
+    // convert the socket to a FILE *
+    FILE *fp = fdopen(w->ifd, "r");
+    if(!fp) {
+        error("STREAM from '%s:%s': failed to get a FILE for FD %d.", w->client_ip, w->client_port, w->ifd);
+        buffer_flush(w->response.data);
+        buffer_sprintf(w->response.data, "Failed to get a FILE for an FD.");
+        return 500;
+    }
+
+    // call the plugins.d processor to receive the metrics
+    size_t count = pluginsd_process(&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
+    fclose(fp);
+    w->ifd = -1;
+    if(w->ofd != -1 && w->ofd != w->ifd) {
+        close(w->ofd);
+        w->ofd = -1;
+    }
+
+    // this will not send anything
+    // the socket is closed
+    buffer_flush(w->response.data);
+    if(count) return 200;
+    return 400;
+}
+
 const char *web_content_type_to_string(uint8_t contenttype) {
     switch(contenttype) {
         case CT_TEXT_HTML:
@@ -2113,7 +2199,7 @@ static inline int web_client_switch_host(RRDHOST *host, struct web_client *w, ch
     }
 
     buffer_flush(w->response.data);
-    buffer_strcat(w->response.data, "Host is not found: ");
+    buffer_strcat(w->response.data, "This netdata does not maintain a database for host: ");
     buffer_strcat_htmlescape(w->response.data, tok?tok:"");
     return 404;
 }
@@ -2127,7 +2213,8 @@ static inline int web_client_process_url(RRDHOST *host, struct web_client *w, ch
             hash_graph = 0,
             hash_list = 0,
             hash_all_json = 0,
-            hash_host = 0;
+            hash_host = 0,
+            hash_stream = 0;
 
 #ifdef NETDATA_INTERNAL_CHECKS
     static uint32_t hash_exit = 0, hash_debug = 0, hash_mirror = 0;
@@ -2142,6 +2229,7 @@ static inline int web_client_process_url(RRDHOST *host, struct web_client *w, ch
         hash_list = simple_hash("list");
         hash_all_json = simple_hash("all.json");
         hash_host = simple_hash("host");
+        hash_stream = simple_hash("stream");
 #ifdef NETDATA_INTERNAL_CHECKS
         hash_exit = simple_hash("exit");
         hash_debug = simple_hash("debug");
@@ -2154,14 +2242,18 @@ static inline int web_client_process_url(RRDHOST *host, struct web_client *w, ch
         uint32_t hash = simple_hash(tok);
         debug(D_WEB_CLIENT, "%llu: Processing command '%s'.", w->id, tok);
 
-        if(unlikely(hash == hash_host && strcmp(tok, "host") == 0)) {
-            debug(D_WEB_CLIENT_ACCESS, "%llu: host switch request ...", w->id);
-            return web_client_switch_host(host, w, url);
-        }
         if(unlikely(hash == hash_api && strcmp(tok, "api") == 0)) {
             debug(D_WEB_CLIENT_ACCESS, "%llu: API request ...", w->id);
             return web_client_api_request(host, w, url);
         }
+        else if(unlikely(hash == hash_host && strcmp(tok, "host") == 0)) {
+            debug(D_WEB_CLIENT_ACCESS, "%llu: host switch request ...", w->id);
+            return web_client_switch_host(host, w, url);
+        }
+        else if(unlikely(hash == hash_stream && strcmp(tok, "stream") == 0)) {
+            debug(D_WEB_CLIENT_ACCESS, "%llu: stream request ...", w->id);
+            return web_client_stream_request(host, w, url);
+        }
         else if(unlikely(hash == hash_netdata_conf && strcmp(tok, "netdata.conf") == 0)) {
             debug(D_WEB_CLIENT_ACCESS, "%llu: Sending netdata.conf ...", w->id);
             w->response.data->contenttype = CT_TEXT_PLAIN;
@@ -2709,7 +2801,8 @@ void *web_client_main(void *ptr)
 
     struct web_client *w = ptr;
     struct pollfd fds[2], *ifd, *ofd;
-    int retval, fdmax = 0, timeout;
+    int retval, timeout;
+    nfds_t fdmax = 0;
 
     log_access("%llu: %s port %s connected on thread task id %d", w->id, w->client_ip, w->client_port, gettid());
 
@@ -2798,6 +2891,10 @@ void *web_client_main(void *ptr)
             if(w->mode == WEB_CLIENT_MODE_NORMAL) {
                 debug(D_WEB_CLIENT, "%llu: Attempting to process received data.", w->id);
                 web_client_process_request(w);
+
+                // if the sockets are closed, may have transferred this client
+                // to plugins.d
+                if(w->ifd == -1 && w->ofd == -1) break;
             }
         }