@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
c 223 lines 6.18 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "exporting_engine.h"
4
5 static struct engine *engine = NULL;
6
7 void analytics_exporting_connectors_ssl(BUFFER *b)
8 {
9 if (netdata_ssl_exporting_ctx) {
10 for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
11 struct simple_connector_data *connector_specific_data = instance->connector_specific_data;
12 if (SSL_connection(&connector_specific_data->ssl)) {
13 buffer_strcat(b, "exporting");
14 break;
15 }
16 }
17 }
18 buffer_strcat(b, "|");
19 }
20
21 void analytics_exporting_connectors(BUFFER *b)
22 {
23 if (!engine)
24 return;
25
26 uint8_t count = 0;
27
28 for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
29 if (count)
30 buffer_strcat(b, "|");
31
32 switch (instance->config.type) {
33 case EXPORTING_CONNECTOR_TYPE_GRAPHITE:
34 buffer_strcat(b, "Graphite");
35 break;
36 case EXPORTING_CONNECTOR_TYPE_GRAPHITE_HTTP:
37 buffer_strcat(b, "GraphiteHTTP");
38 break;
39 case EXPORTING_CONNECTOR_TYPE_JSON:
40 buffer_strcat(b, "JSON");
41 break;
42 case EXPORTING_CONNECTOR_TYPE_JSON_HTTP:
43 buffer_strcat(b, "JSONHTTP");
44 break;
45 case EXPORTING_CONNECTOR_TYPE_OPENTSDB:
46 buffer_strcat(b, "OpenTSDB");
47 break;
48 case EXPORTING_CONNECTOR_TYPE_OPENTSDB_HTTP:
49 buffer_strcat(b, "OpenTSDBHTTP");
50 break;
51 case EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE:
52 #ifdef ENABLE_PROMETHEUS_REMOTE_WRITE
53 buffer_strcat(b, "PrometheusRemoteWrite");
54 #endif
55 break;
56 case EXPORTING_CONNECTOR_TYPE_KINESIS:
57 #if HAVE_KINESIS
58 buffer_strcat(b, "Kinesis");
59 #endif
60 break;
61 case EXPORTING_CONNECTOR_TYPE_PUBSUB:
62 #if ENABLE_EXPORTING_PUBSUB
63 buffer_strcat(b, "Pubsub");
64 #endif
65 break;
66 case EXPORTING_CONNECTOR_TYPE_MONGODB:
67 #ifdef HAVE_MONGOC
68 buffer_strcat(b, "MongoDB");
69 #endif
70 break;
71 default:
72 buffer_strcat(b, "Unknown");
73 }
74
75 count++;
76 }
77 }
78
79 /**
80 * Exporting Clean Engine
81 *
82 * Clean all variables allocated inside engine structure
83 *
84 * @param en a pointer to the structure that will be cleaned.
85 */
86 static void exporting_clean_engine()
87 {
88 if (!engine)
89 return;
90
91 #if HAVE_KINESIS
92 if (engine->aws_sdk_initialized)
93 aws_sdk_shutdown();
94 #endif
95
96 #ifdef ENABLE_PROMETHEUS_REMOTE_WRITE
97 if (engine->protocol_buffers_initialized)
98 protocol_buffers_shutdown();
99 #endif
100
101 //Cleanup web api
102 prometheus_clean_server_root();
103
104 for (struct instance *instance = engine->instance_root; instance;) {
105 struct instance *current_instance = instance;
106 instance = instance->next;
107
108 clean_instance(current_instance);
109 freez(current_instance);
110 }
111
112 freez((void *)engine->config.hostname);
113 freez(engine);
114 }
115
116 /**
117 * Clean up the main exporting thread and all connector workers on Netdata exit
118 *
119 * @param ptr thread data.
120 */
121 static void exporting_main_cleanup(void *pptr)
122 {
123 struct netdata_static_thread *static_thread = CLEANUP_FUNCTION_GET_PTR(pptr);
124 if(!static_thread) return;
125
126 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
127
128 if (!engine) {
129 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
130 return;
131 }
132
133 engine->exit = 1;
134
135 size_t all = 0, exited = 0;
136
137 for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
138 all++;
139
140 if (!instance->exited) {
141 netdata_log_info("EXPORTING: signaling worker '%s' to stop...", instance->config.name);
142 // Lock the mutex before signaling the condition variable
143 netdata_mutex_lock(&instance->mutex);
144 instance->data_is_ready = 1;
145 netdata_cond_signal(&instance->cond_var);
146 netdata_mutex_unlock(&instance->mutex);
147 }
148 else
149 netdata_log_info("EXPORTING: found worker '%s' already stopped", instance->config.name);
150 }
151
152 size_t iterations = 0;
153 while (exited < all) {
154 iterations++;
155 microsleep(10 * USEC_PER_MS);
156
157 exited = 0;
158 for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
159 if (instance->exited) {
160 exited++;
161
162 if(instance->thread) {
163 nd_thread_join(instance->thread);
164 instance->thread = NULL;
165 }
166 }
167 else if(iterations % 100 == 0)
168 netdata_log_info("EXPORTING: still waiting for worker '%s' to exit...", instance->config.name);
169 }
170 }
171
172 // this must be called once all the worker thread have exited
173 exporting_clean_engine();
174
175 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
176 }
177
178 /**
179 * Exporting engine main
180 *
181 * The main thread used to control the exporting engine.
182 *
183 * @param ptr a pointer to netdata_static_structure.
184 *
185 */
186 void exporting_main(void *ptr)
187 {
188 CLEANUP_FUNCTION_REGISTER(exporting_main_cleanup) cleanup_ptr = ptr;
189
190 engine = read_exporting_config();
191 if (!engine) {
192 netdata_log_info("EXPORTING: no exporting connectors configured");
193 return;
194 }
195
196 if (init_connectors(engine) != 0) {
197 netdata_log_error("EXPORTING: cannot initialize exporting connectors");
198 return;
199 }
200
201 RRDSET *st_main_rusage = NULL;
202 RRDDIM *rd_main_user = NULL;
203 RRDDIM *rd_main_system = NULL;
204 create_main_rusage_chart(&st_main_rusage, &rd_main_user, &rd_main_system);
205
206 heartbeat_t hb;
207 heartbeat_init(&hb, localhost->rrd_update_every * USEC_PER_SEC);
208
209 while (service_running(SERVICE_EXPORTERS)) {
210 heartbeat_next(&hb);
211 engine->now = now_realtime_sec();
212
213 if (mark_scheduled_instances(engine))
214 prepare_buffers(engine);
215
216 send_main_rusage(st_main_rusage, rd_main_user, rd_main_system);
217
218 #ifdef UNIT_TESTING
219 break;
220 #endif
221 }
222 service_exits();
223 }