Implement the main flow for the Exporting Engine (#7149)

* Add top level tests

* Add a skeleton for preparing buffers

* Initialize graphite instance

* Prepare buffers for all instances

* Add Grafite collected value formatter

* Add support for exporting.conf read and parsing

* - Use new exporting_config instead of netdata_config

* Implement Grafite worker

* Disable exporting engine compilation if libuv is not available

* Add mutex locks

- Configure connectors as connector_<type> in sections of exporting.conf

- Change exporting_select_type to check for connector_ fields

* - Override exporting_config structure if there no exporting.conf so that
  look ups don't fail and we maintain backwards compatibility

* Separate fixtures in unit tests

* Test exporting_discard_responce

* Test response receiving

* Test buffer sending

* Test simple connector worker

- Instance section has the format connector:instance_name
  e.g graphite:my_graphite_instance

- Connectors with : in their name e.g graphite:plaintext are reserved
  So graphite:plaintext is not accepted because it would activate an
  instance with name "plaintext"
  It should be graphite:plaintext:instance_name

* - Enable the add_connector_instance to cleanup the internal structure
  by passing NULL,not NULL arguments

* Implement configurable update interval

- Add additional check to verify instance uniqueness across connectors

* Add host and chart filters

* Add the value calculation over a database series

* Add the calculated over stored data graphite connector

* Add tests for graphite connector

* Add JSON connector

* Add tests for JSON formatting functions

* Add OpenTSDB connector

* Add tests for the OpenTSDB connector

* Add temporaty notes to the documentation

2 files changed, 190 insertions, 16 deletions

diff --git a/libnetdata/config/appconfig.c b/libnetdata/config/appconfig.c
index 65c36c2818..3774db5549 100644
--- a/libnetdata/config/appconfig.c
+++ b/libnetdata/config/appconfig.c
@@ -42,6 +42,89 @@ struct section {
                             // readers are protected using the rwlock in avl_tree_lock
 };
 
+/*
+ * @Input:
+ *      Connector / instance to add to an internal structure
+ * @Return
+ *      The current head of the linked list of connector_instance
+ *
+ */
+
+_CONNECTOR_INSTANCE *add_connector_instance(struct section *connector, struct section *instance)
+{
+    static struct _connector_instance *global_connector_instance = NULL;
+    struct _connector_instance *local_ci, *local_ci_tmp;
+
+    if (unlikely(!connector)) {
+        if (unlikely(!instance))
+            return global_connector_instance;
+
+        local_ci = global_connector_instance;
+        while (local_ci) {
+            local_ci_tmp = local_ci->next;
+            freez(local_ci);
+            local_ci = local_ci_tmp;
+        }
+        global_connector_instance = NULL;
+        return NULL;
+    }
+
+    local_ci = callocz(1, sizeof(struct _connector_instance));
+    local_ci->instance = instance;
+    local_ci->connector = connector;
+    strncpy(local_ci->instance_name, instance->name, CONFIG_MAX_NAME);
+    strncpy(local_ci->connector_name, connector->name, CONFIG_MAX_NAME);
+    local_ci->next = global_connector_instance;
+    global_connector_instance = local_ci;
+
+    return global_connector_instance;
+}
+
+int is_valid_connector(char *type, int check_reserved)
+{
+    int rc = 1;
+
+    if (unlikely(!type))
+        return 0;
+
+    if (!check_reserved) {
+        if (unlikely(is_valid_connector(type,1))) {
+            return 0;
+        }
+        //if (unlikely(*type == ':')
+        //    return 0;
+        char *separator = strrchr(type, ':');
+        if (likely(separator)) {
+            *separator = '\0';
+            rc = separator - type;
+        } else
+            return 0;
+    }
+//    else {
+//        if (unlikely(is_valid_connector(type,1))) {
+//            error("Section %s invalid -- reserved name", type);
+//            return 0;
+//        }
+//    }
+
+    if (!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
+        return rc;
+    } else if (!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
+        return rc;
+    } else if (!strcmp(type, "opentsdb:http") || !strcmp(type, "opentsdb:https")) {
+        return rc;
+    } else if (!strcmp(type, "json") || !strcmp(type, "json:plaintext")) {
+        return rc;
+    } else if (!strcmp(type, "prometheus_remote_write")) {
+        return rc;
+    } else if (!strcmp(type, "kinesis") || !strcmp(type, "kinesis:plaintext")) {
+        return rc;
+    } else if (!strcmp(type, "mongodb") || !strcmp(type, "mongodb:plaintext")) {
+        return rc;
+    }
+
+    return 0;
+}
 
 // ----------------------------------------------------------------------------
 // locking
@@ -241,6 +324,7 @@ cleanup:
     return ret;
 }
 
+
 char *appconfig_get(struct config *root, const char *section, const char *name, const char *default_value)
 {
     struct config_option *cv;
@@ -440,6 +524,14 @@ int appconfig_load(struct config *root, char *filename, int overwrite_used)
 {
     int line = 0;
     struct section *co = NULL;
+    int is_exporter_config = 0;
+    int is_backend_config = 0;
+    int have_backend_config = 0;
+    int _backends = 0;              // number of backend sections we have
+    char working_instance[CONFIG_MAX_NAME + 1];
+    char working_connector[CONFIG_MAX_NAME + 1];
+    struct section *working_connector_section = NULL;
+    int global_exporting_section = 0;
 
     char buffer[CONFIG_FILE_LINE_MAX + 1], *s;
 
@@ -453,6 +545,8 @@ int appconfig_load(struct config *root, char *filename, int overwrite_used)
         return 0;
     }
 
+    is_exporter_config = (strstr(filename, EXPORTING_CONF) != NULL);
+
     while(fgets(buffer, CONFIG_FILE_LINE_MAX, fp) != NULL) {
         buffer[CONFIG_FILE_LINE_MAX] = '\0';
         line++;
@@ -469,6 +563,46 @@ int appconfig_load(struct config *root, char *filename, int overwrite_used)
             s[len - 1] = '\0';
             s++;
 
+            if (is_exporter_config) {
+                global_exporting_section = !(strcmp(s, CONFIG_SECTION_EXPORTING));
+                if (unlikely(!global_exporting_section)) {
+                    int rc;
+                    rc = is_valid_connector(s, 0);
+                    if (likely(rc)) {
+                        strncpy(working_connector, s, CONFIG_MAX_NAME);
+                        s = s + rc + 1;
+                        if (unlikely(!(*s))) {
+                            _backends++;
+                            sprintf(buffer, "instance_%d", _backends);
+                            s = buffer;
+                        }
+                        strncpy(working_instance, s, CONFIG_MAX_NAME);
+                        working_connector_section = NULL;
+                        if (unlikely(appconfig_section_find(root, working_instance))) {
+                            error("Instance (%s) already exists", working_instance);
+                            co = NULL;
+                            continue;
+                        }
+                    } else {
+                        co = NULL;
+                        error("Section (%s) does not specify a valid connector", s);
+                        continue;
+                    }
+                }
+            }
+
+            is_backend_config = !(strcmp(s, CONFIG_SECTION_BACKEND));
+            if (!have_backend_config)
+                have_backend_config = is_backend_config;
+
+            if (is_backend_config) {
+                if (_backends) {
+                    sprintf(buffer, CONFIG_SECTION_BACKEND "/%d", _backends);
+                    s = buffer;
+                }
+                _backends++;
+            }
+
             co = appconfig_section_find(root, s);
             if(!co) co = appconfig_section_create(root, s);
 
@@ -502,15 +636,32 @@ int appconfig_load(struct config *root, char *filename, int overwrite_used)
 
         struct config_option *cv = appconfig_option_index_find(co, name, 0);
 
-        if(!cv) cv = appconfig_value_create(co, name, value);
-        else {
-            if(((cv->flags & CONFIG_VALUE_USED) && overwrite_used) || !(cv->flags & CONFIG_VALUE_USED)) {
-                debug(D_CONFIG, "CONFIG: line %d of file '%s', overwriting '%s/%s'.", line, filename, co->name, cv->name);
+        if (!cv) {
+            cv = appconfig_value_create(co, name, value);
+            if (likely(is_exporter_config) && unlikely(!global_exporting_section)) {
+                if (unlikely(!working_connector_section)) {
+                    working_connector_section = appconfig_section_find(root, working_connector);
+                    if (!working_connector_section)
+                        working_connector_section = appconfig_section_create(root, working_connector);
+                    if (likely(working_connector_section)) {
+                        add_connector_instance(working_connector_section, co);
+                    }
+                }
+            }
+        } else {
+            if (((cv->flags & CONFIG_VALUE_USED) && overwrite_used) || !(cv->flags & CONFIG_VALUE_USED)) {
+                debug(
+                    D_CONFIG, "CONFIG: line %d of file '%s', overwriting '%s/%s'.", line, filename, co->name, cv->name);
                 freez(cv->value);
                 cv->value = strdupz(value);
-            }
-            else
-                debug(D_CONFIG, "CONFIG: ignoring line %d of file '%s', '%s/%s' is already present and used.", line, filename, co->name, cv->name);
+            } else
+                debug(
+                    D_CONFIG,
+                    "CONFIG: ignoring line %d of file '%s', '%s/%s' is already present and used.",
+                    line,
+                    filename,
+                    co->name,
+                    cv->name);
         }
         cv->flags |= CONFIG_VALUE_LOADED;
     }
diff --git a/libnetdata/config/appconfig.h b/libnetdata/config/appconfig.h
index 32e289f9cd..1a60c46399 100644
--- a/libnetdata/config/appconfig.h
+++ b/libnetdata/config/appconfig.h
@@ -82,15 +82,17 @@
 
 #define CONFIG_FILENAME "netdata.conf"
 
-#define CONFIG_SECTION_GLOBAL   "global"
-#define CONFIG_SECTION_WEB      "web"
-#define CONFIG_SECTION_STATSD   "statsd"
-#define CONFIG_SECTION_PLUGINS  "plugins"
-#define CONFIG_SECTION_CLOUD    "cloud"
-#define CONFIG_SECTION_REGISTRY "registry"
-#define CONFIG_SECTION_HEALTH   "health"
-#define CONFIG_SECTION_BACKEND  "backend"
-#define CONFIG_SECTION_STREAM   "stream"
+#define CONFIG_SECTION_GLOBAL    "global"
+#define CONFIG_SECTION_WEB       "web"
+#define CONFIG_SECTION_STATSD    "statsd"
+#define CONFIG_SECTION_PLUGINS   "plugins"
+#define CONFIG_SECTION_CLOUD     "cloud"
+#define CONFIG_SECTION_REGISTRY  "registry"
+#define CONFIG_SECTION_HEALTH    "health"
+#define CONFIG_SECTION_BACKEND   "backend"
+#define CONFIG_SECTION_STREAM    "stream"
+#define CONFIG_SECTION_EXPORTING "exporting:global"
+#define EXPORTING_CONF           "exporting.conf"
 
 // these are used to limit the configuration names and values lengths
 // they are not enforced by config.c functions (they will strdup() all strings, no matter of their length)
@@ -103,6 +105,11 @@ struct config {
     avl_tree_lock index;
 };
 
+//struct connector_instance {
+//    char instance_name[CONFIG_MAX_NAME + 1];
+//    char connector_name[CONFIG_MAX_NAME + 1];
+//};
+
 #define CONFIG_BOOLEAN_INVALID 100  // an invalid value to check for validity (used as default initialization when needed)
 
 #define CONFIG_BOOLEAN_NO   0       // disabled
@@ -136,4 +143,20 @@ extern int appconfig_section_compare(void *a, void *b);
 
 extern int config_parse_duration(const char* string, int* result);
 
+
+struct connector_instance {
+    char instance_name[CONFIG_MAX_NAME + 1];
+    char connector_name[CONFIG_MAX_NAME + 1];
+};
+
+typedef struct _connector_instance {
+    struct section *connector;        // actual connector
+    struct section *instance;         // This instance
+    char instance_name[CONFIG_MAX_NAME + 1];
+    char connector_name[CONFIG_MAX_NAME + 1];
+    struct _connector_instance *next; // Next instance
+} _CONNECTOR_INSTANCE;
+
+extern _CONNECTOR_INSTANCE *add_connector_instance(struct section *connector, struct section *instance);
+
 #endif /* NETDATA_CONFIG_H */