Dyncfg add streaming support (#15791)

* dyncfg fncnames as constants

* add helper macros to know parser streaming/plugin

* plugins dictionary per RRDHOST

* api_request_v2_config add support for /host/

* streamify pluginsd_register_plugin

* streamify pluginsd_register_module

* streamify report_job_status

* streamify dyncfg get functions

* module_type2str

* add job type and flags

* add DYNCFG_REGISTER_JOB

* implement register job

* push all to parent at startup

* add helper function is_dyncfg_function

* forward virtual functions trough streaming

* separate job2json

* add api/v2/job_statuses

* do cleanup on streaming

* streamify set functions

* support FUNCTION_PAYLOAD trough streaming

* WIP tests

* dont attempt loading non-localhost configs

* move cfg persistence to proper place

* prevent race

* properly update job state at runtime

* cleanup 1

* job2json add missing reason

* add tests

* correct HTTP code

* add test

* streamify delete_job_cb

* add DELETE_JOB keyword

* job delete over streaming

* add tests for create and delete job over parent

* rrdpush common checks to macro

* add missing forwarders

* fix jobs according to test results

* more tests

* review comment 1

* codacy remove valid warning

* codacy ruby fixes

* fix wrong rc check

* minimal test plugin for child

* add test

* dict walk insted of master lock

* minor - english spelling fixes

* thiago comments 1

* minor - rename folder to dynconf

* enable only when built with -DNETDATA_TEST_DYNCFG

* minor - compiler warning

* create dir post daemonization

* stricter URL check

1 files changed, 54 insertions, 5 deletions

diff --git a/streaming/sender.c b/streaming/sender.c
index 2ae3b0cde8..6318464b9a 100644
--- a/streaming/sender.c
+++ b/streaming/sender.c
@@ -955,6 +955,12 @@ void execute_commands(struct sender_state *s) {
     while( start < end && (newline = strchr(start, '\n')) ) {
         *newline = '\0';
 
+        if (s->receiving_function_payload && unlikely(strcmp(start, PLUGINSD_KEYWORD_FUNCTION_PAYLOAD_END) != 0)) {
+            buffer_strcat(s->function_payload.payload, start);
+            start = newline + 1;
+            continue;
+        }
+
         netdata_log_access("STREAM: %d from '%s' for host '%s': %s",
                    gettid(), s->connected_to, rrdhost_hostname(s->host), start);
 
@@ -965,12 +971,12 @@ void execute_commands(struct sender_state *s) {
 
         const char *keyword = get_word(words, num_words, 0);
 
-        if(keyword && strcmp(keyword, PLUGINSD_KEYWORD_FUNCTION) == 0) {
+        if(keyword && (strcmp(keyword, PLUGINSD_KEYWORD_FUNCTION) == 0 || strcmp(keyword, PLUGINSD_KEYWORD_FUNCTION_PAYLOAD_END) == 0)) {
             worker_is_busy(WORKER_SENDER_JOB_FUNCTION_REQUEST);
 
-            char *transaction = get_word(words, num_words, 1);
-            char *timeout_s = get_word(words, num_words, 2);
-            char *function = get_word(words, num_words, 3);
+            char *transaction = s->receiving_function_payload ? s->function_payload.txid : get_word(words, num_words, 1);
+            char *timeout_s = s->receiving_function_payload ? s->function_payload.timeout : get_word(words, num_words, 2);
+            char *function = s->receiving_function_payload ? s->function_payload.fn_name : get_word(words, num_words, 3);
 
             if(!transaction || !*transaction || !timeout_s || !*timeout_s || !function || !*function) {
                 netdata_log_error("STREAM %s [send to %s] %s execution command is incomplete (transaction = '%s', timeout = '%s', function = '%s'). Ignoring it.",
@@ -990,8 +996,9 @@ void execute_commands(struct sender_state *s) {
                 tmp->transaction = string_strdupz(transaction);
                 BUFFER *wb = buffer_create(PLUGINSD_LINE_MAX + 1, &netdata_buffers_statistics.buffers_functions);
 
+                char *payload = s->receiving_function_payload ? (char *)buffer_tostring(s->function_payload.payload) : NULL;
                 int code = rrd_function_run(s->host, wb, timeout, function, false, transaction,
-                                            stream_execute_function_callback, tmp, NULL, NULL);
+                                            stream_execute_function_callback, tmp, NULL, NULL, payload);
 
                 if(code != HTTP_RESP_OK) {
                     if (!buffer_strlen(wb))
@@ -1000,6 +1007,47 @@ void execute_commands(struct sender_state *s) {
                     stream_execute_function_callback(wb, code, tmp);
                 }
             }
+
+            if (s->receiving_function_payload) {
+                s->receiving_function_payload = false;
+
+                buffer_free(s->function_payload.payload);
+                freez(s->function_payload.txid);
+                freez(s->function_payload.timeout);
+                freez(s->function_payload.fn_name);
+
+                memset(&s->function_payload, 0, sizeof(struct function_payload_state));
+            }
+        }
+        else if (keyword && strcmp(keyword, PLUGINSD_KEYWORD_FUNCTION_PAYLOAD) == 0) {
+            if (s->receiving_function_payload) {
+                netdata_log_error("STREAM %s [send to %s] received %s command while already receiving function payload", rrdhost_hostname(s->host), s->connected_to, keyword);
+                s->receiving_function_payload = false;
+                buffer_free(s->function_payload.payload);
+                s->function_payload.payload = NULL;
+
+                // TODO send error response
+            }
+
+            char *transaction = get_word(words, num_words, 1);
+            char *timeout_s = get_word(words, num_words, 2);
+            char *function = get_word(words, num_words, 3);
+
+            if(!transaction || !*transaction || !timeout_s || !*timeout_s || !function || !*function) {
+                netdata_log_error("STREAM %s [send to %s] %s execution command is incomplete (transaction = '%s', timeout = '%s', function = '%s'). Ignoring it.",
+                      rrdhost_hostname(s->host), s->connected_to,
+                      keyword,
+                      transaction?transaction:"(unset)",
+                      timeout_s?timeout_s:"(unset)",
+                      function?function:"(unset)");
+            }
+
+            s->receiving_function_payload = true;
+            s->function_payload.payload = buffer_create(4096, &netdata_buffers_statistics.buffers_functions);
+
+            s->function_payload.txid = strdupz(get_word(words, num_words, 1));
+            s->function_payload.timeout = strdupz(get_word(words, num_words, 2));
+            s->function_payload.fn_name = strdupz(get_word(words, num_words, 3));
         }
         else if(keyword && strcmp(keyword, PLUGINSD_KEYWORD_FUNCTION_CANCEL) == 0) {
             worker_is_busy(WORKER_SENDER_JOB_FUNCTION_REQUEST);
@@ -1343,6 +1391,7 @@ void *rrdpush_sender_thread(void *ptr) {
             rrdpush_send_claimed_id(s->host);
             rrdpush_send_host_labels(s->host);
             rrdpush_send_global_functions(s->host);
+            rrdpush_send_dyncfg(s->host);
             s->replication.oldest_request_after_t = 0;
 
             rrdhost_flag_set(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);