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 = realloc(conf_timer_percentile, sizeof(*conf_timer_percentile) *
 601                                            (conf_timer_percentile_num + 1));
 602   if (tmp == NULL) {
 603     ERROR("statsd plugin: realloc failed.");
 604     return ENOMEM;
 605   }
 606   conf_timer_percentile = tmp;
 607   conf_timer_percentile[conf_timer_percentile_num] = percent;
 608   conf_timer_percentile_num++;
 609
 610   return 0;
 611 } /* }}} int statsd_config_timer_percentile */
 612
 613 static int statsd_config(oconfig_item_t *ci) /* {{{ */
 614 {
 615   for (int i = 0; i < ci->children_num; i++) {
 616     oconfig_item_t *child = ci->children + i;
 617
 618     if (strcasecmp("Host", child->key) == 0)
 619       cf_util_get_string(child, &conf_node);
 620     else if (strcasecmp("Port", child->key) == 0)
 621       cf_util_get_service(child, &conf_service);
 622     else if (strcasecmp("DeleteCounters", child->key) == 0)
 623       cf_util_get_boolean(child, &conf_delete_counters);
 624     else if (strcasecmp("DeleteTimers", child->key) == 0)
 625       cf_util_get_boolean(child, &conf_delete_timers);
 626     else if (strcasecmp("DeleteGauges", child->key) == 0)
 627       cf_util_get_boolean(child, &conf_delete_gauges);
 628     else if (strcasecmp("DeleteSets", child->key) == 0)
 629       cf_util_get_boolean(child, &conf_delete_sets);
 630     else if (strcasecmp("CounterSum", child->key) == 0)
 631       cf_util_get_boolean(child, &conf_counter_sum);
 632     else if (strcasecmp("TimerLower", child->key) == 0)
 633       cf_util_get_boolean(child, &conf_timer_lower);
 634     else if (strcasecmp("TimerUpper", child->key) == 0)
 635       cf_util_get_boolean(child, &conf_timer_upper);
 636     else if (strcasecmp("TimerSum", child->key) == 0)
 637       cf_util_get_boolean(child, &conf_timer_sum);
 638     else if (strcasecmp("TimerCount", child->key) == 0)
 639       cf_util_get_boolean(child, &conf_timer_count);
 640     else if (strcasecmp("TimerPercentile", child->key) == 0)
 641       statsd_config_timer_percentile(child);
 642     else
 643       ERROR("statsd plugin: The \"%s\" config option is not valid.",
 644             child->key);
 645   }
 646
 647   return 0;
 648 } /* }}} int statsd_config */
 649
 650 static int statsd_init(void) /* {{{ */
 651 {
 652   pthread_mutex_lock(&metrics_lock);
 653   if (metrics_tree == NULL)
 654     metrics_tree = c_avl_create((int (*)(const void *, const void *))strcmp);
 655
 656   if (!network_thread_running) {
 657     int status;
 658
 659     status = pthread_create(&network_thread,
 660                             /* attr = */ NULL, statsd_network_thread,
 661                             /* args = */ NULL);
 662     if (status != 0) {
 663       pthread_mutex_unlock(&metrics_lock);
 664       ERROR("statsd plugin: pthread_create failed: %s", STRERRNO);
 665       return status;
 666     }
 667   }
 668   network_thread_running = 1;
 669
 670   pthread_mutex_unlock(&metrics_lock);
 671
 672   return 0;
 673 } /* }}} int statsd_init */
 674
 675 /* Must hold metrics_lock when calling this function. */
 676 static int statsd_metric_clear_set_unsafe(statsd_metric_t *metric) /* {{{ */
 677 {
 678   void *key;
 679   void *value;
 680
 681   if ((metric == NULL) || (metric->type != STATSD_SET))
 682     return EINVAL;
 683
 684   if (metric->set == NULL)
 685     return 0;
 686
 687   while (c_avl_pick(metric->set, &key, &value) == 0) {
 688     sfree(key);
 689     sfree(value);
 690   }
 691
 692   return 0;
 693 } /* }}} int statsd_metric_clear_set_unsafe */
 694
 695 /* Must hold metrics_lock when calling this function. */
 696 static int statsd_metric_submit_unsafe(char const *name,
 697                                        statsd_metric_t *metric) /* {{{ */
 698 {
 699   value_list_t vl = VALUE_LIST_INIT;
 700
 701   vl.values = &(value_t){.gauge = NAN};
 702   vl.values_len = 1;
 703   sstrncpy(vl.plugin, "statsd", sizeof(vl.plugin));
 704
 705   if (metric->type == STATSD_GAUGE)
 706     sstrncpy(vl.type, "gauge", sizeof(vl.type));
 707   else if (metric->type == STATSD_TIMER)
 708     sstrncpy(vl.type, "latency", sizeof(vl.type));
 709   else if (metric->type == STATSD_SET)
 710     sstrncpy(vl.type, "objects", sizeof(vl.type));
 711   else /* if (metric->type == STATSD_COUNTER) */
 712     sstrncpy(vl.type, "derive", sizeof(vl.type));
 713
 714   sstrncpy(vl.type_instance, name, sizeof(vl.type_instance));
 715
 716   if (metric->type == STATSD_GAUGE)
 717     vl.values[0].gauge = (gauge_t)metric->value;
 718   else if (metric->type == STATSD_TIMER) {
 719     _Bool have_events = (metric->updates_num > 0);
 720
 721     /* Make sure all timer metrics share the *same* timestamp. */
 722     vl.time = cdtime();
 723
 724     snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-average", name);
 725     vl.values[0].gauge =
 726         have_events
 727             ? CDTIME_T_TO_DOUBLE(latency_counter_get_average(metric->latency))
 728             : NAN;
 729     plugin_dispatch_values(&vl);
 730
 731     if (conf_timer_lower) {
 732       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-lower", name);
 733       vl.values[0].gauge =
 734           have_events
 735               ? CDTIME_T_TO_DOUBLE(latency_counter_get_min(metric->latency))
 736               : NAN;
 737       plugin_dispatch_values(&vl);
 738     }
 739
 740     if (conf_timer_upper) {
 741       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-upper", name);
 742       vl.values[0].gauge =
 743           have_events
 744               ? CDTIME_T_TO_DOUBLE(latency_counter_get_max(metric->latency))
 745               : NAN;
 746       plugin_dispatch_values(&vl);
 747     }
 748
 749     if (conf_timer_sum) {
 750       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-sum", name);
 751       vl.values[0].gauge =
 752           have_events
 753               ? CDTIME_T_TO_DOUBLE(latency_counter_get_sum(metric->latency))
 754               : NAN;
 755       plugin_dispatch_values(&vl);
 756     }
 757
 758     for (size_t i = 0; i < conf_timer_percentile_num; i++) {
 759       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-percentile-%.0f",
 760                name, conf_timer_percentile[i]);
 761       vl.values[0].gauge =
 762           have_events ? CDTIME_T_TO_DOUBLE(latency_counter_get_percentile(
 763                             metric->latency, conf_timer_percentile[i]))
 764                       : NAN;
 765       plugin_dispatch_values(&vl);
 766     }
 767
 768     /* Keep this at the end, since vl.type is set to "gauge" here. The
 769      * vl.type's above are implicitly set to "latency". */
 770     if (conf_timer_count) {
 771       sstrncpy(vl.type, "gauge", sizeof(vl.type));
 772       snprintf(vl.type_instance, sizeof(vl.type_instance), "%s-count", name);
 773       vl.values[0].gauge = latency_counter_get_num(metric->latency);
 774       plugin_dispatch_values(&vl);
 775     }
 776
 777     latency_counter_reset(metric->latency);
 778     return 0;
 779   } else if (metric->type == STATSD_SET) {
 780     if (metric->set == NULL)
 781       vl.values[0].gauge = 0.0;
 782     else
 783       vl.values[0].gauge = (gauge_t)c_avl_size(metric->set);
 784   } else { /* STATSD_COUNTER */
 785     gauge_t delta = nearbyint(metric->value);
 786
 787     /* Etsy's statsd writes counters as two metrics: a rate and the change since
 788      * the last write. Since collectd does not reset its DERIVE metrics to zero,
 789      * this makes little sense, but we're dispatching a "count" metric here
 790      * anyway - if requested by the user - for compatibility reasons. */
 791     if (conf_counter_sum) {
 792       sstrncpy(vl.type, "count", sizeof(vl.type));
 793       vl.values[0].gauge = delta;
 794       plugin_dispatch_values(&vl);
 795
 796       /* restore vl.type */
 797       sstrncpy(vl.type, "derive", sizeof(vl.type));
 798     }
 799
 800     /* Rather than resetting value to zero, subtract delta so we correctly keep
 801      * track of residuals. */
 802     metric->value -= delta;
 803     metric->counter += (derive_t)delta;
 804
 805     vl.values[0].derive = metric->counter;
 806   }
 807
 808   return plugin_dispatch_values(&vl);
 809 } /* }}} int statsd_metric_submit_unsafe */
 810
 811 static int statsd_read(void) /* {{{ */
 812 {
 813   c_avl_iterator_t *iter;
 814   char *name;
 815   statsd_metric_t *metric;
 816
 817   char **to_be_deleted = NULL;
 818   size_t to_be_deleted_num = 0;
 819
 820   pthread_mutex_lock(&metrics_lock);
 821
 822   if (metrics_tree == NULL) {
 823     pthread_mutex_unlock(&metrics_lock);
 824     return 0;
 825   }
 826
 827   iter = c_avl_get_iterator(metrics_tree);
 828   while (c_avl_iterator_next(iter, (void *)&name, (void *)&metric) == 0) {
 829     if ((metric->updates_num == 0) &&
 830         ((conf_delete_counters && (metric->type == STATSD_COUNTER)) ||
 831          (conf_delete_timers && (metric->type == STATSD_TIMER)) ||
 832          (conf_delete_gauges && (metric->type == STATSD_GAUGE)) ||
 833          (conf_delete_sets && (metric->type == STATSD_SET)))) {
 834       DEBUG("statsd plugin: Deleting metric \"%s\".", name);
 835       strarray_add(&to_be_deleted, &to_be_deleted_num, name);
 836       continue;
 837     }
 838
 839     /* Names have a prefix, e.g. "c:", which determines the (statsd) type.
 840      * Remove this here. */
 841     statsd_metric_submit_unsafe(name + 2, metric);
 842
 843     /* Reset the metric. */
 844     metric->updates_num = 0;
 845     if (metric->type == STATSD_SET)
 846       statsd_metric_clear_set_unsafe(metric);
 847   }
 848   c_avl_iterator_destroy(iter);
 849
 850   for (size_t i = 0; i < to_be_deleted_num; i++) {
 851     int status;
 852
 853     status = c_avl_remove(metrics_tree, to_be_deleted[i], (void *)&name,
 854                           (void *)&metric);
 855     if (status != 0) {
 856       ERROR("stats plugin: c_avl_remove (\"%s\") failed with status %i.",
 857             to_be_deleted[i], status);
 858       continue;
 859     }
 860
 861     sfree(name);
 862     statsd_metric_free(metric);
 863   }
 864
 865   pthread_mutex_unlock(&metrics_lock);
 866
 867   strarray_free(to_be_deleted, to_be_deleted_num);
 868
 869   return 0;
 870 } /* }}} int statsd_read */
 871
 872 static int statsd_shutdown(void) /* {{{ */
 873 {
 874   void *key;
 875   void *value;
 876
 877   if (network_thread_running) {
 878     network_thread_shutdown = 1;
 879     pthread_kill(network_thread, SIGTERM);
 880     pthread_join(network_thread, /* retval = */ NULL);
 881   }
 882   network_thread_running = 0;
 883
 884   pthread_mutex_lock(&metrics_lock);
 885
 886   while (c_avl_pick(metrics_tree, &key, &value) == 0) {
 887     sfree(key);
 888     statsd_metric_free(value);
 889   }
 890   c_avl_destroy(metrics_tree);
 891   metrics_tree = NULL;
 892
 893   sfree(conf_node);
 894   sfree(conf_service);
 895
 896   pthread_mutex_unlock(&metrics_lock);
 897
 898   return 0;
 899 } /* }}} int statsd_shutdown */
 900
 901 void module_register(void) {
 902   plugin_register_complex_config("statsd", statsd_config);
 903   plugin_register_init("statsd", statsd_init);
 904   plugin_register_read("statsd", statsd_read);
 905   plugin_register_shutdown("statsd", statsd_shutdown);
 906 }