Merge remote-tracking branches 'github/pr/392' and 'github/pr/399' into jr/json
[collectd.git] / 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 #include "utils_latency.h"
29
30 #include <pthread.h>
31
32 #include <sys/types.h>
33 #include <sys/socket.h>
34 #include <netdb.h>
35 #include <poll.h>
36
37 #ifndef STATSD_DEFAULT_NODE
38 # define STATSD_DEFAULT_NODE NULL
39 #endif
40
41 #ifndef STATSD_DEFAULT_SERVICE
42 # define STATSD_DEFAULT_SERVICE "8125"
43 #endif
44
45 enum metric_type_e
46 {
47   STATSD_COUNTER,
48   STATSD_TIMER,
49   STATSD_GAUGE,
50   STATSD_SET
51 };
52 typedef enum metric_type_e metric_type_t;
53
54 struct statsd_metric_s
55 {
56   metric_type_t type;
57   double value;
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_timer_lower     = 0;
83 static _Bool conf_timer_upper     = 0;
84 static _Bool conf_timer_sum       = 0;
85 static _Bool conf_timer_count     = 0;
86
87 /* Must hold metrics_lock when calling this function. */
88 static statsd_metric_t *statsd_metric_lookup_unsafe (char const *name, /* {{{ */
89     metric_type_t type)
90 {
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   {
98     case STATSD_COUNTER: key[0] = 'c'; break;
99     case STATSD_TIMER:   key[0] = 't'; break;
100     case STATSD_GAUGE:   key[0] = 'g'; break;
101     case STATSD_SET:     key[0] = 's'; break;
102     default: return (NULL);
103   }
104
105   key[1] = ':';
106   sstrncpy (&key[2], name, sizeof (key) - 2);
107
108   status = c_avl_get (metrics_tree, key, (void *) &metric);
109   if (status == 0)
110     return (metric);
111
112   key_copy = strdup (key);
113   if (key_copy == NULL)
114   {
115     ERROR ("statsd plugin: strdup failed.");
116     return (NULL);
117   }
118
119   metric = malloc (sizeof (*metric));
120   if (metric == NULL)
121   {
122     ERROR ("statsd plugin: malloc failed.");
123     sfree (key_copy);
124     return (NULL);
125   }
126   memset (metric, 0, sizeof (*metric));
127
128   metric->type = type;
129   metric->latency = NULL;
130   metric->set = NULL;
131
132   status = c_avl_insert (metrics_tree, key_copy, metric);
133   if (status != 0)
134   {
135     ERROR ("statsd plugin: c_avl_insert failed.");
136     sfree (key_copy);
137     sfree (metric);
138     return (NULL);
139   }
140
141   return (metric);
142 } /* }}} statsd_metric_lookup_unsafe */
143
144 static int statsd_metric_set (char const *name, double value, /* {{{ */
145     metric_type_t type)
146 {
147   statsd_metric_t *metric;
148
149   pthread_mutex_lock (&metrics_lock);
150
151   metric = statsd_metric_lookup_unsafe (name, type);
152   if (metric == NULL)
153   {
154     pthread_mutex_unlock (&metrics_lock);
155     return (-1);
156   }
157
158   metric->value = value;
159   metric->updates_num++;
160
161   pthread_mutex_unlock (&metrics_lock);
162
163   return (0);
164 } /* }}} int statsd_metric_set */
165
166 static int statsd_metric_add (char const *name, double delta, /* {{{ */
167     metric_type_t type)
168 {
169   statsd_metric_t *metric;
170
171   pthread_mutex_lock (&metrics_lock);
172
173   metric = statsd_metric_lookup_unsafe (name, type);
174   if (metric == NULL)
175   {
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 int statsd_parse_value (char const *str, value_t *ret_value) /* {{{ */
189 {
190   char *endptr = NULL;
191
192   ret_value->gauge = (gauge_t) strtod (str, &endptr);
193   if ((str == endptr) || ((endptr != NULL) && (*endptr != 0)))
194     return (-1);
195
196   return (0);
197 } /* }}} int statsd_parse_value */
198
199 static int statsd_handle_counter (char const *name, /* {{{ */
200     char const *value_str,
201     char const *extra)
202 {
203   value_t value;
204   value_t scale;
205   int status;
206
207   if ((extra != NULL) && (extra[0] != '@'))
208     return (-1);
209
210   scale.gauge = 1.0;
211   if (extra != NULL)
212   {
213     status = statsd_parse_value (extra + 1, &scale);
214     if (status != 0)
215       return (status);
216
217     if (!isfinite (scale.gauge) || (scale.gauge <= 0.0) || (scale.gauge > 1.0))
218       return (-1);
219   }
220
221   value.gauge = 1.0;
222   status = statsd_parse_value (value_str, &value);
223   if (status != 0)
224     return (status);
225
226   return (statsd_metric_add (name, (double) (value.gauge / scale.gauge),
227         STATSD_COUNTER));
228 } /* }}} int statsd_handle_counter */
229
230 static int statsd_handle_gauge (char const *name, /* {{{ */
231     char const *value_str)
232 {
233   value_t value;
234   int status;
235
236   value.gauge = 0;
237   status = statsd_parse_value (value_str, &value);
238   if (status != 0)
239     return (status);
240
241   if ((value_str[0] == '+') || (value_str[0] == '-'))
242     return (statsd_metric_add (name, (double) value.gauge, STATSD_GAUGE));
243   else
244     return (statsd_metric_set (name, (double) value.gauge, STATSD_GAUGE));
245 } /* }}} int statsd_handle_gauge */
246
247 static int statsd_handle_timer (char const *name, /* {{{ */
248     char const *value_str)
249 {
250   statsd_metric_t *metric;
251   value_t value_ms;
252   cdtime_t value;
253   int status;
254
255   value_ms.derive = 0;
256   status = statsd_parse_value (value_str, &value_ms);
257   if (status != 0)
258     return (status);
259
260   value = MS_TO_CDTIME_T (value_ms.gauge);
261
262   pthread_mutex_lock (&metrics_lock);
263
264   metric = statsd_metric_lookup_unsafe (name, STATSD_TIMER);
265   if (metric == NULL)
266   {
267     pthread_mutex_unlock (&metrics_lock);
268     return (-1);
269   }
270
271   if (metric->latency == NULL)
272     metric->latency = latency_counter_create ();
273   if (metric->latency == NULL)
274   {
275     pthread_mutex_unlock (&metrics_lock);
276     return (-1);
277   }
278
279   latency_counter_add (metric->latency, value);
280   metric->updates_num++;
281
282   pthread_mutex_unlock (&metrics_lock);
283   return (0);
284 } /* }}} int statsd_handle_timer */
285
286 static int statsd_handle_set (char const *name, /* {{{ */
287     char const *set_key_orig)
288 {
289   statsd_metric_t *metric = NULL;
290   char *set_key;
291   int status;
292
293   pthread_mutex_lock (&metrics_lock);
294
295   metric = statsd_metric_lookup_unsafe (name, STATSD_SET);
296   if (metric == NULL)
297   {
298     pthread_mutex_unlock (&metrics_lock);
299     return (-1);
300   }
301
302   /* Make sure metric->set exists. */
303   if (metric->set == NULL)
304     metric->set = c_avl_create ((void *) strcmp);
305
306   if (metric->set == NULL)
307   {
308     pthread_mutex_unlock (&metrics_lock);
309     ERROR ("statsd plugin: c_avl_create failed.");
310     return (-1);
311   }
312
313   set_key = strdup (set_key_orig);
314   if (set_key == NULL)
315   {
316     pthread_mutex_unlock (&metrics_lock);
317     ERROR ("statsd plugin: strdup failed.");
318     return (-1);
319   }
320
321   status = c_avl_insert (metric->set, set_key, /* value = */ NULL);
322   if (status < 0)
323   {
324     pthread_mutex_unlock (&metrics_lock);
325     if (status < 0)
326       ERROR ("statsd plugin: c_avl_insert (\"%s\") failed with status %i.",
327           set_key, status);
328     sfree (set_key);
329     return (-1);
330   }
331   else if (status > 0) /* key already exists */
332   {
333     sfree (set_key);
334   }
335
336   metric->updates_num++;
337
338   pthread_mutex_unlock (&metrics_lock);
339   return (0);
340 } /* }}} int statsd_handle_set */
341
342 static int statsd_parse_line (char *buffer) /* {{{ */
343 {
344   char *name = buffer;
345   char *value;
346   char *type;
347   char *extra;
348
349   type = strchr (name, '|');
350   if (type == NULL)
351     return (-1);
352   *type = 0;
353   type++;
354
355   value = strrchr (name, ':');
356   if (value == NULL)
357     return (-1);
358   *value = 0;
359   value++;
360
361   extra = strchr (type, '|');
362   if (extra != NULL)
363   {
364     *extra = 0;
365     extra++;
366   }
367
368   if (strcmp ("c", type) == 0)
369     return (statsd_handle_counter (name, value, extra));
370
371   /* extra is only valid for counters */
372   if (extra != NULL)
373     return (-1);
374
375   if (strcmp ("g", type) == 0)
376     return (statsd_handle_gauge (name, value));
377   else if (strcmp ("ms", type) == 0)
378     return (statsd_handle_timer (name, value));
379   else if (strcmp ("s", type) == 0)
380     return (statsd_handle_set (name, value));
381   else
382     return (-1);
383 } /* }}} void statsd_parse_line */
384
385 static void statsd_parse_buffer (char *buffer) /* {{{ */
386 {
387   while (buffer != NULL)
388   {
389     char orig[64];
390     char *next;
391     int status;
392
393     next = strchr (buffer, '\n');
394     if (next != NULL)
395     {
396       *next = 0;
397       next++;
398     }
399
400     if (*buffer == 0)
401     {
402       buffer = next;
403       continue;
404     }
405
406     sstrncpy (orig, buffer, sizeof (orig));
407
408     status = statsd_parse_line (buffer);
409     if (status != 0)
410       ERROR ("statsd plugin: Unable to parse line: \"%s\"", orig);
411
412     buffer = next;
413   }
414 } /* }}} void statsd_parse_buffer */
415
416 static void statsd_network_read (int fd) /* {{{ */
417 {
418   char buffer[4096];
419   size_t buffer_size;
420   ssize_t status;
421
422   status = recv (fd, buffer, sizeof (buffer), /* flags = */ MSG_DONTWAIT);
423   if (status < 0)
424   {
425     char errbuf[1024];
426
427     if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
428       return;
429
430     ERROR ("statsd plugin: recv(2) failed: %s",
431         sstrerror (errno, errbuf, sizeof (errbuf)));
432     return;
433   }
434
435   buffer_size = (size_t) status;
436   if (buffer_size >= sizeof (buffer))
437     buffer_size = sizeof (buffer) - 1;
438   buffer[buffer_size] = 0;
439
440   statsd_parse_buffer (buffer);
441 } /* }}} void statsd_network_read */
442
443 static int statsd_network_init (struct pollfd **ret_fds, /* {{{ */
444     size_t *ret_fds_num)
445 {
446   struct pollfd *fds = NULL;
447   size_t fds_num = 0;
448
449   struct addrinfo ai_hints;
450   struct addrinfo *ai_list = NULL;
451   struct addrinfo *ai_ptr;
452   int status;
453
454   char const *node = (conf_node != NULL) ? conf_node : STATSD_DEFAULT_NODE;
455   char const *service = (conf_service != NULL)
456     ? conf_service : STATSD_DEFAULT_SERVICE;
457
458   memset (&ai_hints, 0, sizeof (ai_hints));
459   ai_hints.ai_flags = AI_PASSIVE;
460 #ifdef AI_ADDRCONFIG
461   ai_hints.ai_flags |= AI_ADDRCONFIG;
462 #endif
463   ai_hints.ai_family = AF_UNSPEC;
464   ai_hints.ai_socktype = SOCK_DGRAM;
465
466   status = getaddrinfo (node, service, &ai_hints, &ai_list);
467   if (status != 0)
468   {
469     ERROR ("statsd plugin: getaddrinfo (\"%s\", \"%s\") failed: %s",
470         node, service, gai_strerror (status));
471     return (status);
472   }
473
474   for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
475   {
476     int fd;
477     struct pollfd *tmp;
478
479     char dbg_node[NI_MAXHOST];
480     char dbg_service[NI_MAXSERV];
481
482     fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
483     if (fd < 0)
484     {
485       char errbuf[1024];
486       ERROR ("statsd plugin: socket(2) failed: %s",
487           sstrerror (errno, errbuf, sizeof (errbuf)));
488       continue;
489     }
490
491     getnameinfo (ai_ptr->ai_addr, ai_ptr->ai_addrlen,
492         dbg_node, sizeof (dbg_node), dbg_service, sizeof (dbg_service),
493         NI_DGRAM | NI_NUMERICHOST | NI_NUMERICSERV);
494     DEBUG ("statsd plugin: Trying to bind to [%s]:%s ...", dbg_node, dbg_service);
495
496     status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
497     if (status != 0)
498     {
499       char errbuf[1024];
500       ERROR ("statsd plugin: bind(2) failed: %s",
501           sstrerror (errno, errbuf, sizeof (errbuf)));
502       close (fd);
503       continue;
504     }
505
506     tmp = realloc (fds, sizeof (*fds) * (fds_num + 1));
507     if (tmp == NULL)
508     {
509       ERROR ("statsd plugin: realloc failed.");
510       continue;
511     }
512     fds = tmp;
513     tmp = fds + fds_num;
514     fds_num++;
515
516     memset (tmp, 0, sizeof (*tmp));
517     tmp->fd = fd;
518     tmp->events = POLLIN | POLLPRI;
519   }
520
521   freeaddrinfo (ai_list);
522
523   if (fds_num == 0)
524   {
525     ERROR ("statsd plugin: Unable to create listening socket for [%s]:%s.",
526         (node != NULL) ? node : "::", service);
527     return (ENOENT);
528   }
529
530   *ret_fds = fds;
531   *ret_fds_num = fds_num;
532   return (0);
533 } /* }}} int statsd_network_init */
534
535 static void *statsd_network_thread (void *args) /* {{{ */
536 {
537   struct pollfd *fds = NULL;
538   size_t fds_num = 0;
539   int status;
540   size_t i;
541
542   status = statsd_network_init (&fds, &fds_num);
543   if (status != 0)
544   {
545     ERROR ("statsd plugin: Unable to open listening sockets.");
546     pthread_exit ((void *) 0);
547   }
548
549   while (!network_thread_shutdown)
550   {
551     status = poll (fds, (nfds_t) fds_num, /* timeout = */ -1);
552     if (status < 0)
553     {
554       char errbuf[1024];
555
556       if ((errno == EINTR) || (errno == EAGAIN))
557         continue;
558
559       ERROR ("statsd plugin: poll(2) failed: %s",
560           sstrerror (errno, errbuf, sizeof (errbuf)));
561       break;
562     }
563
564     for (i = 0; i < fds_num; i++)
565     {
566       if ((fds[i].revents & (POLLIN | POLLPRI)) == 0)
567         continue;
568
569       statsd_network_read (fds[i].fd);
570       fds[i].revents = 0;
571     }
572   } /* while (!network_thread_shutdown) */
573
574   /* Clean up */
575   for (i = 0; i < fds_num; i++)
576     close (fds[i].fd);
577   sfree (fds);
578
579   return ((void *) 0);
580 } /* }}} void *statsd_network_thread */
581
582 static int statsd_config_timer_percentile (oconfig_item_t *ci) /* {{{ */
583 {
584   double percent = NAN;
585   double *tmp;
586   int status;
587
588   status = cf_util_get_double (ci, &percent);
589   if (status != 0)
590     return (status);
591
592   if ((percent <= 0.0) || (percent >= 100))
593   {
594     ERROR ("statsd plugin: The value for \"%s\" must be between 0 and 100, "
595         "exclusively.", ci->key);
596     return (ERANGE);
597   }
598
599   tmp = realloc (conf_timer_percentile,
600       sizeof (*conf_timer_percentile) * (conf_timer_percentile_num + 1));
601   if (tmp == NULL)
602   {
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   int i;
616
617   for (i = 0; i < ci->children_num; i++)
618   {
619     oconfig_item_t *child = ci->children + i;
620
621     if (strcasecmp ("Host", child->key) == 0)
622       cf_util_get_string (child, &conf_node);
623     else if (strcasecmp ("Port", child->key) == 0)
624       cf_util_get_service (child, &conf_service);
625     else if (strcasecmp ("DeleteCounters", child->key) == 0)
626       cf_util_get_boolean (child, &conf_delete_counters);
627     else if (strcasecmp ("DeleteTimers", child->key) == 0)
628       cf_util_get_boolean (child, &conf_delete_timers);
629     else if (strcasecmp ("DeleteGauges", child->key) == 0)
630       cf_util_get_boolean (child, &conf_delete_gauges);
631     else if (strcasecmp ("DeleteSets", child->key) == 0)
632       cf_util_get_boolean (child, &conf_delete_sets);
633     else if (strcasecmp ("TimerLower", child->key) == 0)
634       cf_util_get_boolean (child, &conf_timer_lower);
635     else if (strcasecmp ("TimerUpper", child->key) == 0)
636       cf_util_get_boolean (child, &conf_timer_upper);
637     else if (strcasecmp ("TimerSum", child->key) == 0)
638       cf_util_get_boolean (child, &conf_timer_sum);
639     else if (strcasecmp ("TimerCount", child->key) == 0)
640       cf_util_get_boolean (child, &conf_timer_count);
641     else if (strcasecmp ("TimerPercentile", child->key) == 0)
642       statsd_config_timer_percentile (child);
643     else
644       ERROR ("statsd plugin: The \"%s\" config option is not valid.",
645           child->key);
646   }
647
648   return (0);
649 } /* }}} int statsd_config */
650
651 static int statsd_init (void) /* {{{ */
652 {
653   pthread_mutex_lock (&metrics_lock);
654   if (metrics_tree == NULL)
655     metrics_tree = c_avl_create ((void *) strcmp);
656
657   if (!network_thread_running)
658   {
659     int status;
660
661     status = pthread_create (&network_thread,
662         /* attr = */ NULL,
663         statsd_network_thread,
664         /* args = */ NULL);
665     if (status != 0)
666     {
667       char errbuf[1024];
668       pthread_mutex_unlock (&metrics_lock);
669       ERROR ("statsd plugin: pthread_create failed: %s",
670           sstrerror (errno, errbuf, sizeof (errbuf)));
671       return (status);
672     }
673   }
674   network_thread_running = 1;
675
676   pthread_mutex_unlock (&metrics_lock);
677
678   return (0);
679 } /* }}} int statsd_init */
680
681 /* Must hold metrics_lock when calling this function. */
682 static int statsd_metric_clear_set_unsafe (statsd_metric_t *metric) /* {{{ */
683 {
684   void *key;
685   void *value;
686
687   if ((metric == NULL) || (metric->type != STATSD_SET))
688     return (EINVAL);
689
690   if (metric->set == NULL)
691     return (0);
692
693   while (c_avl_pick (metric->set, &key, &value) == 0)
694   {
695     sfree (key);
696     sfree (value);
697   }
698
699   return (0);
700 } /* }}} int statsd_metric_clear_set_unsafe */
701
702 /* Must hold metrics_lock when calling this function. */
703 static int statsd_metric_submit_unsafe (char const *name, /* {{{ */
704     statsd_metric_t const *metric)
705 {
706   value_t values[1];
707   value_list_t vl = VALUE_LIST_INIT;
708
709   vl.values = values;
710   vl.values_len = 1;
711   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
712   sstrncpy (vl.plugin, "statsd", sizeof (vl.plugin));
713
714   if (metric->type == STATSD_GAUGE)
715     sstrncpy (vl.type, "gauge", sizeof (vl.type));
716   else if (metric->type == STATSD_TIMER)
717     sstrncpy (vl.type, "latency", sizeof (vl.type));
718   else if (metric->type == STATSD_SET)
719     sstrncpy (vl.type, "objects", sizeof (vl.type));
720   else /* if (metric->type == STATSD_COUNTER) */
721     sstrncpy (vl.type, "derive", sizeof (vl.type));
722
723   sstrncpy (vl.type_instance, name, sizeof (vl.type_instance));
724
725   if (metric->type == STATSD_GAUGE)
726     values[0].gauge = (gauge_t) metric->value;
727   else if (metric->type == STATSD_TIMER)
728   {
729     size_t i;
730
731     if (metric->updates_num == 0)
732       return (0);
733
734     vl.time = cdtime ();
735
736     ssnprintf (vl.type_instance, sizeof (vl.type_instance),
737         "%s-average", name);
738     values[0].gauge = CDTIME_T_TO_DOUBLE (
739         latency_counter_get_average (metric->latency));
740     plugin_dispatch_values (&vl);
741
742     if (conf_timer_lower) {
743       ssnprintf (vl.type_instance, sizeof (vl.type_instance),
744           "%s-lower", name);
745       values[0].gauge = CDTIME_T_TO_DOUBLE (
746           latency_counter_get_min (metric->latency));
747       plugin_dispatch_values (&vl);
748     }
749
750     if (conf_timer_upper) {
751       ssnprintf (vl.type_instance, sizeof (vl.type_instance),
752           "%s-upper", name);
753       values[0].gauge = CDTIME_T_TO_DOUBLE (
754           latency_counter_get_max (metric->latency));
755       plugin_dispatch_values (&vl);
756     }
757
758     if (conf_timer_sum) {
759       ssnprintf (vl.type_instance, sizeof (vl.type_instance),
760           "%s-sum", name);
761       values[0].gauge = CDTIME_T_TO_DOUBLE (
762           latency_counter_get_sum (metric->latency));
763       plugin_dispatch_values (&vl);
764     }
765
766     for (i = 0; i < conf_timer_percentile_num; i++)
767     {
768       ssnprintf (vl.type_instance, sizeof (vl.type_instance),
769           "%s-percentile-%.0f", name, conf_timer_percentile[i]);
770       values[0].gauge = CDTIME_T_TO_DOUBLE (
771           latency_counter_get_percentile (
772             metric->latency, conf_timer_percentile[i]));
773       plugin_dispatch_values (&vl);
774     }
775
776     /* Keep this at the end, since vl.type is set to "gauge" here. The
777      * vl.type's above are implicitly set to "latency". */
778     if (conf_timer_count) {
779       sstrncpy (vl.type, "gauge", sizeof (vl.type));
780       ssnprintf (vl.type_instance, sizeof (vl.type_instance),
781           "%s-count", name);
782       values[0].gauge = latency_counter_get_num (metric->latency);
783       plugin_dispatch_values (&vl);
784     }
785
786     latency_counter_reset (metric->latency);
787     return (0);
788   }
789   else if (metric->type == STATSD_SET)
790   {
791     if (metric->set == NULL)
792       values[0].gauge = 0.0;
793     else
794       values[0].gauge = (gauge_t) c_avl_size (metric->set);
795   }
796   else
797     values[0].derive = (derive_t) metric->value;
798
799   return (plugin_dispatch_values (&vl));
800 } /* }}} int statsd_metric_submit_unsafe */
801
802 static int statsd_read (void) /* {{{ */
803 {
804   c_avl_iterator_t *iter;
805   char *name;
806   statsd_metric_t *metric;
807
808   char **to_be_deleted = NULL;
809   size_t to_be_deleted_num = 0;
810   size_t i;
811
812   pthread_mutex_lock (&metrics_lock);
813
814   if (metrics_tree == NULL)
815   {
816     pthread_mutex_unlock (&metrics_lock);
817     return (0);
818   }
819
820   iter = c_avl_get_iterator (metrics_tree);
821   while (c_avl_iterator_next (iter, (void *) &name, (void *) &metric) == 0)
822   {
823     if ((metric->updates_num == 0)
824         && ((conf_delete_counters && (metric->type == STATSD_COUNTER))
825           || (conf_delete_timers && (metric->type == STATSD_TIMER))
826           || (conf_delete_gauges && (metric->type == STATSD_GAUGE))
827           || (conf_delete_sets && (metric->type == STATSD_SET))))
828     {
829       DEBUG ("statsd plugin: Deleting metric \"%s\".", name);
830       strarray_add (&to_be_deleted, &to_be_deleted_num, name);
831       continue;
832     }
833
834     /* Names have a prefix, e.g. "c:", which determines the (statsd) type.
835      * Remove this here. */
836     statsd_metric_submit_unsafe (name + 2, metric);
837
838     /* Reset the metric. */
839     metric->updates_num = 0;
840     if (metric->type == STATSD_SET)
841       statsd_metric_clear_set_unsafe (metric);
842   }
843   c_avl_iterator_destroy (iter);
844
845   for (i = 0; i < to_be_deleted_num; i++)
846   {
847     int status;
848
849     status = c_avl_remove (metrics_tree, to_be_deleted[i],
850         (void *) &name, (void *) &metric);
851     if (status != 0)
852     {
853       ERROR ("stats plugin: c_avl_remove (\"%s\") failed with status %i.",
854           to_be_deleted[i], status);
855       continue;
856     }
857
858     sfree (name);
859     sfree (metric);
860   }
861
862   pthread_mutex_unlock (&metrics_lock);
863
864   strarray_free (to_be_deleted, to_be_deleted_num);
865
866   return (0);
867 } /* }}} int statsd_read */
868
869 static int statsd_shutdown (void) /* {{{ */
870 {
871   void *key;
872   void *value;
873
874   pthread_mutex_lock (&metrics_lock);
875
876   if (network_thread_running)
877   {
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   while (c_avl_pick (metrics_tree, &key, &value) == 0)
885   {
886     sfree (key);
887     sfree (value);
888   }
889   c_avl_destroy (metrics_tree);
890   metrics_tree = NULL;
891
892   sfree (conf_node);
893   sfree (conf_service);
894
895   pthread_mutex_unlock (&metrics_lock);
896
897   return (0);
898 } /* }}} int statsd_shutdown */
899
900 void module_register (void)
901 {
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 }
907
908 /* vim: set sw=2 sts=2 et fdm=marker : */