Virtual hosts for data collection (#14464)

* support multiple hosts at pluginsd structures

* cleanup obsolete code

* use a lookup hashtable to quickly find the keyword to execute, without traversing the whole linked list of keywords

* more cleanup

* move new hash function to inlined.h

* minimize comparisons, eliminate a pre-parsing of the first keyword for each line

* cleanup parser from old code

* move parser into libnetdata

* unique entries in parser keywords hashtable

* move all hashing functions to inlined.h, name their sources, simple_hash() now defaults to FNV1a, it was FNV1

* small_hash() for parser

* plugins.d now can switch hosts, and also create/update them

* update hash function and hashtable size

* updated message

* unittest all hashing functions

* reset the chart when setting a new host

* remove host tags

* enable archived hosts when a collector pushes host info

* do not need localhost to swtich to localhost

* disable ARAL and OWA with -DFSANITIZE_ADDRESS=1

3 files changed, 5 insertions, 15 deletions

diff --git a/streaming/receiver.c b/streaming/receiver.c
index 9378d2d825..0eed7b0fb2 100644
--- a/streaming/receiver.c
+++ b/streaming/receiver.c
@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
 #include "rrdpush.h"
-#include "parser/parser.h"
 
 // IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly
 #define WORKER_RECEIVER_JOB_BYTES_READ (WORKER_PARSER_FIRST_JOB - 1)
@@ -332,10 +331,6 @@ static void streaming_parser_thread_cleanup(void *ptr) {
 
 bool plugin_is_enabled(struct plugind *cd);
 
-void streaming_parser_cleanup(void *user) {
-    pluginsd_cleanup_v2(user);
-}
-
 static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, int fd, void *ssl) {
     size_t result;
 
@@ -347,7 +342,10 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, i
         .trust_durations = 1
     };
 
-    PARSER *parser = parser_init(rpt->host, &user, streaming_parser_cleanup, NULL, NULL, fd, PARSER_INPUT_SPLIT, ssl);
+    PARSER *parser = parser_init(&user, NULL, NULL, fd,
+                                 PARSER_INPUT_SPLIT, ssl);
+
+    pluginsd_keywords_init(parser, PARSER_INIT_STREAMING);
 
     rrd_collector_started();
 
@@ -728,16 +726,12 @@ static int rrdpush_receive(struct receiver_state *rpt)
 
     struct plugind cd = {
             .update_every = default_rrd_update_every,
-            .serial_failures = 0,
-            .successful_collections = 0,
             .unsafe = {
                     .spinlock = NETDATA_SPINLOCK_INITIALIZER,
                     .running = true,
                     .enabled = true,
             },
             .started_t = now_realtime_sec(),
-            .next = NULL,
-            .capabilities = 0,
     };
 
     // put the client IP and port into the buffers used by plugins.d
@@ -808,8 +802,6 @@ static int rrdpush_receive(struct receiver_state *rpt)
 
     rrdpush_receive_log_status(rpt, "ready to receive data", "CONNECTED");
 
-    cd.capabilities = rpt->capabilities;
-
 #ifdef ENABLE_ACLK
     // in case we have cloud connection we inform cloud
     // new child connected
diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c
index 3ba9a1da84..0fbe80257d 100644
--- a/streaming/rrdpush.c
+++ b/streaming/rrdpush.c
@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
 #include "rrdpush.h"
-#include "parser/parser.h"
 
 /*
  * rrdpush
diff --git a/streaming/sender.c b/streaming/sender.c
index d1ce9b0f86..1bae576e34 100644
--- a/streaming/sender.c
+++ b/streaming/sender.c
@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
 #include "rrdpush.h"
-#include "parser/parser.h"
 
 #define WORKER_SENDER_JOB_CONNECT                    0
 #define WORKER_SENDER_JOB_PIPE_READ                  1
@@ -937,7 +936,7 @@ void execute_commands(struct sender_state *s) {
         // internal_error(true, "STREAM %s [send to %s] received command over connection: %s", rrdhost_hostname(s->host), s->connected_to, start);
 
         char *words[PLUGINSD_MAX_WORDS] = { NULL };
-        size_t num_words = pluginsd_split_words(start, words, PLUGINSD_MAX_WORDS, NULL, NULL, 0);
+        size_t num_words = pluginsd_split_words(start, words, PLUGINSD_MAX_WORDS);
 
         const char *keyword = get_word(words, num_words, 0);