added backends monitoring charts
Costa Tsaousis (ktsaou) committed
Nov 20, 2016 at 15:20 UTC
551906753018487b8e7688d5643127651e55a839
1 file changed
+152
-15
src/backends.c
+152
-15
@@ -210,41 +210,49 @@ static inline calculated_number backend_duration_average(RRDSET *st, RRDDIM *rd,
210
return sum / (calculated_number)counter;
211
}
212
213
-static inline void format_dimension_collected_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
213
+static inline int format_dimension_collected_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
214
(void)host;
215
(void)after;
216
(void)before;
217
(void)options;
218
buffer_sprintf(b, "%s.%s.%s.%s " COLLECTED_NUMBER_FORMAT " %u\n", prefix, hostname, st->id, rd->id, rd->last_collected_value, (uint32_t)rd->last_collected_time.tv_sec);
219
+ return 1;
220
}
221
221
-static inline void format_dimension_stored_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
222
+static inline int format_dimension_stored_graphite_plaintext(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
223
(void)host;
224
calculated_number value = backend_duration_average(st, rd, after, before, options);
224
- if(!isnan(value))
225
- buffer_sprintf(b, "%s.%s.%s.%s " CALCULATED_NUMBER_FORMAT " %u\n", prefix, hostname, st->id, rd->id, value, (uint32_t)before);
225
+ if(!isnan(value)) {
226
+ buffer_sprintf(b, "%s.%s.%s.%s " CALCULATED_NUMBER_FORMAT " %u\n", prefix, hostname, st->id, rd->id, value, (uint32_t) before);
227
+ return 1;
228
+ }
229
+ return 0;
230
}
231
228
-static inline void format_dimension_collected_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
232
+static inline int format_dimension_collected_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
233
(void)host;
234
(void)after;
235
(void)before;
236
(void)options;
237
buffer_sprintf(b, "put %s.%s.%s %u " COLLECTED_NUMBER_FORMAT " host=%s\n", prefix, st->id, rd->id, (uint32_t)rd->last_collected_time.tv_sec, rd->last_collected_value, hostname);
238
+ return 1;
239
}
240
236
-static inline void format_dimension_stored_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
241
+static inline int format_dimension_stored_opentsdb_telnet(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
242
(void)host;
243
calculated_number value = backend_duration_average(st, rd, after, before, options);
239
- if(!isnan(value))
240
- buffer_sprintf(b, "put %s.%s.%s %u " CALCULATED_NUMBER_FORMAT " host=%s\n", prefix, st->id, rd->id, (uint32_t)before, value, hostname);
244
+ if(!isnan(value)) {
245
+ buffer_sprintf(b, "put %s.%s.%s %u " CALCULATED_NUMBER_FORMAT " host=%s\n", prefix, st->id, rd->id, (uint32_t) before, value, hostname);
246
+ return 1;
247
+ }
248
+ return 0;
249
}
250
251
void *backends_main(void *ptr) {
252
(void)ptr;
253
254
BUFFER *b = buffer_create(1);
247
- void (*formatter)(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options);
255
+ int (*formatter)(BUFFER *b, const char *prefix, RRDHOST *host, const char *hostname, RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options);
256
257
info("BACKEND thread created with task id %d", gettid());
258
@@ -254,6 +262,9 @@ void *backends_main(void *ptr) {
262
if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
263
error("Cannot set pthread cancel state to ENABLE.");
264
265
+ // ------------------------------------------------------------------------
266
+ // collect configuration options
267
+
268
int default_port = 0;
269
int sock = -1;
270
uint32_t options = BACKEND_SOURCE_DATA_AVERAGE;
@@ -266,6 +277,9 @@ void *backends_main(void *ptr) {
277
int frequency = (int)config_get_number("backend", "update every", 10);
278
int buffer_on_failures = (int)config_get_number("backend", "buffer on failures", 10);
279
280
+ // ------------------------------------------------------------------------
281
+ // validate configuration options
282
+ // and prepare for sending data to our backend
283
if(!enabled || frequency < 1)
284
goto cleanup;
285
@@ -302,15 +316,61 @@ void *backends_main(void *ptr) {
316
goto cleanup;
317
}
318
319
+ // ------------------------------------------------------------------------
320
+ // prepare the charts for monitoring the backend
321
+
322
+ collected_number
323
+ chart_buffered_metrics = 0,
324
+ chart_lost_metrics = 0,
325
+ chart_sent_metrics = 0,
326
+ chart_buffered_bytes = 0,
327
+ chart_sent_bytes = 0,
328
+ chart_transmission_successes = 0,
329
+ chart_transmission_failures = 0,
330
+ chart_data_lost_events = 0,
331
+ chart_lost_bytes = 0,
332
+ chart_backend_reconnects = 0;
333
+
334
+ RRDSET *chart_metrics = rrdset_find("netdata.backend_metrics");
335
+ if(!chart_metrics) {
336
+ chart_metrics = rrdset_create("netdata", "backend_metrics", NULL, "backend", NULL, "Netdata Buffered Metrics", "metrics", 130600, frequency, RRDSET_TYPE_LINE);
337
+ rrddim_add(chart_metrics, "buffered", NULL, 1, 1, RRDDIM_ABSOLUTE);
338
+ rrddim_add(chart_metrics, "lost", NULL, 1, 1, RRDDIM_ABSOLUTE);
339
+ rrddim_add(chart_metrics, "sent", NULL, 1, 1, RRDDIM_ABSOLUTE);
340
+ }
341
+
342
+ RRDSET *chart_bytes = rrdset_find("netdata.backend_bytes");
343
+ if(!chart_bytes) {
344
+ chart_bytes = rrdset_create("netdata", "backend_bytes", NULL, "backend", NULL, "Netdata Backend Data Size", "KB", 130610, frequency, RRDSET_TYPE_AREA);
345
+ rrddim_add(chart_bytes, "buffered", NULL, 1, 1024, RRDDIM_ABSOLUTE);
346
+ rrddim_add(chart_bytes, "lost", NULL, 1, 1024, RRDDIM_ABSOLUTE);
347
+ rrddim_add(chart_bytes, "sent", NULL, 1, 1024, RRDDIM_ABSOLUTE);
348
+ }
349
+
350
+ RRDSET *chart_ops = rrdset_find("netdata.backend_ops");
351
+ if(!chart_ops) {
352
+ chart_ops = rrdset_create("netdata", "backend_ops", NULL, "backend", NULL, "Netdata Backend Operations", "operations", 130620, frequency, RRDSET_TYPE_LINE);
353
+ rrddim_add(chart_ops, "write", NULL, 1, 1, RRDDIM_ABSOLUTE);
354
+ rrddim_add(chart_ops, "discard", NULL, 1, 1, RRDDIM_ABSOLUTE);
355
+ rrddim_add(chart_ops, "reconnect", NULL, 1, 1, RRDDIM_ABSOLUTE);
356
+ rrddim_add(chart_ops, "failure", NULL, 1, 1, RRDDIM_ABSOLUTE);
357
+ }
358
+
359
+ // ------------------------------------------------------------------------
360
+ // prepare the backend main loop
361
+
362
+ info("BACKEND configured ('%s' on '%s' sending '%s' data, every %d seconds, as host '%s', with prefix '%s')", type, destination, source, frequency, hostname, prefix);
363
+
364
unsigned long long step_ut = frequency * 1000000ULL;
365
unsigned long long random_ut = time_usec() % (step_ut / 2);
366
time_t before = (time_t)((time_usec() - step_ut) / 10000000ULL);
367
time_t after = before;
368
int failures = 0;
369
311
- info("BACKEND configured ('%s' on '%s' sending '%s' data, every %d seconds, as host '%s', with prefix '%s')", type, destination, source, frequency, hostname, prefix);
312
-
370
for(;;) {
371
+ // ------------------------------------------------------------------------
372
+ // wait for the next iteration point
373
+
374
unsigned long long now_ut = time_usec();
375
unsigned long long next_ut = now_ut - (now_ut % step_ut) + step_ut;
376
before = (time_t)(next_ut / 1000000ULL);
@@ -323,6 +383,9 @@ void *backends_main(void *ptr) {
383
now_ut = time_usec();
384
}
385
386
+ // ------------------------------------------------------------------------
387
+ // add to the buffer the data we need to send to the backend
388
+
389
RRDSET *st;
390
int pthreadoldcancelstate;
391
@@ -336,7 +399,7 @@ void *backends_main(void *ptr) {
399
RRDDIM *rd;
400
for(rd = st->dimensions; rd ;rd = rd->next) {
401
if(rd->last_collected_time.tv_sec >= after)
339
- formatter(b, prefix, &localhost, hostname, st, rd, after, before, options);
402
+ chart_buffered_metrics += formatter(b, prefix, &localhost, hostname, st, rd, after, before, options);
403
}
404
405
pthread_rwlock_unlock(&st->rwlock);
@@ -346,11 +409,26 @@ void *backends_main(void *ptr) {
409
if(unlikely(pthread_setcancelstate(pthreadoldcancelstate, NULL) != 0))
410
error("Cannot set pthread cancel state to RESTORE (%d).", pthreadoldcancelstate);
411
412
+ chart_buffered_bytes = (collected_number)buffer_strlen(b);
413
+
414
+ // reset the monitoring chart counters
415
+ chart_sent_bytes =
416
+ chart_sent_metrics =
417
+ chart_lost_metrics =
418
+ chart_transmission_successes =
419
+ chart_transmission_failures =
420
+ chart_data_lost_events =
421
+ chart_lost_bytes =
422
+ chart_backend_reconnects = 0;
423
+
424
if(unlikely(netdata_exit)) break;
425
351
- // fprintf(stderr, "\nBACKEND BEGIN:\n%s\nBACKEND END\n", buffer_tostring(b)); // FIXME
426
+ //fprintf(stderr, "\nBACKEND BEGIN:\n%s\nBACKEND END\n", buffer_tostring(b)); // FIXME
427
//fprintf(stderr, "after = %lu, before = %lu\n", after, before);
428
429
+ // ------------------------------------------------------------------------
430
+ // connect to a backend server
431
+
432
if(unlikely(sock == -1)) {
433
const char *s = destination;
434
while(*s) {
@@ -367,6 +445,7 @@ void *backends_main(void *ptr) {
445
446
char buf[e - s + 1];
447
strncpyz(buf, s, e - s);
448
+ chart_backend_reconnects++;
449
sock = connect_to_one(buf, default_port);
450
if(sock != -1) break;
451
s = e;
@@ -375,34 +454,92 @@ void *backends_main(void *ptr) {
454
455
if(unlikely(netdata_exit)) break;
456
457
+ // ------------------------------------------------------------------------
458
+ // send our buffer to the backend server
459
+
460
if(likely(sock != -1)) {
461
size_t len = buffer_strlen(b);
462
ssize_t written = write(sock, buffer_tostring(b), len);
463
if(written != -1 && (size_t)written == len) {
464
+ // we sent the data successfully
465
+ chart_transmission_successes++;
466
+ chart_sent_bytes += written;
467
+ chart_sent_metrics = chart_buffered_metrics;
468
+
469
+ // reset the failures count
470
failures = 0;
471
+
472
+ // empty the buffer
473
buffer_flush(b);
474
}
475
else {
386
- failures++;
476
+ // oops! we couldn't send (all or some of the) data
477
error("Failed to write data to database backend '%s'. Willing to write %zu bytes, wrote %zd bytes. Will re-connect.", destination, len, written);
478
+ chart_transmission_failures++;
479
+
480
+ if(written != -1)
481
+ chart_sent_bytes += written;
482
+
483
+ // increment the counter we check for data loss
484
+ failures++;
485
+
486
+ // close the socket - we will re-open it next time
487
close(sock);
488
sock = -1;
489
}
490
491
+ // either the buffer is empty
492
+ // or is holding the data we couldn't send
493
+ // so, make sure the next iteration will continue
494
+ // from where we are now
495
after = before;
496
}
497
else {
395
- failures++;
498
error("Failed to update database backend '%s'", destination);
499
+ chart_transmission_failures++;
500
+
501
+ // increment the counter we check for data loss
502
+ failures++;
503
}
504
505
if(failures > buffer_on_failures) {
506
+ // too bad! we are going to lose data
507
+ chart_lost_bytes += buffer_strlen(b);
508
error("Reached %d backend failures. Flushing buffers to protect this host - this results in data loss on back-end server '%s'", failures, destination);
509
buffer_flush(b);
510
failures = 0;
511
+ chart_data_lost_events++;
512
+ chart_lost_metrics = chart_buffered_metrics;
513
}
514
515
if(unlikely(netdata_exit)) break;
516
+
517
+ // ------------------------------------------------------------------------
518
+ // update the monitoring charts
519
+
520
+ if(chart_ops->counter_done) rrdset_next(chart_ops);
521
+ rrddim_set(chart_ops, "write", chart_transmission_successes);
522
+ rrddim_set(chart_ops, "discard", chart_data_lost_events);
523
+ rrddim_set(chart_ops, "failure", chart_transmission_failures);
524
+ rrddim_set(chart_ops, "reconnect", chart_backend_reconnects);
525
+ rrdset_done(chart_ops);
526
+
527
+ if(chart_metrics->counter_done) rrdset_next(chart_metrics);
528
+ rrddim_set(chart_metrics, "buffered", chart_buffered_metrics);
529
+ rrddim_set(chart_metrics, "lost", chart_lost_metrics);
530
+ rrddim_set(chart_metrics, "sent", chart_sent_metrics);
531
+ rrdset_done(chart_metrics);
532
+
533
+ if(chart_bytes->counter_done) rrdset_next(chart_bytes);
534
+ rrddim_set(chart_bytes, "buffered", chart_buffered_bytes);
535
+ rrddim_set(chart_bytes, "lost", chart_lost_bytes);
536
+ rrddim_set(chart_bytes, "sent", chart_sent_bytes);
537
+ rrdset_done(chart_bytes);
538
+
539
+ if(likely(buffer_strlen(b) == 0))
540
+ chart_buffered_metrics = 0;
541
+
542
+ if(unlikely(netdata_exit)) break;
543
}
544
545
cleanup: