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