Merge branch 'st/python'
[collectd.git] / src / gmond.c
1 /**
2  * collectd - src/gmond.c
3  * Copyright (C) 2009  Florian octo Forster
4  *
5  * This program is free software; you can redistribute it and/or modify it
6  * under the terms of the GNU General Public License as published by the
7  * Free Software Foundation; only version 2 of the License is applicable.
8  *
9  * This program is distributed in the hope that it will be useful, but
10  * WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
12  * General Public License for more details.
13  *
14  * You should have received a copy of the GNU General Public License along
15  * with this program; if not, write to the Free Software Foundation, Inc.,
16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
17  *
18  * Authors:
19  *   Florian octo Forster <octo at verplant.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
28 #if HAVE_PTHREAD_H
29 # include <pthread.h>
30 #endif
31 #if HAVE_SYS_SOCKET_H
32 # include <sys/socket.h>
33 #endif
34 #if HAVE_NETDB_H
35 # include <netdb.h>
36 #endif
37 #if HAVE_NETINET_IN_H
38 # include <netinet/in.h>
39 #endif
40 #if HAVE_ARPA_INET_H
41 # include <arpa/inet.h>
42 #endif
43 #if HAVE_POLL_H
44 # include <poll.h>
45 #endif
46
47 #include <gm_protocol.h>
48
49 #ifndef IPV6_ADD_MEMBERSHIP
50 # ifdef IPV6_JOIN_GROUP
51 #  define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
52 # else
53 #  error "Neither IP_ADD_MEMBERSHIP nor IPV6_JOIN_GROUP is defined"
54 # endif
55 #endif /* !IP_ADD_MEMBERSHIP */
56
57 #ifdef GANGLIA_MAX_MESSAGE_LEN
58 # define BUFF_SIZE GANGLIA_MAX_MESSAGE_LEN
59 #else
60 # define BUFF_SIZE 1400
61 #endif
62
63 struct socket_entry_s
64 {
65   int                     fd;
66   struct sockaddr_storage addr;
67   socklen_t               addrlen;
68 };
69 typedef struct socket_entry_s socket_entry_t;
70
71 struct staging_entry_s
72 {
73   char key[2 * DATA_MAX_NAME_LEN];
74   value_list_t vl;
75   int flags;
76 };
77 typedef struct staging_entry_s staging_entry_t;
78
79 struct metric_map_s
80 {
81   char *ganglia_name;
82   char *type;
83   char *type_instance;
84   char *ds_name;
85   int   ds_type;
86   int   ds_index;
87 };
88 typedef struct metric_map_s metric_map_t;
89
90 #define MC_RECEIVE_GROUP_DEFAULT "239.2.11.71"
91 static char          *mc_receive_group = NULL;
92 #define MC_RECEIVE_PORT_DEFAULT "8649"
93 static char          *mc_receive_port = NULL;
94
95 static struct pollfd *mc_receive_sockets = NULL;
96 static size_t         mc_receive_sockets_num = 0;
97
98 static socket_entry_t  *mc_send_sockets = NULL;
99 static size_t           mc_send_sockets_num = 0;
100 static pthread_mutex_t  mc_send_sockets_lock = PTHREAD_MUTEX_INITIALIZER;
101
102 static int            mc_receive_thread_loop    = 0;
103 static int            mc_receive_thread_running = 0;
104 static pthread_t      mc_receive_thread_id;
105
106 static metric_map_t metric_map_default[] =
107 { /*---------------+-------------+-----------+-------------+------+-----*
108    * ganglia_name  ! type        ! type_inst ! data_source ! type ! idx *
109    *---------------+-------------+-----------+-------------+------+-----*/
110   { "load_one",     "load",       "",         "shortterm",     -1,   -1 },
111   { "load_five",    "load",       "",         "midterm",       -1,   -1 },
112   { "load_fifteen", "load",       "",         "longterm",      -1,   -1 },
113   { "cpu_user",     "cpu",        "user",     "value",         -1,   -1 },
114   { "cpu_system",   "cpu",        "system",   "value",         -1,   -1 },
115   { "cpu_idle",     "cpu",        "idle",     "value",         -1,   -1 },
116   { "cpu_nice",     "cpu",        "nice",     "value",         -1,   -1 },
117   { "cpu_wio",      "cpu",        "wait",     "value",         -1,   -1 },
118   { "mem_free",     "memory",     "free",     "value",         -1,   -1 },
119   { "mem_shared",   "memory",     "shared",   "value",         -1,   -1 },
120   { "mem_buffers",  "memory",     "buffered", "value",         -1,   -1 },
121   { "mem_cached",   "memory",     "cached",   "value",         -1,   -1 },
122   { "mem_total",    "memory",     "total",    "value",         -1,   -1 },
123   { "bytes_in",     "if_octets",  "",         "rx",            -1,   -1 },
124   { "bytes_out",    "if_octets",  "",         "tx",            -1,   -1 },
125   { "pkts_in",      "if_packets", "",         "rx",            -1,   -1 },
126   { "pkts_out",     "if_packets", "",         "tx",            -1,   -1 }
127 };
128 static size_t metric_map_len_default = STATIC_ARRAY_SIZE (metric_map_default);
129
130 static metric_map_t *metric_map = NULL;
131 static size_t        metric_map_len = 0;
132
133 static c_avl_tree_t   *staging_tree;
134 static pthread_mutex_t staging_lock = PTHREAD_MUTEX_INITIALIZER;
135
136 static metric_map_t *metric_lookup (const char *key) /* {{{ */
137 {
138   metric_map_t *map;
139   size_t map_len;
140   size_t i;
141
142   /* Search the user-supplied table first.. */
143   map = metric_map;
144   map_len = metric_map_len;
145   for (i = 0; i < map_len; i++)
146     if (strcmp (map[i].ganglia_name, key) == 0)
147       break;
148
149   /* .. and fall back to the built-in table if nothing is found. */
150   if (i >= map_len)
151   {
152     map = metric_map_default;
153     map_len = metric_map_len_default;
154
155     for (i = 0; i < map_len; i++)
156       if (strcmp (map[i].ganglia_name, key) == 0)
157         break;
158   }
159
160   if (i >= map_len)
161     return (NULL);
162
163   /* Look up the DS type and ds_index. */
164   if ((map[i].ds_type < 0) || (map[i].ds_index < 0)) /* {{{ */
165   {
166     const data_set_t *ds;
167
168     ds = plugin_get_ds (map[i].type);
169     if (ds == NULL)
170     {
171       WARNING ("gmond plugin: Type not defined: %s", map[i].type);
172       return (NULL);
173     }
174
175     if ((map[i].ds_name == NULL) && (ds->ds_num != 1))
176     {
177       WARNING ("gmond plugin: No data source name defined for metric %s, "
178           "but type %s has more than one data source.",
179           map[i].ganglia_name, map[i].type);
180       return (NULL);
181     }
182
183     if (map[i].ds_name == NULL)
184     {
185       map[i].ds_index = 0;
186     }
187     else
188     {
189       int j;
190
191       for (j = 0; j < ds->ds_num; j++)
192         if (strcasecmp (ds->ds[j].name, map[i].ds_name) == 0)
193           break;
194
195       if (j >= ds->ds_num)
196       {
197         WARNING ("gmond plugin: There is no data source "
198             "named `%s' in type `%s'.",
199             map[i].ds_name, ds->type);
200         return (NULL);
201       }
202       map[i].ds_index = j;
203     }
204
205     map[i].ds_type = ds->ds[map[i].ds_index].type;
206   } /* }}} if ((map[i].ds_type < 0) || (map[i].ds_index < 0)) */
207
208   return (map + i);
209 } /* }}} metric_map_t *metric_lookup */
210
211 static int create_sockets (socket_entry_t **ret_sockets, /* {{{ */
212     size_t *ret_sockets_num,
213     const char *node, const char *service, int listen)
214 {
215   struct addrinfo  ai_hints;
216   struct addrinfo *ai_list;
217   struct addrinfo *ai_ptr;
218   int              ai_return;
219
220   socket_entry_t *sockets;
221   size_t          sockets_num;
222
223   int status;
224     
225   sockets     = *ret_sockets;
226   sockets_num = *ret_sockets_num;
227
228   memset (&ai_hints, 0, sizeof (ai_hints));
229   ai_hints.ai_flags    = 0;
230 #ifdef AI_PASSIVE
231   ai_hints.ai_flags |= AI_PASSIVE;
232 #endif
233 #ifdef AI_ADDRCONFIG
234   ai_hints.ai_flags |= AI_ADDRCONFIG;
235 #endif
236   ai_hints.ai_family   = AF_UNSPEC;
237   ai_hints.ai_socktype = SOCK_DGRAM;
238   ai_hints.ai_protocol = IPPROTO_UDP;
239
240   ai_return = getaddrinfo (node, service, &ai_hints, &ai_list);
241   if (ai_return != 0)
242   {
243     char errbuf[1024];
244     ERROR ("gmond plugin: getaddrinfo (%s, %s) failed: %s",
245         (node == NULL) ? "(null)" : node,
246         (service == NULL) ? "(null)" : service,
247         (ai_return == EAI_SYSTEM)
248         ? sstrerror (errno, errbuf, sizeof (errbuf))
249         : gai_strerror (ai_return));
250     return (-1);
251   }
252
253   for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) /* {{{ */
254   {
255     socket_entry_t *tmp;
256
257     tmp = realloc (sockets, (sockets_num + 1) * sizeof (*sockets));
258     if (tmp == NULL)
259     {
260       ERROR ("gmond plugin: realloc failed.");
261       continue;
262     }
263     sockets = tmp;
264
265     sockets[sockets_num].fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype,
266         ai_ptr->ai_protocol);
267     if (sockets[sockets_num].fd < 0)
268     {
269       char errbuf[1024];
270       ERROR ("gmond plugin: socket failed: %s",
271           sstrerror (errno, errbuf, sizeof (errbuf)));
272       continue;
273     }
274
275     assert (sizeof (sockets[sockets_num].addr) >= ai_ptr->ai_addrlen);
276     memcpy (&sockets[sockets_num].addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
277     sockets[sockets_num].addrlen = ai_ptr->ai_addrlen;
278
279     /* Sending socket: Open only one socket and don't bind it. */
280     if (listen == 0)
281     {
282       sockets_num++;
283       break;
284     }
285     else
286     {
287       int yes = 1;
288
289       setsockopt (sockets[sockets_num].fd, SOL_SOCKET, SO_REUSEADDR,
290           (void *) &yes, sizeof (yes));
291     }
292
293     status = bind (sockets[sockets_num].fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
294     if (status != 0)
295     {
296       char errbuf[1024];
297       ERROR ("gmond plugin: bind failed: %s",
298           sstrerror (errno, errbuf, sizeof (errbuf)));
299       close (sockets[sockets_num].fd);
300       continue;
301     }
302
303     if (ai_ptr->ai_family == AF_INET)
304     {
305       struct sockaddr_in *addr;
306       struct ip_mreq mreq;
307       int loop;
308
309       addr = (struct sockaddr_in *) ai_ptr->ai_addr;
310
311       if (!IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
312       {
313         sockets_num++;
314         continue;
315       }
316
317       loop = 1;
318       setsockopt (sockets[sockets_num].fd, IPPROTO_IP, IP_MULTICAST_LOOP,
319           (void *) &loop, sizeof (loop));
320
321       memset (&mreq, 0, sizeof (mreq));
322       mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
323       mreq.imr_interface.s_addr = htonl (INADDR_ANY);
324       setsockopt (sockets[sockets_num].fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
325           (void *) &mreq, sizeof (mreq));
326     } /* if (ai_ptr->ai_family == AF_INET) */
327     else if (ai_ptr->ai_family == AF_INET6)
328     {
329       struct sockaddr_in6 *addr;
330       struct ipv6_mreq mreq;
331       int loop;
332
333       addr = (struct sockaddr_in6 *) ai_ptr->ai_addr;
334
335       if (!IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
336       {
337         sockets_num++;
338         continue;
339       }
340
341       loop = 1;
342       setsockopt (sockets[sockets_num].fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
343           (void *) &loop, sizeof (loop));
344
345       memset (&mreq, 0, sizeof (mreq));
346       memcpy (&mreq.ipv6mr_multiaddr,
347           &addr->sin6_addr, sizeof (addr->sin6_addr));
348       mreq.ipv6mr_interface = 0; /* any */
349       setsockopt (sockets[sockets_num].fd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP,
350           (void *) &mreq, sizeof (mreq));
351     } /* if (ai_ptr->ai_family == AF_INET6) */
352
353     sockets_num++;
354   } /* }}} for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) */
355
356   freeaddrinfo (ai_list);
357
358   if ((*ret_sockets_num) >= sockets_num)
359     return (-1);
360
361   *ret_sockets = sockets;
362   *ret_sockets_num = sockets_num;
363   return (0);
364 } /* }}} int create_sockets */
365
366 static int request_meta_data (const char *host, const char *name) /* {{{ */
367 {
368   Ganglia_metadata_msg msg;
369   char buffer[BUFF_SIZE];
370   unsigned int buffer_size;
371   XDR xdr;
372   size_t i;
373
374   memset (&msg, 0, sizeof (msg));
375
376   msg.id = gmetadata_request;
377   msg.Ganglia_metadata_msg_u.grequest.metric_id.host = strdup (host);
378   msg.Ganglia_metadata_msg_u.grequest.metric_id.name = strdup (name);
379
380   if ((msg.Ganglia_metadata_msg_u.grequest.metric_id.host == NULL)
381       || (msg.Ganglia_metadata_msg_u.grequest.metric_id.name == NULL))
382   {
383     sfree (msg.Ganglia_metadata_msg_u.grequest.metric_id.host);
384     sfree (msg.Ganglia_metadata_msg_u.grequest.metric_id.name);
385     return (-1);
386   }
387
388   memset (buffer, 0, sizeof (buffer));
389   xdrmem_create (&xdr, buffer, sizeof (buffer), XDR_ENCODE);
390
391   if (!xdr_Ganglia_metadata_msg (&xdr, &msg))
392   {
393     sfree (msg.Ganglia_metadata_msg_u.grequest.metric_id.host);
394     sfree (msg.Ganglia_metadata_msg_u.grequest.metric_id.name);
395     return (-1);
396   }
397
398   buffer_size = xdr_getpos (&xdr);
399
400   DEBUG ("gmond plugin: Requesting meta data for %s/%s.",
401       host, name);
402
403   pthread_mutex_lock (&mc_send_sockets_lock);
404   for (i = 0; i < mc_send_sockets_num; i++)
405     sendto (mc_send_sockets[i].fd, buffer, (size_t) buffer_size,
406         /* flags = */ 0,
407         (struct sockaddr *) &mc_send_sockets[i].addr,
408         mc_send_sockets[i].addrlen);
409   pthread_mutex_unlock (&mc_send_sockets_lock);
410
411   sfree (msg.Ganglia_metadata_msg_u.grequest.metric_id.host);
412   sfree (msg.Ganglia_metadata_msg_u.grequest.metric_id.name);
413   return (0);
414 } /* }}} int request_meta_data */
415
416 static staging_entry_t *staging_entry_get (const char *host, /* {{{ */
417     const char *name,
418     const char *type, const char *type_instance,
419     int values_len)
420 {
421   char key[2 * DATA_MAX_NAME_LEN];
422   staging_entry_t *se;
423   int status;
424
425   if (staging_tree == NULL)
426     return (NULL);
427
428   ssnprintf (key, sizeof (key), "%s/%s/%s", host, type,
429       (type_instance != NULL) ? type_instance : "");
430
431   se = NULL;
432   status = c_avl_get (staging_tree, key, (void *) &se);
433   if (status == 0)
434     return (se);
435
436   /* insert new entry */
437   se = (staging_entry_t *) malloc (sizeof (*se));
438   if (se == NULL)
439     return (NULL);
440   memset (se, 0, sizeof (*se));
441
442   sstrncpy (se->key, key, sizeof (se->key));
443   se->flags = 0;
444
445   se->vl.values = (value_t *) calloc (values_len, sizeof (*se->vl.values));
446   if (se->vl.values == NULL)
447   {
448     sfree (se);
449     return (NULL);
450   }
451   se->vl.values_len = values_len;
452
453   se->vl.time = 0;
454   se->vl.interval = 0;
455   sstrncpy (se->vl.host, host, sizeof (se->vl.host));
456   sstrncpy (se->vl.plugin, "gmond", sizeof (se->vl.plugin));
457   sstrncpy (se->vl.type, type, sizeof (se->vl.type));
458   if (type_instance != NULL)
459     sstrncpy (se->vl.type_instance, type_instance,
460         sizeof (se->vl.type_instance));
461
462   status = c_avl_insert (staging_tree, se->key, se);
463   if (status != 0)
464   {
465     ERROR ("gmond plugin: c_avl_insert failed.");
466     sfree (se->vl.values);
467     sfree (se);
468     return (NULL);
469   }
470
471   return (se);
472 } /* }}} staging_entry_t *staging_entry_get */
473
474 static int staging_entry_submit (const char *host, const char *name, /* {{{ */
475     staging_entry_t *se)
476 {
477   value_list_t vl;
478   value_t values[se->vl.values_len];
479
480   if (se->vl.interval == 0)
481   {
482     /* No meta data has been received for this metric yet. */
483     se->flags = 0;
484     pthread_mutex_unlock (&staging_lock);
485     request_meta_data (host, name);
486     return (0);
487   }
488
489   se->flags = 0;
490
491   memcpy (values, se->vl.values, sizeof (values));
492   memcpy (&vl, &se->vl, sizeof (vl));
493
494   /* Unlock before calling `plugin_dispatch_values'.. */
495   pthread_mutex_unlock (&staging_lock);
496
497   vl.values = values;
498
499   plugin_dispatch_values (&vl);
500
501   return (0);
502 } /* }}} int staging_entry_submit */
503
504 static int staging_entry_update (const char *host, const char *name, /* {{{ */
505     const char *type, const char *type_instance,
506     int ds_index, int ds_type, value_t value)
507 {
508   const data_set_t *ds;
509   staging_entry_t *se;
510
511   ds = plugin_get_ds (type);
512   if (ds == NULL)
513   {
514     ERROR ("gmond plugin: Looking up type %s failed.", type);
515     return (-1);
516   }
517
518   if (ds->ds_num <= ds_index)
519   {
520     ERROR ("gmond plugin: Invalid index %i: %s has only %i data source(s).",
521         ds_index, ds->type, ds->ds_num);
522     return (-1);
523   }
524
525   pthread_mutex_lock (&staging_lock);
526
527   se = staging_entry_get (host, name, type, type_instance, ds->ds_num);
528   if (se == NULL)
529   {
530     pthread_mutex_unlock (&staging_lock);
531     ERROR ("gmond plugin: staging_entry_get failed.");
532     return (-1);
533   }
534   if (se->vl.values_len != ds->ds_num)
535   {
536     pthread_mutex_unlock (&staging_lock);
537     return (-1);
538   }
539
540   if (ds_type == DS_TYPE_COUNTER)
541     se->vl.values[ds_index].counter += value.counter;
542   else if (ds_type == DS_TYPE_GAUGE)
543     se->vl.values[ds_index].gauge = value.gauge;
544   else if (ds_type == DS_TYPE_DERIVE)
545     se->vl.values[ds_index].derive += value.derive;
546   else if (ds_type == DS_TYPE_ABSOLUTE)
547     se->vl.values[ds_index].absolute = value.absolute;
548
549   se->flags |= (0x01 << ds_index);
550
551   /* Check if all values have been set and submit if so. */
552   if (se->flags == ((0x01 << se->vl.values_len) - 1))
553   {
554     /* `staging_lock' is unlocked in `staging_entry_submit'. */
555     staging_entry_submit (host, name, se);
556   }
557   else
558   {
559     pthread_mutex_unlock (&staging_lock);
560   }
561
562   return (0);
563 } /* }}} int staging_entry_update */
564
565 static int mc_handle_value_msg (Ganglia_value_msg *msg) /* {{{ */
566 {
567   const char *host;
568   const char *name;
569   metric_map_t *map;
570
571   value_t value_counter;
572   value_t value_gauge;
573   value_t value_derive;
574
575   /* Fill in `host', `name', `value_counter', and `value_gauge' according to
576    * the value type, or return with an error. */
577   switch (msg->id) /* {{{ */
578   {
579     case gmetric_uint:
580     {
581       Ganglia_gmetric_uint msg_uint;
582
583       msg_uint = msg->Ganglia_value_msg_u.gu_int;
584
585       host = msg_uint.metric_id.host;
586       name = msg_uint.metric_id.name;
587       value_counter.counter = (counter_t) msg_uint.ui;
588       value_gauge.gauge = (gauge_t) msg_uint.ui;
589       value_derive.derive = (derive_t) msg_uint.ui;
590       break;
591     }
592
593     case gmetric_string:
594     {
595       Ganglia_gmetric_string msg_string;
596       char *endptr;
597
598       msg_string = msg->Ganglia_value_msg_u.gstr;
599
600       host = msg_string.metric_id.host;
601       name = msg_string.metric_id.name;
602
603       endptr = NULL;
604       errno = 0;
605       value_counter.counter = (counter_t) strtoull (msg_string.str,
606           &endptr, /* base = */ 0);
607       if ((endptr == msg_string.str) || (errno != 0))
608         value_counter.counter = -1;
609
610       endptr = NULL;
611       errno = 0;
612       value_gauge.gauge = (gauge_t) strtod (msg_string.str, &endptr);
613       if ((endptr == msg_string.str) || (errno != 0))
614         value_gauge.gauge = NAN;
615
616       endptr = NULL;
617       errno = 0;
618       value_derive.derive = (derive_t) strtoll (msg_string.str,
619           &endptr, /* base = */ 0);
620       if ((endptr == msg_string.str) || (errno != 0))
621         value_derive.derive = 0;
622
623       break;
624     }
625
626     case gmetric_float:
627     {
628       Ganglia_gmetric_float msg_float;
629
630       msg_float = msg->Ganglia_value_msg_u.gf;
631
632       host = msg_float.metric_id.host;
633       name = msg_float.metric_id.name;
634       value_counter.counter = (counter_t) msg_float.f;
635       value_gauge.gauge = (gauge_t) msg_float.f;
636       value_derive.derive = (derive_t) msg_float.f;
637       break;
638     }
639
640     case gmetric_double:
641     {
642       Ganglia_gmetric_double msg_double;
643
644       msg_double = msg->Ganglia_value_msg_u.gd;
645
646       host = msg_double.metric_id.host;
647       name = msg_double.metric_id.name;
648       value_counter.counter = (counter_t) msg_double.d;
649       value_gauge.gauge = (gauge_t) msg_double.d;
650       value_derive.derive = (derive_t) msg_double.d;
651       break;
652     }
653     default:
654       DEBUG ("gmond plugin: Value type not handled: %i", msg->id);
655       return (-1);
656   } /* }}} switch (msg->id) */
657
658   assert (host != NULL);
659   assert (name != NULL);
660
661   map = metric_lookup (name);
662   if (map != NULL)
663   {
664     value_t val_copy;
665
666     val_copy = value_counter;
667     if (map->ds_type == DS_TYPE_GAUGE)
668       val_copy = value_gauge;
669     else if (map->ds_type == DS_TYPE_DERIVE)
670       val_copy = value_derive;
671
672     return (staging_entry_update (host, name,
673           map->type, map->type_instance,
674           map->ds_index, map->ds_type,
675           val_copy));
676   }
677
678   DEBUG ("gmond plugin: Cannot find a translation for %s.", name);
679   return (-1);
680 } /* }}} int mc_handle_value_msg */
681
682 static int mc_handle_metadata_msg (Ganglia_metadata_msg *msg) /* {{{ */
683 {
684   switch (msg->id)
685   {
686     case gmetadata_full:
687     {
688       Ganglia_metadatadef msg_meta;
689       staging_entry_t *se;
690       const data_set_t *ds;
691       metric_map_t *map;
692
693       msg_meta = msg->Ganglia_metadata_msg_u.gfull;
694
695       if (msg_meta.metric.tmax <= 0)
696         return (-1);
697
698       map = metric_lookup (msg_meta.metric_id.name);
699       if (map == NULL)
700       {
701         DEBUG ("gmond plugin: Not handling meta data %s.",
702             msg_meta.metric_id.name);
703         return (0);
704       }
705
706       ds = plugin_get_ds (map->type);
707       if (ds == NULL)
708       {
709         WARNING ("gmond plugin: Could not find data set %s.", map->type);
710         return (-1);
711       }
712
713       DEBUG ("gmond plugin: Received meta data for %s/%s.",
714           msg_meta.metric_id.host, msg_meta.metric_id.name);
715
716       pthread_mutex_lock (&staging_lock);
717       se = staging_entry_get (msg_meta.metric_id.host,
718           msg_meta.metric_id.name,
719           map->type, map->type_instance,
720           ds->ds_num);
721       if (se != NULL)
722         se->vl.interval = (int) msg_meta.metric.tmax;
723       pthread_mutex_unlock (&staging_lock);
724
725       if (se == NULL)
726       {
727         ERROR ("gmond plugin: staging_entry_get failed.");
728         return (-1);
729       }
730
731       break;
732     }
733
734     default:
735     {
736       return (-1);
737     }
738   }
739
740   return (0);
741 } /* }}} int mc_handle_metadata_msg */
742
743 static int mc_handle_metric (void *buffer, size_t buffer_size) /* {{{ */
744 {
745   XDR xdr;
746   Ganglia_msg_formats format;
747
748   xdrmem_create (&xdr, buffer, buffer_size, XDR_DECODE);
749
750   xdr_Ganglia_msg_formats (&xdr, &format);
751   xdr_setpos (&xdr, 0);
752
753   switch (format)
754   {
755     case gmetric_ushort:
756     case gmetric_short:
757     case gmetric_int:
758     case gmetric_uint:
759     case gmetric_string:
760     case gmetric_float:
761     case gmetric_double:
762     {
763       Ganglia_value_msg msg;
764
765       memset (&msg, 0, sizeof (msg));
766       if (xdr_Ganglia_value_msg (&xdr, &msg))
767         mc_handle_value_msg (&msg);
768       break;
769     }
770
771     case gmetadata_full:
772     case gmetadata_request:
773     {
774       Ganglia_metadata_msg msg;
775       memset (&msg, 0, sizeof (msg));
776       if (xdr_Ganglia_metadata_msg (&xdr, &msg))
777         mc_handle_metadata_msg (&msg);
778       break;
779     }
780
781     default:
782       DEBUG ("gmond plugin: Unknown format: %i", format);
783       return (-1);
784   } /* switch (format) */
785
786
787   return (0);
788 } /* }}} int mc_handle_metric */
789
790 static int mc_handle_socket (struct pollfd *p) /* {{{ */
791 {
792   char buffer[BUFF_SIZE];
793   ssize_t buffer_size;
794
795   if ((p->revents & (POLLIN | POLLPRI)) == 0)
796   {
797     p->revents = 0;
798     return (-1);
799   }
800
801   buffer_size = recv (p->fd, buffer, sizeof (buffer), /* flags = */ 0);
802   if (buffer_size <= 0)
803   {
804     char errbuf[1024];
805     ERROR ("gmond plugin: recv failed: %s",
806         sstrerror (errno, errbuf, sizeof (errbuf)));
807     p->revents = 0;
808     return (-1);
809   }
810
811   mc_handle_metric (buffer, (size_t) buffer_size);
812   return (0);
813 } /* }}} int mc_handle_socket */
814
815 static void *mc_receive_thread (void *arg) /* {{{ */
816 {
817   socket_entry_t *mc_receive_socket_entries;
818   int status;
819   size_t i;
820
821   mc_receive_socket_entries = NULL;
822   status = create_sockets (&mc_receive_socket_entries, &mc_receive_sockets_num,
823       (mc_receive_group != NULL) ? mc_receive_group : MC_RECEIVE_GROUP_DEFAULT,
824       (mc_receive_port != NULL) ? mc_receive_port : MC_RECEIVE_PORT_DEFAULT,
825       /* listen = */ 1);
826   if (status != 0)
827   {
828     ERROR ("gmond plugin: create_sockets failed.");
829     return ((void *) -1);
830   }
831
832   mc_receive_sockets = (struct pollfd *) calloc (mc_receive_sockets_num,
833       sizeof (*mc_receive_sockets));
834   if (mc_receive_sockets == NULL)
835   {
836     ERROR ("gmond plugin: calloc failed.");
837     for (i = 0; i < mc_receive_sockets_num; i++)
838       close (mc_receive_socket_entries[i].fd);
839     free (mc_receive_socket_entries);
840     mc_receive_socket_entries = NULL;
841     mc_receive_sockets_num = 0;
842     return ((void *) -1);
843   }
844
845   for (i = 0; i < mc_receive_sockets_num; i++)
846   {
847     mc_receive_sockets[i].fd = mc_receive_socket_entries[i].fd;
848     mc_receive_sockets[i].events = POLLIN | POLLPRI;
849     mc_receive_sockets[i].revents = 0;
850   }
851
852   while (mc_receive_thread_loop != 0)
853   {
854     status = poll (mc_receive_sockets, mc_receive_sockets_num, -1);
855     if (status <= 0)
856     {
857       char errbuf[1024];
858       if (errno == EINTR)
859         continue;
860       ERROR ("gmond plugin: poll failed: %s",
861           sstrerror (errno, errbuf, sizeof (errbuf)));
862       break;
863     }
864
865     for (i = 0; i < mc_receive_sockets_num; i++)
866     {
867       if (mc_receive_sockets[i].revents != 0)
868         mc_handle_socket (mc_receive_sockets + i);
869     }
870   } /* while (mc_receive_thread_loop != 0) */
871
872   return ((void *) 0);
873 } /* }}} void *mc_receive_thread */
874
875 static int mc_receive_thread_start (void) /* {{{ */
876 {
877   int status;
878
879   if (mc_receive_thread_running != 0)
880     return (-1);
881
882   mc_receive_thread_loop = 1;
883
884   status = pthread_create (&mc_receive_thread_id, /* attr = */ NULL,
885       mc_receive_thread, /* args = */ NULL);
886   if (status != 0)
887   {
888     ERROR ("gmond plugin: Starting receive thread failed.");
889     mc_receive_thread_loop = 0;
890     return (-1);
891   }
892
893   mc_receive_thread_running = 1;
894   return (0);
895 } /* }}} int start_receive_thread */
896
897 static int mc_receive_thread_stop (void) /* {{{ */
898 {
899   if (mc_receive_thread_running == 0)
900     return (-1);
901
902   mc_receive_thread_loop = 0;
903
904   INFO ("gmond plugin: Stopping receive thread.");
905   pthread_kill (mc_receive_thread_id, SIGTERM);
906   pthread_join (mc_receive_thread_id, /* return value = */ NULL);
907   memset (&mc_receive_thread_id, 0, sizeof (mc_receive_thread_id));
908
909   mc_receive_thread_running = 0;
910
911   return (0);
912 } /* }}} int mc_receive_thread_stop */
913
914 /* 
915  * Config:
916  *
917  * <Plugin gmond>
918  *   MCReceiveFrom "239.2.11.71" "8649"
919  *   <Metric "load_one">
920  *     Type "load"
921  *     [TypeInstance "foo"]
922  *     [DataSource "bar"]
923  *   </Metric>
924  * </Plugin>
925  */
926 static int gmond_config_set_string (oconfig_item_t *ci, char **str) /* {{{ */
927 {
928   char *tmp;
929
930   if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING))
931   {
932     WARNING ("gmond plugin: The `%s' option needs "
933         "exactly one string argument.", ci->key);
934     return (-1);
935   }
936
937   tmp = strdup (ci->values[0].value.string);
938   if (tmp == NULL)
939   {
940     ERROR ("gmond plugin: strdup failed.");
941     return (-1);
942   }
943
944   sfree (*str);
945   *str = tmp;
946   return (0);
947 } /* }}} int gmond_config_set_string */
948
949 static int gmond_config_add_metric (oconfig_item_t *ci) /* {{{ */
950 {
951   metric_map_t *map;
952   int i;
953
954   if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING))
955   {
956     WARNING ("gmond plugin: `Metric' blocks need "
957         "exactly one string argument.");
958     return (-1);
959   }
960
961   map = realloc (metric_map, (metric_map_len + 1) * sizeof (*metric_map));
962   if (map == NULL)
963   {
964     ERROR ("gmond plugin: realloc failed.");
965     return (-1);
966   }
967   metric_map = map;
968   map = metric_map + metric_map_len;
969
970   memset (map, 0, sizeof (*map));
971   map->type = NULL;
972   map->type_instance = NULL;
973   map->ds_name = NULL;
974   map->ds_type = -1;
975   map->ds_index = -1;
976
977   map->ganglia_name = strdup (ci->values[0].value.string);
978   if (map->ganglia_name == NULL)
979   {
980     ERROR ("gmond plugin: strdup failed.");
981     return (-1);
982   }
983
984   for (i = 0; i < ci->children_num; i++)
985   {
986     oconfig_item_t *child = ci->children + i;
987     if (strcasecmp ("Type", child->key) == 0)
988       gmond_config_set_string (child, &map->type);
989     else if (strcasecmp ("TypeInstance", child->key) == 0)
990       gmond_config_set_string (child, &map->type_instance);
991     else if (strcasecmp ("DataSource", child->key) == 0)
992       gmond_config_set_string (child, &map->ds_name);
993     else
994     {
995       WARNING ("gmond plugin: Unknown configuration option `%s' ignored.",
996           child->key);
997     }
998   }
999
1000   if (map->type == NULL)
1001   {
1002     ERROR ("gmond plugin: No type is set for metric %s.",
1003         map->ganglia_name);
1004     sfree (map->ganglia_name);
1005     sfree (map->type_instance);
1006     return (-1);
1007   }
1008
1009   metric_map_len++;
1010   return (0);
1011 } /* }}} int gmond_config_add_metric */
1012
1013 static int gmond_config_set_address (oconfig_item_t *ci, /* {{{ */
1014     char **ret_addr, char **ret_port)
1015 {
1016   char *addr;
1017   char *port;
1018
1019   if ((ci->values_num != 1) && (ci->values_num != 2))
1020   {
1021     WARNING ("gmond plugin: The `%s' config option needs "
1022         "one or two string arguments.",
1023         ci->key);
1024     return (-1);
1025   }
1026   if ((ci->values[0].type != OCONFIG_TYPE_STRING)
1027       || ((ci->values_num == 2)
1028         && (ci->values[1].type != OCONFIG_TYPE_STRING)))
1029   {
1030     WARNING ("gmond plugin: The `%s' config option needs "
1031         "one or two string arguments.",
1032         ci->key);
1033     return (-1);
1034   }
1035
1036   addr = strdup (ci->values[0].value.string);
1037   if (ci->values_num == 2)
1038     port = strdup (ci->values[1].value.string);
1039   else
1040     port = NULL;
1041
1042   if ((addr == NULL) || ((ci->values_num == 2) && (port == NULL)))
1043   {
1044     ERROR ("gmond plugin: strdup failed.");
1045     sfree (addr);
1046     sfree (port);
1047     return (-1);
1048   }
1049
1050   sfree (*ret_addr);
1051   sfree (*ret_port);
1052
1053   *ret_addr = addr;
1054   *ret_port = port;
1055
1056   return (0);
1057 } /* }}} int gmond_config_set_address */
1058
1059 static int gmond_config (oconfig_item_t *ci) /* {{{ */
1060 {
1061   int i;
1062
1063   for (i = 0; i < ci->children_num; i++)
1064   {
1065     oconfig_item_t *child = ci->children + i;
1066     if (strcasecmp ("MCReceiveFrom", child->key) == 0)
1067       gmond_config_set_address (child, &mc_receive_group, &mc_receive_port);
1068     else if (strcasecmp ("Metric", child->key) == 0)
1069       gmond_config_add_metric (child);
1070     else
1071     {
1072       WARNING ("gmond plugin: Unknown configuration option `%s' ignored.",
1073           child->key);
1074     }
1075   }
1076
1077   return (0);
1078 } /* }}} int gmond_config */
1079
1080 static int gmond_init (void) /* {{{ */
1081 {
1082   create_sockets (&mc_send_sockets, &mc_send_sockets_num,
1083       (mc_receive_group != NULL) ? mc_receive_group : MC_RECEIVE_GROUP_DEFAULT,
1084       (mc_receive_port != NULL) ? mc_receive_port : MC_RECEIVE_PORT_DEFAULT,
1085       /* listen = */ 0);
1086
1087   staging_tree = c_avl_create ((void *) strcmp);
1088   if (staging_tree == NULL)
1089   {
1090     ERROR ("gmond plugin: c_avl_create failed.");
1091     return (-1);
1092   }
1093
1094   mc_receive_thread_start ();
1095
1096   return (0);
1097 } /* }}} int gmond_init */
1098
1099 static int gmond_shutdown (void) /* {{{ */
1100 {
1101   size_t i;
1102
1103   mc_receive_thread_stop ();
1104
1105   pthread_mutex_lock (&mc_send_sockets_lock);
1106   for (i = 0; i < mc_send_sockets_num; i++)
1107   {
1108     close (mc_send_sockets[i].fd);
1109     mc_send_sockets[i].fd = -1;
1110   }
1111   sfree (mc_send_sockets);
1112   mc_send_sockets_num = 0;
1113   pthread_mutex_unlock (&mc_send_sockets_lock);
1114
1115
1116   return (0);
1117 } /* }}} int gmond_shutdown */
1118
1119 void module_register (void)
1120 {
1121   plugin_register_complex_config ("gmond", gmond_config);
1122   plugin_register_init ("gmond", gmond_init);
1123   plugin_register_shutdown ("gmond", gmond_shutdown);
1124 }
1125
1126 /* vim: set sw=2 sts=2 et fdm=marker : */