Exporting cleanup (#9098)

Cleanup allocated variables for the majority of the databases.

12 files changed, 145 insertions, 35 deletions

diff --git a/exporting/aws_kinesis/aws_kinesis.c b/exporting/aws_kinesis/aws_kinesis.c
index e8a9b540aa..a32d805d66 100644
--- a/exporting/aws_kinesis/aws_kinesis.c
+++ b/exporting/aws_kinesis/aws_kinesis.c
@@ -3,6 +3,29 @@
 #include "aws_kinesis.h"
 
 /**
+ * Clean AWS Kinesis *
+ */
+void aws_kinesis_cleanup(struct instance *instance)
+{
+    info("EXPORTING: cleaning up instance %s ...", instance->config.name);
+    kinesis_shutdown(instance->connector_specific_data);
+
+    freez(instance->connector_specific_data);
+
+    struct aws_kinesis_specific_config *connector_specific_config = instance->config.connector_specific_config;
+    if (connector_specific_config) {
+        freez(connector_specific_config->auth_key_id);
+        freez(connector_specific_config->secure_key);
+        freez(connector_specific_config->stream_name);
+
+        freez(connector_specific_config);
+    }
+
+    info("EXPORTING: instance %s exited", instance->config.name);
+    instance->exited = 1;
+}
+
+/**
  * Initialize AWS Kinesis connector instance
  *
  * @param instance an instance data structure.
@@ -68,12 +91,16 @@ void aws_kinesis_connector_worker(void *instance_p)
     struct aws_kinesis_specific_config *connector_specific_config = instance->config.connector_specific_config;
     struct aws_kinesis_specific_data *connector_specific_data = instance->connector_specific_data;
 
-    while (!netdata_exit) {
+    while (!instance->engine->exit) {
         unsigned long long partition_key_seq = 0;
         struct stats *stats = &instance->stats;
 
         uv_mutex_lock(&instance->mutex);
         uv_cond_wait(&instance->cond_var, &instance->mutex);
+        if (unlikely(instance->engine->exit)) {
+            uv_mutex_unlock(&instance->mutex);
+            break;
+        }
 
         // reset the monitoring chart counters
         stats->received_bytes =
@@ -155,7 +182,7 @@ void aws_kinesis_connector_worker(void *instance_p)
                 stats->receptions++;
             }
 
-            if (unlikely(netdata_exit))
+            if (unlikely(instance->engine->exit))
                 break;
         }
 
@@ -172,7 +199,9 @@ void aws_kinesis_connector_worker(void *instance_p)
         uv_mutex_unlock(&instance->mutex);
 
 #ifdef UNIT_TESTING
-        break;
+        return;
 #endif
     }
+
+    aws_kinesis_cleanup(instance);
 }
diff --git a/exporting/clean_connectors.c b/exporting/clean_connectors.c
index 7cbdf8ac25..1cd4b42db5 100644
--- a/exporting/clean_connectors.c
+++ b/exporting/clean_connectors.c
@@ -3,23 +3,21 @@
 #include "exporting_engine.h"
 
 /**
- * Clean the instance config
+ * Clean the instance config.
  *
  * @param config an instance config structure.
  */
 static void clean_instance_config(struct instance_config *config)
 {
-    if (config->name)
-        freez((void *)config->name);
+    if(!config)
+        return;
 
-    if (config->destination)
-        freez((void *)config->destination);
+    freez((void *)config->name);
+    freez((void *)config->destination);
 
-    if (config->charts_pattern)
-        simple_pattern_free(config->charts_pattern);
+    simple_pattern_free(config->charts_pattern);
 
-    if (config->hosts_pattern)
-        simple_pattern_free(config->hosts_pattern);
+    simple_pattern_free(config->hosts_pattern);
 }
 
 /**
@@ -30,8 +28,7 @@ static void clean_instance_config(struct instance_config *config)
 void clean_instance(struct instance *instance)
 {
     clean_instance_config(&instance->config);
-    if (instance->labels)
-        buffer_free(instance->labels);
+    buffer_free(instance->labels);
 
     uv_cond_destroy(&instance->cond_var);
     // uv_mutex_destroy(&instance->mutex);
diff --git a/exporting/exporting_engine.c b/exporting/exporting_engine.c
index e8dd8d64ea..f1c4377f1f 100644
--- a/exporting/exporting_engine.c
+++ b/exporting/exporting_engine.c
@@ -5,6 +5,43 @@
 static struct engine *engine = NULL;
 
 /**
+ * Exporting Clean Engine
+ *
+ * Clean all variables allocated inside engine structure
+ *
+ * @param en a pointer to the strcuture that will be cleaned.
+ */
+static void exporting_clean_engine()
+{
+    if (!engine)
+        return;
+
+#if HAVE_KINESIS
+    if (engine->aws_sdk_initialized)
+        aws_sdk_shutdown();
+#endif
+
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+    if (engine->protocol_buffers_initialized)
+        protocol_buffers_shutdown();
+#endif
+
+    //Cleanup web api
+    prometheus_clean_server_root();
+
+    for (struct instance *instance = engine->instance_root; instance;) {
+        struct instance *current_instance = instance;
+        instance = instance->next;
+
+        clean_instance(current_instance);
+    }
+
+    freez((void *)engine->config.prefix);
+    freez((void *)engine->config.hostname);
+    freez(engine);
+}
+
+/**
  * Clean up the main exporting thread and all connector workers on Netdata exit
  *
  * @param ptr thread data.
@@ -48,18 +85,7 @@ static void exporting_main_cleanup(void *ptr)
         }
     }
 
-    for (struct instance *instance = engine->instance_root; instance;) {
-        struct instance *current_instance = instance;
-        instance = instance->next;
-        clean_instance(current_instance);
-    }
-
-    if (engine->config.prefix)
-        freez((void *)engine->config.prefix);
-    if (engine->config.hostname)
-        freez((void *)engine->config.hostname);
-    freez(engine);
-
+    exporting_clean_engine();
     static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
 }
 
diff --git a/exporting/exporting_engine.h b/exporting/exporting_engine.h
index 9869cd7e20..b82dc9084c 100644
--- a/exporting/exporting_engine.h
+++ b/exporting/exporting_engine.h
@@ -192,6 +192,7 @@ struct engine {
     time_t now;
 
     int aws_sdk_initialized;
+    int protocol_buffers_initialized;
     int mongoc_initialized;
 
     struct instance *instance_root;
@@ -251,5 +252,12 @@ static inline void disable_instance(struct instance *instance)
 }
 
 #include "exporting/prometheus/prometheus.h"
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+#include "exporting/prometheus/remote_write/remote_write.h"
+#endif
+
+#if HAVE_KINESIS
+#include "exporting/aws_kinesis/aws_kinesis.h"
+#endif
 
 #endif /* NETDATA_EXPORTING_ENGINE_H */
diff --git a/exporting/graphite/graphite.c b/exporting/graphite/graphite.c
index 88b61343fb..8404722569 100644
--- a/exporting/graphite/graphite.c
+++ b/exporting/graphite/graphite.c
@@ -12,7 +12,7 @@ int init_graphite_instance(struct instance *instance)
 {
     instance->worker = simple_connector_worker;
 
-    struct simple_connector_config *connector_specific_config = mallocz(sizeof(struct simple_connector_config));
+    struct simple_connector_config *connector_specific_config = callocz(1, sizeof(struct simple_connector_config));
     instance->config.connector_specific_config = (void *)connector_specific_config;
     connector_specific_config->default_port = 2003;
 
diff --git a/exporting/json/json.c b/exporting/json/json.c
index 5534791f32..57bf6a76c6 100644
--- a/exporting/json/json.c
+++ b/exporting/json/json.c
@@ -12,7 +12,7 @@ int init_json_instance(struct instance *instance)
 {
     instance->worker = simple_connector_worker;
 
-    struct simple_connector_config *connector_specific_config = mallocz(sizeof(struct simple_connector_config));
+    struct simple_connector_config *connector_specific_config = callocz(1, sizeof(struct simple_connector_config));
     instance->config.connector_specific_config = (void *)connector_specific_config;
     connector_specific_config->default_port = 5448;
 
diff --git a/exporting/opentsdb/opentsdb.c b/exporting/opentsdb/opentsdb.c
index 3bfb145517..cf55713ed8 100644
--- a/exporting/opentsdb/opentsdb.c
+++ b/exporting/opentsdb/opentsdb.c
@@ -12,7 +12,7 @@ int init_opentsdb_telnet_instance(struct instance *instance)
 {
     instance->worker = simple_connector_worker;
 
-    struct simple_connector_config *connector_specific_config = mallocz(sizeof(struct simple_connector_config));
+    struct simple_connector_config *connector_specific_config = callocz(1, sizeof(struct simple_connector_config));
     instance->config.connector_specific_config = (void *)connector_specific_config;
     connector_specific_config->default_port = 4242;
 
@@ -53,7 +53,7 @@ int init_opentsdb_http_instance(struct instance *instance)
 {
     instance->worker = simple_connector_worker;
 
-    struct simple_connector_config *connector_specific_config = mallocz(sizeof(struct simple_connector_config));
+    struct simple_connector_config *connector_specific_config = callocz(1, sizeof(struct simple_connector_config));
     instance->config.connector_specific_config = (void *)connector_specific_config;
     connector_specific_config->default_port = 4242;
 
diff --git a/exporting/prometheus/prometheus.c b/exporting/prometheus/prometheus.c
index 6cfe279013..17cf902ebb 100644
--- a/exporting/prometheus/prometheus.c
+++ b/exporting/prometheus/prometheus.c
@@ -69,6 +69,30 @@ static struct prometheus_server {
     struct prometheus_server *next;
 } *prometheus_server_root = NULL;
 
+static netdata_mutex_t prometheus_server_root_mutex = NETDATA_MUTEX_INITIALIZER;
+
+/**
+ * Clean server root local structure
+ */
+void prometheus_clean_server_root()
+{
+    if (prometheus_server_root) {
+        netdata_mutex_lock(&prometheus_server_root_mutex);
+
+        struct prometheus_server *ps;
+        for (ps = prometheus_server_root; ps; ) {
+            struct prometheus_server *current = ps;
+            ps = ps->next;
+            if(current->server)
+                freez((void *)current->server);
+
+            freez(current);
+        }
+        prometheus_server_root = NULL;
+        netdata_mutex_unlock(&prometheus_server_root_mutex);
+    }
+}
+
 /**
  * Get the last time when a Prometheus server scraped the Netdata Prometheus exporter.
  *
@@ -82,8 +106,6 @@ static inline time_t prometheus_server_last_access(const char *server, RRDHOST *
 #ifdef UNIT_TESTING
     return 0;
 #endif
-    static netdata_mutex_t prometheus_server_root_mutex = NETDATA_MUTEX_INITIALIZER;
-
     uint32_t hash = simple_hash(server);
 
     netdata_mutex_lock(&prometheus_server_root_mutex);
diff --git a/exporting/prometheus/prometheus.h b/exporting/prometheus/prometheus.h
index 85bcc7a7f8..2f0845ce98 100644
--- a/exporting/prometheus/prometheus.h
+++ b/exporting/prometheus/prometheus.h
@@ -36,4 +36,6 @@ char *prometheus_units_copy(char *d, const char *s, size_t usable, int showoldun
 
 void format_host_labels_prometheus(struct instance *instance, RRDHOST *host);
 
+extern void prometheus_clean_server_root();
+
 #endif //NETDATA_EXPORTING_PROMETHEUS_H
diff --git a/exporting/prometheus/remote_write/remote_write.c b/exporting/prometheus/remote_write/remote_write.c
index 939b98fb73..2a94fc44e7 100644
--- a/exporting/prometheus/remote_write/remote_write.c
+++ b/exporting/prometheus/remote_write/remote_write.c
@@ -83,6 +83,20 @@ int process_prometheus_remote_write_response(BUFFER *buffer, struct instance *in
 }
 
 /**
+ * Release specific data allocated.
+ *
+ * @param instance an instance data structure.
+ */
+void clean_prometheus_remote_write(struct instance *instance)
+{
+    freez(instance->connector_specific_data);
+
+    struct prometheus_remote_write_specific_config *connector_specific_config =
+        instance->config.connector_specific_config;
+    freez(connector_specific_config->remote_write_path);
+}
+
+/**
  * Initialize Prometheus Remote Write connector instance
  *
  * @param instance an instance data structure.
@@ -117,6 +131,8 @@ int init_prometheus_remote_write_instance(struct instance *instance)
 
     connector_specific_data->write_request = init_write_request();
 
+    instance->engine->protocol_buffers_initialized = 1;
+
     return 0;
 }
 
diff --git a/exporting/prometheus/remote_write/remote_write.h b/exporting/prometheus/remote_write/remote_write.h
index 050eb0c1a8..3bd212414a 100644
--- a/exporting/prometheus/remote_write/remote_write.h
+++ b/exporting/prometheus/remote_write/remote_write.h
@@ -8,6 +8,7 @@
 #include "remote_write_request.h"
 
 int init_prometheus_remote_write_instance(struct instance *instance);
+extern void clean_prometheus_remote_write(struct instance *instance);
 
 int format_host_prometheus_remote_write(struct instance *instance, RRDHOST *host);
 int format_chart_prometheus_remote_write(struct instance *instance, RRDSET *st);
diff --git a/exporting/send_data.c b/exporting/send_data.c
index 3a749cdbb8..8c579d8fce 100644
--- a/exporting/send_data.c
+++ b/exporting/send_data.c
@@ -149,7 +149,8 @@ void simple_connector_cleanup(struct instance *instance)
 {
     info("EXPORTING: cleaning up instance %s ...", instance->config.name);
 
-    // TODO free allocated resources
+    buffer_free(instance->buffer);
+    freez(instance->config.connector_specific_config);
 
     info("EXPORTING: instance %s exited", instance->config.name);
     instance->exited = 1;
@@ -218,7 +219,10 @@ void simple_connector_worker(void *instance_p)
         uv_mutex_lock(&instance->mutex);
         uv_cond_wait(&instance->cond_var, &instance->mutex);
 
-        if(unlikely(instance->engine->exit)) break;
+        if (unlikely(instance->engine->exit)) {
+            uv_mutex_unlock(&instance->mutex);
+            break;
+        }
 
         if (likely(sock != -1)) {
             simple_connector_send_buffer(&sock, &failures, instance);
@@ -252,9 +256,14 @@ void simple_connector_worker(void *instance_p)
         uv_mutex_unlock(&instance->mutex);
 
 #ifdef UNIT_TESTING
-        break;
+        return;
 #endif
     }
 
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
+    if (instance->config.type == EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE)
+        clean_prometheus_remote_write(instance);
+#endif
+
     simple_connector_cleanup(instance);
 }