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