Fix error handling in exporting connector (#8910)
Vladimir Kobal committed
May 14, 2020 at 11:42 UTC
a606a27f164b1c704d850c838a7b89d6c6e0c17c
8 files changed
+47
-107
CMakeLists.txt
-1
@@ -1152,7 +1152,6 @@ endif()
1152
-Wl,--wrap=rrdset_is_exportable
1153
-Wl,--wrap=exporting_calculate_value_from_stored_data
1154
-Wl,--wrap=prepare_buffers
1155
- -Wl,--wrap=notify_workers
1155
-Wl,--wrap=send_internal_metrics
1156
-Wl,--wrap=now_realtime_sec
1157
-Wl,--wrap=uv_thread_set_name_np
Makefile.am
-1
@@ -878,7 +878,6 @@ if ENABLE_UNITTESTS
878
-Wl,--wrap=rrdset_is_exportable \
879
-Wl,--wrap=exporting_calculate_value_from_stored_data \
880
-Wl,--wrap=prepare_buffers \
881
- -Wl,--wrap=notify_workers \
881
-Wl,--wrap=send_internal_metrics \
882
-Wl,--wrap=now_realtime_sec \
883
-Wl,--wrap=uv_thread_set_name_np \
exporting/exporting_engine.c
+2
-11
@@ -48,17 +48,8 @@ void *exporting_main(void *ptr)
48
heartbeat_next(&hb, step_ut);
49
engine->now = now_realtime_sec();
50
51
- if (mark_scheduled_instances(engine)) {
52
- if (prepare_buffers(engine) != 0) {
53
- error("EXPORTING: cannot prepare data to send");
54
- break;
55
- }
56
- }
57
-
58
- if (notify_workers(engine) != 0) {
59
- error("EXPORTING: cannot communicate with exporting connector instance working threads");
60
- break;
61
- }
51
+ if (mark_scheduled_instances(engine))
52
+ prepare_buffers(engine);
53
54
send_main_rusage(st_main_rusage, rd_main_user, rd_main_system);
55
exporting/exporting_engine.h
+17
-9
@@ -151,6 +151,7 @@ struct instance {
151
struct stats stats;
152
153
int scheduled;
154
+ int disabled;
155
int skip_host;
156
int skip_chart;
157
@@ -203,8 +204,7 @@ EXPORTING_CONNECTOR_TYPE exporting_select_type(const char *type);
204
int init_connectors(struct engine *engine);
205
206
int mark_scheduled_instances(struct engine *engine);
206
-int prepare_buffers(struct engine *engine);
207
-int notify_workers(struct engine *engine);
207
+void prepare_buffers(struct engine *engine);
208
209
size_t exporting_name_copy(char *dst, const char *src, size_t max_len);
210
@@ -216,13 +216,13 @@ calculated_number exporting_calculate_value_from_stored_data(
216
RRDDIM *rd,
217
time_t *last_timestamp);
218
219
-int start_batch_formatting(struct engine *engine);
220
-int start_host_formatting(struct engine *engine, RRDHOST *host);
221
-int start_chart_formatting(struct engine *engine, RRDSET *st);
222
-int metric_formatting(struct engine *engine, RRDDIM *rd);
223
-int end_chart_formatting(struct engine *engine, RRDSET *st);
224
-int end_host_formatting(struct engine *engine, RRDHOST *host);
225
-int end_batch_formatting(struct engine *engine);
219
+void start_batch_formatting(struct engine *engine);
220
+void start_host_formatting(struct engine *engine, RRDHOST *host);
221
+void start_chart_formatting(struct engine *engine, RRDSET *st);
222
+void metric_formatting(struct engine *engine, RRDDIM *rd);
223
+void end_chart_formatting(struct engine *engine, RRDSET *st);
224
+void end_host_formatting(struct engine *engine, RRDHOST *host);
225
+void end_batch_formatting(struct engine *engine);
226
int flush_host_labels(struct instance *instance, RRDHOST *host);
227
int simple_connector_update_buffered_bytes(struct instance *instance);
228
@@ -235,6 +235,14 @@ void create_main_rusage_chart(RRDSET **st_rusage, RRDDIM **rd_user, RRDDIM **rd_
235
void send_main_rusage(RRDSET *st_rusage, RRDDIM *rd_user, RRDDIM *rd_system);
236
void send_internal_metrics(struct instance *instance);
237
238
+static inline void disable_instance(struct instance *instance)
239
+{
240
+ instance->disabled = 1;
241
+ instance->scheduled = 0;
242
+ uv_mutex_unlock(&instance->mutex);
243
+ error("EXPORTING: Instance %s disabled", instance->config.name);
244
+}
245
+
246
#include "exporting/prometheus/prometheus.h"
247
248
#endif /* NETDATA_EXPORTING_ENGINE_H */
exporting/process_data.c
+28
-72
@@ -43,7 +43,7 @@ int mark_scheduled_instances(struct engine *engine)
43
int instances_were_scheduled = 0;
44
45
for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
46
- if (engine->now % instance->config.update_every < localhost->rrd_update_every) {
46
+ if (!instance->disabled && (engine->now % instance->config.update_every < localhost->rrd_update_every)) {
47
instance->scheduled = 1;
48
instances_were_scheduled = 1;
49
instance->before = engine->now;
@@ -160,21 +160,18 @@ calculated_number exporting_calculate_value_from_stored_data(
160
* Start batch formatting for every connector instance's buffer
161
*
162
* @param engine an engine data structure.
163
- * @return Returns 0 on success, 1 on failure.
163
*/
165
-int start_batch_formatting(struct engine *engine)
164
+void start_batch_formatting(struct engine *engine)
165
{
166
for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
167
if (instance->scheduled) {
168
uv_mutex_lock(&instance->mutex);
169
if (instance->start_batch_formatting && instance->start_batch_formatting(instance) != 0) {
170
error("EXPORTING: cannot start batch formatting for %s", instance->config.name);
172
- return 1;
171
+ disable_instance(instance);
172
}
173
}
174
}
176
-
177
- return 0;
175
}
176
177
/**
@@ -182,24 +179,21 @@ int start_batch_formatting(struct engine *engine)
179
*
180
* @param engine an engine data structure.
181
* @param host a data collecting host.
185
- * @return Returns 0 on success, 1 on failure.
182
*/
187
-int start_host_formatting(struct engine *engine, RRDHOST *host)
183
+void start_host_formatting(struct engine *engine, RRDHOST *host)
184
{
185
for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
186
if (instance->scheduled) {
187
if (rrdhost_is_exportable(instance, host)) {
188
if (instance->start_host_formatting && instance->start_host_formatting(instance, host) != 0) {
189
error("EXPORTING: cannot start host formatting for %s", instance->config.name);
194
- return 1;
190
+ disable_instance(instance);
191
}
192
} else {
193
instance->skip_host = 1;
194
}
195
}
196
}
201
-
202
- return 0;
197
}
198
199
/**
@@ -207,24 +201,21 @@ int start_host_formatting(struct engine *engine, RRDHOST *host)
201
*
202
* @param engine an engine data structure.
203
* @param st a chart.
210
- * @return Returns 0 on success, 1 on failure.
204
*/
212
-int start_chart_formatting(struct engine *engine, RRDSET *st)
205
+void start_chart_formatting(struct engine *engine, RRDSET *st)
206
{
207
for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
208
if (instance->scheduled && !instance->skip_host) {
209
if (rrdset_is_exportable(instance, st)) {
210
if (instance->start_chart_formatting && instance->start_chart_formatting(instance, st) != 0) {
211
error("EXPORTING: cannot start chart formatting for %s", instance->config.name);
219
- return 1;
212
+ disable_instance(instance);
213
}
214
} else {
215
instance->skip_chart = 1;
216
}
217
}
218
}
226
-
227
- return 0;
219
}
220
221
/**
@@ -232,21 +223,19 @@ int start_chart_formatting(struct engine *engine, RRDSET *st)
223
*
224
* @param engine an engine data structure.
225
* @param rd a dimension(metric) in the Netdata database.
235
- * @return Returns 0 on success, 1 on failure.
226
*/
237
-int metric_formatting(struct engine *engine, RRDDIM *rd)
227
+void metric_formatting(struct engine *engine, RRDDIM *rd)
228
{
229
for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
230
if (instance->scheduled && !instance->skip_host && !instance->skip_chart) {
231
if (instance->metric_formatting && instance->metric_formatting(instance, rd) != 0) {
232
error("EXPORTING: cannot format metric for %s", instance->config.name);
243
- return 1;
233
+ disable_instance(instance);
234
+ continue;
235
}
236
instance->stats.buffered_metrics++;
237
}
238
}
248
-
249
- return 0;
239
}
240
241
/**
@@ -254,21 +243,19 @@ int metric_formatting(struct engine *engine, RRDDIM *rd)
243
*
244
* @param engine an engine data structure.
245
* @param a chart.
257
- * @return Returns 0 on success, 1 on failure.
246
*/
259
-int end_chart_formatting(struct engine *engine, RRDSET *st)
247
+void end_chart_formatting(struct engine *engine, RRDSET *st)
248
{
249
for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
250
if (instance->scheduled && !instance->skip_host && !instance->skip_chart) {
251
if (instance->end_chart_formatting && instance->end_chart_formatting(instance, st) != 0) {
252
error("EXPORTING: cannot end chart formatting for %s", instance->config.name);
265
- return 1;
253
+ disable_instance(instance);
254
+ continue;
255
}
256
}
257
instance->skip_chart = 0;
258
}
270
-
271
- return 0;
259
}
260
261
/**
@@ -276,36 +263,34 @@ int end_chart_formatting(struct engine *engine, RRDSET *st)
263
*
264
* @param engine an engine data structure.
265
* @param host a data collecting host.
279
- * @return Returns 0 on success, 1 on failure.
266
*/
281
-int end_host_formatting(struct engine *engine, RRDHOST *host)
267
+void end_host_formatting(struct engine *engine, RRDHOST *host)
268
{
269
for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
270
if (instance->scheduled && !instance->skip_host) {
271
if (instance->end_host_formatting && instance->end_host_formatting(instance, host) != 0) {
272
error("EXPORTING: cannot end host formatting for %s", instance->config.name);
287
- return 1;
273
+ disable_instance(instance);
274
+ continue;
275
}
276
}
277
instance->skip_host = 0;
278
}
292
-
293
- return 0;
279
}
280
281
/**
282
* End batch formatting for every connector instance's buffer
283
*
284
* @param engine an engine data structure.
300
- * @return Returns 0 on success, 1 on failure.
285
*/
302
-int end_batch_formatting(struct engine *engine)
286
+void end_batch_formatting(struct engine *engine)
287
{
288
for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
289
if (instance->scheduled) {
290
if (instance->end_batch_formatting && instance->end_batch_formatting(instance) != 0) {
291
error("EXPORTING: cannot end batch formatting for %s", instance->config.name);
308
- return 1;
292
+ disable_instance(instance);
293
+ continue;
294
}
295
uv_mutex_unlock(&instance->mutex);
296
uv_cond_signal(&instance->cond_var);
@@ -314,8 +299,6 @@ int end_batch_formatting(struct engine *engine)
299
instance->after = instance->before;
300
}
301
}
317
-
318
- return 0;
302
}
303
304
/**
@@ -325,51 +308,39 @@ int end_batch_formatting(struct engine *engine)
308
* configured rules.
309
*
310
* @param engine an engine data structure.
328
- * @return Returns 0 on success, 1 on failure.
311
*/
330
-int prepare_buffers(struct engine *engine)
312
+void prepare_buffers(struct engine *engine)
313
{
332
- if (start_batch_formatting(engine) != 0)
333
- return 1;
334
-
314
netdata_thread_disable_cancelability();
315
+ start_batch_formatting(engine);
316
+
317
rrd_rdlock();
318
RRDHOST *host;
319
rrdhost_foreach_read(host)
320
{
321
rrdhost_rdlock(host);
341
- if (start_host_formatting(engine, host) != 0)
342
- return 1;
322
+ start_host_formatting(engine, host);
323
RRDSET *st;
324
rrdset_foreach_read(st, host)
325
{
326
rrdset_rdlock(st);
347
- if (start_chart_formatting(engine, st) != 0)
348
- return 1;
327
+ start_chart_formatting(engine, st);
328
329
RRDDIM *rd;
330
rrddim_foreach_read(rd, st)
352
- {
353
- if (metric_formatting(engine, rd) != 0)
354
- return 1;
355
- }
331
+ metric_formatting(engine, rd);
332
357
- if (end_chart_formatting(engine, st) != 0)
358
- return 1;
333
+ end_chart_formatting(engine, st);
334
rrdset_unlock(st);
335
}
336
362
- if (end_host_formatting(engine, host) != 0)
363
- return 1;
337
+ end_host_formatting(engine, host);
338
rrdhost_unlock(host);
339
}
340
rrd_unlock();
341
netdata_thread_enable_cancelability();
342
369
- if (end_batch_formatting(engine) != 0)
370
- return 1;
371
-
372
- return 0;
343
+ end_batch_formatting(engine);
344
}
345
346
/**
@@ -401,18 +372,3 @@ int simple_connector_update_buffered_bytes(struct instance *instance)
372
373
return 0;
374
}
404
-
405
-/**
406
- * Notify workers
407
- *
408
- * Notify exporting connector instance working threads that data is ready to send.
409
- *
410
- * @param engine an engine data structure.
411
- * @return Returns 0 on success, 1 on failure.
412
- */
413
-int notify_workers(struct engine *engine)
414
-{
415
- (void)engine;
416
-
417
- return 0;
418
-}
exporting/tests/exporting_doubles.c
-7
@@ -75,13 +75,6 @@ int __wrap_prepare_buffers(struct engine *engine)
75
return mock_type(int);
76
}
77
78
-int __wrap_notify_workers(struct engine *engine)
79
-{
80
- function_called();
81
- check_expected_ptr(engine);
82
- return mock_type(int);
83
-}
84
-
78
void __wrap_create_main_rusage_chart(RRDSET **st_rusage, RRDDIM **rd_user, RRDDIM **rd_system)
79
{
80
function_called();
exporting/tests/test_exporting_engine.c
-4
@@ -60,10 +60,6 @@ static void test_exporting_engine(void **state)
60
expect_memory(__wrap_prepare_buffers, engine, engine, sizeof(struct engine));
61
will_return(__wrap_prepare_buffers, 0);
62
63
- expect_function_call(__wrap_notify_workers);
64
- expect_memory(__wrap_notify_workers, engine, engine, sizeof(struct engine));
65
- will_return(__wrap_notify_workers, 0);
66
-
63
expect_function_call(__wrap_send_main_rusage);
64
expect_value(__wrap_send_main_rusage, st_rusage, NULL);
65
expect_value(__wrap_send_main_rusage, rd_user, NULL);
exporting/tests/test_exporting_engine.h
-2
@@ -93,8 +93,6 @@ calculated_number __wrap_exporting_calculate_value_from_stored_data(
93
int __real_prepare_buffers(struct engine *engine);
94
int __wrap_prepare_buffers(struct engine *engine);
95
96
-int __wrap_notify_workers(struct engine *engine);
97
-
96
void __real_create_main_rusage_chart(RRDSET **st_rusage, RRDDIM **rd_user, RRDDIM **rd_system);
97
void __wrap_create_main_rusage_chart(RRDSET **st_rusage, RRDDIM **rd_user, RRDDIM **rd_system);
98