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   STATSD_SET
  50 };
  51 typedef enum metric_type_e metric_type_t;
  52
  53 struct statsd_metric_s
  54 {
  55   metric_type_t type;
  56   int64_t value;
  57   c_avl_tree_t *set;
  58   unsigned long updates_num;
  59 };
  60 typedef struct statsd_metric_s statsd_metric_t;
  61
  62 static c_avl_tree_t   *metrics_tree = NULL;
  63 static pthread_mutex_t metrics_lock = PTHREAD_MUTEX_INITIALIZER;
  64
  65 static pthread_t network_thread;
  66 static _Bool     network_thread_running = 0;
  67 static _Bool     network_thread_shutdown = 0;
  68
  69 static char *conf_node = NULL;
  70 static char *conf_service = NULL;
  71
  72 static _Bool conf_delete_counters = 0;
  73 static _Bool conf_delete_timers   = 0;
  74 static _Bool conf_delete_gauges   = 0;
  75 static _Bool conf_delete_sets     = 0;
  76
  77 /* Must hold metrics_lock when calling this function. */
  78 static int statsd_metric_set_unsafe (char const *name, int64_t value, /* {{{ */
  79     metric_type_t type)
  80 {
  81   statsd_metric_t *metric;
  82   char *key;
  83   int status;
  84
  85   status = c_avl_get (metrics_tree, name, (void *) &metric);
  86   if (status == 0)
  87   {
  88     metric->value = value;
  89     metric->updates_num++;
  90
  91     return (0);
  92   }
  93
  94   DEBUG ("stats plugin: Adding new metric \"%s\".", name);
  95   key = strdup (name);
  96   metric = calloc (1, sizeof (*metric));
  97   if ((key == NULL) || (metric == NULL))
  98   {
  99     sfree (key);
 100     sfree (metric);
 101     return (-1);
 102   }
 103
 104   metric->type = type;
 105   metric->value = value;
 106   metric->updates_num = 1;
 107
 108   status = c_avl_insert (metrics_tree, key, metric);
 109   if (status != 0)
 110   {
 111     sfree (key);
 112     sfree (metric);
 113
 114     return (-1);
 115   }
 116
 117   return (0);
 118 } /* }}} int statsd_metric_set_unsafe */
 119
 120 static int statsd_metric_set (char const *name, int64_t value, /* {{{ */
 121     metric_type_t type)
 122 {
 123   int status;
 124
 125   pthread_mutex_lock (&metrics_lock);
 126   status = statsd_metric_set_unsafe (name, value, type);
 127   pthread_mutex_unlock (&metrics_lock);
 128
 129   return (status);
 130 } /* }}} int statsd_metric_set */
 131
 132 static int statsd_metric_add (char const *name, int64_t delta, /* {{{ */
 133     metric_type_t type)
 134 {
 135   statsd_metric_t *metric;
 136   int status;
 137
 138   pthread_mutex_lock (&metrics_lock);
 139
 140   status = c_avl_get (metrics_tree, name, (void *) &metric);
 141   if (status == 0)
 142   {
 143     metric->value += delta;
 144     metric->updates_num++;
 145
 146     pthread_mutex_unlock (&metrics_lock);
 147     return (0);
 148   }
 149   else /* no such value yet */
 150   {
 151     status = statsd_metric_set_unsafe (name, delta, type);
 152
 153     pthread_mutex_unlock (&metrics_lock);
 154     return (status);
 155   }
 156 } /* }}} int statsd_metric_add */
 157
 158 static int statsd_handle_counter (char const *name, /* {{{ */
 159     char const *value_str,
 160     char const *extra)
 161 {
 162   char key[DATA_MAX_NAME_LEN + 2];
 163   value_t value;
 164   value_t scale;
 165   int status;
 166
 167   if ((extra != NULL) && (extra[0] != '@'))
 168     return (-1);
 169
 170   scale.gauge = 1.0;
 171   if (extra != NULL)
 172   {
 173     status = parse_value (extra + 1, &scale, DS_TYPE_GAUGE);
 174     if (status != 0)
 175       return (status);
 176
 177     if (!isfinite (scale.gauge) || (scale.gauge <= 0.0) || (scale.gauge > 1.0))
 178       return (-1);
 179   }
 180
 181   value.derive = 1;
 182   status = parse_value (value_str, &value, DS_TYPE_DERIVE);
 183   if (status != 0)
 184     return (status);
 185
 186   if (value.derive < 1)
 187     return (-1);
 188
 189   ssnprintf (key, sizeof (key), "c:%s", name);
 190
 191   return (statsd_metric_add (key,
 192         (int64_t) (((gauge_t) value.derive) / scale.gauge),
 193         STATSD_COUNTER));
 194 } /* }}} int statsd_handle_counter */
 195
 196 static int statsd_handle_gauge (char const *name, /* {{{ */
 197     char const *value_str)
 198 {
 199   char key[DATA_MAX_NAME_LEN + 2];
 200   value_t value;
 201   int status;
 202
 203   value.derive = 0;
 204   status = parse_value (value_str, &value, DS_TYPE_DERIVE);
 205   if (status != 0)
 206     return (status);
 207
 208   ssnprintf (key, sizeof (key), "g:%s", name);
 209
 210   if ((value_str[0] == '+') || (value_str[0] == '-'))
 211     return (statsd_metric_add (key, (int64_t) value.derive, STATSD_GAUGE));
 212   else
 213     return (statsd_metric_set (key, (int64_t) value.derive, STATSD_GAUGE));
 214 } /* }}} int statsd_handle_gauge */
 215
 216 static int statsd_handle_timer (char const *name, /* {{{ */
 217     char const *value_str)
 218 {
 219   char key[DATA_MAX_NAME_LEN + 2];
 220   value_t value;
 221   int status;
 222
 223   value.derive = 0;
 224   status = parse_value (value_str, &value, DS_TYPE_DERIVE);
 225   if (status != 0)
 226     return (status);
 227
 228   ssnprintf (key, sizeof (key), "t:%s", name);
 229
 230   return (statsd_metric_add (key, (int64_t) value.derive, STATSD_TIMER));
 231 } /* }}} int statsd_handle_timer */
 232
 233 static int statsd_handle_set (char const *key_orig, /* {{{ */
 234     char const *name_orig)
 235 {
 236   char key[DATA_MAX_NAME_LEN + 2];
 237   char *name;
 238   statsd_metric_t *metric = NULL;
 239   int status;
 240
 241   ssnprintf (key, sizeof (key), "s:%s", key_orig);
 242
 243   pthread_mutex_lock (&metrics_lock);
 244
 245   status = c_avl_get (metrics_tree, key, (void *) &metric);
 246   if (status != 0) /* Create a new metric */
 247   {
 248     char *key_copy;
 249
 250     DEBUG ("stats plugin: Adding new metric \"%s\".", key);
 251     key_copy = strdup (key);
 252     if (key_copy == NULL)
 253     {
 254       pthread_mutex_unlock (&metrics_lock);
 255       ERROR ("statsd plugin: strdup failed.");
 256       return (-1);
 257     }
 258
 259     metric = calloc (1, sizeof (*metric));
 260     if (metric == NULL)
 261     {
 262       pthread_mutex_unlock (&metrics_lock);
 263       ERROR ("statsd plugin: calloc failed.");
 264       sfree (key_copy);
 265       return (-1);
 266     }
 267     metric->type = STATSD_SET;
 268     metric->set = NULL;
 269
 270     status = c_avl_insert (metrics_tree, key_copy, metric);
 271     if (status != 0)
 272     {
 273       pthread_mutex_unlock (&metrics_lock);
 274       ERROR ("statsd plugin: c_avl_insert (\"%s\") failed with status %i.",
 275           key_copy, status);
 276       sfree (key_copy);
 277       sfree (metric);
 278       return (-1);
 279     }
 280   }
 281   assert (metric != NULL);
 282
 283   /* Make sure metric->set exists. */
 284   if (metric->set == NULL)
 285     metric->set = c_avl_create ((void *) strcmp);
 286
 287   if (metric->set == NULL)
 288   {
 289     pthread_mutex_unlock (&metrics_lock);
 290     ERROR ("statsd plugin: c_avl_create failed.");
 291     return (-1);
 292   }
 293
 294   name = strdup (name_orig);
 295   if (name == NULL)
 296   {
 297     pthread_mutex_unlock (&metrics_lock);
 298     ERROR ("statsd plugin: strdup failed.");
 299     return (-1);
 300   }
 301
 302   status = c_avl_insert (metric->set, name, /* value = */ NULL);
 303   if (status < 0)
 304   {
 305     pthread_mutex_unlock (&metrics_lock);
 306     if (status < 0)
 307       ERROR ("statsd plugin: c_avl_insert (\"%s\") failed with status %i.",
 308           name, status);
 309     sfree (name);
 310     return (-1);
 311   }
 312   else if (status > 0) /* key already exists */
 313   {
 314     sfree (name);
 315   }
 316
 317   metric->updates_num++;
 318
 319   pthread_mutex_unlock (&metrics_lock);
 320   return (0);
 321 } /* }}} int statsd_handle_set */
 322
 323 static int statsd_parse_line (char *buffer) /* {{{ */
 324 {
 325   char *name = buffer;
 326   char *value;
 327   char *type;
 328   char *extra;
 329
 330   type = strchr (name, '|');
 331   if (type == NULL)
 332     return (-1);
 333   *type = 0;
 334   type++;
 335
 336   value = strrchr (name, ':');
 337   if (value == NULL)
 338     return (-1);
 339   *value = 0;
 340   value++;
 341
 342   extra = strchr (type, '|');
 343   if (extra != NULL)
 344   {
 345     *extra = 0;
 346     extra++;
 347   }
 348
 349   if (strcmp ("c", type) == 0)
 350     return (statsd_handle_counter (name, value, extra));
 351
 352   /* extra is only valid for counters */
 353   if (extra != NULL)
 354     return (-1);
 355
 356   if (strcmp ("g", type) == 0)
 357     return (statsd_handle_gauge (name, value));
 358   else if (strcmp ("ms", type) == 0)
 359     return (statsd_handle_timer (name, value));
 360   else if (strcmp ("s", type) == 0)
 361     return (statsd_handle_set (name, value));
 362   else
 363     return (-1);
 364 } /* }}} void statsd_parse_line */
 365
 366 static void statsd_parse_buffer (char *buffer) /* {{{ */
 367 {
 368   char *dummy;
 369   char *saveptr = NULL;
 370   char *ptr;
 371
 372   for (dummy = buffer;
 373       (ptr = strtok_r (dummy, "\r\n", &saveptr)) != NULL;
 374       dummy = NULL)
 375   {
 376     char *line_orig = sstrdup (ptr);
 377     int status;
 378
 379     status = statsd_parse_line (ptr);
 380     if (status != 0)
 381       ERROR ("statsd plugin: Unable to parse line: \"%s\"", line_orig);
 382
 383     sfree (line_orig);
 384   }
 385 } /* }}} void statsd_parse_buffer */
 386
 387 static void statsd_network_read (int fd) /* {{{ */
 388 {
 389   char buffer[4096];
 390   size_t buffer_size;
 391   ssize_t status;
 392
 393   status = recv (fd, buffer, sizeof (buffer), /* flags = */ MSG_DONTWAIT);
 394   if (status < 0)
 395   {
 396     char errbuf[1024];
 397
 398     if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
 399       return;
 400
 401     ERROR ("statsd plugin: recv(2) failed: %s",
 402         sstrerror (errno, errbuf, sizeof (errbuf)));
 403     return;
 404   }
 405
 406   buffer_size = (size_t) status;
 407   if (buffer_size >= sizeof (buffer))
 408     buffer_size = sizeof (buffer) - 1;
 409   buffer[buffer_size] = 0;
 410
 411   statsd_parse_buffer (buffer);
 412 } /* }}} void statsd_network_read */
 413
 414 static int statsd_network_init (struct pollfd **ret_fds, /* {{{ */
 415     size_t *ret_fds_num)
 416 {
 417   struct pollfd *fds = NULL;
 418   size_t fds_num = 0;
 419
 420   struct addrinfo ai_hints;
 421   struct addrinfo *ai_list = NULL;
 422   struct addrinfo *ai_ptr;
 423   int status;
 424
 425   char const *node = (conf_node != NULL) ? conf_node : STATSD_DEFAULT_NODE;
 426   char const *service = (conf_service != NULL)
 427     ? conf_service : STATSD_DEFAULT_SERVICE;
 428
 429   memset (&ai_hints, 0, sizeof (ai_hints));
 430   ai_hints.ai_flags = AI_PASSIVE;
 431 #ifdef AI_ADDRCONFIG
 432   ai_hints.ai_flags |= AI_ADDRCONFIG;
 433 #endif
 434   ai_hints.ai_family = AF_UNSPEC;
 435   ai_hints.ai_socktype = SOCK_DGRAM;
 436
 437   status = getaddrinfo (node, service, &ai_hints, &ai_list);
 438   if (status != 0)
 439   {
 440     ERROR ("statsd plugin: getaddrinfo (\"%s\", \"%s\") failed: %s",
 441         node, service, gai_strerror (status));
 442     return (status);
 443   }
 444
 445   for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 446   {
 447     int fd;
 448     struct pollfd *tmp;
 449
 450     char dbg_node[NI_MAXHOST];
 451     char dbg_service[NI_MAXSERV];
 452
 453     fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
 454     if (fd < 0)
 455     {
 456       char errbuf[1024];
 457       ERROR ("statsd plugin: socket(2) failed: %s",
 458           sstrerror (errno, errbuf, sizeof (errbuf)));
 459       continue;
 460     }
 461
 462     getnameinfo (ai_ptr->ai_addr, ai_ptr->ai_addrlen,
 463         dbg_node, sizeof (dbg_node), dbg_service, sizeof (dbg_service),
 464         NI_DGRAM | NI_NUMERICHOST | NI_NUMERICSERV);
 465     DEBUG ("statsd plugin: Trying to bind to [%s]:%s ...", dbg_node, dbg_service);
 466
 467     status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 468     if (status != 0)
 469     {
 470       char errbuf[1024];
 471       ERROR ("statsd plugin: bind(2) failed: %s",
 472           sstrerror (errno, errbuf, sizeof (errbuf)));
 473       close (fd);
 474       continue;
 475     }
 476
 477     tmp = realloc (fds, sizeof (*fds) * (fds_num + 1));
 478     if (tmp == NULL)
 479     {
 480       ERROR ("statsd plugin: realloc failed.");
 481       continue;
 482     }
 483     fds = tmp;
 484     tmp = fds + fds_num;
 485     fds_num++;
 486
 487     memset (tmp, 0, sizeof (*tmp));
 488     tmp->fd = fd;
 489     tmp->events = POLLIN | POLLPRI;
 490   }
 491
 492   freeaddrinfo (ai_list);
 493
 494   if (fds_num == 0)
 495   {
 496     ERROR ("statsd plugin: Unable to create listening socket for [%s]:%s.",
 497         (node != NULL) ? node : "::", service);
 498     return (ENOENT);
 499   }
 500
 501   *ret_fds = fds;
 502   *ret_fds_num = fds_num;
 503   return (0);
 504 } /* }}} int statsd_network_init */
 505
 506 static void *statsd_network_thread (void *args) /* {{{ */
 507 {
 508   struct pollfd *fds = NULL;
 509   size_t fds_num = 0;
 510   int status;
 511   size_t i;
 512
 513   status = statsd_network_init (&fds, &fds_num);
 514   if (status != 0)
 515   {
 516     ERROR ("statsd plugin: Unable to open listening sockets.");
 517     pthread_exit ((void *) 0);
 518   }
 519
 520   while (!network_thread_shutdown)
 521   {
 522     status = poll (fds, (nfds_t) fds_num, /* timeout = */ -1);
 523     if (status < 0)
 524     {
 525       char errbuf[1024];
 526
 527       if ((errno == EINTR) || (errno == EAGAIN))
 528         continue;
 529
 530       ERROR ("statsd plugin: poll(2) failed: %s",
 531           sstrerror (errno, errbuf, sizeof (errbuf)));
 532       break;
 533     }
 534
 535     for (i = 0; i < fds_num; i++)
 536     {
 537       if ((fds[i].revents & (POLLIN | POLLPRI)) == 0)
 538         continue;
 539
 540       statsd_network_read (fds[i].fd);
 541       fds[i].revents = 0;
 542     }
 543   } /* while (!network_thread_shutdown) */
 544
 545   /* Clean up */
 546   for (i = 0; i < fds_num; i++)
 547     close (fds[i].fd);
 548   sfree (fds);
 549
 550   return ((void *) 0);
 551 } /* }}} void *statsd_network_thread */
 552
 553 static int statsd_config (oconfig_item_t *ci) /* {{{ */
 554 {
 555   int i;
 556
 557   for (i = 0; i < ci->children_num; i++)
 558   {
 559     oconfig_item_t *child = ci->children + i;
 560
 561     if (strcasecmp ("Host", child->key) == 0)
 562       cf_util_get_string (child, &conf_node);
 563     else if (strcasecmp ("Port", child->key) == 0)
 564       cf_util_get_service (child, &conf_service);
 565     else if (strcasecmp ("DeleteCounters", child->key) == 0)
 566       cf_util_get_boolean (child, &conf_delete_counters);
 567     else if (strcasecmp ("DeleteTimers", child->key) == 0)
 568       cf_util_get_boolean (child, &conf_delete_timers);
 569     else if (strcasecmp ("DeleteGauges", child->key) == 0)
 570       cf_util_get_boolean (child, &conf_delete_gauges);
 571     else if (strcasecmp ("DeleteSets", child->key) == 0)
 572       cf_util_get_boolean (child, &conf_delete_sets);
 573     else
 574       ERROR ("statsd plugin: The \"%s\" config option is not valid.",
 575           child->key);
 576   }
 577
 578   return (0);
 579 } /* }}} int statsd_config */
 580
 581 static int statsd_init (void) /* {{{ */
 582 {
 583   pthread_mutex_lock (&metrics_lock);
 584   if (metrics_tree == NULL)
 585     metrics_tree = c_avl_create ((void *) strcasecmp);
 586
 587   if (!network_thread_running)
 588   {
 589     int status;
 590
 591     status = pthread_create (&network_thread,
 592         /* attr = */ NULL,
 593         statsd_network_thread,
 594         /* args = */ NULL);
 595     if (status != 0)
 596     {
 597       char errbuf[1024];
 598       pthread_mutex_unlock (&metrics_lock);
 599       ERROR ("statsd plugin: pthread_create failed: %s",
 600           sstrerror (errno, errbuf, sizeof (errbuf)));
 601       return (status);
 602     }
 603   }
 604   network_thread_running = 1;
 605
 606   pthread_mutex_unlock (&metrics_lock);
 607
 608   return (0);
 609 } /* }}} int statsd_init */
 610
 611 /* Must hold metrics_lock when calling this function. */
 612 static int statsd_metric_clear_set_unsafe (statsd_metric_t *metric) /* {{{ */
 613 {
 614   void *key;
 615   void *value;
 616
 617   if ((metric == NULL) || (metric->type != STATSD_SET))
 618     return (EINVAL);
 619
 620   if (metric->set == NULL)
 621     return (0);
 622
 623   while (c_avl_pick (metric->set, &key, &value) == 0)
 624   {
 625     sfree (key);
 626     sfree (value);
 627   }
 628
 629   return (0);
 630 } /* }}} int statsd_metric_clear_set_unsafe */
 631
 632 /* Must hold metrics_lock when calling this function. */
 633 static int statsd_metric_submit_unsafe (char const *name, /* {{{ */
 634     statsd_metric_t const *metric)
 635 {
 636   value_t values[1];
 637   value_list_t vl = VALUE_LIST_INIT;
 638
 639   if (metric->type == STATSD_GAUGE)
 640     values[0].gauge = (gauge_t) metric->value;
 641   else if (metric->type == STATSD_TIMER)
 642   {
 643     if (metric->updates_num == 0)
 644       values[0].gauge = NAN;
 645     else
 646       values[0].gauge =
 647         ((gauge_t) metric->value) / ((gauge_t) metric->updates_num);
 648   }
 649   else if (metric->type == STATSD_SET)
 650   {
 651     if (metric->set == NULL)
 652       values[0].gauge = 0.0;
 653     else
 654       values[0].gauge = (gauge_t) c_avl_size (metric->set);
 655   }
 656   else
 657     values[0].derive = (derive_t) metric->value;
 658
 659   vl.values = values;
 660   vl.values_len = 1;
 661   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 662   sstrncpy (vl.plugin, "statsd", sizeof (vl.plugin));
 663
 664   if (metric->type == STATSD_GAUGE)
 665     sstrncpy (vl.type, "gauge", sizeof (vl.type));
 666   else if (metric->type == STATSD_TIMER)
 667     sstrncpy (vl.type, "latency", sizeof (vl.type));
 668   else if (metric->type == STATSD_SET)
 669     sstrncpy (vl.type, "objects", sizeof (vl.type));
 670   else /* if (metric->type == STATSD_COUNTER) */
 671     sstrncpy (vl.type, "derive", sizeof (vl.type));
 672
 673   sstrncpy (vl.type_instance, name, sizeof (vl.type_instance));
 674
 675   return (plugin_dispatch_values (&vl));
 676 } /* }}} int statsd_metric_submit_unsafe */
 677
 678 static int statsd_read (void) /* {{{ */
 679 {
 680   c_avl_iterator_t *iter;
 681   char *name;
 682   statsd_metric_t *metric;
 683
 684   char **to_be_deleted = NULL;
 685   size_t to_be_deleted_num = 0;
 686   size_t i;
 687
 688   pthread_mutex_lock (&metrics_lock);
 689
 690   if (metrics_tree == NULL)
 691   {
 692     pthread_mutex_unlock (&metrics_lock);
 693     return (0);
 694   }
 695
 696   iter = c_avl_get_iterator (metrics_tree);
 697   while (c_avl_iterator_next (iter, (void *) &name, (void *) &metric) == 0)
 698   {
 699     if ((metric->updates_num == 0)
 700         && ((conf_delete_counters && (metric->type == STATSD_COUNTER))
 701           || (conf_delete_timers && (metric->type == STATSD_TIMER))
 702           || (conf_delete_gauges && (metric->type == STATSD_GAUGE))
 703           || (conf_delete_sets && (metric->type == STATSD_SET))))
 704     {
 705       DEBUG ("statsd plugin: Deleting metric \"%s\".", name);
 706       strarray_add (&to_be_deleted, &to_be_deleted_num, name);
 707       continue;
 708     }
 709
 710     /* Names have a prefix, e.g. "c:", which determines the (statsd) type.
 711      * Remove this here. */
 712     statsd_metric_submit_unsafe (name + 2, metric);
 713
 714     /* Reset the metric. */
 715     metric->updates_num = 0;
 716     if (metric->type == STATSD_SET)
 717       statsd_metric_clear_set_unsafe (metric);
 718   }
 719   c_avl_iterator_destroy (iter);
 720
 721   for (i = 0; i < to_be_deleted_num; i++)
 722   {
 723     int status;
 724
 725     status = c_avl_remove (metrics_tree, to_be_deleted[i],
 726         (void *) &name, (void *) &metric);
 727     if (status != 0)
 728     {
 729       ERROR ("stats plugin: c_avl_remove (\"%s\") failed with status %i.",
 730           to_be_deleted[i], status);
 731       continue;
 732     }
 733
 734     sfree (name);
 735     sfree (metric);
 736   }
 737
 738   pthread_mutex_unlock (&metrics_lock);
 739
 740   strarray_free (to_be_deleted, to_be_deleted_num);
 741
 742   return (0);
 743 } /* }}} int statsd_read */
 744
 745 static int statsd_shutdown (void) /* {{{ */
 746 {
 747   void *key;
 748   void *value;
 749
 750   pthread_mutex_lock (&metrics_lock);
 751
 752   if (network_thread_running)
 753   {
 754     network_thread_shutdown = 1;
 755     pthread_kill (network_thread, SIGTERM);
 756     pthread_join (network_thread, /* retval = */ NULL);
 757   }
 758   network_thread_running = 0;
 759
 760   while (c_avl_pick (metrics_tree, &key, &value) == 0)
 761   {
 762     sfree (key);
 763     sfree (value);
 764   }
 765   c_avl_destroy (metrics_tree);
 766   metrics_tree = NULL;
 767
 768   sfree (conf_node);
 769   sfree (conf_service);
 770
 771   pthread_mutex_unlock (&metrics_lock);
 772
 773   return (0);
 774 } /* }}} int statsd_shutdown */
 775
 776 void module_register (void)
 777 {
 778   plugin_register_complex_config ("statsd", statsd_config);
 779   plugin_register_init ("statsd", statsd_init);
 780   plugin_register_read ("statsd", statsd_read);
 781   plugin_register_shutdown ("statsd", statsd_shutdown);
 782 }
 783
 784 /* vim: set sw=2 sts=2 et fdm=marker : */