src/backends.c

   1 #include "common.h"
   2
   3 #define BACKEND_SOURCE_DATA_AS_COLLECTED 0x00000001
   4 #define BACKEND_SOURCE_DATA_AVERAGE      0x00000002
   5 #define BACKEND_SOURCE_DATA_SUM          0x00000004
   6
   7 int connect_to_socket4(const char *ip, int port, struct timeval *timeout) {
   8     int sock;
   9
  10     debug(D_LISTENER, "IPv4 connecting to ip '%s' port %d", ip, port);
  11
  12     sock = socket(AF_INET, SOCK_STREAM, 0);
  13     if(sock < 0) {
  14         error("IPv4 socket() on ip '%s' port %d failed.", ip, port);
  15         return -1;
  16     }
  17
  18     if(setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (char *)timeout, sizeof(struct timeval)) < 0)
  19         error("Failed to set timeout on the socket to ip '%s' port %d", ip, port);
  20
  21     struct sockaddr_in name;
  22     memset(&name, 0, sizeof(struct sockaddr_in));
  23     name.sin_family = AF_INET;
  24     name.sin_port = htons(port);
  25
  26     int ret = inet_pton(AF_INET, ip, (void *)&name.sin_addr.s_addr);
  27     if(ret != 1) {
  28         error("Failed to convert '%s' to a valid IPv4 address.", ip);
  29         close(sock);
  30         return -1;
  31     }
  32
  33     if(connect(sock, (struct sockaddr *) &name, sizeof(name)) < 0) {
  34         close(sock);
  35         error("IPv4 failed to connect to '%s', port %d", ip, port);
  36         return -1;
  37     }
  38
  39     debug(D_LISTENER, "Connected to IPv4 ip '%s' port %d", ip, port);
  40     return sock;
  41 }
  42
  43 int connect_to_socket6(const char *ip, int port, struct timeval *timeout) {
  44     int sock = -1;
  45     int ipv6only = 1;
  46
  47     debug(D_LISTENER, "IPv6 connecting to ip '%s' port %d", ip, port);
  48
  49     sock = socket(AF_INET6, SOCK_STREAM, 0);
  50     if (sock < 0) {
  51         error("IPv6 socket() on ip '%s' port %d failed.", ip, port);
  52         return -1;
  53     }
  54
  55     if(setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (char *)timeout, sizeof(struct timeval)) < 0)
  56         error("Failed to set timeout on the socket to ip '%s' port %d", ip, port);
  57
  58     /* IPv6 only */
  59     if(setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (void*)&ipv6only, sizeof(ipv6only)) != 0)
  60         error("Cannot set IPV6_V6ONLY on ip '%s' port's %d.", ip, port);
  61
  62     struct sockaddr_in6 name;
  63     memset(&name, 0, sizeof(struct sockaddr_in6));
  64     name.sin6_family = AF_INET6;
  65     name.sin6_port = htons ((uint16_t) port);
  66
  67     int ret = inet_pton(AF_INET6, ip, (void *)&name.sin6_addr.s6_addr);
  68     if(ret != 1) {
  69         error("Failed to convert IP '%s' to a valid IPv6 address.", ip);
  70         close(sock);
  71         return -1;
  72     }
  73
  74     name.sin6_scope_id = 0;
  75
  76     if(connect(sock, (struct sockaddr *)&name, sizeof(name)) < 0) {
  77         close(sock);
  78         error("IPv6 failed to connect to '%s', port %d", ip, port);
  79         return -1;
  80     }
  81
  82     debug(D_LISTENER, "Connected to IPv6 ip '%s' port %d", ip, port);
  83     return sock;
  84 }
  85
  86
  87 static inline int connect_to_one(const char *definition, int default_port, struct timeval *timeout) {
  88     struct addrinfo hints;
  89     struct addrinfo *result = NULL, *rp = NULL;
  90
  91     char buffer[strlen(definition) + 1];
  92     strcpy(buffer, definition);
  93
  94     char buffer2[10 + 1];
  95     snprintfz(buffer2, 10, "%d", default_port);
  96
  97     char *ip = buffer, *port = buffer2;
  98
  99     char *e = ip;
 100     if(*e == '[') {
 101         e = ++ip;
 102         while(*e && *e != ']') e++;
 103         if(*e == ']') {
 104             *e = '\0';
 105             e++;
 106         }
 107     }
 108     else {
 109         while(*e && *e != ':') e++;
 110     }
 111
 112     if(*e == ':') {
 113         port = e + 1;
 114         *e = '\0';
 115     }
 116
 117     if(!*ip)
 118         return -1;
 119
 120     if(!*port)
 121         port = buffer2;
 122
 123     memset(&hints, 0, sizeof(struct addrinfo));
 124     hints.ai_family = AF_UNSPEC;    /* Allow IPv4 or IPv6 */
 125     hints.ai_socktype = SOCK_DGRAM; /* Datagram socket */
 126     hints.ai_flags = AI_PASSIVE;    /* For wildcard IP address */
 127     hints.ai_protocol = 0;          /* Any protocol */
 128     hints.ai_canonname = NULL;
 129     hints.ai_addr = NULL;
 130     hints.ai_next = NULL;
 131
 132     int r = getaddrinfo(ip, port, &hints, &result);
 133     if (r != 0) {
 134         error("Cannot resolve host '%s', port '%s': %s\n", ip, port, gai_strerror(r));
 135         return -1;
 136     }
 137
 138     int fd = -1;
 139     for (rp = result; rp != NULL && fd == -1; rp = rp->ai_next) {
 140         char rip[INET_ADDRSTRLEN + INET6_ADDRSTRLEN] = "INVALID";
 141         int rport;
 142
 143         switch (rp->ai_addr->sa_family) {
 144             case AF_INET: {
 145                 struct sockaddr_in *sin = (struct sockaddr_in *) rp->ai_addr;
 146                 inet_ntop(AF_INET, &sin->sin_addr, rip, INET_ADDRSTRLEN);
 147                 rport = ntohs(sin->sin_port);
 148                 fd = connect_to_socket4(rip, rport, timeout);
 149                 break;
 150             }
 151
 152             case AF_INET6: {
 153                 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *) rp->ai_addr;
 154                 inet_ntop(AF_INET6, &sin6->sin6_addr, rip, INET6_ADDRSTRLEN);
 155                 rport = ntohs(sin6->sin6_port);
 156                 fd = connect_to_socket6(rip, rport, timeout);
 157                 break;
 158             }
 159         }
 160     }
 161
 162     freeaddrinfo(result);
 163
 164     return fd;
 165 }
 166
 167 static inline calculated_number backend_calculate_value_from_stored_data(RRDSET *st, RRDDIM *rd, time_t after, time_t before, uint32_t options) {
 168     time_t first_t = rrdset_first_entry_t(st);
 169     time_t last_t = rrdset_last_entry_t(st);
 170
 171     if(unlikely(before - after < st->update_every && after != after - after % st->update_every))
 172         // when st->update_every is bigger than the frequency we send data to backend
 173         // skip the iterations that are not aligned to the database
 174         return NAN;
 175
 176     // align the time-frame
 177     // for 'after' also skip the first value by adding st->update_every
 178     after  = after  - after  % st->update_every + st->update_every;
 179     before = before - before % st->update_every;
 180
 181     if(unlikely(after < first_t))
 182         after = first_t;
 183
 184     if(unlikely(after > before))
 185         // this can happen when the st->update_every > before - after
 186         before = after;
 187
 188     if(unlikely(before > last_t))
 189         before = last_t;
 190
 191     size_t counter = 0;
 192     calculated_number sum = 0;
 193
 194     long    start_at_slot = rrdset_time2slot(st, before),
 195             stop_at_slot  = rrdset_time2slot(st, after),
 196             slot, stop_now = 0;
 197
 198     for(slot = start_at_slot; !stop_now ; slot--) {
 199         if(unlikely(slot < 0)) slot = st->entries - 1;
 200         if(unlikely(slot == stop_at_slot)) stop_now = 1;
 201
 202         storage_number n = rd->values[slot];
 203         if(unlikely(!does_storage_number_exist(n))) continue;
 204
 205         calculated_number value = unpack_storage_number(n);
 206         sum += value;
 207         counter++;
 208     }
 209
 210     if(unlikely(!counter))
 211         return NAN;
 212
 213     if(unlikely(options & BACKEND_SOURCE_DATA_SUM))
 214         return sum;
 215
 216     return sum / (calculated_number)counter;
 217 }
 218
 219 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) {
 220     (void)host;
 221     (void)after;
 222     (void)before;
 223     (void)options;
 224     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);
 225     return 1;
 226 }
 227
 228 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) {
 229     (void)host;
 230     calculated_number value = backend_calculate_value_from_stored_data(st, rd, after, before, options);
 231     if(!isnan(value)) {
 232         buffer_sprintf(b, "%s.%s.%s.%s " CALCULATED_NUMBER_FORMAT " %u\n", prefix, hostname, st->id, rd->id, value, (uint32_t) before);
 233         return 1;
 234     }
 235     return 0;
 236 }
 237
 238 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) {
 239     (void)host;
 240     (void)after;
 241     (void)before;
 242     (void)options;
 243     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);
 244     return 1;
 245 }
 246
 247 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) {
 248     (void)host;
 249     calculated_number value = backend_calculate_value_from_stored_data(st, rd, after, before, options);
 250     if(!isnan(value)) {
 251         buffer_sprintf(b, "put %s.%s.%s %u " CALCULATED_NUMBER_FORMAT " host=%s\n", prefix, st->id, rd->id, (uint32_t) before, value, hostname);
 252         return 1;
 253     }
 254     return 0;
 255 }
 256
 257 void *backends_main(void *ptr) {
 258     (void)ptr;
 259
 260     BUFFER *b = buffer_create(1);
 261     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);
 262
 263     info("BACKEND thread created with task id %d", gettid());
 264
 265     if(pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL) != 0)
 266         error("Cannot set pthread cancel type to DEFERRED.");
 267
 268     if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
 269         error("Cannot set pthread cancel state to ENABLE.");
 270
 271     // ------------------------------------------------------------------------
 272     // collect configuration options
 273
 274     struct timeval timeout = {
 275             .tv_sec = 0,
 276             .tv_usec = 0
 277     };
 278     int default_port = 0;
 279     int sock = -1;
 280     uint32_t options;
 281     int enabled = config_get_boolean("backend", "enabled", 0);
 282     const char *source = config_get("backend", "data source", "average");
 283     const char *type = config_get("backend", "type", "graphite");
 284     const char *destination = config_get("backend", "destination", "localhost");
 285     const char *prefix = config_get("backend", "prefix", "netdata");
 286     const char *hostname = config_get("backend", "hostname", localhost.hostname);
 287     int frequency = (int)config_get_number("backend", "update every", 10);
 288     int buffer_on_failures = (int)config_get_number("backend", "buffer on failures", 10);
 289     long timeoutms = config_get_number("backend", "timeout ms", frequency * 2 * 1000);
 290
 291     // ------------------------------------------------------------------------
 292     // validate configuration options
 293     // and prepare for sending data to our backend
 294     if(!enabled || frequency < 1)
 295         goto cleanup;
 296
 297     if(!strcmp(source, "as collected")) {
 298         options = BACKEND_SOURCE_DATA_AS_COLLECTED;
 299     }
 300     else if(!strcmp(source, "average")) {
 301         options = BACKEND_SOURCE_DATA_AVERAGE;
 302     }
 303     else if(!strcmp(source, "sum") || !strcmp(source, "volume")) {
 304         options = BACKEND_SOURCE_DATA_SUM;
 305     }
 306     else {
 307         error("Invalid data source method '%s' for backend given. Disabling backed.", source);
 308         goto cleanup;
 309     }
 310
 311     if(!strcmp(type, "graphite") || !strcmp(type, "graphite:plaintext")) {
 312         default_port = 2003;
 313         if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
 314             formatter = format_dimension_collected_graphite_plaintext;
 315         else
 316             formatter = format_dimension_stored_graphite_plaintext;
 317     }
 318     else if(!strcmp(type, "opentsdb") || !strcmp(type, "opentsdb:telnet")) {
 319         default_port = 4242;
 320         if(options == BACKEND_SOURCE_DATA_AS_COLLECTED)
 321             formatter = format_dimension_collected_opentsdb_telnet;
 322         else
 323             formatter = format_dimension_stored_opentsdb_telnet;
 324     }
 325     else {
 326         error("Unknown backend type '%s'", type);
 327         goto cleanup;
 328     }
 329
 330     if(timeoutms < 1) {
 331         error("BACKED invalid timeout %ld ms given. Assuming %d ms.", timeoutms, frequency * 2 * 1000);
 332         timeoutms = frequency * 2 * 1000;
 333     }
 334     timeout.tv_sec  = (timeoutms * 1000) / 1000000;
 335     timeout.tv_usec = (timeoutms * 1000) % 1000000;
 336
 337     // ------------------------------------------------------------------------
 338     // prepare the charts for monitoring the backend
 339
 340     struct rusage thread;
 341
 342     collected_number
 343             chart_buffered_metrics = 0,
 344             chart_lost_metrics = 0,
 345             chart_sent_metrics = 0,
 346             chart_buffered_bytes = 0,
 347             chart_sent_bytes = 0,
 348             chart_transmission_successes = 0,
 349             chart_transmission_failures = 0,
 350             chart_data_lost_events = 0,
 351             chart_lost_bytes = 0,
 352             chart_backend_reconnects = 0,
 353             chart_backend_latency = 0;
 354
 355     RRDSET *chart_metrics = rrdset_find("netdata.backend_metrics");
 356     if(!chart_metrics) {
 357         chart_metrics = rrdset_create("netdata", "backend_metrics", NULL, "backend", NULL, "Netdata Buffered Metrics", "metrics", 130600, frequency, RRDSET_TYPE_LINE);
 358         rrddim_add(chart_metrics, "buffered", NULL,   1, 1, RRDDIM_ABSOLUTE);
 359         rrddim_add(chart_metrics, "lost",     NULL,   1, 1, RRDDIM_ABSOLUTE);
 360         rrddim_add(chart_metrics, "sent",     NULL,   1, 1, RRDDIM_ABSOLUTE);
 361     }
 362
 363     RRDSET *chart_bytes = rrdset_find("netdata.backend_bytes");
 364     if(!chart_bytes) {
 365         chart_bytes = rrdset_create("netdata", "backend_bytes", NULL, "backend", NULL, "Netdata Backend Data Size", "KB", 130610, frequency, RRDSET_TYPE_AREA);
 366         rrddim_add(chart_bytes, "buffered", NULL,  1, 1024, RRDDIM_ABSOLUTE);
 367         rrddim_add(chart_bytes, "lost",     NULL,  1, 1024, RRDDIM_ABSOLUTE);
 368         rrddim_add(chart_bytes, "sent",     NULL,  1, 1024, RRDDIM_ABSOLUTE);
 369     }
 370
 371     RRDSET *chart_ops = rrdset_find("netdata.backend_ops");
 372     if(!chart_ops) {
 373         chart_ops = rrdset_create("netdata", "backend_ops", NULL, "backend", NULL, "Netdata Backend Operations", "operations", 130630, frequency, RRDSET_TYPE_LINE);
 374         rrddim_add(chart_ops, "write",     NULL,  1, 1, RRDDIM_ABSOLUTE);
 375         rrddim_add(chart_ops, "discard",   NULL,  1, 1, RRDDIM_ABSOLUTE);
 376         rrddim_add(chart_ops, "reconnect", NULL,  1, 1, RRDDIM_ABSOLUTE);
 377         rrddim_add(chart_ops, "failure",   NULL,  1, 1, RRDDIM_ABSOLUTE);
 378     }
 379
 380     RRDSET *chart_latency = rrdset_find("netdata.backend_latency");
 381     if(!chart_latency) {
 382         chart_latency = rrdset_create("netdata", "backend_latency", NULL, "backend", NULL, "Netdata Backend Latency", "ms", 130620, frequency, RRDSET_TYPE_AREA);
 383         rrddim_add(chart_latency, "latency",   NULL,  1, 1000, RRDDIM_ABSOLUTE);
 384     }
 385
 386     RRDSET *chart_rusage = rrdset_find("netdata.backend_thread_cpu");
 387     if(!chart_rusage) {
 388         chart_rusage = rrdset_create("netdata", "backend_thread_cpu", NULL, "backend", NULL, "NetData Backend Thread CPU usage", "milliseconds/s", 130630, frequency, RRDSET_TYPE_STACKED);
 389         rrddim_add(chart_rusage, "user",   NULL, 1, 1000, RRDDIM_INCREMENTAL);
 390         rrddim_add(chart_rusage, "system", NULL, 1, 1000, RRDDIM_INCREMENTAL);
 391     }
 392
 393     // ------------------------------------------------------------------------
 394     // prepare the backend main loop
 395
 396     info("BACKEND configured ('%s' on '%s' sending '%s' data, every %d seconds, as host '%s', with prefix '%s')", type, destination, source, frequency, hostname, prefix);
 397
 398     usec_t step_ut = frequency * USEC_PER_SEC;
 399     usec_t random_ut = now_realtime_usec() % (step_ut / 2);
 400     time_t before = (time_t)((now_realtime_usec() - step_ut) / USEC_PER_SEC);
 401     time_t after = before;
 402     int failures = 0;
 403
 404     for(;;) {
 405         // ------------------------------------------------------------------------
 406         // wait for the next iteration point
 407
 408         usec_t now_ut = now_realtime_usec();
 409         usec_t next_ut = now_ut - (now_ut % step_ut) + step_ut;
 410         before = (time_t)(next_ut / USEC_PER_SEC);
 411
 412         // add a little delay (1/4 of the step) plus some randomness
 413         next_ut += (step_ut / 4) + random_ut;
 414
 415         while(now_ut < next_ut) {
 416             sleep_usec(next_ut - now_ut);
 417             now_ut = now_realtime_usec();
 418         }
 419
 420         // ------------------------------------------------------------------------
 421         // add to the buffer the data we need to send to the backend
 422
 423         RRDSET *st;
 424         int pthreadoldcancelstate;
 425
 426         if(unlikely(pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &pthreadoldcancelstate) != 0))
 427             error("Cannot set pthread cancel state to DISABLE.");
 428
 429         rrdhost_rdlock(&localhost);
 430         for(st = localhost.rrdset_root; st ;st = st->next) {
 431             pthread_rwlock_rdlock(&st->rwlock);
 432
 433             RRDDIM *rd;
 434             for(rd = st->dimensions; rd ;rd = rd->next) {
 435                 if(rd->last_collected_time.tv_sec >= after)
 436                     chart_buffered_metrics += formatter(b, prefix, &localhost, hostname, st, rd, after, before, options);
 437             }
 438
 439             pthread_rwlock_unlock(&st->rwlock);
 440         }
 441         rrdhost_unlock(&localhost);
 442
 443         if(unlikely(pthread_setcancelstate(pthreadoldcancelstate, NULL) != 0))
 444             error("Cannot set pthread cancel state to RESTORE (%d).", pthreadoldcancelstate);
 445
 446         chart_buffered_bytes = (collected_number)buffer_strlen(b);
 447
 448         // reset the monitoring chart counters
 449         chart_sent_bytes =
 450         chart_sent_metrics =
 451         chart_lost_metrics =
 452         chart_transmission_successes =
 453         chart_transmission_failures =
 454         chart_data_lost_events =
 455         chart_lost_bytes =
 456         chart_backend_reconnects =
 457         chart_backend_latency = 0;
 458
 459         if(unlikely(netdata_exit)) break;
 460
 461         //fprintf(stderr, "\nBACKEND BEGIN:\n%s\nBACKEND END\n", buffer_tostring(b)); // FIXME
 462         //fprintf(stderr, "after = %lu, before = %lu\n", after, before);
 463
 464         // ------------------------------------------------------------------------
 465         // connect to a backend server
 466
 467         if(unlikely(sock == -1)) {
 468             usec_t start_ut = now_realtime_usec();
 469             const char *s = destination;
 470             while(*s) {
 471                 const char *e = s;
 472
 473                 // skip separators, moving both s(tart) and e(nd)
 474                 while(isspace(*e) || *e == ',') s = ++e;
 475
 476                 // move e(nd) to the first separator
 477                 while(*e && !isspace(*e) && *e != ',') e++;
 478
 479                 // is there anything?
 480                 if(!*s || s == e) break;
 481
 482                 char buf[e - s + 1];
 483                 strncpyz(buf, s, e - s);
 484                 chart_backend_reconnects++;
 485                 sock = connect_to_one(buf, default_port, &timeout);
 486                 if(sock != -1) break;
 487                 s = e;
 488             }
 489             chart_backend_latency += now_realtime_usec() - start_ut;
 490         }
 491
 492         if(unlikely(netdata_exit)) break;
 493
 494         // ------------------------------------------------------------------------
 495         // send our buffer to the backend server
 496
 497         if(likely(sock != -1)) {
 498             size_t len = buffer_strlen(b);
 499             usec_t start_ut = now_realtime_usec();
 500             int flags = 0;
 501 #ifdef MSG_NOSIGNAL
 502             flags += MSG_NOSIGNAL;
 503 #endif
 504             ssize_t written = send(sock, buffer_tostring(b), len, flags);
 505             chart_backend_latency += now_realtime_usec() - start_ut;
 506             if(written != -1 && (size_t)written == len) {
 507                 // we sent the data successfully
 508                 chart_transmission_successes++;
 509                 chart_sent_bytes += written;
 510                 chart_sent_metrics = chart_buffered_metrics;
 511
 512                 // reset the failures count
 513                 failures = 0;
 514
 515                 // empty the buffer
 516                 buffer_flush(b);
 517             }
 518             else {
 519                 // oops! we couldn't send (all or some of the) data
 520                 error("Failed to write data to database backend '%s'. Willing to write %zu bytes, wrote %zd bytes. Will re-connect.", destination, len, written);
 521                 chart_transmission_failures++;
 522
 523                 if(written != -1)
 524                     chart_sent_bytes += written;
 525
 526                 // increment the counter we check for data loss
 527                 failures++;
 528
 529                 // close the socket - we will re-open it next time
 530                 close(sock);
 531                 sock = -1;
 532             }
 533
 534             // either the buffer is empty
 535             // or is holding the data we couldn't send
 536             // so, make sure the next iteration will continue
 537             // from where we are now
 538             after = before;
 539         }
 540         else {
 541             error("Failed to update database backend '%s'", destination);
 542             chart_transmission_failures++;
 543
 544             // increment the counter we check for data loss
 545             failures++;
 546         }
 547
 548         if(failures > buffer_on_failures) {
 549             // too bad! we are going to lose data
 550             chart_lost_bytes += buffer_strlen(b);
 551             error("Reached %d backend failures. Flushing buffers to protect this host - this results in data loss on back-end server '%s'", failures, destination);
 552             buffer_flush(b);
 553             failures = 0;
 554             chart_data_lost_events++;
 555             chart_lost_metrics = chart_buffered_metrics;
 556         }
 557
 558         if(unlikely(netdata_exit)) break;
 559
 560         // ------------------------------------------------------------------------
 561         // update the monitoring charts
 562
 563         if(chart_ops->counter_done) rrdset_next(chart_ops);
 564         rrddim_set(chart_ops, "write",        chart_transmission_successes);
 565         rrddim_set(chart_ops, "discard",      chart_data_lost_events);
 566         rrddim_set(chart_ops, "failure",      chart_transmission_failures);
 567         rrddim_set(chart_ops, "reconnect",    chart_backend_reconnects);
 568         rrdset_done(chart_ops);
 569
 570         if(chart_metrics->counter_done) rrdset_next(chart_metrics);
 571         rrddim_set(chart_metrics, "buffered", chart_buffered_metrics);
 572         rrddim_set(chart_metrics, "lost",     chart_lost_metrics);
 573         rrddim_set(chart_metrics, "sent",     chart_sent_metrics);
 574         rrdset_done(chart_metrics);
 575
 576         if(chart_bytes->counter_done) rrdset_next(chart_bytes);
 577         rrddim_set(chart_bytes, "buffered",   chart_buffered_bytes);
 578         rrddim_set(chart_bytes, "lost",       chart_lost_bytes);
 579         rrddim_set(chart_bytes, "sent",       chart_sent_bytes);
 580         rrdset_done(chart_bytes);
 581
 582         if(chart_latency->counter_done) rrdset_next(chart_latency);
 583         rrddim_set(chart_latency, "latency",  chart_backend_latency);
 584         rrdset_done(chart_latency);
 585
 586         getrusage(RUSAGE_THREAD, &thread);
 587         if(chart_rusage->counter_done) rrdset_next(chart_rusage);
 588         rrddim_set(chart_rusage, "user",   thread.ru_utime.tv_sec * 1000000ULL + thread.ru_utime.tv_usec);
 589         rrddim_set(chart_rusage, "system", thread.ru_stime.tv_sec * 1000000ULL + thread.ru_stime.tv_usec);
 590         rrdset_done(chart_rusage);
 591
 592         if(likely(buffer_strlen(b) == 0))
 593             chart_buffered_metrics = 0;
 594
 595         if(unlikely(netdata_exit)) break;
 596     }
 597
 598 cleanup:
 599     if(sock != -1)
 600         close(sock);
 601
 602     info("BACKEND thread exiting");
 603
 604     pthread_exit(NULL);
 605     return NULL;
 606 }