src/statsd.c

   1 /**
   2  * collectd - src/statsd.c
   3  * Copyright (C) 2013       Florian octo Forster
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a
   6  * copy of this software and associated documentation files (the "Software"),
   7  * to deal in the Software without restriction, including without limitation
   8  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
   9  * and/or sell copies of the Software, and to permit persons to whom the
  10  * Software is furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  20  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  21  * DEALINGS IN THE SOFTWARE.
  22  *
  23  * Authors:
  24  *   Florian octo Forster <octo at collectd.org>
  25  */
  26
  27 #include "collectd.h"
  28
  29 #include "common.h"
  30 #include "plugin.h"
  31 #include "utils_avltree.h"
  32 #include "utils_latency.h"
  33
  34 #include <netdb.h>
  35 #include <poll.h>
  36 #include <sys/types.h>
  37
  38 /* AIX doesn't have MSG_DONTWAIT */
  39 #ifndef MSG_DONTWAIT
  40 #define MSG_DONTWAIT MSG_NONBLOCK
  41 #endif
  42
  43 #ifndef STATSD_DEFAULT_NODE
  44 #define STATSD_DEFAULT_NODE NULL
  45 #endif
  46
  47 #ifndef STATSD_DEFAULT_SERVICE
  48 #define STATSD_DEFAULT_SERVICE "8125"
  49 #endif
  50
  51 enum metric_type_e { STATSD_COUNTER, STATSD_TIMER, STATSD_GAUGE, STATSD_SET };
  52 typedef enum metric_type_e metric_type_t;
  53
  54 struct statsd_metric_s {
  55   metric_type_t type;
  56   double value;
  57   derive_t counter;
  58   latency_counter_t *latency;
  59   c_avl_tree_t *set;
  60   unsigned long updates_num;
  61 };
  62 typedef struct statsd_metric_s statsd_metric_t;
  63
  64 static c_avl_tree_t *metrics_tree = NULL;
  65 static pthread_mutex_t metrics_lock = PTHREAD_MUTEX_INITIALIZER;
  66
  67 static pthread_t network_thread;
  68 static _Bool network_thread_running = 0;
  69 static _Bool network_thread_shutdown = 0;
  70
  71 static char *conf_node = NULL;
  72 static char *conf_service = NULL;
  73
  74 static _Bool conf_delete_counters = 0;
  75 static _Bool conf_delete_timers = 0;
  76 static _Bool conf_delete_gauges = 0;
  77 static _Bool conf_delete_sets = 0;
  78
  79 static double *conf_timer_percentile = NULL;
  80 static size_t conf_timer_percentile_num = 0;
  81
  82 static _Bool conf_counter_sum = 0;
  83 static _Bool conf_timer_lower = 0;
  84 static _Bool conf_timer_upper = 0;
  85 static _Bool conf_timer_sum = 0;
  86 static _Bool conf_timer_count = 0;
  87
  88 /* Must hold metrics_lock when calling this function. */
  89 static statsd_metric_t *statsd_metric_lookup_unsafe(char const *name, /* {{{ */
  90                                                     metric_type_t type) {
  91   char key[DATA_MAX_NAME_LEN + 2];
  92   char *key_copy;
  93   statsd_metric_t *metric;
  94   int status;
  95
  96   switch (type) {
  97   case STATSD_COUNTER:
  98     key[0] = 'c';
  99     break;
 100   case STATSD_TIMER:
 101     key[0] = 't';
 102     break;
 103   case STATSD_GAUGE:
 104     key[0] = 'g';
 105     break;
 106   case STATSD_SET:
 107     key[0] = 's';
 108     break;
 109   default:
 110     return NULL;
 111   }
 112
 113   key[1] = ':';
 114   sstrncpy(&key[2], name, sizeof(key) - 2);
 115
 116   status = c_avl_get(metrics_tree, key, (void *)&metric);
 117   if (status == 0)
 118     return metric;
 119
 120   key_copy = strdup(key);
 121   if (key_copy == NULL) {
 122     ERROR("statsd plugin: strdup failed.");
 123     return NULL;
 124   }
 125
 126   metric = calloc(1, sizeof(*metric));
 127   if (metric == NULL) {
 128     ERROR("statsd plugin: calloc failed.");
 129     sfree(key_copy);
 130     return NULL;
 131   }
 132
 133   metric->type = type;
 134   metric->latency = NULL;
 135   metric->set = NULL;
 136
 137   status = c_avl_insert(metrics_tree, key_copy, metric);
 138   if (status != 0) {
 139     ERROR("statsd plugin: c_avl_insert failed.");
 140     sfree(key_copy);
 141     sfree(metric);
 142     return NULL;
 143   }
 144
 145   return metric;
 146 } /* }}} statsd_metric_lookup_unsafe */
 147
 148 static int statsd_metric_set(char const *name, double value, /* {{{ */
 149                              metric_type_t type) {
 150   statsd_metric_t *metric;
 151
 152   pthread_mutex_lock(&metrics_lock);
 153
 154   metric = statsd_metric_lookup_unsafe(name, type);
 155   if (metric == NULL) {
 156     pthread_mutex_unlock(&metrics_lock);
 157     return -1;
 158   }
 159
 160   metric->value = value;
 161   metric->updates_num++;
 162
 163   pthread_mutex_unlock(&metrics_lock);
 164
 165   return 0;
 166 } /* }}} int statsd_metric_set */
 167
 168 static int statsd_metric_add(char const *name, double delta, /* {{{ */
 169                              metric_type_t type) {
 170   statsd_metric_t *metric;
 171
 172   pthread_mutex_lock(&metrics_lock);
 173
 174   metric = statsd_metric_lookup_unsafe(name, type);
 175   if (metric == NULL) {
 176     pthread_mutex_unlock(&metrics_lock);
 177     return -1;
 178   }
 179
 180   metric->value += delta;
 181   metric->updates_num++;
 182
 183   pthread_mutex_unlock(&metrics_lock);
 184
 185   return 0;
 186 } /* }}} int statsd_metric_add */
 187
 188 static void statsd_metric_free(statsd_metric_t *metric) /* {{{ */
 189 {
 190   if (metric == NULL)
 191     return;
 192
 193   if (metric->latency != NULL) {
 194     latency_counter_destroy(metric->latency);
 195     metric->latency = NULL;
 196   }
 197
 198   if (metric->set != NULL) {
 199     void *key;
 200     void *value;
 201
 202     while (c_avl_pick(metric->set, &key, &value) == 0) {
 203       sfree(key);
 204       assert(value == NULL);
 205     }
 206
 207     c_avl_destroy(metric->set);
 208     metric->set = NULL;
 209   }
 210
 211   sfree(metric);
 212 } /* }}} void statsd_metric_free */
 213
 214 static int statsd_parse_value(char const *str, value_t *ret_value) /* {{{ */
 215 {
 216   char *endptr = NULL;
 217
 218   ret_value->gauge = (gauge_t)strtod(str, &endptr);
 219   if ((str == endptr) || ((endptr != NULL) && (*endptr != 0)))
 220     return -1;
 221
 222   return 0;
 223 } /* }}} int statsd_parse_value */
 224
 225 static int statsd_handle_counter(char const *name, /* {{{ */
 226                                  char const *value_str, char const *extra) {
 227   value_t value;
 228   value_t scale;
 229   int status;
 230
 231   if ((extra != NULL) && (extra[0] != '@'))
 232     return -1;
 233
 234   scale.gauge = 1.0;
 235   if (extra != NULL) {
 236     status = statsd_parse_value(extra + 1, &scale);
 237     if (status != 0)
 238       return status;
 239
 240     if (!isfinite(scale.gauge) || (scale.gauge <= 0.0) || (scale.gauge > 1.0))
 241       return -1;
 242   }
 243
 244   value.gauge = 1.0;
 245   status = statsd_parse_value(value_str, &value);
 246   if (status != 0)
 247     return status;
 248
 249   /* Changes to the counter are added to (statsd_metric_t*)->value. ->counter is
 250    * only updated in statsd_metric_submit_unsafe(). */
 251   return statsd_metric_add(name, (double)(value.gauge / scale.gauge),
 252                            STATSD_COUNTER);
 253 } /* }}} int statsd_handle_counter */
 254
 255 static int statsd_handle_gauge(char const *name, /* {{{ */
 256                                char const *value_str) {
 257   value_t value;
 258   int status;
 259
 260   value.gauge = 0;
 261   status = statsd_parse_value(value_str, &value);
 262   if (status != 0)
 263     return status;
 264
 265   if ((value_str[0] == '+') || (value_str[0] == '-'))
 266     return statsd_metric_add(name, (double)value.gauge, STATSD_GAUGE);
 267   else
 268     return statsd_metric_set(name, (double)value.gauge, STATSD_GAUGE);
 269 } /* }}} int statsd_handle_gauge */
 270
 271 static int statsd_handle_timer(char const *name, /* {{{ */
 272                                char const *value_str, char const *extra) {
 273   statsd_metric_t *metric;
 274   value_t value_ms;
 275   value_t scale;
 276   cdtime_t value;
 277   int status;
 278
 279   if ((extra != NULL) && (extra[0] != '@'))
 280     return -1;
 281
 282   scale.gauge = 1.0;
 283   if (extra != NULL) {
 284     status = statsd_parse_value(extra + 1, &scale);
 285     if (status != 0)
 286       return status;
 287
 288     if (!isfinite(scale.gauge) || (scale.gauge <= 0.0) || (scale.gauge > 1.0))
 289       return -1;
 290   }
 291
 292   value_ms.derive = 0;
 293   status = statsd_parse_value(value_str, &value_ms);
 294   if (status != 0)
 295     return status;
 296
 297   value = MS_TO_CDTIME_T(value_ms.gauge / scale.gauge);
 298
 299   pthread_mutex_lock(&metrics_lock);
 300
 301   metric = statsd_metric_lookup_unsafe(name, STATSD_TIMER);
 302   if (metric == NULL) {
 303     pthread_mutex_unlock(&metrics_lock);
 304     return -1;
 305   }
 306
 307   if (metric->latency == NULL)
 308     metric->latency = latency_counter_create();
 309   if (metric->latency == NULL) {
 310     pthread_mutex_unlock(&metrics_lock);
 311     return -1;
 312   }
 313
 314   latency_counter_add(metric->latency, value);
 315   metric->updates_num++;
 316
 317   pthread_mutex_unlock(&metrics_lock);
 318   return 0;
 319 } /* }}} int statsd_handle_timer */
 320
 321 static int statsd_handle_set(char const *name, /* {{{ */
 322                              char const *set_key_orig) {
 323   statsd_metric_t *metric = NULL;
 324   char *set_key;
 325   int status;
 326
 327   pthread_mutex_lock(&metrics_lock);
 328
 329   metric = statsd_metric_lookup_unsafe(name, STATSD_SET);
 330   if (metric == NULL) {
 331     pthread_mutex_unlock(&metrics_lock);
 332     return -1;
 333   }
 334
 335   /* Make sure metric->set exists. */
 336   if (metric->set == NULL)
 337     metric->set = c_avl_create((int (*)(const void *, const void *))strcmp);
 338
 339   if (metric->set == NULL) {
 340     pthread_mutex_unlock(&metrics_lock);
 341     ERROR("statsd plugin: c_avl_create failed.");
 342     return -1;
 343   }
 344
 345   set_key = strdup(set_key_orig);
 346   if (set_key == NULL) {
 347     pthread_mutex_unlock(&metrics_lock);
 348     ERROR("statsd plugin: strdup failed.");
 349     return -1;
 350   }
 351
 352   status = c_avl_insert(metric->set, set_key, /* value = */ NULL);
 353   if (status < 0) {
 354     pthread_mutex_unlock(&metrics_lock);
 355     if (status < 0)
 356       ERROR("statsd plugin: c_avl_insert (\"%s\") failed with status %i.",
 357             set_key, status);
 358     sfree(set_key);
 359     return -1;
 360   } else if (status > 0) /* key already exists */
 361   {
 362     sfree(set_key);
 363   }
 364
 365   metric->updates_num++;
 366
 367   pthread_mutex_unlock(&metrics_lock);
 368   return 0;
 369 } /* }}} int statsd_handle_set */
 370
 371 static int statsd_parse_line(char *buffer) /* {{{ */
 372 {
 373   char *name = buffer;
 374   char *value;
 375   char *type;
 376   char *extra;
 377
 378   type = strchr(name, '|');
 379   if (type == NULL)
 380     return -1;
 381   *type = 0;
 382   type++;
 383
 384   value = strrchr(name, ':');
 385   if (value == NULL)
 386     return -1;
 387   *value = 0;
 388   value++;
 389
 390   extra = strchr(type, '|');
 391   if (extra != NULL) {
 392     *extra = 0;
 393     extra++;
 394   }
 395
 396   if (strcmp("c", type) == 0)
 397     return statsd_handle_counter(name, value, extra);
 398   else if (strcmp("ms", type) == 0)
 399     return statsd_handle_timer(name, value, extra);
 400
 401   /* extra is only valid for counters and timers */
 402   if (extra != NULL)
 403     return -1;
 404
 405   if (strcmp("g", type) == 0)
 406     return statsd_handle_gauge(name, value);
 407   else if (strcmp("s", type) == 0)
 408     return statsd_handle_set(name, value);
 409   else
 410     return -1;
 411 } /* }}} void statsd_parse_line */
 412
 413 static void statsd_parse_buffer(char *buffer) /* {{{ */
 414 {
 415   while (buffer != NULL) {
 416     char orig[64];
 417     char *next;
 418     int status;
 419
 420     next = strchr(buffer, '\n');
 421     if (next != NULL) {
 422       *next = 0;
 423       next++;
 424     }
 425
 426     if (*buffer == 0) {
 427       buffer = next;
 428       continue;
 429     }
 430
 431     sstrncpy(orig, buffer, sizeof(orig));
 432
 433     status = statsd_parse_line(buffer);
 434     if (status != 0)
 435       ERROR("statsd plugin: Unable to parse line: \"%s\"", orig);
 436
 437     buffer = next;
 438   }
 439 } /* }}} void statsd_parse_buffer */
 440
 441 static void statsd_network_read(int fd) /* {{{ */
 442 {
 443   char buffer[4096];
 444   size_t buffer_size;
 445   ssize_t status;
 446
 447   status = recv(fd, buffer, sizeof(buffer), /* flags = */ MSG_DONTWAIT);
 448   if (status < 0) {
 449
 450     if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
 451       return;
 452
 453     ERROR("statsd plugin: recv(2) failed: %s", STRERRNO);
 454     return;
 455   }
 456
 457   buffer_size = (size_t)status;
 458   if (buffer_size >= sizeof(buffer))
 459     buffer_size = sizeof(buffer) - 1;
 460   buffer[buffer_size] = 0;
 461
 462   statsd_parse_buffer(buffer);
 463 } /* }}} void statsd_network_read */
 464
 465 static int statsd_network_init(struct pollfd **ret_fds, /* {{{ */
 466                                size_t *ret_fds_num) {
 467   struct pollfd *fds = NULL;
 468   size_t fds_num = 0;
 469
 470   struct addrinfo *ai_list;
 471   int status;
 472
 473   char const *node = (conf_node != NULL) ? conf_node : STATSD_DEFAULT_NODE;
 474   char const *service =
 475       (conf_service != NULL) ? conf_service : STATSD_DEFAULT_SERVICE;
 476
 477   struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
 478                               .ai_flags = AI_PASSIVE | AI_ADDRCONFIG,
 479                               .ai_socktype = SOCK_DGRAM};
 480
 481   status = getaddrinfo(node, service, &ai_hints, &ai_list);
 482   if (status != 0) {
 483     ERROR("statsd plugin: getaddrinfo (\"%s\", \"%s\") failed: %s", node,
 484           service, gai_strerror(status));
 485     return status;
 486   }
 487
 488   for (struct addrinfo *ai_ptr = ai_list; ai_ptr != NULL;
 489        ai_ptr = ai_ptr->ai_next) {
 490     int fd;
 491     struct pollfd *tmp;
 492
 493     char dbg_node[NI_MAXHOST];
 494     char dbg_service[NI_MAXSERV];
 495
 496     fd = socket(ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
 497     if (fd < 0) {
 498       ERROR("statsd plugin: socket(2) failed: %s", STRERRNO);
 499       continue;
 500     }
 501
 502     getnameinfo(ai_ptr->ai_addr, ai_ptr->ai_addrlen, dbg_node, sizeof(dbg_node),
 503                 dbg_service, sizeof(dbg_service),
 504                 NI_DGRAM | NI_NUMERICHOST | NI_NUMERICSERV);
 505     DEBUG("statsd plugin: Trying to bind to [%s]:%s ...", dbg_node,
 506           dbg_service);
 507
 508     status = bind(fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 509     if (status != 0) {
 510       ERROR("statsd plugin: bind(2) failed: %s", STRERRNO);
 511       close(fd);
 512       continue;
 513     }
 514
 515     tmp = realloc(fds, sizeof(*fds) * (fds_num + 1));
 516     if (tmp == NULL) {
 517       ERROR("statsd plugin: realloc failed.");
 518       close(fd);
 519       continue;
 520     }
 521     fds = tmp;
 522     tmp = fds + fds_num;
 523     fds_num++;
 524
 525     memset(tmp, 0, sizeof(*tmp));
 526     tmp->fd = fd;
 527     tmp->events = POLLIN | POLLPRI;
 528   }
 529
 530   freeaddrinfo(ai_list);
 531
 532   if (fds_num == 0) {
 533     ERROR("statsd plugin: Unable to create listening socket for [%s]:%s.",
 534           (node != NULL) ? node : "::", service);
 535     return ENOENT;
 536   }
 537
 538   *ret_fds = fds;
 539   *ret_fds_num = fds_num;
 540   return 0;
 541 } /* }}} int statsd_network_init */
 542
 543 static void *statsd_network_thread(void *args) /* {{{ */
 544 {
 545   struct pollfd *fds = NULL;
 546   size_t fds_num = 0;
 547   int status;
 548
 549   status = statsd_network_init(&fds, &fds_num);
 550   if (status != 0) {
 551     ERROR("statsd plugin: Unable to open listening sockets.");
 552     pthread_exit((void *)0);
 553   }
 554
 555   while (!network_thread_shutdown) {
 556     status = poll(fds, (nfds_t)fds_num, /* timeout = */ -1);
 557     if (status < 0) {
 558
 559       if ((errno == EINTR) || (errno == EAGAIN))
 560         continue;
 561
 562       ERROR("statsd plugin: poll(2) failed: %s", STRERRNO);
 563       break;
 564     }
 565
 566     for (size_t i = 0; i < fds_num; i++) {
 567       if ((fds[i].revents & (POLLIN | POLLPRI)) == 0)
 568         continue;
 569
 570       statsd_network_read(fds[i].fd);
 571       fds[i].revents = 0;
 572     }
 573   } /* while (!network_thread_shutdown) */
 574
 575   /* Clean up */
 576   for (size_t i = 0; i < fds_num; i++)
 577     close(fds[i].fd);
 578   sfree(fds);
 579
 580   return (void *)0;
 581 } /* }}} void *statsd_network_thread */
 582
 583 static int statsd_config_timer_percentile(oconfig_item_t *ci) /* {{{ */
 584 {
 585   double percent = NAN;
 586   double *tmp;
 587   int status;
 588
 589   status = cf_util_get_double(ci, &percent);
 590   if (status != 0)
 591     return status;
 592
 593   if ((percent <= 0.0) || (percent >= 100)) {
 594     ERROR("statsd plugin: The value for \"%s\" must be between 0 and 100, "
 595           "exclusively.",
 596           ci->key);
 597     return ERANGE;
 598   }
 599
 600   tmp =
 601       realloc(conf_timer_percentile,
 602               sizeof(*conf_timer_percentile) * (conf_timer_percentile_num + 1));
 603   if (tmp == NULL) {
 604     ERROR("statsd plugin: realloc failed.");
 605     return ENOMEM;
 606   }
 607   conf_timer_percentile = tmp;
 608   conf_timer_percentile[conf_timer_percentile_num] = percent;
 609   conf_timer_percentile_num++;
 610
 611   return 0;
 612 } /* }}} int statsd_config_timer_percentile */
 613
 614 static int statsd_config(oconfig_item_t *ci) /* {{{ */
 615 {
 616   for (int i = 0; i < ci->children_num; i++) {
 617     oconfig_item_t *child = ci->children + i;
 618
 619     if (strcasecmp("Host", child->key) == 0)
 620       cf_util_get_string(child, &conf_node);
 621     else if (strcasecmp("Port", child->key) == 0)
 622       cf_util_get_service(child, &conf_service);
 623     else if (strcasecmp("DeleteCounters", child->key) == 0)
 624       cf_util_get_boolean(child, &conf_delete_counters);
 625     else if (strcasecmp("DeleteTimers", child->key) == 0)
 626       cf_util_get_boolean(child, &conf_delete_timers);
 627     else if (strcasecmp("DeleteGauges", child->key) == 0)
 628       cf_util_get_boolean(child, &conf_delete_gauges);
 629     else if (strcasecmp("DeleteSets", child->key) == 0)
 630       cf_util_get_boolean(child, &conf_delete_sets);
 631     else if (strcasecmp("CounterSum", child->key) == 0)
 632       cf_util_get_boolean(child, &conf_counter_sum);
 633     else if (strcasecmp("TimerLower", child->key) == 0)
 634       cf_util_get_boolean(child, &conf_timer_lower);
 635     else if (strcasecmp("TimerUpper", child->key) == 0)
 636       cf_util_get_boolean(child, &conf_timer_upper);
 637     else if (strcasecmp("TimerSum", child->key) == 0)
 638       cf_util_get_boolean(child, &conf_timer_sum);
 639     else if (strcasecmp("TimerCount", child->key) == 0)
 640       cf_util_get_boolean(child, &conf_timer_count);
 641     else if (strcasecmp("TimerPercentile", child->key) == 0)
 642       statsd_config_timer_percentile(child);
 643     else
 644       ERROR("statsd plugin: The \"%s\" config option is not valid.",
 645             child->key);
 646   }
 647
 648   return 0;
 649 } /* }}} int statsd_config */
 650
 651 static int statsd_init(void) /* {{{ */
 652 {
 653   pthread_mutex_lock(&metrics_lock);
 654   if (metrics_tree == NULL)
 655     metrics_tree = c_avl_create((int (*)(const void *, const void *))strcmp);
 656
 657   if (!network_thread_running) {
 658     int status;
 659
 660     status = pthread_create(&network_thread,
 661                             /* attr = */ NULL, statsd_network_thread,
 662                             /* args = */ NULL);
 663     if (status != 0) {
 664       pthread_mutex_unlock(&metrics_lock);
 665       ERROR("statsd plugin: pthread_create failed: %s", STRERRNO);
 666       return status;
 667     }
 668   }
 669   network_thread_running = 1;
 670
 671   pthread_mutex_unlock(&metrics_lock);
 672
 673   return 0;
 674 } /* }}} int statsd_init */
 675
 676 /* Must hold metrics_lock when calling this function. */
 677 static int statsd_metric_clear_set_unsafe(statsd_metric_t *metric) /* {{{ */
 678 {
 679   void *key;
 680   void *value;
 681
 682   if ((metric == NULL) || (metric->type != STATSD_SET))
 683     return EINVAL;
 684
 685   if (metric->set == NULL)
 686     return 0;
 687
 688   while (c_avl_pick(metric->set, &key, &value) == 0) {
 689     sfree(key);
 690     sfree(value);
 691   }
 692
 693   return 0;
 694 } /* }}} int statsd_metric_clear_set_unsafe */
 695
 696 /* Must hold metrics_lock when calling this function. */
 697 static int statsd_metric_submit_unsafe(char const *name,
 698                                        statsd_metric_t *metric) /* {{{ */
 699 {
 700   value_list_t vl = VALUE_LIST_INIT;
 701
 702   vl.values = &(value_t){.gauge = NAN};
 703   vl.values_len = 1;
 704   sstrncpy(vl.plugin, "statsd", sizeof(vl.plugin));
 705
 706   if (metric->type == STATSD_GAUGE)
 707     sstrncpy(vl.type, "gauge", sizeof(vl.type));
 708   else if (metric->type == STATSD_TIMER)
 709     sstrncpy(vl.type, "latency", sizeof(vl.type));
 710   else if (metric->type == STATSD_SET)
 711     sstrncpy(vl.type, "objects", sizeof(vl.type));
 712   else /* if (metric->type == STATSD_COUNTER) */
 713     sstrncpy(vl.type, "derive", sizeof(vl.type));
 714
 715   sstrncpy(vl.type_instance, name, sizeof(vl.type_instance));
 716
 717   if (metric->type == STATSD_GAUGE)
 718     vl.values[0].gauge = (gauge_t)metric->value;
 719   else if (metric->type == STATSD_TIMER) {
 720     _Bool have_events = (metric->updates_num > 0);
 721
 722     /* Make sure all timer metrics share the *same* timestamp. */
 723     vl.time = cdtime();
 724
 725     snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-average", name);
 726     vl.values[0].gauge =
 727         have_events
 728             ? CDTIME_T_TO_DOUBLE(latency_counter_get_average(metric->latency))
 729             : NAN;
 730     plugin_dispatch_values(&vl);
 731
 732     if (conf_timer_lower) {
 733       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-lower", name);
 734       vl.values[0].gauge =
 735           have_events
 736               ? CDTIME_T_TO_DOUBLE(latency_counter_get_min(metric->latency))
 737               : NAN;
 738       plugin_dispatch_values(&vl);
 739     }
 740
 741     if (conf_timer_upper) {
 742       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-upper", name);
 743       vl.values[0].gauge =
 744           have_events
 745               ? CDTIME_T_TO_DOUBLE(latency_counter_get_max(metric->latency))
 746               : NAN;
 747       plugin_dispatch_values(&vl);
 748     }
 749
 750     if (conf_timer_sum) {
 751       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-sum", name);
 752       vl.values[0].gauge =
 753           have_events
 754               ? CDTIME_T_TO_DOUBLE(latency_counter_get_sum(metric->latency))
 755               : NAN;
 756       plugin_dispatch_values(&vl);
 757     }
 758
 759     for (size_t i = 0; i < conf_timer_percentile_num; i++) {
 760       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-percentile-%.0f",
 761                name, conf_timer_percentile[i]);
 762       vl.values[0].gauge =
 763           have_events ? CDTIME_T_TO_DOUBLE(latency_counter_get_percentile(
 764                             metric->latency, conf_timer_percentile[i]))
 765                       : NAN;
 766       plugin_dispatch_values(&vl);
 767     }
 768
 769     /* Keep this at the end, since vl.type is set to "gauge" here. The
 770      * vl.type's above are implicitly set to "latency". */
 771     if (conf_timer_count) {
 772       sstrncpy(vl.type, "gauge", sizeof(vl.type));
 773       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-count", name);
 774       vl.values[0].gauge = latency_counter_get_num(metric->latency);
 775       plugin_dispatch_values(&vl);
 776     }
 777
 778     latency_counter_reset(metric->latency);
 779     return 0;
 780   } else if (metric->type == STATSD_SET) {
 781     if (metric->set == NULL)
 782       vl.values[0].gauge = 0.0;
 783     else
 784       vl.values[0].gauge = (gauge_t)c_avl_size(metric->set);
 785   } else { /* STATSD_COUNTER */
 786     gauge_t delta = nearbyint(metric->value);
 787
 788     /* Etsy's statsd writes counters as two metrics: a rate and the change since
 789      * the last write. Since collectd does not reset its DERIVE metrics to zero,
 790      * this makes little sense, but we're dispatching a "count" metric here
 791      * anyway - if requested by the user - for compatibility reasons. */
 792     if (conf_counter_sum) {
 793       sstrncpy(vl.type, "count", sizeof(vl.type));
 794       vl.values[0].gauge = delta;
 795       plugin_dispatch_values(&vl);
 796
 797       /* restore vl.type */
 798       sstrncpy(vl.type, "derive", sizeof(vl.type));
 799     }
 800
 801     /* Rather than resetting value to zero, subtract delta so we correctly keep
 802      * track of residuals. */
 803     metric->value -= delta;
 804     metric->counter += (derive_t)delta;
 805
 806     vl.values[0].derive = metric->counter;
 807   }
 808
 809   return plugin_dispatch_values(&vl);
 810 } /* }}} int statsd_metric_submit_unsafe */
 811
 812 static int statsd_read(void) /* {{{ */
 813 {
 814   c_avl_iterator_t *iter;
 815   char *name;
 816   statsd_metric_t *metric;
 817
 818   char **to_be_deleted = NULL;
 819   size_t to_be_deleted_num = 0;
 820
 821   pthread_mutex_lock(&metrics_lock);
 822
 823   if (metrics_tree == NULL) {
 824     pthread_mutex_unlock(&metrics_lock);
 825     return 0;
 826   }
 827
 828   iter = c_avl_get_iterator(metrics_tree);
 829   while (c_avl_iterator_next(iter, (void *)&name, (void *)&metric) == 0) {
 830     if ((metric->updates_num == 0) &&
 831         ((conf_delete_counters && (metric->type == STATSD_COUNTER)) ||
 832          (conf_delete_timers && (metric->type == STATSD_TIMER)) ||
 833          (conf_delete_gauges && (metric->type == STATSD_GAUGE)) ||
 834          (conf_delete_sets && (metric->type == STATSD_SET)))) {
 835       DEBUG("statsd plugin: Deleting metric \"%s\".", name);
 836       strarray_add(&to_be_deleted, &to_be_deleted_num, name);
 837       continue;
 838     }
 839
 840     /* Names have a prefix, e.g. "c:", which determines the (statsd) type.
 841      * Remove this here. */
 842     statsd_metric_submit_unsafe(name + 2, metric);
 843
 844     /* Reset the metric. */
 845     metric->updates_num = 0;
 846     if (metric->type == STATSD_SET)
 847       statsd_metric_clear_set_unsafe(metric);
 848   }
 849   c_avl_iterator_destroy(iter);
 850
 851   for (size_t i = 0; i < to_be_deleted_num; i++) {
 852     int status;
 853
 854     status = c_avl_remove(metrics_tree, to_be_deleted[i], (void *)&name,
 855                           (void *)&metric);
 856     if (status != 0) {
 857       ERROR("stats plugin: c_avl_remove (\"%s\") failed with status %i.",
 858             to_be_deleted[i], status);
 859       continue;
 860     }
 861
 862     sfree(name);
 863     statsd_metric_free(metric);
 864   }
 865
 866   pthread_mutex_unlock(&metrics_lock);
 867
 868   strarray_free(to_be_deleted, to_be_deleted_num);
 869
 870   return 0;
 871 } /* }}} int statsd_read */
 872
 873 static int statsd_shutdown(void) /* {{{ */
 874 {
 875   void *key;
 876   void *value;
 877
 878   if (network_thread_running) {
 879     network_thread_shutdown = 1;
 880     pthread_kill(network_thread, SIGTERM);
 881     pthread_join(network_thread, /* retval = */ NULL);
 882   }
 883   network_thread_running = 0;
 884
 885   pthread_mutex_lock(&metrics_lock);
 886
 887   while (c_avl_pick(metrics_tree, &key, &value) == 0) {
 888     sfree(key);
 889     statsd_metric_free(value);
 890   }
 891   c_avl_destroy(metrics_tree);
 892   metrics_tree = NULL;
 893
 894   sfree(conf_node);
 895   sfree(conf_service);
 896
 897   pthread_mutex_unlock(&metrics_lock);
 898
 899   return 0;
 900 } /* }}} int statsd_shutdown */
 901
 902 void module_register(void) {
 903   plugin_register_complex_config("statsd", statsd_config);
 904   plugin_register_init("statsd", statsd_init);
 905   plugin_register_read("statsd", statsd_read);
 906   plugin_register_shutdown("statsd", statsd_shutdown);
 907 }