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     char errbuf[1024];
 450
 451     if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
 452       return;
 453
 454     ERROR("statsd plugin: recv(2) failed: %s",
 455           sstrerror(errno, errbuf, sizeof(errbuf)));
 456     return;
 457   }
 458
 459   buffer_size = (size_t)status;
 460   if (buffer_size >= sizeof(buffer))
 461     buffer_size = sizeof(buffer) - 1;
 462   buffer[buffer_size] = 0;
 463
 464   statsd_parse_buffer(buffer);
 465 } /* }}} void statsd_network_read */
 466
 467 static int statsd_network_init(struct pollfd **ret_fds, /* {{{ */
 468                                size_t *ret_fds_num) {
 469   struct pollfd *fds = NULL;
 470   size_t fds_num = 0;
 471
 472   struct addrinfo *ai_list;
 473   int status;
 474
 475   char const *node = (conf_node != NULL) ? conf_node : STATSD_DEFAULT_NODE;
 476   char const *service =
 477       (conf_service != NULL) ? conf_service : STATSD_DEFAULT_SERVICE;
 478
 479   struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
 480                               .ai_flags = AI_PASSIVE | AI_ADDRCONFIG,
 481                               .ai_socktype = SOCK_DGRAM};
 482
 483   status = getaddrinfo(node, service, &ai_hints, &ai_list);
 484   if (status != 0) {
 485     ERROR("statsd plugin: getaddrinfo (\"%s\", \"%s\") failed: %s", node,
 486           service, gai_strerror(status));
 487     return status;
 488   }
 489
 490   for (struct addrinfo *ai_ptr = ai_list; ai_ptr != NULL;
 491        ai_ptr = ai_ptr->ai_next) {
 492     int fd;
 493     struct pollfd *tmp;
 494
 495     char dbg_node[NI_MAXHOST];
 496     char dbg_service[NI_MAXSERV];
 497
 498     fd = socket(ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
 499     if (fd < 0) {
 500       char errbuf[1024];
 501       ERROR("statsd plugin: socket(2) failed: %s",
 502             sstrerror(errno, errbuf, sizeof(errbuf)));
 503       continue;
 504     }
 505
 506     getnameinfo(ai_ptr->ai_addr, ai_ptr->ai_addrlen, dbg_node, sizeof(dbg_node),
 507                 dbg_service, sizeof(dbg_service),
 508                 NI_DGRAM | NI_NUMERICHOST | NI_NUMERICSERV);
 509     DEBUG("statsd plugin: Trying to bind to [%s]:%s ...", dbg_node,
 510           dbg_service);
 511
 512     status = bind(fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 513     if (status != 0) {
 514       char errbuf[1024];
 515       ERROR("statsd plugin: bind(2) failed: %s",
 516             sstrerror(errno, errbuf, sizeof(errbuf)));
 517       close(fd);
 518       continue;
 519     }
 520
 521     tmp = realloc(fds, sizeof(*fds) * (fds_num + 1));
 522     if (tmp == NULL) {
 523       ERROR("statsd plugin: realloc failed.");
 524       close(fd);
 525       continue;
 526     }
 527     fds = tmp;
 528     tmp = fds + fds_num;
 529     fds_num++;
 530
 531     memset(tmp, 0, sizeof(*tmp));
 532     tmp->fd = fd;
 533     tmp->events = POLLIN | POLLPRI;
 534   }
 535
 536   freeaddrinfo(ai_list);
 537
 538   if (fds_num == 0) {
 539     ERROR("statsd plugin: Unable to create listening socket for [%s]:%s.",
 540           (node != NULL) ? node : "::", service);
 541     return ENOENT;
 542   }
 543
 544   *ret_fds = fds;
 545   *ret_fds_num = fds_num;
 546   return 0;
 547 } /* }}} int statsd_network_init */
 548
 549 static void *statsd_network_thread(void *args) /* {{{ */
 550 {
 551   struct pollfd *fds = NULL;
 552   size_t fds_num = 0;
 553   int status;
 554
 555   status = statsd_network_init(&fds, &fds_num);
 556   if (status != 0) {
 557     ERROR("statsd plugin: Unable to open listening sockets.");
 558     pthread_exit((void *)0);
 559   }
 560
 561   while (!network_thread_shutdown) {
 562     status = poll(fds, (nfds_t)fds_num, /* timeout = */ -1);
 563     if (status < 0) {
 564       char errbuf[1024];
 565
 566       if ((errno == EINTR) || (errno == EAGAIN))
 567         continue;
 568
 569       ERROR("statsd plugin: poll(2) failed: %s",
 570             sstrerror(errno, errbuf, sizeof(errbuf)));
 571       break;
 572     }
 573
 574     for (size_t i = 0; i < fds_num; i++) {
 575       if ((fds[i].revents & (POLLIN | POLLPRI)) == 0)
 576         continue;
 577
 578       statsd_network_read(fds[i].fd);
 579       fds[i].revents = 0;
 580     }
 581   } /* while (!network_thread_shutdown) */
 582
 583   /* Clean up */
 584   for (size_t i = 0; i < fds_num; i++)
 585     close(fds[i].fd);
 586   sfree(fds);
 587
 588   return (void *)0;
 589 } /* }}} void *statsd_network_thread */
 590
 591 static int statsd_config_timer_percentile(oconfig_item_t *ci) /* {{{ */
 592 {
 593   double percent = NAN;
 594   double *tmp;
 595   int status;
 596
 597   status = cf_util_get_double(ci, &percent);
 598   if (status != 0)
 599     return status;
 600
 601   if ((percent <= 0.0) || (percent >= 100)) {
 602     ERROR("statsd plugin: The value for \"%s\" must be between 0 and 100, "
 603           "exclusively.",
 604           ci->key);
 605     return ERANGE;
 606   }
 607
 608   tmp =
 609       realloc(conf_timer_percentile,
 610               sizeof(*conf_timer_percentile) * (conf_timer_percentile_num + 1));
 611   if (tmp == NULL) {
 612     ERROR("statsd plugin: realloc failed.");
 613     return ENOMEM;
 614   }
 615   conf_timer_percentile = tmp;
 616   conf_timer_percentile[conf_timer_percentile_num] = percent;
 617   conf_timer_percentile_num++;
 618
 619   return 0;
 620 } /* }}} int statsd_config_timer_percentile */
 621
 622 static int statsd_config(oconfig_item_t *ci) /* {{{ */
 623 {
 624   for (int i = 0; i < ci->children_num; i++) {
 625     oconfig_item_t *child = ci->children + i;
 626
 627     if (strcasecmp("Host", child->key) == 0)
 628       cf_util_get_string(child, &conf_node);
 629     else if (strcasecmp("Port", child->key) == 0)
 630       cf_util_get_service(child, &conf_service);
 631     else if (strcasecmp("DeleteCounters", child->key) == 0)
 632       cf_util_get_boolean(child, &conf_delete_counters);
 633     else if (strcasecmp("DeleteTimers", child->key) == 0)
 634       cf_util_get_boolean(child, &conf_delete_timers);
 635     else if (strcasecmp("DeleteGauges", child->key) == 0)
 636       cf_util_get_boolean(child, &conf_delete_gauges);
 637     else if (strcasecmp("DeleteSets", child->key) == 0)
 638       cf_util_get_boolean(child, &conf_delete_sets);
 639     else if (strcasecmp("CounterSum", child->key) == 0)
 640       cf_util_get_boolean(child, &conf_counter_sum);
 641     else if (strcasecmp("TimerLower", child->key) == 0)
 642       cf_util_get_boolean(child, &conf_timer_lower);
 643     else if (strcasecmp("TimerUpper", child->key) == 0)
 644       cf_util_get_boolean(child, &conf_timer_upper);
 645     else if (strcasecmp("TimerSum", child->key) == 0)
 646       cf_util_get_boolean(child, &conf_timer_sum);
 647     else if (strcasecmp("TimerCount", child->key) == 0)
 648       cf_util_get_boolean(child, &conf_timer_count);
 649     else if (strcasecmp("TimerPercentile", child->key) == 0)
 650       statsd_config_timer_percentile(child);
 651     else
 652       ERROR("statsd plugin: The \"%s\" config option is not valid.",
 653             child->key);
 654   }
 655
 656   return 0;
 657 } /* }}} int statsd_config */
 658
 659 static int statsd_init(void) /* {{{ */
 660 {
 661   pthread_mutex_lock(&metrics_lock);
 662   if (metrics_tree == NULL)
 663     metrics_tree = c_avl_create((int (*)(const void *, const void *))strcmp);
 664
 665   if (!network_thread_running) {
 666     int status;
 667
 668     status = pthread_create(&network_thread,
 669                             /* attr = */ NULL, statsd_network_thread,
 670                             /* args = */ NULL);
 671     if (status != 0) {
 672       char errbuf[1024];
 673       pthread_mutex_unlock(&metrics_lock);
 674       ERROR("statsd plugin: pthread_create failed: %s",
 675             sstrerror(errno, errbuf, sizeof(errbuf)));
 676       return status;
 677     }
 678   }
 679   network_thread_running = 1;
 680
 681   pthread_mutex_unlock(&metrics_lock);
 682
 683   return 0;
 684 } /* }}} int statsd_init */
 685
 686 /* Must hold metrics_lock when calling this function. */
 687 static int statsd_metric_clear_set_unsafe(statsd_metric_t *metric) /* {{{ */
 688 {
 689   void *key;
 690   void *value;
 691
 692   if ((metric == NULL) || (metric->type != STATSD_SET))
 693     return EINVAL;
 694
 695   if (metric->set == NULL)
 696     return 0;
 697
 698   while (c_avl_pick(metric->set, &key, &value) == 0) {
 699     sfree(key);
 700     sfree(value);
 701   }
 702
 703   return 0;
 704 } /* }}} int statsd_metric_clear_set_unsafe */
 705
 706 /* Must hold metrics_lock when calling this function. */
 707 static int statsd_metric_submit_unsafe(char const *name,
 708                                        statsd_metric_t *metric) /* {{{ */
 709 {
 710   value_list_t vl = VALUE_LIST_INIT;
 711
 712   vl.values = &(value_t){.gauge = NAN};
 713   vl.values_len = 1;
 714   sstrncpy(vl.plugin, "statsd", sizeof(vl.plugin));
 715
 716   if (metric->type == STATSD_GAUGE)
 717     sstrncpy(vl.type, "gauge", sizeof(vl.type));
 718   else if (metric->type == STATSD_TIMER)
 719     sstrncpy(vl.type, "latency", sizeof(vl.type));
 720   else if (metric->type == STATSD_SET)
 721     sstrncpy(vl.type, "objects", sizeof(vl.type));
 722   else /* if (metric->type == STATSD_COUNTER) */
 723     sstrncpy(vl.type, "derive", sizeof(vl.type));
 724
 725   sstrncpy(vl.type_instance, name, sizeof(vl.type_instance));
 726
 727   if (metric->type == STATSD_GAUGE)
 728     vl.values[0].gauge = (gauge_t)metric->value;
 729   else if (metric->type == STATSD_TIMER) {
 730     _Bool have_events = (metric->updates_num > 0);
 731
 732     /* Make sure all timer metrics share the *same* timestamp. */
 733     vl.time = cdtime();
 734
 735     snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-average", name);
 736     vl.values[0].gauge =
 737         have_events
 738             ? CDTIME_T_TO_DOUBLE(latency_counter_get_average(metric->latency))
 739             : NAN;
 740     plugin_dispatch_values(&vl);
 741
 742     if (conf_timer_lower) {
 743       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-lower", name);
 744       vl.values[0].gauge =
 745           have_events
 746               ? CDTIME_T_TO_DOUBLE(latency_counter_get_min(metric->latency))
 747               : NAN;
 748       plugin_dispatch_values(&vl);
 749     }
 750
 751     if (conf_timer_upper) {
 752       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-upper", name);
 753       vl.values[0].gauge =
 754           have_events
 755               ? CDTIME_T_TO_DOUBLE(latency_counter_get_max(metric->latency))
 756               : NAN;
 757       plugin_dispatch_values(&vl);
 758     }
 759
 760     if (conf_timer_sum) {
 761       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-sum", name);
 762       vl.values[0].gauge =
 763           have_events
 764               ? CDTIME_T_TO_DOUBLE(latency_counter_get_sum(metric->latency))
 765               : NAN;
 766       plugin_dispatch_values(&vl);
 767     }
 768
 769     for (size_t i = 0; i < conf_timer_percentile_num; i++) {
 770       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-percentile-%.0f",
 771                name, conf_timer_percentile[i]);
 772       vl.values[0].gauge =
 773           have_events ? CDTIME_T_TO_DOUBLE(latency_counter_get_percentile(
 774                             metric->latency, conf_timer_percentile[i]))
 775                       : NAN;
 776       plugin_dispatch_values(&vl);
 777     }
 778
 779     /* Keep this at the end, since vl.type is set to "gauge" here. The
 780      * vl.type's above are implicitly set to "latency". */
 781     if (conf_timer_count) {
 782       sstrncpy(vl.type, "gauge", sizeof(vl.type));
 783       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-count", name);
 784       vl.values[0].gauge = latency_counter_get_num(metric->latency);
 785       plugin_dispatch_values(&vl);
 786     }
 787
 788     latency_counter_reset(metric->latency);
 789     return 0;
 790   } else if (metric->type == STATSD_SET) {
 791     if (metric->set == NULL)
 792       vl.values[0].gauge = 0.0;
 793     else
 794       vl.values[0].gauge = (gauge_t)c_avl_size(metric->set);
 795   } else { /* STATSD_COUNTER */
 796     gauge_t delta = nearbyint(metric->value);
 797
 798     /* Etsy's statsd writes counters as two metrics: a rate and the change since
 799      * the last write. Since collectd does not reset its DERIVE metrics to zero,
 800      * this makes little sense, but we're dispatching a "count" metric here
 801      * anyway - if requested by the user - for compatibility reasons. */
 802     if (conf_counter_sum) {
 803       sstrncpy(vl.type, "count", sizeof(vl.type));
 804       vl.values[0].gauge = delta;
 805       plugin_dispatch_values(&vl);
 806
 807       /* restore vl.type */
 808       sstrncpy(vl.type, "derive", sizeof(vl.type));
 809     }
 810
 811     /* Rather than resetting value to zero, subtract delta so we correctly keep
 812      * track of residuals. */
 813     metric->value -= delta;
 814     metric->counter += (derive_t)delta;
 815
 816     vl.values[0].derive = metric->counter;
 817   }
 818
 819   return plugin_dispatch_values(&vl);
 820 } /* }}} int statsd_metric_submit_unsafe */
 821
 822 static int statsd_read(void) /* {{{ */
 823 {
 824   c_avl_iterator_t *iter;
 825   char *name;
 826   statsd_metric_t *metric;
 827
 828   char **to_be_deleted = NULL;
 829   size_t to_be_deleted_num = 0;
 830
 831   pthread_mutex_lock(&metrics_lock);
 832
 833   if (metrics_tree == NULL) {
 834     pthread_mutex_unlock(&metrics_lock);
 835     return 0;
 836   }
 837
 838   iter = c_avl_get_iterator(metrics_tree);
 839   while (c_avl_iterator_next(iter, (void *)&name, (void *)&metric) == 0) {
 840     if ((metric->updates_num == 0) &&
 841         ((conf_delete_counters && (metric->type == STATSD_COUNTER)) ||
 842          (conf_delete_timers && (metric->type == STATSD_TIMER)) ||
 843          (conf_delete_gauges && (metric->type == STATSD_GAUGE)) ||
 844          (conf_delete_sets && (metric->type == STATSD_SET)))) {
 845       DEBUG("statsd plugin: Deleting metric \"%s\".", name);
 846       strarray_add(&to_be_deleted, &to_be_deleted_num, name);
 847       continue;
 848     }
 849
 850     /* Names have a prefix, e.g. "c:", which determines the (statsd) type.
 851      * Remove this here. */
 852     statsd_metric_submit_unsafe(name + 2, metric);
 853
 854     /* Reset the metric. */
 855     metric->updates_num = 0;
 856     if (metric->type == STATSD_SET)
 857       statsd_metric_clear_set_unsafe(metric);
 858   }
 859   c_avl_iterator_destroy(iter);
 860
 861   for (size_t i = 0; i < to_be_deleted_num; i++) {
 862     int status;
 863
 864     status = c_avl_remove(metrics_tree, to_be_deleted[i], (void *)&name,
 865                           (void *)&metric);
 866     if (status != 0) {
 867       ERROR("stats plugin: c_avl_remove (\"%s\") failed with status %i.",
 868             to_be_deleted[i], status);
 869       continue;
 870     }
 871
 872     sfree(name);
 873     statsd_metric_free(metric);
 874   }
 875
 876   pthread_mutex_unlock(&metrics_lock);
 877
 878   strarray_free(to_be_deleted, to_be_deleted_num);
 879
 880   return 0;
 881 } /* }}} int statsd_read */
 882
 883 static int statsd_shutdown(void) /* {{{ */
 884 {
 885   void *key;
 886   void *value;
 887
 888   if (network_thread_running) {
 889     network_thread_shutdown = 1;
 890     pthread_kill(network_thread, SIGTERM);
 891     pthread_join(network_thread, /* retval = */ NULL);
 892   }
 893   network_thread_running = 0;
 894
 895   pthread_mutex_lock(&metrics_lock);
 896
 897   while (c_avl_pick(metrics_tree, &key, &value) == 0) {
 898     sfree(key);
 899     statsd_metric_free(value);
 900   }
 901   c_avl_destroy(metrics_tree);
 902   metrics_tree = NULL;
 903
 904   sfree(conf_node);
 905   sfree(conf_service);
 906
 907   pthread_mutex_unlock(&metrics_lock);
 908
 909   return 0;
 910 } /* }}} int statsd_shutdown */
 911
 912 void module_register(void) {
 913   plugin_register_complex_config("statsd", statsd_config);
 914   plugin_register_init("statsd", statsd_init);
 915   plugin_register_read("statsd", statsd_read);
 916   plugin_register_shutdown("statsd", statsd_shutdown);
 917 }