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