src/statsd.c

   1 /**
   2  * collectd - src/statsd.c
   3  *
   4  * Copyright (C) 2013       Florian octo Forster
   5  *
   6  * Permission to use, copy, modify, and distribute this software for any
   7  * purpose with or without fee is hereby granted, provided that the above
   8  * copyright notice and this permission notice appear in all copies.
   9  *
  10  * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
  11  * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  12  * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
  13  * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  14  * WHATSOEVER RESULTING FROM LOSS OF MIND, USE, DATA OR PROFITS, WHETHER
  15  * IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING
  16  * OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  17  *
  18  * Authors:
  19  *   Florian octo Forster <octo at collectd.org>
  20  */
  21
  22 #include "collectd.h"
  23 #include "plugin.h"
  24 #include "common.h"
  25 #include "configfile.h"
  26 #include "utils_avltree.h"
  27 #include "utils_complain.h"
  28
  29 #include <pthread.h>
  30
  31 #include <sys/types.h>
  32 #include <sys/socket.h>
  33 #include <netdb.h>
  34 #include <poll.h>
  35
  36 #ifndef STATSD_DEFAULT_NODE
  37 # define STATSD_DEFAULT_NODE NULL
  38 #endif
  39
  40 #ifndef STATSD_DEFAULT_SERVICE
  41 # define STATSD_DEFAULT_SERVICE "8125"
  42 #endif
  43
  44 enum metric_type_e
  45 {
  46   STATSD_COUNTER,
  47   STATSD_TIMER,
  48   STATSD_GAUGE
  49 };
  50 typedef enum metric_type_e metric_type_t;
  51
  52 struct statsd_metric_s
  53 {
  54   metric_type_t type;
  55   int64_t value;
  56   unsigned long updates_num;
  57 };
  58 typedef struct statsd_metric_s statsd_metric_t;
  59
  60 static c_avl_tree_t   *metrics_tree = NULL;
  61 static pthread_mutex_t metrics_lock = PTHREAD_MUTEX_INITIALIZER;
  62
  63 static pthread_t network_thread;
  64 static _Bool     network_thread_running = 0;
  65 static _Bool     network_thread_shutdown = 0;
  66
  67 static char *conf_node = NULL;
  68 static char *conf_service = NULL;
  69
  70 static _Bool conf_delete_counters = 0;
  71 static _Bool conf_delete_timers   = 0;
  72 static _Bool conf_delete_gauges   = 0;
  73
  74 /* Must hold metrics_lock when calling this function. */
  75 static int statsd_metric_set_unsafe (char const *name, int64_t value, /* {{{ */
  76     metric_type_t type)
  77 {
  78   statsd_metric_t *metric;
  79   char *key;
  80   int status;
  81
  82   status = c_avl_get (metrics_tree, name, (void *) &metric);
  83   if (status == 0)
  84   {
  85     metric->value = value;
  86     metric->updates_num++;
  87
  88     return (0);
  89   }
  90
  91   DEBUG ("stats plugin: Adding new metric \"%s\".", name);
  92   key = strdup (name);
  93   metric = calloc (1, sizeof (*metric));
  94   if ((key == NULL) || (metric == NULL))
  95   {
  96     sfree (key);
  97     sfree (metric);
  98     return (-1);
  99   }
 100
 101   metric->type = type;
 102   metric->value = value;
 103   metric->updates_num = 1;
 104
 105   status = c_avl_insert (metrics_tree, key, metric);
 106   if (status != 0)
 107   {
 108     sfree (key);
 109     sfree (metric);
 110
 111     return (-1);
 112   }
 113
 114   return (0);
 115 } /* }}} int statsd_metric_set_unsafe */
 116
 117 static int statsd_metric_set (char const *name, int64_t value, /* {{{ */
 118     metric_type_t type)
 119 {
 120   int status;
 121
 122   pthread_mutex_lock (&metrics_lock);
 123   status = statsd_metric_set_unsafe (name, value, type);
 124   pthread_mutex_unlock (&metrics_lock);
 125
 126   return (status);
 127 } /* }}} int statsd_metric_set */
 128
 129 static int statsd_metric_add (char const *name, int64_t delta, /* {{{ */
 130     metric_type_t type)
 131 {
 132   statsd_metric_t *metric;
 133   int status;
 134
 135   pthread_mutex_lock (&metrics_lock);
 136
 137   status = c_avl_get (metrics_tree, name, (void *) &metric);
 138   if (status == 0)
 139   {
 140     metric->value += delta;
 141     metric->updates_num++;
 142
 143     pthread_mutex_unlock (&metrics_lock);
 144     return (0);
 145   }
 146   else /* no such value yet */
 147   {
 148     status = statsd_metric_set_unsafe (name, delta, type);
 149
 150     pthread_mutex_unlock (&metrics_lock);
 151     return (status);
 152   }
 153 } /* }}} int statsd_metric_add */
 154
 155 static int statsd_handle_counter (char const *name, /* {{{ */
 156     char const *value_str,
 157     char const *extra)
 158 {
 159   char key[DATA_MAX_NAME_LEN + 2];
 160   value_t value;
 161   value_t scale;
 162   int status;
 163
 164   if ((extra != NULL) && (extra[0] != '@'))
 165     return (-1);
 166
 167   scale.gauge = 1.0;
 168   if (extra != NULL)
 169   {
 170     status = parse_value (extra + 1, &scale, DS_TYPE_GAUGE);
 171     if (status != 0)
 172       return (status);
 173
 174     if (!isfinite (scale.gauge) || (scale.gauge <= 0.0) || (scale.gauge > 1.0))
 175       return (-1);
 176   }
 177
 178   value.derive = 1;
 179   status = parse_value (value_str, &value, DS_TYPE_DERIVE);
 180   if (status != 0)
 181     return (status);
 182
 183   if (value.derive < 1)
 184     return (-1);
 185
 186   ssnprintf (key, sizeof (key), "c:%s", name);
 187
 188   return (statsd_metric_add (key,
 189         (int64_t) (((gauge_t) value.derive) / scale.gauge),
 190         STATSD_COUNTER));
 191 } /* }}} int statsd_handle_counter */
 192
 193 static int statsd_handle_gauge (char const *name, /* {{{ */
 194     char const *value_str)
 195 {
 196   char key[DATA_MAX_NAME_LEN + 2];
 197   value_t value;
 198   int status;
 199
 200   value.derive = 0;
 201   status = parse_value (value_str, &value, DS_TYPE_DERIVE);
 202   if (status != 0)
 203     return (status);
 204
 205   ssnprintf (key, sizeof (key), "g:%s", name);
 206
 207   if ((value_str[0] == '+') || (value_str[0] == '-'))
 208     return (statsd_metric_add (key, (int64_t) value.derive, STATSD_GAUGE));
 209   else
 210     return (statsd_metric_set (key, (int64_t) value.derive, STATSD_GAUGE));
 211 } /* }}} int statsd_handle_gauge */
 212
 213 static int statsd_handle_timer (char const *name, /* {{{ */
 214     char const *value_str)
 215 {
 216   char key[DATA_MAX_NAME_LEN + 2];
 217   value_t value;
 218   int status;
 219
 220   value.derive = 0;
 221   status = parse_value (value_str, &value, DS_TYPE_DERIVE);
 222   if (status != 0)
 223     return (status);
 224
 225   ssnprintf (key, sizeof (key), "t:%s", name);
 226
 227   return (statsd_metric_add (key, (int64_t) value.derive, STATSD_TIMER));
 228 } /* }}} int statsd_handle_timer */
 229
 230 static int statsd_handle_set (char const *name __attribute__((unused)), /* {{{ */
 231     char const *value_str __attribute__((unused)))
 232 {
 233   static c_complain_t c = C_COMPLAIN_INIT_STATIC;
 234
 235   c_complain (LOG_WARNING, &c,
 236       "statsd plugin: Support for sets is not yet implemented.");
 237
 238   return (0);
 239 } /* }}} int statsd_handle_set */
 240
 241 static int statsd_parse_line (char *buffer) /* {{{ */
 242 {
 243   char *name = buffer;
 244   char *value;
 245   char *type;
 246   char *extra;
 247
 248   type = strchr (name, '|');
 249   if (type == NULL)
 250     return (-1);
 251   *type = 0;
 252   type++;
 253
 254   value = strrchr (name, ':');
 255   if (value == NULL)
 256     return (-1);
 257   *value = 0;
 258   value++;
 259
 260   extra = strchr (type, '|');
 261   if (extra != NULL)
 262   {
 263     *extra = 0;
 264     extra++;
 265   }
 266
 267   if (strcmp ("c", type) == 0)
 268     return (statsd_handle_counter (name, value, extra));
 269
 270   /* extra is only valid for counters */
 271   if (extra != NULL)
 272     return (-1);
 273
 274   if (strcmp ("g", type) == 0)
 275     return (statsd_handle_gauge (name, value));
 276   else if (strcmp ("ms", type) == 0)
 277     return (statsd_handle_timer (name, value));
 278   else if (strcmp ("s", type) == 0)
 279     return (statsd_handle_set (name, value));
 280   else
 281     return (-1);
 282 } /* }}} void statsd_parse_line */
 283
 284 static void statsd_parse_buffer (char *buffer) /* {{{ */
 285 {
 286   char *dummy;
 287   char *saveptr = NULL;
 288   char *ptr;
 289
 290   for (dummy = buffer;
 291       (ptr = strtok_r (dummy, "\r\n", &saveptr)) != NULL;
 292       dummy = NULL)
 293   {
 294     char *line_orig = sstrdup (ptr);
 295     int status;
 296
 297     status = statsd_parse_line (ptr);
 298     if (status != 0)
 299       ERROR ("statsd plugin: Unable to parse line: \"%s\"", line_orig);
 300
 301     sfree (line_orig);
 302   }
 303 } /* }}} void statsd_parse_buffer */
 304
 305 static void statsd_network_read (int fd) /* {{{ */
 306 {
 307   char buffer[4096];
 308   size_t buffer_size;
 309   ssize_t status;
 310
 311   status = recv (fd, buffer, sizeof (buffer), /* flags = */ MSG_DONTWAIT);
 312   if (status < 0)
 313   {
 314     char errbuf[1024];
 315
 316     if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
 317       return;
 318
 319     ERROR ("statsd plugin: recv(2) failed: %s",
 320         sstrerror (errno, errbuf, sizeof (errbuf)));
 321     return;
 322   }
 323
 324   buffer_size = (size_t) status;
 325   if (buffer_size >= sizeof (buffer))
 326     buffer_size = sizeof (buffer) - 1;
 327   buffer[buffer_size] = 0;
 328
 329   statsd_parse_buffer (buffer);
 330 } /* }}} void statsd_network_read */
 331
 332 static int statsd_network_init (struct pollfd **ret_fds, /* {{{ */
 333     size_t *ret_fds_num)
 334 {
 335   struct pollfd *fds = NULL;
 336   size_t fds_num = 0;
 337
 338   struct addrinfo ai_hints;
 339   struct addrinfo *ai_list = NULL;
 340   struct addrinfo *ai_ptr;
 341   int status;
 342
 343   char const *node = (conf_node != NULL) ? conf_node : STATSD_DEFAULT_NODE;
 344   char const *service = (conf_service != NULL)
 345     ? conf_service : STATSD_DEFAULT_SERVICE;
 346
 347   memset (&ai_hints, 0, sizeof (ai_hints));
 348   ai_hints.ai_flags = AI_PASSIVE;
 349 #ifdef AI_ADDRCONFIG
 350   ai_hints.ai_flags |= AI_ADDRCONFIG;
 351 #endif
 352   ai_hints.ai_family = AF_UNSPEC;
 353   ai_hints.ai_socktype = SOCK_DGRAM;
 354
 355   status = getaddrinfo (node, service, &ai_hints, &ai_list);
 356   if (status != 0)
 357   {
 358     ERROR ("statsd plugin: getaddrinfo (\"%s\", \"%s\") failed: %s",
 359         node, service, gai_strerror (status));
 360     return (status);
 361   }
 362
 363   for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 364   {
 365     int fd;
 366     struct pollfd *tmp;
 367
 368     char dbg_node[NI_MAXHOST];
 369     char dbg_service[NI_MAXSERV];
 370
 371     fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
 372     if (fd < 0)
 373     {
 374       char errbuf[1024];
 375       ERROR ("statsd plugin: socket(2) failed: %s",
 376           sstrerror (errno, errbuf, sizeof (errbuf)));
 377       continue;
 378     }
 379
 380     getnameinfo (ai_ptr->ai_addr, ai_ptr->ai_addrlen,
 381         dbg_node, sizeof (dbg_node), dbg_service, sizeof (dbg_service),
 382         NI_DGRAM | NI_NUMERICHOST | NI_NUMERICSERV);
 383     DEBUG ("statsd plugin: Trying to bind to [%s]:%s ...", dbg_node, dbg_service);
 384
 385     status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 386     if (status != 0)
 387     {
 388       char errbuf[1024];
 389       ERROR ("statsd plugin: bind(2) failed: %s",
 390           sstrerror (errno, errbuf, sizeof (errbuf)));
 391       close (fd);
 392       continue;
 393     }
 394
 395     tmp = realloc (fds, sizeof (*fds) * (fds_num + 1));
 396     if (tmp == NULL)
 397     {
 398       ERROR ("statsd plugin: realloc failed.");
 399       continue;
 400     }
 401     fds = tmp;
 402     tmp = fds + fds_num;
 403     fds_num++;
 404
 405     memset (tmp, 0, sizeof (*tmp));
 406     tmp->fd = fd;
 407     tmp->events = POLLIN | POLLPRI;
 408   }
 409
 410   freeaddrinfo (ai_list);
 411
 412   if (fds_num == 0)
 413   {
 414     ERROR ("statsd plugin: Unable to create listening socket for [%s]:%s.",
 415         (node != NULL) ? node : "::", service);
 416     return (ENOENT);
 417   }
 418
 419   *ret_fds = fds;
 420   *ret_fds_num = fds_num;
 421   return (0);
 422 } /* }}} int statsd_network_init */
 423
 424 static void *statsd_network_thread (void *args) /* {{{ */
 425 {
 426   struct pollfd *fds = NULL;
 427   size_t fds_num = 0;
 428   int status;
 429   size_t i;
 430
 431   status = statsd_network_init (&fds, &fds_num);
 432   if (status != 0)
 433   {
 434     ERROR ("statsd plugin: Unable to open listening sockets.");
 435     pthread_exit ((void *) 0);
 436   }
 437
 438   while (!network_thread_shutdown)
 439   {
 440     status = poll (fds, (nfds_t) fds_num, /* timeout = */ -1);
 441     if (status < 0)
 442     {
 443       char errbuf[1024];
 444
 445       if ((errno == EINTR) || (errno == EAGAIN))
 446         continue;
 447
 448       ERROR ("statsd plugin: poll(2) failed: %s",
 449           sstrerror (errno, errbuf, sizeof (errbuf)));
 450       break;
 451     }
 452
 453     for (i = 0; i < fds_num; i++)
 454     {
 455       if ((fds[i].revents & (POLLIN | POLLPRI)) == 0)
 456         continue;
 457
 458       statsd_network_read (fds[i].fd);
 459       fds[i].revents = 0;
 460     }
 461   } /* while (!network_thread_shutdown) */
 462
 463   /* Clean up */
 464   for (i = 0; i < fds_num; i++)
 465     close (fds[i].fd);
 466   sfree (fds);
 467
 468   return ((void *) 0);
 469 } /* }}} void *statsd_network_thread */
 470
 471 static int statsd_config (oconfig_item_t *ci) /* {{{ */
 472 {
 473   int i;
 474
 475   for (i = 0; i < ci->children_num; i++)
 476   {
 477     oconfig_item_t *child = ci->children + i;
 478
 479     if (strcasecmp ("Host", child->key) == 0)
 480       cf_util_get_string (child, &conf_node);
 481     else if (strcasecmp ("Port", child->key) == 0)
 482       cf_util_get_service (child, &conf_service);
 483     else if (strcasecmp ("DeleteCounters", child->key) == 0)
 484       cf_util_get_boolean (child, &conf_delete_counters);
 485     else if (strcasecmp ("DeleteTimers", child->key) == 0)
 486       cf_util_get_boolean (child, &conf_delete_timers);
 487     else if (strcasecmp ("DeleteGauges", child->key) == 0)
 488       cf_util_get_boolean (child, &conf_delete_gauges);
 489     else
 490       ERROR ("statsd plugin: The \"%s\" config option is not valid.",
 491           child->key);
 492   }
 493
 494   return (0);
 495 } /* }}} int statsd_config */
 496
 497 static int statsd_init (void) /* {{{ */
 498 {
 499   pthread_mutex_lock (&metrics_lock);
 500   if (metrics_tree == NULL)
 501     metrics_tree = c_avl_create ((void *) strcasecmp);
 502
 503   if (!network_thread_running)
 504   {
 505     int status;
 506
 507     status = pthread_create (&network_thread,
 508         /* attr = */ NULL,
 509         statsd_network_thread,
 510         /* args = */ NULL);
 511     if (status != 0)
 512     {
 513       char errbuf[1024];
 514       pthread_mutex_unlock (&metrics_lock);
 515       ERROR ("statsd plugin: pthread_create failed: %s",
 516           sstrerror (errno, errbuf, sizeof (errbuf)));
 517       return (status);
 518     }
 519   }
 520   network_thread_running = 1;
 521
 522   pthread_mutex_unlock (&metrics_lock);
 523
 524   return (0);
 525 } /* }}} int statsd_init */
 526
 527 static int statsd_metric_submit (char const *name, /* {{{ */
 528     statsd_metric_t const *metric)
 529 {
 530   value_t values[1];
 531   value_list_t vl = VALUE_LIST_INIT;
 532
 533   if (metric->type == STATSD_GAUGE)
 534     values[0].gauge = (gauge_t) metric->value;
 535   else if (metric->type == STATSD_TIMER)
 536   {
 537     if (metric->updates_num == 0)
 538       values[0].gauge = NAN;
 539     else
 540       values[0].gauge =
 541         ((gauge_t) metric->value) / ((gauge_t) metric->updates_num);
 542   }
 543   else
 544     values[0].derive = (derive_t) metric->value;
 545
 546   vl.values = values;
 547   vl.values_len = 1;
 548   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 549   sstrncpy (vl.plugin, "statsd", sizeof (vl.plugin));
 550
 551   if (metric->type == STATSD_GAUGE)
 552     sstrncpy (vl.type, "gauge", sizeof (vl.type));
 553   else if (metric->type == STATSD_TIMER)
 554     sstrncpy (vl.type, "latency", sizeof (vl.type));
 555   else /* if (metric->type == STATSD_COUNTER) */
 556     sstrncpy (vl.type, "derive", sizeof (vl.type));
 557
 558   sstrncpy (vl.type_instance, name, sizeof (vl.type_instance));
 559
 560   return (plugin_dispatch_values (&vl));
 561 } /* }}} int statsd_metric_submit */
 562
 563 static int statsd_read (void) /* {{{ */
 564 {
 565   c_avl_iterator_t *iter;
 566   char *name;
 567   statsd_metric_t *metric;
 568
 569   char **to_be_deleted = NULL;
 570   size_t to_be_deleted_num = 0;
 571   size_t i;
 572
 573   pthread_mutex_lock (&metrics_lock);
 574
 575   if (metrics_tree == NULL)
 576   {
 577     pthread_mutex_unlock (&metrics_lock);
 578     return (0);
 579   }
 580
 581   iter = c_avl_get_iterator (metrics_tree);
 582   while (c_avl_iterator_next (iter, (void *) &name, (void *) &metric) == 0)
 583   {
 584     if ((metric->updates_num == 0)
 585         && ((conf_delete_counters && (metric->type == STATSD_COUNTER))
 586           || (conf_delete_timers && (metric->type == STATSD_TIMER))
 587           || (conf_delete_gauges && (metric->type == STATSD_GAUGE))))
 588     {
 589       DEBUG ("statsd plugin: Deleting metric \"%s\".", name);
 590       strarray_add (&to_be_deleted, &to_be_deleted_num, name);
 591       continue;
 592     }
 593
 594     /* Names have a prefix, e.g. "c:", which determines the (statsd) type.
 595      * Remove this here. */
 596     statsd_metric_submit (name + 2, metric);
 597     metric->updates_num = 0;
 598   }
 599   c_avl_iterator_destroy (iter);
 600
 601   for (i = 0; i < to_be_deleted_num; i++)
 602   {
 603     int status;
 604
 605     status = c_avl_remove (metrics_tree, to_be_deleted[i],
 606         (void *) &name, (void *) &metric);
 607     if (status != 0)
 608     {
 609       ERROR ("stats plugin: c_avl_remove (\"%s\") failed with status %i.",
 610           to_be_deleted[i], status);
 611       continue;
 612     }
 613
 614     sfree (name);
 615     sfree (metric);
 616   }
 617
 618   pthread_mutex_unlock (&metrics_lock);
 619
 620   strarray_free (to_be_deleted, to_be_deleted_num);
 621
 622   return (0);
 623 } /* }}} int statsd_read */
 624
 625 static int statsd_shutdown (void) /* {{{ */
 626 {
 627   void *key;
 628   void *value;
 629
 630   pthread_mutex_lock (&metrics_lock);
 631
 632   if (network_thread_running)
 633   {
 634     network_thread_shutdown = 1;
 635     pthread_kill (network_thread, SIGTERM);
 636     pthread_join (network_thread, /* retval = */ NULL);
 637   }
 638   network_thread_running = 0;
 639
 640   while (c_avl_pick (metrics_tree, &key, &value) == 0)
 641   {
 642     sfree (key);
 643     sfree (value);
 644   }
 645   c_avl_destroy (metrics_tree);
 646   metrics_tree = NULL;
 647
 648   sfree (conf_node);
 649   sfree (conf_service);
 650
 651   pthread_mutex_unlock (&metrics_lock);
 652
 653   return (0);
 654 } /* }}} int statsd_shutdown */
 655
 656 void module_register (void)
 657 {
 658   plugin_register_complex_config ("statsd", statsd_config);
 659   plugin_register_init ("statsd", statsd_init);
 660   plugin_register_read ("statsd", statsd_read);
 661   plugin_register_shutdown ("statsd", statsd_shutdown);
 662 }
 663
 664 /* vim: set sw=2 sts=2 et fdm=marker : */