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, 19 insertions, 0 deletions

diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h
index f8d6926039..c3c14233fa 100644
--- a/streaming/rrdpush.h
+++ b/streaming/rrdpush.h
@@ -45,6 +45,7 @@ typedef enum {
     STREAM_CAP_INTERPOLATED     = (1 << 14), // streaming supports interpolated streaming of values
     STREAM_CAP_IEEE754          = (1 << 15), // streaming supports binary/hex transfer of double values
     STREAM_CAP_DATA_WITH_ML     = (1 << 16), // streaming supports transferring anomaly bit
+    STREAM_CAP_DYNCFG           = (1 << 17), // dynamic configuration of plugins trough streaming
 
     STREAM_CAP_INVALID          = (1 << 30), // used as an invalid value for capabilities when this is set
     // this must be signed int, so don't use the last bit
@@ -232,6 +233,13 @@ typedef enum __attribute__((packed)) {
     SENDER_FLAG_COMPRESSION  = (1 << 1), // The stream needs to have and has compression
 } SENDER_FLAGS;
 
+struct function_payload_state {
+    BUFFER *payload;
+    char *txid;
+    char *fn_name;
+    char *timeout;
+};
+
 struct sender_state {
     RRDHOST *host;
     pid_t tid;                              // the thread id of the sender, from gettid()
@@ -261,6 +269,9 @@ struct sender_state {
     int rrdpush_sender_pipe[2];                     // collector to sender thread signaling
     int rrdpush_sender_socket;
 
+    int receiving_function_payload;
+    struct function_payload_state function_payload; // state when receiving function with payload
+
     uint16_t hops;
 
 #ifdef ENABLE_RRDPUSH_COMPRESSION
@@ -482,6 +493,7 @@ void *rrdpush_sender_thread(void *ptr);
 void rrdpush_send_host_labels(RRDHOST *host);
 void rrdpush_send_claimed_id(RRDHOST *host);
 void rrdpush_send_global_functions(RRDHOST *host);
+void rrdpush_send_dyncfg(RRDHOST *host);
 
 #define THREAD_TAG_STREAM_RECEIVER "RCVR" // "[host]" is appended
 #define THREAD_TAG_STREAM_SENDER "SNDR" // "[host]" is appended
@@ -764,4 +776,11 @@ typedef struct rrdhost_status {
 void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s);
 bool rrdhost_state_cloud_emulation(RRDHOST *host);
 
+void rrdpush_send_job_status_update(RRDHOST *host, const char *plugin_name, const char *module_name, struct job *job);
+void rrdpush_send_job_deleted(RRDHOST *host, const char *plugin_name, const char *module_name, const char *job_name);
+
+void rrdpush_send_dyncfg_enable(RRDHOST *host, const char *plugin_name);
+void rrdpush_send_dyncfg_reg_module(RRDHOST *host, const char *plugin_name, const char *module_name, enum module_type type);
+void rrdpush_send_dyncfg_reg_job(RRDHOST *host, const char *plugin_name, const char *module_name, const char *job_name, enum job_type type, uint32_t flags);//x
+
 #endif //NETDATA_RRDPUSH_H