stream paths propagated to children and parents (#18430)

* stream paths propagates to children and parents

* do not send the stream paths back to the parent

* cleanup paths when the child disconnects

* fix version placement

* removed last_time_t

* remove deadlock

* prevent recurring sender cleanups when the sender cannot connect to parent

* split sender.c to multiple files

* fix json generation

* allow null for uuids

* update stream_paths when the retention of the node changes

* clear missing claim_id

* validations

* fix uuid json parsing

* fixed left-over uuid

* keep the array sorted

* fix sending nodes calculation

* send stream path updates when node_id is set or changed

* stream path capabilities are now the full capabilities as a sender, independently of the current streaming connections

CMakeLists.txt

@@ -1264,6 +1264,14 @@ set(STREAMING_PLUGIN_FILES
         src/streaming/protocol/commands.c
         src/streaming/protocol/commands.h
         src/streaming/protocol/command-claimed_id.c
+        src/streaming/stream_path.c
+        src/streaming/stream_path.h
+        src/streaming/stream_capabilities.c
+        src/streaming/stream_capabilities.h
+        src/streaming/sender_connect.c
+        src/streaming/sender_internals.h
+        src/streaming/sender_execute.c
+        src/streaming/sender_commit.c
 )
 
 set(WEB_PLUGIN_FILES

src/aclk/aclk.c

@@ -839,16 +839,16 @@ exit:
 
 void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)
 {
-    nd_uuid_t node_id;
+    ND_UUID node_id;
 
     if (!aclk_online())
         return;
 
-    if (!uuid_is_null(host->node_id)) {
-        uuid_copy(node_id, host->node_id);
+    if (!UUIDiszero(host->node_id)) {
+        node_id = host->node_id;
     }
     else {
-        int ret = get_node_id(&host->host_uuid, &node_id);
+        int ret = get_node_id(&host->host_id.uuid, &node_id.uuid);
         if (ret > 0) {
             // this means we were not able to check if node_id already present
             netdata_log_error("Unable to check for node_id. Ignoring the host state update.");
@@ -887,7 +887,7 @@ void aclk_host_state_update(RRDHOST *host, int cmd, int queryable)
         .session_id = aclk_session_newarch
     };
     node_state_update.node_id = mallocz(UUID_STR_LEN);
-    uuid_unparse_lower(node_id, (char*)node_state_update.node_id);
+    uuid_unparse_lower(node_id.uuid, (char*)node_state_update.node_id);
 
     node_state_update.capabilities = aclk_get_agent_capas();
 
@@ -1059,12 +1059,12 @@ char *aclk_state(void)
             else
                 buffer_strcat(wb, "null");
 
-            if (uuid_is_null(host->node_id))
+            if (UUIDiszero(host->node_id))
                 buffer_strcat(wb, "\n\tNode ID: null\n");
             else {
-                char node_id[GUID_LEN + 1];
-                uuid_unparse_lower(host->node_id, node_id);
-                buffer_sprintf(wb, "\n\tNode ID: %s\n", node_id);
+                char node_id_str[UUID_STR_LEN];
+                uuid_unparse_lower(host->node_id.uuid, node_id_str);
+                buffer_sprintf(wb, "\n\tNode ID: %s\n", node_id_str);
             }
 
             buffer_sprintf(wb, "\tStreaming Hops: %d\n\tRelationship: %s", host->system_info->hops, host == localhost ? "self" : "child");
@@ -1192,12 +1192,12 @@ char *aclk_state_json(void)
         } else
             json_object_object_add(nodeinstance, "claimed_id", NULL);
 
-        if (uuid_is_null(host->node_id)) {
+        if (UUIDiszero(host->node_id)) {
             json_object_object_add(nodeinstance, "node-id", NULL);
         } else {
-            char node_id[GUID_LEN + 1];
-            uuid_unparse_lower(host->node_id, node_id);
-            tmp = json_object_new_string(node_id);
+            char node_id_str[UUID_STR_LEN];
+            uuid_unparse_lower(host->node_id.uuid, node_id_str);
+            tmp = json_object_new_string(node_id_str);
             json_object_object_add(nodeinstance, "node-id", tmp);
         }
 

src/claim/claim.c

@@ -166,8 +166,8 @@ bool load_claiming_state(void) {
         have_claimed_id = true;
     }
 
-    invalidate_node_instances(&localhost->host_uuid, have_claimed_id ? &uuid.uuid : NULL);
-    metaqueue_store_claim_id(&localhost->host_uuid, have_claimed_id ? &uuid.uuid : NULL);
+    invalidate_node_instances(&localhost->host_id.uuid, have_claimed_id ? &uuid.uuid : NULL);
+    metaqueue_store_claim_id(&localhost->host_id.uuid, have_claimed_id ? &uuid.uuid : NULL);
 
     errno_clear();
 
@@ -197,7 +197,7 @@ CLOUD_STATUS claim_reload_and_wait_online(void) {
         int ms = 0;
         do {
             status = cloud_status();
-            if ((status == CLOUD_STATUS_ONLINE || status == CLOUD_STATUS_INDIRECT) && !uuid_is_null(localhost->host_uuid))
+            if ((status == CLOUD_STATUS_ONLINE || status == CLOUD_STATUS_INDIRECT) && !UUIDiszero(localhost->host_id))
                 break;
 
             sleep_usec(50 * USEC_PER_MS);

src/claim/cloud-status.c

@@ -32,7 +32,7 @@ CLOUD_STATUS cloud_status(void) {
     if(localhost->sender &&
         rrdhost_flag_check(localhost, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS) &&
         stream_has_capability(localhost->sender, STREAM_CAP_NODE_ID) &&
-        !uuid_is_null(localhost->node_id) &&
+        !UUIDiszero(localhost->node_id) &&
         !UUIDiszero(localhost->aclk.claim_id_of_parent))
         return CLOUD_STATUS_INDIRECT;
 

src/collectors/plugins.d/gperf-config.txt

@@ -35,6 +35,8 @@
 #define PLUGINSD_KEYWORD_ID_RSET                   21
 #define PLUGINSD_KEYWORD_ID_RSSTATE                24
 
+#define PLUGINSD_KEYWORD_ID_JSON                   80
+
 #define PLUGINSD_KEYWORD_ID_DYNCFG_ENABLE          901
 #define PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_MODULE 902
 #define PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_JOB    903
@@ -102,11 +104,15 @@ REND,                 PLUGINSD_KEYWORD_ID_REND,                 PARSER_INIT_STRE
 RSET,                 PLUGINSD_KEYWORD_ID_RSET,                 PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 30
 RSSTATE,              PLUGINSD_KEYWORD_ID_RSSTATE,              PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 31
 #
+# JSON
+#
+JSON,                 PLUGINSD_KEYWORD_ID_JSON,                 PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 32
+#
 # obsolete - do nothing commands
 #
-DYNCFG_ENABLE,          PLUGINSD_KEYWORD_ID_DYNCFG_ENABLE,          PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 32
-DYNCFG_REGISTER_MODULE, PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_MODULE, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 33
-DYNCFG_REGISTER_JOB,    PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_JOB,    PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 34
-DYNCFG_RESET,           PLUGINSD_KEYWORD_ID_DYNCFG_RESET,           PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 35
-REPORT_JOB_STATUS,      PLUGINSD_KEYWORD_ID_REPORT_JOB_STATUS,      PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 36
-DELETE_JOB,             PLUGINSD_KEYWORD_ID_DELETE_JOB,             PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 37
+DYNCFG_ENABLE,          PLUGINSD_KEYWORD_ID_DYNCFG_ENABLE,          PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 33
+DYNCFG_REGISTER_MODULE, PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_MODULE, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 34
+DYNCFG_REGISTER_JOB,    PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_JOB,    PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 35
+DYNCFG_RESET,           PLUGINSD_KEYWORD_ID_DYNCFG_RESET,           PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 36
+REPORT_JOB_STATUS,      PLUGINSD_KEYWORD_ID_REPORT_JOB_STATUS,      PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 37
+DELETE_JOB,             PLUGINSD_KEYWORD_ID_DELETE_JOB,             PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 38

src/collectors/plugins.d/gperf-hashtable.h

@@ -67,6 +67,8 @@
 #define PLUGINSD_KEYWORD_ID_RSET                   21
 #define PLUGINSD_KEYWORD_ID_RSSTATE                24
 
+#define PLUGINSD_KEYWORD_ID_JSON                   80
+
 #define PLUGINSD_KEYWORD_ID_DYNCFG_ENABLE          901
 #define PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_MODULE 902
 #define PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_JOB    903
@@ -75,12 +77,12 @@
 #define PLUGINSD_KEYWORD_ID_DELETE_JOB             906
 
 
-#define GPERF_PARSER_TOTAL_KEYWORDS 37
+#define GPERF_PARSER_TOTAL_KEYWORDS 38
 #define GPERF_PARSER_MIN_WORD_LENGTH 3
 #define GPERF_PARSER_MAX_WORD_LENGTH 22
-#define GPERF_PARSER_MIN_HASH_VALUE 7
-#define GPERF_PARSER_MAX_HASH_VALUE 52
-/* maximum key range = 46, duplicates = 0 */
+#define GPERF_PARSER_MIN_HASH_VALUE 4
+#define GPERF_PARSER_MAX_HASH_VALUE 53
+/* maximum key range = 50, duplicates = 0 */
 
 #ifdef __GNUC__
 __inline
@@ -94,32 +96,32 @@ gperf_keyword_hash_function (register const char *str, register size_t len)
 {
   static const unsigned char asso_values[] =
     {
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53,  6, 24,  3,  9,  6,
-       0, 53,  3, 27, 53, 53, 33, 53, 42,  0,
-      53, 53,  0, 30, 53, 12,  3, 53,  9,  0,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53, 53, 53, 53, 53,
-      53, 53, 53, 53, 53, 53
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 31, 28,  2,  4,  0,
+       5, 54,  0, 25, 20, 54, 17, 54, 27,  0,
+      54, 54,  1, 16, 54, 15,  0, 54,  2,  0,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54, 54, 54, 54, 54,
+      54, 54, 54, 54, 54, 54
     };
   return len + asso_values[(unsigned char)str[1]] + asso_values[(unsigned char)str[0]];
 }
@@ -130,92 +132,94 @@ static const PARSER_KEYWORD gperf_keywords[] =
     {(char*)0,0,PARSER_INIT_PLUGINSD,0},
     {(char*)0,0,PARSER_INIT_PLUGINSD,0},
     {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-#line 67 "gperf-config.txt"
+#line 69 "gperf-config.txt"
     {"HOST",            PLUGINSD_KEYWORD_ID_HOST,            PARSER_INIT_PLUGINSD|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 4},
-    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-#line 87 "gperf-config.txt"
-    {"CONFIG",                PLUGINSD_KEYWORD_ID_CONFIG,                PARSER_INIT_PLUGINSD|PARSER_REP_METADATA,                       WORKER_PARSER_FIRST_JOB + 21},
-#line 101 "gperf-config.txt"
+#line 103 "gperf-config.txt"
     {"REND",                 PLUGINSD_KEYWORD_ID_REND,                 PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 29},
-#line 75 "gperf-config.txt"
+#line 68 "gperf-config.txt"
+    {"EXIT",            PLUGINSD_KEYWORD_ID_EXIT,            PARSER_INIT_PLUGINSD,                     WORKER_PARSER_FIRST_JOB + 3},
+#line 77 "gperf-config.txt"
     {"CHART",                 PLUGINSD_KEYWORD_ID_CHART,                 PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 9},
-#line 84 "gperf-config.txt"
+#line 89 "gperf-config.txt"
+    {"CONFIG",                PLUGINSD_KEYWORD_ID_CONFIG,                PARSER_INIT_PLUGINSD|PARSER_REP_METADATA,                       WORKER_PARSER_FIRST_JOB + 21},
+#line 86 "gperf-config.txt"
     {"OVERWRITE",             PLUGINSD_KEYWORD_ID_OVERWRITE,             PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 18},
-#line 70 "gperf-config.txt"
+#line 72 "gperf-config.txt"
     {"HOST_LABEL",      PLUGINSD_KEYWORD_ID_HOST_LABEL,      PARSER_INIT_PLUGINSD|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 7},
-#line 68 "gperf-config.txt"
+#line 70 "gperf-config.txt"
     {"HOST_DEFINE",     PLUGINSD_KEYWORD_ID_HOST_DEFINE,     PARSER_INIT_PLUGINSD|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 5},
-    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-#line 100 "gperf-config.txt"
+#line 102 "gperf-config.txt"
     {"RDSTATE",              PLUGINSD_KEYWORD_ID_RDSTATE,              PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 28},
-#line 86 "gperf-config.txt"
-    {"VARIABLE",              PLUGINSD_KEYWORD_ID_VARIABLE,              PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 20},
-#line 69 "gperf-config.txt"
-    {"HOST_DEFINE_END", PLUGINSD_KEYWORD_ID_HOST_DEFINE_END, PARSER_INIT_PLUGINSD|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 6},
-#line 66 "gperf-config.txt"
-    {"EXIT",            PLUGINSD_KEYWORD_ID_EXIT,            PARSER_INIT_PLUGINSD,                     WORKER_PARSER_FIRST_JOB + 3},
-#line 80 "gperf-config.txt"
-    {"FUNCTION",              PLUGINSD_KEYWORD_ID_FUNCTION,              PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 14},
-#line 110 "gperf-config.txt"
-    {"DYNCFG_RESET",           PLUGINSD_KEYWORD_ID_DYNCFG_RESET,           PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 35},
-#line 107 "gperf-config.txt"
-    {"DYNCFG_ENABLE",          PLUGINSD_KEYWORD_ID_DYNCFG_ENABLE,          PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 32},
-#line 111 "gperf-config.txt"
-    {"REPORT_JOB_STATUS",      PLUGINSD_KEYWORD_ID_REPORT_JOB_STATUS,      PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 36},
-    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-#line 112 "gperf-config.txt"
-    {"DELETE_JOB",             PLUGINSD_KEYWORD_ID_DELETE_JOB,             PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 37},
-#line 98 "gperf-config.txt"
-    {"CHART_DEFINITION_END", PLUGINSD_KEYWORD_ID_CHART_DEFINITION_END, PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 26},
     {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-#line 109 "gperf-config.txt"
-    {"DYNCFG_REGISTER_JOB",    PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_JOB,    PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 34},
-#line 82 "gperf-config.txt"
-    {"FUNCTION_PROGRESS",     PLUGINSD_KEYWORD_ID_FUNCTION_PROGRESS,     PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 16},
-#line 99 "gperf-config.txt"
-    {"RBEGIN",               PLUGINSD_KEYWORD_ID_RBEGIN,               PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 27},
-#line 108 "gperf-config.txt"
-    {"DYNCFG_REGISTER_MODULE", PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_MODULE, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 33},
-    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-#line 81 "gperf-config.txt"
-    {"FUNCTION_RESULT_BEGIN", PLUGINSD_KEYWORD_ID_FUNCTION_RESULT_BEGIN, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 15},
-#line 102 "gperf-config.txt"
+#line 118 "gperf-config.txt"
+    {"DELETE_JOB",             PLUGINSD_KEYWORD_ID_DELETE_JOB,             PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 38},
+#line 71 "gperf-config.txt"
+    {"HOST_DEFINE_END", PLUGINSD_KEYWORD_ID_HOST_DEFINE_END, PARSER_INIT_PLUGINSD|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 6},
+#line 116 "gperf-config.txt"
+    {"DYNCFG_RESET",           PLUGINSD_KEYWORD_ID_DYNCFG_RESET,           PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 36},
+#line 113 "gperf-config.txt"
+    {"DYNCFG_ENABLE",          PLUGINSD_KEYWORD_ID_DYNCFG_ENABLE,          PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 33},
+#line 117 "gperf-config.txt"
+    {"REPORT_JOB_STATUS",      PLUGINSD_KEYWORD_ID_REPORT_JOB_STATUS,      PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 37},
+#line 87 "gperf-config.txt"
+    {"SET",                   PLUGINSD_KEYWORD_ID_SET,                   PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 19},
+#line 95 "gperf-config.txt"
+    {"SET2",       PLUGINSD_KEYWORD_ID_SET2,       PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 24},
+#line 104 "gperf-config.txt"
     {"RSET",                 PLUGINSD_KEYWORD_ID_RSET,                 PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 30},
-#line 74 "gperf-config.txt"
-    {"BEGIN",                 PLUGINSD_KEYWORD_ID_BEGIN,                 PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 8},
-#line 92 "gperf-config.txt"
-    {"BEGIN2",     PLUGINSD_KEYWORD_ID_BEGIN2,     PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 23},
-#line 103 "gperf-config.txt"
+#line 100 "gperf-config.txt"
+    {"CHART_DEFINITION_END", PLUGINSD_KEYWORD_ID_CHART_DEFINITION_END, PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 26},
+#line 115 "gperf-config.txt"
+    {"DYNCFG_REGISTER_JOB",    PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_JOB,    PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 35},
+#line 105 "gperf-config.txt"
     {"RSSTATE",              PLUGINSD_KEYWORD_ID_RSSTATE,              PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 31},
-#line 64 "gperf-config.txt"
+#line 78 "gperf-config.txt"
+    {"CLABEL",                PLUGINSD_KEYWORD_ID_CLABEL,                PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 10},
+#line 114 "gperf-config.txt"
+    {"DYNCFG_REGISTER_MODULE", PLUGINSD_KEYWORD_ID_DYNCFG_REGISTER_MODULE, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 34},
+#line 66 "gperf-config.txt"
     {"FLUSH",           PLUGINSD_KEYWORD_ID_FLUSH,           PARSER_INIT_PLUGINSD,                     WORKER_PARSER_FIRST_JOB + 1},
-#line 85 "gperf-config.txt"
-    {"SET",                   PLUGINSD_KEYWORD_ID_SET,                   PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 19},
+#line 82 "gperf-config.txt"
+    {"FUNCTION",              PLUGINSD_KEYWORD_ID_FUNCTION,              PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 14},
 #line 93 "gperf-config.txt"
-    {"SET2",       PLUGINSD_KEYWORD_ID_SET2,       PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 24},
-    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+    {"CLAIMED_ID", PLUGINSD_KEYWORD_ID_CLAIMED_ID, PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 22},
+#line 81 "gperf-config.txt"
+    {"END",                   PLUGINSD_KEYWORD_ID_END,                   PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 13},
+#line 96 "gperf-config.txt"
+    {"END2",       PLUGINSD_KEYWORD_ID_END2,       PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 25},
+#line 79 "gperf-config.txt"
+    {"CLABEL_COMMIT",         PLUGINSD_KEYWORD_ID_CLABEL_COMMIT,         PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 11},
 #line 76 "gperf-config.txt"
-    {"CLABEL",                PLUGINSD_KEYWORD_ID_CLABEL,                PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 10},
-#line 65 "gperf-config.txt"
+    {"BEGIN",                 PLUGINSD_KEYWORD_ID_BEGIN,                 PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 8},
+#line 94 "gperf-config.txt"
+    {"BEGIN2",     PLUGINSD_KEYWORD_ID_BEGIN2,     PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 23},
+#line 101 "gperf-config.txt"
+    {"RBEGIN",               PLUGINSD_KEYWORD_ID_RBEGIN,               PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 27},
+#line 67 "gperf-config.txt"
     {"DISABLE",         PLUGINSD_KEYWORD_ID_DISABLE,         PARSER_INIT_PLUGINSD,                     WORKER_PARSER_FIRST_JOB + 2},
-#line 83 "gperf-config.txt"
-    {"LABEL",                 PLUGINSD_KEYWORD_ID_LABEL,                 PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 17},
-#line 78 "gperf-config.txt"
+#line 84 "gperf-config.txt"
+    {"FUNCTION_PROGRESS",     PLUGINSD_KEYWORD_ID_FUNCTION_PROGRESS,     PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 16},
+#line 80 "gperf-config.txt"
     {"DIMENSION",             PLUGINSD_KEYWORD_ID_DIMENSION,             PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 12},
-#line 91 "gperf-config.txt"
-    {"CLAIMED_ID", PLUGINSD_KEYWORD_ID_CLAIMED_ID, PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 22},
+#line 88 "gperf-config.txt"
+    {"VARIABLE",              PLUGINSD_KEYWORD_ID_VARIABLE,              PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 20},
+#line 109 "gperf-config.txt"
+    {"JSON",                 PLUGINSD_KEYWORD_ID_JSON,                 PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 32},
+#line 83 "gperf-config.txt"
+    {"FUNCTION_RESULT_BEGIN", PLUGINSD_KEYWORD_ID_FUNCTION_RESULT_BEGIN, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 15},
     {(char*)0,0,PARSER_INIT_PLUGINSD,0},
     {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-#line 77 "gperf-config.txt"
-    {"CLABEL_COMMIT",         PLUGINSD_KEYWORD_ID_CLABEL_COMMIT,         PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 11},
     {(char*)0,0,PARSER_INIT_PLUGINSD,0},
-#line 79 "gperf-config.txt"
-    {"END",                   PLUGINSD_KEYWORD_ID_END,                   PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 13},
-#line 94 "gperf-config.txt"
-    {"END2",       PLUGINSD_KEYWORD_ID_END2,       PARSER_INIT_STREAMING,                     WORKER_PARSER_FIRST_JOB + 25}
+    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+    {(char*)0,0,PARSER_INIT_PLUGINSD,0},
+#line 85 "gperf-config.txt"
+    {"LABEL",                 PLUGINSD_KEYWORD_ID_LABEL,                 PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING|PARSER_REP_METADATA, WORKER_PARSER_FIRST_JOB + 17}
   };
 
 const PARSER_KEYWORD *

src/collectors/plugins.d/pluginsd_parser.c

@@ -1081,6 +1081,35 @@ static inline PARSER_RC pluginsd_exit(char **words __maybe_unused, size_t num_wo
     return PARSER_RC_STOP;
 }
 
+static void pluginsd_json_stream_paths(PARSER *parser, void *action_data __maybe_unused) {
+    stream_path_set_from_json(parser->user.host, buffer_tostring(parser->defer.response), false);
+    buffer_free(parser->defer.response);
+}
+
+static void pluginsd_json_dev_null(PARSER *parser, void *action_data __maybe_unused) {
+    buffer_free(parser->defer.response);
+}
+
+static PARSER_RC pluginsd_json(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
+    RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_JSON);
+    if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
+
+    char *keyword = get_word(words, num_words, 1);
+
+    parser->defer.response = buffer_create(0, NULL);
+    parser->defer.end_keyword = PLUGINSD_KEYWORD_JSON_END;
+    parser->defer.action = pluginsd_json_dev_null;
+    parser->defer.action_data = NULL;
+    parser->flags |= PARSER_DEFER_UNTIL_KEYWORD;
+
+    if(strcmp(keyword, PLUGINSD_KEYWORD_STREAM_PATH) == 0)
+        parser->defer.action = pluginsd_json_stream_paths;
+    else
+        netdata_log_error("PLUGINSD: invalid JSON payload keyword '%s'", keyword);
+
+    return PARSER_RC_OK;
+}
+
 PARSER_RC rrdpush_receiver_pluginsd_claimed_id(char **words, size_t num_words, PARSER *parser);
 
 // ----------------------------------------------------------------------------
@@ -1215,6 +1244,8 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, int fd_input,
 #include "gperf-hashtable.h"
 
 PARSER_RC parser_execute(PARSER *parser, const PARSER_KEYWORD *keyword, char **words, size_t num_words) {
+    // put all the keywords ordered by the frequency they are used
+
     switch(keyword->id) {
         case PLUGINSD_KEYWORD_ID_SET2:
             return pluginsd_set_v2(words, num_words, parser);
@@ -1254,6 +1285,8 @@ PARSER_RC parser_execute(PARSER *parser, const PARSER_KEYWORD *keyword, char **w
             return pluginsd_function_result_begin(words, num_words, parser);
         case PLUGINSD_KEYWORD_ID_FUNCTION_PROGRESS:
             return pluginsd_function_progress(words, num_words, parser);
+        case PLUGINSD_KEYWORD_ID_JSON:
+            return pluginsd_json(words, num_words, parser);
         case PLUGINSD_KEYWORD_ID_LABEL:
             return pluginsd_label(words, num_words, parser);
         case PLUGINSD_KEYWORD_ID_OVERWRITE:

src/collectors/plugins.d/pluginsd_parser.h

@@ -93,6 +93,8 @@ typedef struct parser_user_object {
     } v2;
 } PARSER_USER_OBJECT;
 
+typedef void (*parser_deferred_action_t)(struct parser *parser, void *action_data);
+
 struct parser {
     uint8_t version;                // Parser version
     PARSER_REPERTOIRE repertoire;
@@ -115,7 +117,7 @@ struct parser {
     struct {
         const char *end_keyword;
         BUFFER *response;
-        void (*action)(struct parser *parser, void *action_data);
+        parser_deferred_action_t action;
         void *action_data;
     } defer;
 

src/daemon/commands.c

@@ -372,10 +372,10 @@ static int remove_ephemeral_host(BUFFER *wb, RRDHOST *host, bool report_error)
 
     if (!rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST)) {
         rrdhost_option_set(host, RRDHOST_OPTION_EPHEMERAL_HOST);
-        sql_set_host_label(&host->host_uuid, "_is_ephemeral", "true");
+        sql_set_host_label(&host->host_id.uuid, "_is_ephemeral", "true");
         aclk_host_state_update(host, 0, 0);
         unregister_node(host->machine_guid);
-        uuid_clear(host->node_id);
+        host->node_id = UUID_ZERO;
         buffer_sprintf(wb, "Unregistering node with machine guid %s, hostname = %s", host->machine_guid, rrdhost_hostname(host));
         rrd_wrlock();
         rrdhost_free___while_having_rrd_wrlock(host, true);
@@ -517,7 +517,7 @@ static void pipe_write_cb(uv_write_t* req, int status)
 
 static inline void add_char_to_command_reply(BUFFER *reply_string, unsigned *reply_string_size, char character)
 {
-    buffer_fast_charcat(reply_string, character);
+    buffer_putc(reply_string, character);
     *reply_string_size +=1;
 }
 

src/daemon/config/dyncfg-tree.c

@@ -71,12 +71,10 @@ static void dyncfg_tree_for_host(RRDHOST *host, BUFFER *wb, const char *path, co
     if(id && *id)
         template = string_strdupz(id);
 
-    ND_UUID host_uuid = uuid2UUID(host->host_uuid);
-
     size_t path_len = strlen(path);
     DYNCFG *df;
     dfe_start_read(dyncfg_globals.nodes, df) {
-        if(!UUIDeq(df->host_uuid, host_uuid))
+        if(!UUIDeq(df->host_uuid, host->host_id))
             continue;
 
         if(strncmp(string2str(df->path), path, path_len) != 0)