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