Exporting cleanup (#9098)
Cleanup allocated variables for the majority of the databases.
thiagoftsm committed
May 25, 2020 at 14:27 UTC
4fb41597da6961813679c1b795793600ddc0a402
12 files changed
+145
-35
exporting/aws_kinesis/aws_kinesis.c
+32
-3
@@ -2,6 +2,29 @@
2
3
#include "aws_kinesis.h"
4
5
+/**
6
+ * Clean AWS Kinesis *
7
+ */
8
+void aws_kinesis_cleanup(struct instance *instance)
9
+{
10
+ info("EXPORTING: cleaning up instance %s ...", instance->config.name);
11
+ kinesis_shutdown(instance->connector_specific_data);
12
+
13
+ freez(instance->connector_specific_data);
14
+
15
+ struct aws_kinesis_specific_config *connector_specific_config = instance->config.connector_specific_config;
16
+ if (connector_specific_config) {
17
+ freez(connector_specific_config->auth_key_id);
18
+ freez(connector_specific_config->secure_key);
19
+ freez(connector_specific_config->stream_name);
20
+
21
+ freez(connector_specific_config);
22
+ }
23
+
24
+ info("EXPORTING: instance %s exited", instance->config.name);
25
+ instance->exited = 1;
26
+}
27
+
28
/**
29
* Initialize AWS Kinesis connector instance
30
*
@@ -68,12 +91,16 @@ void aws_kinesis_connector_worker(void *instance_p)
91
struct aws_kinesis_specific_config *connector_specific_config = instance->config.connector_specific_config;
92
struct aws_kinesis_specific_data *connector_specific_data = instance->connector_specific_data;
93
71
- while (!netdata_exit) {
94
+ while (!instance->engine->exit) {
95
unsigned long long partition_key_seq = 0;
96
struct stats *stats = &instance->stats;
97
98
uv_mutex_lock(&instance->mutex);
99
uv_cond_wait(&instance->cond_var, &instance->mutex);
100
+ if (unlikely(instance->engine->exit)) {
101
+ uv_mutex_unlock(&instance->mutex);
102
+ break;
103
+ }
104
105
// reset the monitoring chart counters
106
stats->received_bytes =
@@ -155,7 +182,7 @@ void aws_kinesis_connector_worker(void *instance_p)
182
stats->receptions++;
183
}
184
158
- if (unlikely(netdata_exit))
185
+ if (unlikely(instance->engine->exit))
186
break;
187
}
188
@@ -172,7 +199,9 @@ void aws_kinesis_connector_worker(void *instance_p)
199
uv_mutex_unlock(&instance->mutex);
200
201
#ifdef UNIT_TESTING
175
- break;
202
+ return;
203
#endif
204
}
205
+
206
+ aws_kinesis_cleanup(instance);
207
}
exporting/clean_connectors.c
+8
-11
@@ -3,23 +3,21 @@
3
#include "exporting_engine.h"
4
5
/**
6
- * Clean the instance config
6
+ * Clean the instance config.
7
*
8
* @param config an instance config structure.
9
*/
10
static void clean_instance_config(struct instance_config *config)
11
{
12
- if (config->name)
13
- freez((void *)config->name);
12
+ if(!config)
13
+ return;
14
15
- if (config->destination)
16
- freez((void *)config->destination);
15
+ freez((void *)config->name);
16
+ freez((void *)config->destination);
17
18
- if (config->charts_pattern)
19
- simple_pattern_free(config->charts_pattern);
18
+ simple_pattern_free(config->charts_pattern);
19
21
- if (config->hosts_pattern)
22
- simple_pattern_free(config->hosts_pattern);
20
+ simple_pattern_free(config->hosts_pattern);
21
}
22
23
/**
@@ -30,8 +28,7 @@ static void clean_instance_config(struct instance_config *config)
28
void clean_instance(struct instance *instance)
29
{
30
clean_instance_config(&instance->config);
33
- if (instance->labels)
34
- buffer_free(instance->labels);
31
+ buffer_free(instance->labels);
32
33
uv_cond_destroy(&instance->cond_var);
34
// uv_mutex_destroy(&instance->mutex);
exporting/exporting_engine.c
+38
-12
@@ -4,6 +4,43 @@
4
5
static struct engine *engine = NULL;
6
7
+/**
8
+ * Exporting Clean Engine
9
+ *
10
+ * Clean all variables allocated inside engine structure
11
+ *
12
+ * @param en a pointer to the strcuture that will be cleaned.
13
+ */
14
+static void exporting_clean_engine()
15
+{
16
+ if (!engine)
17
+ return;
18
+
19
+#if HAVE_KINESIS
20
+ if (engine->aws_sdk_initialized)
21
+ aws_sdk_shutdown();
22
+#endif
23
+
24
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
25
+ if (engine->protocol_buffers_initialized)
26
+ protocol_buffers_shutdown();
27
+#endif
28
+
29
+ //Cleanup web api
30
+ prometheus_clean_server_root();
31
+
32
+ for (struct instance *instance = engine->instance_root; instance;) {
33
+ struct instance *current_instance = instance;
34
+ instance = instance->next;
35
+
36
+ clean_instance(current_instance);
37
+ }
38
+
39
+ freez((void *)engine->config.prefix);
40
+ freez((void *)engine->config.hostname);
41
+ freez(engine);
42
+}
43
+
44
/**
45
* Clean up the main exporting thread and all connector workers on Netdata exit
46
*
@@ -48,18 +85,7 @@ static void exporting_main_cleanup(void *ptr)
85
}
86
}
87
51
- for (struct instance *instance = engine->instance_root; instance;) {
52
- struct instance *current_instance = instance;
53
- instance = instance->next;
54
- clean_instance(current_instance);
55
- }
56
-
57
- if (engine->config.prefix)
58
- freez((void *)engine->config.prefix);
59
- if (engine->config.hostname)
60
- freez((void *)engine->config.hostname);
61
- freez(engine);
62
-
88
+ exporting_clean_engine();
89
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
90
}
91
exporting/exporting_engine.h
+8
@@ -192,6 +192,7 @@ struct engine {
192
time_t now;
193
194
int aws_sdk_initialized;
195
+ int protocol_buffers_initialized;
196
int mongoc_initialized;
197
198
struct instance *instance_root;
@@ -251,5 +252,12 @@ static inline void disable_instance(struct instance *instance)
252
}
253
254
#include "exporting/prometheus/prometheus.h"
255
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
256
+#include "exporting/prometheus/remote_write/remote_write.h"
257
+#endif
258
+
259
+#if HAVE_KINESIS
260
+#include "exporting/aws_kinesis/aws_kinesis.h"
261
+#endif
262
263
#endif /* NETDATA_EXPORTING_ENGINE_H */
exporting/graphite/graphite.c
+1
-1
@@ -12,7 +12,7 @@ int init_graphite_instance(struct instance *instance)
12
{
13
instance->worker = simple_connector_worker;
14
15
- struct simple_connector_config *connector_specific_config = mallocz(sizeof(struct simple_connector_config));
15
+ struct simple_connector_config *connector_specific_config = callocz(1, sizeof(struct simple_connector_config));
16
instance->config.connector_specific_config = (void *)connector_specific_config;
17
connector_specific_config->default_port = 2003;
18
exporting/json/json.c
+1
-1
@@ -12,7 +12,7 @@ int init_json_instance(struct instance *instance)
12
{
13
instance->worker = simple_connector_worker;
14
15
- struct simple_connector_config *connector_specific_config = mallocz(sizeof(struct simple_connector_config));
15
+ struct simple_connector_config *connector_specific_config = callocz(1, sizeof(struct simple_connector_config));
16
instance->config.connector_specific_config = (void *)connector_specific_config;
17
connector_specific_config->default_port = 5448;
18
exporting/opentsdb/opentsdb.c
+2
-2
@@ -12,7 +12,7 @@ int init_opentsdb_telnet_instance(struct instance *instance)
12
{
13
instance->worker = simple_connector_worker;
14
15
- struct simple_connector_config *connector_specific_config = mallocz(sizeof(struct simple_connector_config));
15
+ struct simple_connector_config *connector_specific_config = callocz(1, sizeof(struct simple_connector_config));
16
instance->config.connector_specific_config = (void *)connector_specific_config;
17
connector_specific_config->default_port = 4242;
18
@@ -53,7 +53,7 @@ int init_opentsdb_http_instance(struct instance *instance)
53
{
54
instance->worker = simple_connector_worker;
55
56
- struct simple_connector_config *connector_specific_config = mallocz(sizeof(struct simple_connector_config));
56
+ struct simple_connector_config *connector_specific_config = callocz(1, sizeof(struct simple_connector_config));
57
instance->config.connector_specific_config = (void *)connector_specific_config;
58
connector_specific_config->default_port = 4242;
59
exporting/prometheus/prometheus.c
+24
-2
@@ -69,6 +69,30 @@ static struct prometheus_server {
69
struct prometheus_server *next;
70
} *prometheus_server_root = NULL;
71
72
+static netdata_mutex_t prometheus_server_root_mutex = NETDATA_MUTEX_INITIALIZER;
73
+
74
+/**
75
+ * Clean server root local structure
76
+ */
77
+void prometheus_clean_server_root()
78
+{
79
+ if (prometheus_server_root) {
80
+ netdata_mutex_lock(&prometheus_server_root_mutex);
81
+
82
+ struct prometheus_server *ps;
83
+ for (ps = prometheus_server_root; ps; ) {
84
+ struct prometheus_server *current = ps;
85
+ ps = ps->next;
86
+ if(current->server)
87
+ freez((void *)current->server);
88
+
89
+ freez(current);
90
+ }
91
+ prometheus_server_root = NULL;
92
+ netdata_mutex_unlock(&prometheus_server_root_mutex);
93
+ }
94
+}
95
+
96
/**
97
* Get the last time when a Prometheus server scraped the Netdata Prometheus exporter.
98
*
@@ -82,8 +106,6 @@ static inline time_t prometheus_server_last_access(const char *server, RRDHOST *
106
#ifdef UNIT_TESTING
107
return 0;
108
#endif
85
- static netdata_mutex_t prometheus_server_root_mutex = NETDATA_MUTEX_INITIALIZER;
86
-
109
uint32_t hash = simple_hash(server);
110
111
netdata_mutex_lock(&prometheus_server_root_mutex);
exporting/prometheus/prometheus.h
+2
@@ -36,4 +36,6 @@ char *prometheus_units_copy(char *d, const char *s, size_t usable, int showoldun
36
37
void format_host_labels_prometheus(struct instance *instance, RRDHOST *host);
38
39
+extern void prometheus_clean_server_root();
40
+
41
#endif //NETDATA_EXPORTING_PROMETHEUS_H
exporting/prometheus/remote_write/remote_write.c
+16
@@ -82,6 +82,20 @@ int process_prometheus_remote_write_response(BUFFER *buffer, struct instance *in
82
return exporting_discard_response(buffer, instance);
83
}
84
85
+/**
86
+ * Release specific data allocated.
87
+ *
88
+ * @param instance an instance data structure.
89
+ */
90
+void clean_prometheus_remote_write(struct instance *instance)
91
+{
92
+ freez(instance->connector_specific_data);
93
+
94
+ struct prometheus_remote_write_specific_config *connector_specific_config =
95
+ instance->config.connector_specific_config;
96
+ freez(connector_specific_config->remote_write_path);
97
+}
98
+
99
/**
100
* Initialize Prometheus Remote Write connector instance
101
*
@@ -117,6 +131,8 @@ int init_prometheus_remote_write_instance(struct instance *instance)
131
132
connector_specific_data->write_request = init_write_request();
133
134
+ instance->engine->protocol_buffers_initialized = 1;
135
+
136
return 0;
137
}
138
exporting/prometheus/remote_write/remote_write.h
+1
@@ -8,6 +8,7 @@
8
#include "remote_write_request.h"
9
10
int init_prometheus_remote_write_instance(struct instance *instance);
11
+extern void clean_prometheus_remote_write(struct instance *instance);
12
13
int format_host_prometheus_remote_write(struct instance *instance, RRDHOST *host);
14
int format_chart_prometheus_remote_write(struct instance *instance, RRDSET *st);
exporting/send_data.c
+12
-3
@@ -149,7 +149,8 @@ void simple_connector_cleanup(struct instance *instance)
149
{
150
info("EXPORTING: cleaning up instance %s ...", instance->config.name);
151
152
- // TODO free allocated resources
152
+ buffer_free(instance->buffer);
153
+ freez(instance->config.connector_specific_config);
154
155
info("EXPORTING: instance %s exited", instance->config.name);
156
instance->exited = 1;
@@ -218,7 +219,10 @@ void simple_connector_worker(void *instance_p)
219
uv_mutex_lock(&instance->mutex);
220
uv_cond_wait(&instance->cond_var, &instance->mutex);
221
221
- if(unlikely(instance->engine->exit)) break;
222
+ if (unlikely(instance->engine->exit)) {
223
+ uv_mutex_unlock(&instance->mutex);
224
+ break;
225
+ }
226
227
if (likely(sock != -1)) {
228
simple_connector_send_buffer(&sock, &failures, instance);
@@ -252,9 +256,14 @@ void simple_connector_worker(void *instance_p)
256
uv_mutex_unlock(&instance->mutex);
257
258
#ifdef UNIT_TESTING
255
- break;
259
+ return;
260
#endif
261
}
262
263
+#if ENABLE_PROMETHEUS_REMOTE_WRITE
264
+ if (instance->config.type == EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE)
265
+ clean_prometheus_remote_write(instance);
266
+#endif
267
+
268
simple_connector_cleanup(instance);
269
}