src/network.c

   1 /**
   2  * collectd - src/network.c
   3  * Copyright (C) 2005-2007  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 #include "network.h"
  29
  30 #if HAVE_PTHREAD_H
  31 # include <pthread.h>
  32 #endif
  33 #if HAVE_SYS_SOCKET_H
  34 # include <sys/socket.h>
  35 #endif
  36 #if HAVE_NETDB_H
  37 # include <netdb.h>
  38 #endif
  39 #if HAVE_NETINET_IN_H
  40 # include <netinet/in.h>
  41 #endif
  42 #if HAVE_ARPA_INET_H
  43 # include <arpa/inet.h>
  44 #endif
  45 #if HAVE_POLL_H
  46 # include <poll.h>
  47 #endif
  48
  49 /* 1500 - 40 - 8  =  Ethernet packet - IPv6 header - UDP header */
  50 /* #define BUFF_SIZE 1452 */
  51
  52 #ifndef IPV6_ADD_MEMBERSHIP
  53 # ifdef IPV6_JOIN_GROUP
  54 #  define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
  55 # else
  56 #  error "Neither IP_ADD_MEMBERSHIP nor IPV6_JOIN_GROUP is defined"
  57 # endif
  58 #endif /* !IP_ADD_MEMBERSHIP */
  59
  60 #define BUFF_SIZE 1024
  61
  62 /*
  63  * Private data types
  64  */
  65 typedef struct sockent
  66 {
  67         int                      fd;
  68         struct sockaddr_storage *addr;
  69         socklen_t                addrlen;
  70         struct sockent          *next;
  71 } sockent_t;
  72
  73 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
  74  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  75  * +-------+-----------------------+-------------------------------+
  76  * ! Ver.  !                       ! Length                        !
  77  * +-------+-----------------------+-------------------------------+
  78  */
  79 struct part_header_s
  80 {
  81         uint16_t type;
  82         uint16_t length;
  83 };
  84 typedef struct part_header_s part_header_t;
  85
  86 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
  87  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  88  * +-------------------------------+-------------------------------+
  89  * ! Type                          ! Length                        !
  90  * +-------------------------------+-------------------------------+
  91  * : (Length - 4) Bytes                                            :
  92  * +---------------------------------------------------------------+
  93  */
  94 struct part_string_s
  95 {
  96         part_header_t *head;
  97         char *value;
  98 };
  99 typedef struct part_string_s part_string_t;
 100
 101 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
 102  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 103  * +-------------------------------+-------------------------------+
 104  * ! Type                          ! Length                        !
 105  * +-------------------------------+-------------------------------+
 106  * : (Length - 4 == 2 || 4 || 8) Bytes                             :
 107  * +---------------------------------------------------------------+
 108  */
 109 struct part_number_s
 110 {
 111         part_header_t *head;
 112         uint64_t *value;
 113 };
 114 typedef struct part_number_s part_number_t;
 115
 116 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
 117  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 118  * +-------------------------------+-------------------------------+
 119  * ! Type                          ! Length                        !
 120  * +-------------------------------+---------------+---------------+
 121  * ! Num of values                 ! Type0         ! Type1         !
 122  * +-------------------------------+---------------+---------------+
 123  * ! Value0                                                        !
 124  * !                                                               !
 125  * +---------------------------------------------------------------+
 126  * ! Value1                                                        !
 127  * !                                                               !
 128  * +---------------------------------------------------------------+
 129  */
 130 struct part_values_s
 131 {
 132         part_header_t *head;
 133         uint16_t *num_values;
 134         uint8_t  *values_types;
 135         value_t  *values;
 136 };
 137 typedef struct part_values_s part_values_t;
 138
 139 /*
 140  * Private variables
 141  */
 142 static const char *config_keys[] =
 143 {
 144         "Listen",
 145         "Server",
 146         "TimeToLive",
 147         "Forward"
 148 };
 149 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
 150
 151 static int network_config_ttl = 0;
 152 static int network_config_forward = 0;
 153
 154 static sockent_t *sending_sockets = NULL;
 155
 156 static struct pollfd *listen_sockets = NULL;
 157 static int listen_sockets_num = 0;
 158 static pthread_t listen_thread = 0;
 159 static int listen_loop = 0;
 160
 161 static char         send_buffer[BUFF_SIZE];
 162 static char        *send_buffer_ptr;
 163 static int          send_buffer_fill;
 164 static value_list_t send_buffer_vl = VALUE_LIST_INIT;
 165 static char         send_buffer_type[DATA_MAX_NAME_LEN];
 166 static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER;
 167
 168 static avl_tree_t      *cache_tree = NULL;
 169 static pthread_mutex_t  cache_lock = PTHREAD_MUTEX_INITIALIZER;
 170
 171 /*
 172  * Private functions
 173  */
 174 static int cache_check (const char *type, const value_list_t *vl)
 175 {
 176         char key[1024];
 177         time_t *value = NULL;
 178         int retval = -1;
 179
 180         if (cache_tree == NULL)
 181                 return (-1);
 182
 183         if (format_name (key, sizeof (key), vl->host, vl->plugin,
 184                                 vl->plugin_instance, type, vl->type_instance))
 185                 return (-1);
 186
 187         pthread_mutex_lock (&cache_lock);
 188
 189         if (avl_get (cache_tree, key, (void *) &value) == 0)
 190         {
 191                 if (*value < vl->time)
 192                 {
 193                         *value = vl->time;
 194                         retval = 0;
 195                 }
 196                 else
 197                 {
 198                         DEBUG ("network plugin: cache_check: *value = %i >= vl->time = %i",
 199                                         (int) *value, (int) vl->time);
 200                         retval = 1;
 201                 }
 202         }
 203         else
 204         {
 205                 char *key_copy = strdup (key);
 206                 value = malloc (sizeof (time_t));
 207                 if ((key_copy != NULL) && (value != NULL))
 208                 {
 209                         *value = vl->time;
 210                         avl_insert (cache_tree, key_copy, value);
 211                         retval = 0;
 212                 }
 213                 else
 214                 {
 215                         sfree (key_copy);
 216                         sfree (value);
 217                 }
 218         }
 219
 220         /* TODO: Flush cache */
 221
 222         pthread_mutex_unlock (&cache_lock);
 223
 224         DEBUG ("network plugin: cache_check: key = %s; time = %i; retval = %i",
 225                         key, (int) vl->time, retval);
 226
 227         return (retval);
 228 } /* int cache_check */
 229
 230 static int write_part_values (char **ret_buffer, int *ret_buffer_len,
 231                 const data_set_t *ds, const value_list_t *vl)
 232 {
 233         part_values_t pv;
 234         int i;
 235
 236         i = 6 + (9 * vl->values_len);
 237         if (*ret_buffer_len < i)
 238                 return (-1);
 239         *ret_buffer_len -= i;
 240
 241         pv.head = (part_header_t *) *ret_buffer;
 242         pv.num_values = (uint16_t *) (pv.head + 1);
 243         pv.values_types = (uint8_t *) (pv.num_values + 1);
 244         pv.values = (value_t *) (pv.values_types + vl->values_len);
 245         *ret_buffer = (void *) (pv.values + vl->values_len);
 246
 247         pv.head->type = htons (TYPE_VALUES);
 248         pv.head->length = htons (6 + (9 * vl->values_len));
 249         *pv.num_values = htons ((uint16_t) vl->values_len);
 250
 251         for (i = 0; i < vl->values_len; i++)
 252         {
 253                 if (ds->ds[i].type == DS_TYPE_COUNTER)
 254                 {
 255                         pv.values_types[i] = DS_TYPE_COUNTER;
 256                         pv.values[i].counter = htonll (vl->values[i].counter);
 257                 }
 258                 else
 259                 {
 260                         pv.values_types[i] = DS_TYPE_GAUGE;
 261                         pv.values[i].gauge = vl->values[i].gauge;
 262                 }
 263         } /* for (values) */
 264
 265         return (0);
 266 } /* int write_part_values */
 267
 268 static int write_part_number (char **ret_buffer, int *ret_buffer_len,
 269                 int type, uint64_t value)
 270 {
 271         part_number_t pn;
 272
 273         if (*ret_buffer_len < 12)
 274                 return (-1);
 275
 276         pn.head = (part_header_t *) *ret_buffer;
 277         pn.value = (uint64_t *) (pn.head + 1);
 278
 279         pn.head->type = htons (type);
 280         pn.head->length = htons (12);
 281         *pn.value = htonll (value);
 282
 283         *ret_buffer = (char *) (pn.value + 1);
 284         *ret_buffer_len -= 12;
 285
 286         return (0);
 287 } /* int write_part_number */
 288
 289 static int write_part_string (char **ret_buffer, int *ret_buffer_len,
 290                 int type, const char *str, int str_len)
 291 {
 292         part_string_t ps;
 293         int len;
 294
 295         len = 4 + str_len + 1;
 296         if (*ret_buffer_len < len)
 297                 return (-1);
 298         *ret_buffer_len -= len;
 299
 300         ps.head = (part_header_t *) *ret_buffer;
 301         ps.value = (char *) (ps.head + 1);
 302
 303         ps.head->type = htons ((uint16_t) type);
 304         ps.head->length = htons ((uint16_t) str_len + 5);
 305         if (str_len > 0)
 306                 memcpy (ps.value, str, str_len);
 307         ps.value[str_len] = '\0';
 308         *ret_buffer = (void *) (ps.value + (str_len + 1));
 309
 310         return (0);
 311 } /* int write_part_string */
 312
 313 static int parse_part_values (void **ret_buffer, int *ret_buffer_len,
 314                 value_t **ret_values, int *ret_num_values)
 315 {
 316         char *buffer = *ret_buffer;
 317         int   buffer_len = *ret_buffer_len;
 318         part_values_t pv;
 319         int   i;
 320
 321         uint16_t h_length;
 322         uint16_t h_type;
 323         uint16_t h_num;
 324
 325         if (buffer_len < (15))
 326         {
 327                 DEBUG ("packet is too short: buffer_len = %i", buffer_len);
 328                 return (-1);
 329         }
 330
 331         pv.head = (part_header_t *) buffer;
 332         h_length = ntohs (pv.head->length);
 333         h_type = ntohs (pv.head->type);
 334
 335         assert (h_type == TYPE_VALUES);
 336
 337         pv.num_values = (uint16_t *) (pv.head + 1);
 338         h_num = ntohs (*pv.num_values);
 339
 340         if (h_num != ((h_length - 6) / 9))
 341         {
 342                 DEBUG ("`length' and `num of values' don't match");
 343                 return (-1);
 344         }
 345
 346         pv.values_types = (uint8_t *) (pv.num_values + 1);
 347         pv.values = (value_t *) (pv.values_types + h_num);
 348
 349         for (i = 0; i < h_num; i++)
 350                 if (pv.values_types[i] == DS_TYPE_COUNTER)
 351                         pv.values[i].counter = ntohll (pv.values[i].counter);
 352
 353         *ret_buffer     = (void *) (pv.values + h_num);
 354         *ret_buffer_len = buffer_len - h_length;
 355         *ret_num_values = h_num;
 356         *ret_values     = pv.values;
 357
 358         return (0);
 359 } /* int parse_part_values */
 360
 361 static int parse_part_number (void **ret_buffer, int *ret_buffer_len,
 362                 uint64_t *value)
 363 {
 364         part_number_t pn;
 365         uint16_t len;
 366
 367         pn.head = (part_header_t *) *ret_buffer;
 368         pn.value = (uint64_t *) (pn.head + 1);
 369
 370         len = ntohs (pn.head->length);
 371         if (len != 12)
 372                 return (-1);
 373         if (len > *ret_buffer_len)
 374                 return (-1);
 375         *value = ntohll (*pn.value);
 376
 377         *ret_buffer = (void *) (pn.value + 1);
 378         *ret_buffer_len -= len;
 379
 380         return (0);
 381 } /* int parse_part_number */
 382
 383 static int parse_part_string (void **ret_buffer, int *ret_buffer_len,
 384                 char *output, int output_len)
 385 {
 386         char *buffer = *ret_buffer;
 387         int   buffer_len = *ret_buffer_len;
 388         part_string_t ps;
 389
 390         uint16_t h_length;
 391         uint16_t h_type;
 392
 393         DEBUG ("network plugin: parse_part_string: ret_buffer = %p;"
 394                         " ret_buffer_len = %i; output = %p; output_len = %i;",
 395                         *ret_buffer, *ret_buffer_len,
 396                         (void *) output, output_len);
 397
 398         ps.head = (part_header_t *) buffer;
 399
 400         h_length = ntohs (ps.head->length);
 401         h_type = ntohs (ps.head->type);
 402
 403         DEBUG ("network plugin: parse_part_string: length = %hu; type = %hu;",
 404                         h_length, h_type);
 405
 406         if (buffer_len < h_length)
 407         {
 408                 DEBUG ("packet is too short");
 409                 return (-1);
 410         }
 411         assert ((h_type == TYPE_HOST)
 412                         || (h_type == TYPE_PLUGIN)
 413                         || (h_type == TYPE_PLUGIN_INSTANCE)
 414                         || (h_type == TYPE_TYPE)
 415                         || (h_type == TYPE_TYPE_INSTANCE));
 416
 417         ps.value = buffer + 4;
 418         if (ps.value[h_length - 5] != '\0')
 419         {
 420                 DEBUG ("String does not end with a nullbyte");
 421                 return (-1);
 422         }
 423
 424         if (output_len < (h_length - 4))
 425         {
 426                 DEBUG ("output buffer is too small");
 427                 return (-1);
 428         }
 429         strcpy (output, ps.value);
 430
 431         DEBUG ("network plugin: parse_part_string: output = %s", output);
 432
 433         *ret_buffer = (void *) (buffer + h_length);
 434         *ret_buffer_len = buffer_len - h_length;
 435
 436         return (0);
 437 } /* int parse_part_string */
 438
 439 static int parse_packet (void *buffer, int buffer_len)
 440 {
 441         part_header_t *header;
 442         int status;
 443
 444         value_list_t vl = VALUE_LIST_INIT;
 445         char type[DATA_MAX_NAME_LEN];
 446
 447         DEBUG ("network plugin: parse_packet: buffer = %p; buffer_len = %i;",
 448                         buffer, buffer_len);
 449
 450         memset (&vl, '\0', sizeof (vl));
 451         memset (&type, '\0', sizeof (type));
 452         status = 0;
 453
 454         while ((status == 0) && (buffer_len > sizeof (part_header_t)))
 455         {
 456                 header = (part_header_t *) buffer;
 457
 458                 if (ntohs (header->length) > buffer_len)
 459                         break;
 460                 /* Assure that this loop terminates eventually */
 461                 if (ntohs (header->length) < 4)
 462                         break;
 463
 464                 if (ntohs (header->type) == TYPE_VALUES)
 465                 {
 466                         status = parse_part_values (&buffer, &buffer_len,
 467                                         &vl.values, &vl.values_len);
 468
 469                         if (status != 0)
 470                         {
 471                                 DEBUG ("parse_part_values failed.");
 472                                 break;
 473                         }
 474
 475                         if ((vl.time > 0)
 476                                         && (strlen (vl.host) > 0)
 477                                         && (strlen (vl.plugin) > 0)
 478                                         && (strlen (type) > 0)
 479                                         && (cache_check (type, &vl) == 0))
 480                         {
 481                                 DEBUG ("network plugin: parse_packet:"
 482                                                 " dispatching values");
 483                                 plugin_dispatch_values (type, &vl);
 484                         }
 485                         else
 486                         {
 487                                 DEBUG ("network plugin: parse_packet:"
 488                                                 " NOT dispatching values");
 489                         }
 490                 }
 491                 else if (ntohs (header->type) == TYPE_TIME)
 492                 {
 493                         uint64_t tmp = 0;
 494                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 495                         if (status == 0)
 496                                 vl.time = (time_t) tmp;
 497                 }
 498                 else if (ntohs (header->type) == TYPE_HOST)
 499                 {
 500                         status = parse_part_string (&buffer, &buffer_len,
 501                                         vl.host, sizeof (vl.host));
 502                         DEBUG ("network plugin: parse_packet: vl.host = %s", vl.host);
 503                 }
 504                 else if (ntohs (header->type) == TYPE_PLUGIN)
 505                 {
 506                         status = parse_part_string (&buffer, &buffer_len,
 507                                         vl.plugin, sizeof (vl.plugin));
 508                         DEBUG ("network plugin: parse_packet: vl.plugin = %s", vl.plugin);
 509                 }
 510                 else if (ntohs (header->type) == TYPE_PLUGIN_INSTANCE)
 511                 {
 512                         status = parse_part_string (&buffer, &buffer_len,
 513                                         vl.plugin_instance, sizeof (vl.plugin_instance));
 514                         DEBUG ("network plugin: parse_packet: vl.plugin_instance = %s", vl.plugin_instance);
 515                 }
 516                 else if (ntohs (header->type) == TYPE_TYPE)
 517                 {
 518                         status = parse_part_string (&buffer, &buffer_len,
 519                                         type, sizeof (type));
 520                         DEBUG ("network plugin: parse_packet: type = %s", type);
 521                 }
 522                 else if (ntohs (header->type) == TYPE_TYPE_INSTANCE)
 523                 {
 524                         status = parse_part_string (&buffer, &buffer_len,
 525                                         vl.type_instance, sizeof (vl.type_instance));
 526                         DEBUG ("network plugin: parse_packet: vl.type_instance = %s", vl.type_instance);
 527                 }
 528                 else
 529                 {
 530                         DEBUG ("network plugin: parse_packet: Unknown part"
 531                                         " type: 0x%0hx", ntohs (header->type));
 532                         buffer = ((char *) buffer) + ntohs (header->length);
 533                 }
 534         } /* while (buffer_len > sizeof (part_header_t)) */
 535
 536         return (0);
 537 } /* int parse_packet */
 538
 539 static void free_sockent (sockent_t *se)
 540 {
 541         sockent_t *next;
 542         while (se != NULL)
 543         {
 544                 next = se->next;
 545                 free (se->addr);
 546                 free (se);
 547                 se = next;
 548         }
 549 } /* void free_sockent */
 550
 551 /*
 552  * int network_set_ttl
 553  *
 554  * Set the `IP_MULTICAST_TTL', `IP_TTL', `IPV6_MULTICAST_HOPS' or
 555  * `IPV6_UNICAST_HOPS', depending on which option is applicable.
 556  *
 557  * The `struct addrinfo' is used to destinguish between unicast and multicast
 558  * sockets.
 559  */
 560 static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 561 {
 562         if ((network_config_ttl < 1) || (network_config_ttl > 255))
 563                 return (-1);
 564
 565         DEBUG ("ttl = %i", network_config_ttl);
 566
 567         if (ai->ai_family == AF_INET)
 568         {
 569                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 570                 int optname;
 571
 572                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 573                         optname = IP_MULTICAST_TTL;
 574                 else
 575                         optname = IP_TTL;
 576
 577                 if (setsockopt (se->fd, IPPROTO_IP, optname,
 578                                         &network_config_ttl,
 579                                         sizeof (network_config_ttl)) == -1)
 580                 {
 581                         char errbuf[1024];
 582                         ERROR ("setsockopt: %s",
 583                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 584                         return (-1);
 585                 }
 586         }
 587         else if (ai->ai_family == AF_INET6)
 588         {
 589                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 590                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 591                 int optname;
 592
 593                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 594                         optname = IPV6_MULTICAST_HOPS;
 595                 else
 596                         optname = IPV6_UNICAST_HOPS;
 597
 598                 if (setsockopt (se->fd, IPPROTO_IPV6, optname,
 599                                         &network_config_ttl,
 600                                         sizeof (network_config_ttl)) == -1)
 601                 {
 602                         char errbuf[1024];
 603                         ERROR ("setsockopt: %s",
 604                                         sstrerror (errno, errbuf,
 605                                                 sizeof (errbuf)));
 606                         return (-1);
 607                 }
 608         }
 609
 610         return (0);
 611 } /* int network_set_ttl */
 612
 613 static int network_bind_socket (const sockent_t *se, const struct addrinfo *ai)
 614 {
 615         int loop = 0;
 616
 617         DEBUG ("fd = %i; calling `bind'", se->fd);
 618
 619         if (bind (se->fd, ai->ai_addr, ai->ai_addrlen) == -1)
 620         {
 621                 char errbuf[1024];
 622                 ERROR ("bind: %s",
 623                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 624                 return (-1);
 625         }
 626
 627         if (ai->ai_family == AF_INET)
 628         {
 629                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 630                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 631                 {
 632                         struct ip_mreq mreq;
 633
 634                         DEBUG ("fd = %i; IPv4 multicast address found", se->fd);
 635
 636                         mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
 637                         mreq.imr_interface.s_addr = htonl (INADDR_ANY);
 638
 639                         if (setsockopt (se->fd, IPPROTO_IP, IP_MULTICAST_LOOP,
 640                                                 &loop, sizeof (loop)) == -1)
 641                         {
 642                                 char errbuf[1024];
 643                                 ERROR ("setsockopt: %s",
 644                                                 sstrerror (errno, errbuf,
 645                                                         sizeof (errbuf)));
 646                                 return (-1);
 647                         }
 648
 649                         if (setsockopt (se->fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 650                                                 &mreq, sizeof (mreq)) == -1)
 651                         {
 652                                 char errbuf[1024];
 653                                 ERROR ("setsockopt: %s",
 654                                                 sstrerror (errno, errbuf,
 655                                                         sizeof (errbuf)));
 656                                 return (-1);
 657                         }
 658                 }
 659         }
 660         else if (ai->ai_family == AF_INET6)
 661         {
 662                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 663                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 664                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 665                 {
 666                         struct ipv6_mreq mreq;
 667
 668                         DEBUG ("fd = %i; IPv6 multicast address found", se->fd);
 669
 670                         memcpy (&mreq.ipv6mr_multiaddr,
 671                                         &addr->sin6_addr,
 672                                         sizeof (addr->sin6_addr));
 673
 674                         /* http://developer.apple.com/documentation/Darwin/Reference/ManPages/man4/ip6.4.html
 675                          * ipv6mr_interface may be set to zeroes to
 676                          * choose the default multicast interface or to
 677                          * the index of a particular multicast-capable
 678                          * interface if the host is multihomed.
 679                          * Membership is associ-associated with a
 680                          * single interface; programs running on
 681                          * multihomed hosts may need to join the same
 682                          * group on more than one interface.*/
 683                         mreq.ipv6mr_interface = 0;
 684
 685                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
 686                                                 &loop, sizeof (loop)) == -1)
 687                         {
 688                                 char errbuf[1024];
 689                                 ERROR ("setsockopt: %s",
 690                                                 sstrerror (errno, errbuf,
 691                                                         sizeof (errbuf)));
 692                                 return (-1);
 693                         }
 694
 695                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP,
 696                                                 &mreq, sizeof (mreq)) == -1)
 697                         {
 698                                 char errbuf[1024];
 699                                 ERROR ("setsockopt: %s",
 700                                                 sstrerror (errno, errbuf,
 701                                                         sizeof (errbuf)));
 702                                 return (-1);
 703                         }
 704                 }
 705         }
 706
 707         return (0);
 708 } /* int network_bind_socket */
 709
 710 static sockent_t *network_create_socket (const char *node,
 711                 const char *service,
 712                 int listen)
 713 {
 714         struct addrinfo  ai_hints;
 715         struct addrinfo *ai_list, *ai_ptr;
 716         int              ai_return;
 717
 718         sockent_t *se_head = NULL;
 719         sockent_t *se_tail = NULL;
 720
 721         DEBUG ("node = %s, service = %s", node, service);
 722
 723         memset (&ai_hints, '\0', sizeof (ai_hints));
 724         ai_hints.ai_flags    = 0;
 725 #ifdef AI_PASSIVE
 726         ai_hints.ai_flags |= AI_PASSIVE;
 727 #endif
 728 #ifdef AI_ADDRCONFIG
 729         ai_hints.ai_flags |= AI_ADDRCONFIG;
 730 #endif
 731         ai_hints.ai_family   = AF_UNSPEC;
 732         ai_hints.ai_socktype = SOCK_DGRAM;
 733         ai_hints.ai_protocol = IPPROTO_UDP;
 734
 735         ai_return = getaddrinfo (node, service, &ai_hints, &ai_list);
 736         if (ai_return != 0)
 737         {
 738                 char errbuf[1024];
 739                 ERROR ("getaddrinfo (%s, %s): %s",
 740                                 (node == NULL) ? "(null)" : node,
 741                                 (service == NULL) ? "(null)" : service,
 742                                 (ai_return == EAI_SYSTEM)
 743                                 ? sstrerror (errno, errbuf, sizeof (errbuf))
 744                                 : gai_strerror (ai_return));
 745                 return (NULL);
 746         }
 747
 748         for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 749         {
 750                 sockent_t *se;
 751
 752                 if ((se = (sockent_t *) malloc (sizeof (sockent_t))) == NULL)
 753                 {
 754                         char errbuf[1024];
 755                         ERROR ("malloc: %s",
 756                                         sstrerror (errno, errbuf,
 757                                                 sizeof (errbuf)));
 758                         continue;
 759                 }
 760
 761                 if ((se->addr = (struct sockaddr_storage *) malloc (sizeof (struct sockaddr_storage))) == NULL)
 762                 {
 763                         char errbuf[1024];
 764                         ERROR ("malloc: %s",
 765                                         sstrerror (errno, errbuf,
 766                                                 sizeof (errbuf)));
 767                         free (se);
 768                         continue;
 769                 }
 770
 771                 assert (sizeof (struct sockaddr_storage) >= ai_ptr->ai_addrlen);
 772                 memset (se->addr, '\0', sizeof (struct sockaddr_storage));
 773                 memcpy (se->addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 774                 se->addrlen = ai_ptr->ai_addrlen;
 775
 776                 se->fd   = socket (ai_ptr->ai_family,
 777                                 ai_ptr->ai_socktype,
 778                                 ai_ptr->ai_protocol);
 779                 se->next = NULL;
 780
 781                 if (se->fd == -1)
 782                 {
 783                         char errbuf[1024];
 784                         ERROR ("socket: %s",
 785                                         sstrerror (errno, errbuf,
 786                                                 sizeof (errbuf)));
 787                         free (se->addr);
 788                         free (se);
 789                         continue;
 790                 }
 791
 792                 if (listen != 0)
 793                 {
 794                         if (network_bind_socket (se, ai_ptr) != 0)
 795                         {
 796                                 close (se->fd);
 797                                 free (se->addr);
 798                                 free (se);
 799                                 continue;
 800                         }
 801                 }
 802                 else /* listen == 0 */
 803                 {
 804                         network_set_ttl (se, ai_ptr);
 805                 }
 806
 807                 if (se_tail == NULL)
 808                 {
 809                         se_head = se;
 810                         se_tail = se;
 811                 }
 812                 else
 813                 {
 814                         se_tail->next = se;
 815                         se_tail = se;
 816                 }
 817
 818                 /* We don't open more than one write-socket per node/service pair.. */
 819                 if (listen == 0)
 820                         break;
 821         }
 822
 823         freeaddrinfo (ai_list);
 824
 825         return (se_head);
 826 } /* sockent_t *network_create_socket */
 827
 828 static sockent_t *network_create_default_socket (int listen)
 829 {
 830         sockent_t *se_ptr  = NULL;
 831         sockent_t *se_head = NULL;
 832         sockent_t *se_tail = NULL;
 833
 834         se_ptr = network_create_socket (NET_DEFAULT_V6_ADDR,
 835                         NET_DEFAULT_PORT, listen);
 836
 837         /* Don't send to the same machine in IPv6 and IPv4 if both are available. */
 838         if ((listen == 0) && (se_ptr != NULL))
 839                 return (se_ptr);
 840
 841         if (se_ptr != NULL)
 842         {
 843                 se_head = se_ptr;
 844                 se_tail = se_ptr;
 845                 while (se_tail->next != NULL)
 846                         se_tail = se_tail->next;
 847         }
 848
 849         se_ptr = network_create_socket (NET_DEFAULT_V4_ADDR, NET_DEFAULT_PORT, listen);
 850
 851         if (se_tail == NULL)
 852                 return (se_ptr);
 853
 854         se_tail->next = se_ptr;
 855         return (se_head);
 856 } /* sockent_t *network_create_default_socket */
 857
 858 static int network_add_listen_socket (const char *node, const char *service)
 859 {
 860         sockent_t *se;
 861         sockent_t *se_ptr;
 862         int se_num = 0;
 863
 864         if (service == NULL)
 865                 service = NET_DEFAULT_PORT;
 866
 867         if (node == NULL)
 868                 se = network_create_default_socket (1 /* listen == true */);
 869         else
 870                 se = network_create_socket (node, service, 1 /* listen == true */);
 871
 872         if (se == NULL)
 873                 return (-1);
 874
 875         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
 876                 se_num++;
 877
 878         listen_sockets = (struct pollfd *) realloc (listen_sockets,
 879                         (listen_sockets_num + se_num)
 880                         * sizeof (struct pollfd));
 881
 882         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
 883         {
 884                 listen_sockets[listen_sockets_num].fd = se_ptr->fd;
 885                 listen_sockets[listen_sockets_num].events = POLLIN | POLLPRI;
 886                 listen_sockets[listen_sockets_num].revents = 0;
 887                 listen_sockets_num++;
 888         } /* for (se) */
 889
 890         free_sockent (se);
 891         return (0);
 892 } /* int network_add_listen_socket */
 893
 894 static int network_add_sending_socket (const char *node, const char *service)
 895 {
 896         sockent_t *se;
 897         sockent_t *se_ptr;
 898
 899         if (service == NULL)
 900                 service = NET_DEFAULT_PORT;
 901
 902         if (node == NULL)
 903                 se = network_create_default_socket (0 /* listen == false */);
 904         else
 905                 se = network_create_socket (node, service, 0 /* listen == false */);
 906
 907         if (se == NULL)
 908                 return (-1);
 909
 910         if (sending_sockets == NULL)
 911         {
 912                 sending_sockets = se;
 913                 return (0);
 914         }
 915
 916         for (se_ptr = sending_sockets; se_ptr->next != NULL; se_ptr = se_ptr->next)
 917                 /* seek end */;
 918
 919         se_ptr->next = se;
 920         return (0);
 921 } /* int network_get_listen_socket */
 922
 923 int network_receive (void)
 924 {
 925         char buffer[BUFF_SIZE];
 926         int  buffer_len;
 927
 928         int i;
 929         int status;
 930
 931         if (listen_sockets_num == 0)
 932                 network_add_listen_socket (NULL, NULL);
 933
 934         if (listen_sockets_num == 0)
 935         {
 936                 ERROR ("network: Failed to open a listening socket.");
 937                 return (-1);
 938         }
 939
 940         while (listen_loop == 0)
 941         {
 942                 status = poll (listen_sockets, listen_sockets_num, -1);
 943
 944                 if (status <= 0)
 945                 {
 946                         char errbuf[1024];
 947                         if (errno == EINTR)
 948                                 continue;
 949                         ERROR ("poll failed: %s",
 950                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 951                         return (-1);
 952                 }
 953
 954                 for (i = 0; (i < listen_sockets_num) && (status > 0); i++)
 955                 {
 956                         if ((listen_sockets[i].revents & (POLLIN | POLLPRI)) == 0)
 957                                 continue;
 958                         status--;
 959
 960                         buffer_len = recv (listen_sockets[i].fd,
 961                                         buffer, sizeof (buffer),
 962                                         0 /* no flags */);
 963                         if (buffer_len < 0)
 964                         {
 965                                 char errbuf[1024];
 966                                 ERROR ("recv failed: %s",
 967                                                 sstrerror (errno, errbuf,
 968                                                         sizeof (errbuf)));
 969                                 return (-1);
 970                         }
 971
 972                         parse_packet (buffer, buffer_len);
 973                 } /* for (listen_sockets) */
 974         } /* while (listen_loop == 0) */
 975
 976         return (0);
 977 }
 978
 979 static void *receive_thread (void *arg)
 980 {
 981         return (network_receive () ? (void *) 1 : (void *) 0);
 982 } /* void *receive_thread */
 983
 984 static void network_send_buffer (const char *buffer, int buffer_len)
 985 {
 986         sockent_t *se;
 987         int status;
 988
 989         DEBUG ("buffer_len = %i", buffer_len);
 990
 991         for (se = sending_sockets; se != NULL; se = se->next)
 992         {
 993                 while (42)
 994                 {
 995                         status = sendto (se->fd, buffer, buffer_len, 0 /* no flags */,
 996                                         (struct sockaddr *) se->addr, se->addrlen);
 997                         if (status < 0)
 998                         {
 999                                 char errbuf[1024];
1000                                 if (errno == EINTR)
1001                                         continue;
1002                                 ERROR ("network plugin: sendto failed: %s",
1003                                                 sstrerror (errno, errbuf,
1004                                                         sizeof (errbuf)));
1005                                 break;
1006                         }
1007
1008                         break;
1009                 } /* while (42) */
1010         } /* for (sending_sockets) */
1011 } /* void network_send_buffer */
1012
1013 static int add_to_buffer (char *buffer, int buffer_size,
1014                 value_list_t *vl_def, char *type_def,
1015                 const data_set_t *ds, const value_list_t *vl)
1016 {
1017         char *buffer_orig = buffer;
1018
1019         if (strcmp (vl_def->host, vl->host) != 0)
1020         {
1021                 if (write_part_string (&buffer, &buffer_size, TYPE_HOST,
1022                                         vl->host, strlen (vl->host)) != 0)
1023                         return (-1);
1024                 strcpy (vl_def->host, vl->host);
1025                 DEBUG ("network plugin: add_to_buffer: host = %s", vl->host);
1026         }
1027
1028         if (vl_def->time != vl->time)
1029         {
1030                 if (write_part_number (&buffer, &buffer_size, TYPE_TIME,
1031                                         (uint64_t) vl->time))
1032                         return (-1);
1033                 vl_def->time = vl->time;
1034                 DEBUG ("network plugin: add_to_buffer: time = %u",
1035                                 (unsigned int) vl->time);
1036         }
1037
1038         if (strcmp (vl_def->plugin, vl->plugin) != 0)
1039         {
1040                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN,
1041                                         vl->plugin, strlen (vl->plugin)) != 0)
1042                         return (-1);
1043                 strcpy (vl_def->plugin, vl->plugin);
1044                 DEBUG ("network plugin: add_to_buffer: plugin = %s",
1045                                 vl->plugin);
1046         }
1047
1048         if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0)
1049         {
1050                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN_INSTANCE,
1051                                         vl->plugin_instance,
1052                                         strlen (vl->plugin_instance)) != 0)
1053                         return (-1);
1054                 strcpy (vl_def->plugin_instance, vl->plugin_instance);
1055                 DEBUG ("network plugin: add_to_buffer: plugin_instance = %s",
1056                                 vl->plugin_instance);
1057         }
1058
1059         if (strcmp (type_def, ds->type) != 0)
1060         {
1061                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE,
1062                                         ds->type, strlen (ds->type)) != 0)
1063                         return (-1);
1064                 strcpy (type_def, ds->type);
1065                 DEBUG ("network plugin: add_to_buffer: type = %s", ds->type);
1066         }
1067
1068         if (strcmp (vl_def->type_instance, vl->type_instance) != 0)
1069         {
1070                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE_INSTANCE,
1071                                         vl->type_instance,
1072                                         strlen (vl->type_instance)) != 0)
1073                         return (-1);
1074                 strcpy (vl_def->type_instance, vl->type_instance);
1075                 DEBUG ("network plugin: add_to_buffer: type_instance = %s",
1076                                 vl->type_instance);
1077         }
1078
1079         if (write_part_values (&buffer, &buffer_size, ds, vl) != 0)
1080                 return (-1);
1081
1082         return (buffer - buffer_orig);
1083 } /* int add_to_buffer */
1084
1085 static void flush_buffer (void)
1086 {
1087         DEBUG ("network plugin: flush_buffer: send_buffer_fill = %i",
1088                         send_buffer_fill);
1089
1090         network_send_buffer (send_buffer, send_buffer_fill);
1091         send_buffer_ptr  = send_buffer;
1092         send_buffer_fill = 0;
1093         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1094         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1095 }
1096
1097 static int network_write (const data_set_t *ds, const value_list_t *vl)
1098 {
1099         int status;
1100
1101         /* If the value is already in the cache, we have received it via the
1102          * network. We write it again if forwarding is activated. It's then in
1103          * the cache and should we receive it again we will ignore it. */
1104         status = cache_check (ds->type, vl);
1105         if ((network_config_forward == 0)
1106                         && (status != 0))
1107                 return (0);
1108
1109         pthread_mutex_lock (&send_buffer_lock);
1110
1111         status = add_to_buffer (send_buffer_ptr,
1112                         sizeof (send_buffer) - send_buffer_fill,
1113                         &send_buffer_vl, send_buffer_type,
1114                         ds, vl);
1115         if (status >= 0)
1116         {
1117                 /* status == bytes added to the buffer */
1118                 send_buffer_fill += status;
1119                 send_buffer_ptr  += status;
1120         }
1121         else
1122         {
1123                 flush_buffer ();
1124
1125                 status = add_to_buffer (send_buffer_ptr,
1126                                 sizeof (send_buffer) - send_buffer_fill,
1127                                 &send_buffer_vl, send_buffer_type,
1128                                 ds, vl);
1129
1130                 if (status >= 0)
1131                 {
1132                         send_buffer_fill += status;
1133                         send_buffer_ptr  += status;
1134                 }
1135         }
1136
1137         if (status < 0)
1138         {
1139                 ERROR ("network plugin: Unable to append to the "
1140                                 "buffer for some weird reason");
1141         }
1142         else if ((sizeof (send_buffer) - send_buffer_fill) < 15)
1143         {
1144                 flush_buffer ();
1145         }
1146
1147         pthread_mutex_unlock (&send_buffer_lock);
1148
1149         return ((status < 0) ? -1 : 0);
1150 } /* int network_write */
1151
1152 static int network_config (const char *key, const char *val)
1153 {
1154         char *node;
1155         char *service;
1156
1157         char *fields[3];
1158         int   fields_num;
1159
1160         if ((strcasecmp ("Listen", key) == 0)
1161                         || (strcasecmp ("Server", key) == 0))
1162         {
1163                 char *val_cpy = strdup (val);
1164                 if (val_cpy == NULL)
1165                         return (1);
1166
1167                 service = NET_DEFAULT_PORT;
1168                 fields_num = strsplit (val_cpy, fields, 3);
1169                 if ((fields_num != 1)
1170                                 && (fields_num != 2))
1171                         return (1);
1172                 else if (fields_num == 2)
1173                         service = fields[1];
1174                 node = fields[0];
1175
1176                 if (strcasecmp ("Listen", key) == 0)
1177                         network_add_listen_socket (node, service);
1178                 else
1179                         network_add_sending_socket (node, service);
1180         }
1181         else if (strcasecmp ("TimeToLive", key) == 0)
1182         {
1183                 int tmp = atoi (val);
1184                 if ((tmp > 0) && (tmp < 256))
1185                         network_config_ttl = tmp;
1186                 else
1187                         return (1);
1188         }
1189         else if (strcasecmp ("Forward", key) == 0)
1190         {
1191                 if ((strcasecmp ("true", val) == 0)
1192                                 || (strcasecmp ("yes", val) == 0)
1193                                 || (strcasecmp ("on", val) == 0))
1194                         network_config_forward = 1;
1195                 else
1196                         network_config_forward = 0;
1197         }
1198         else
1199         {
1200                 return (-1);
1201         }
1202         return (0);
1203 }
1204
1205 static int network_shutdown (void)
1206 {
1207         DEBUG ("Shutting down.");
1208
1209         listen_loop++;
1210
1211         if (listen_thread != (pthread_t) 0)
1212         {
1213                 pthread_kill (listen_thread, SIGTERM);
1214                 pthread_join (listen_thread, NULL /* no return value */);
1215                 listen_thread = (pthread_t) 0;
1216         }
1217
1218         listen_thread = 0;
1219
1220         if (cache_tree != NULL)
1221         {
1222                 void *key;
1223                 void *value;
1224
1225                 while (avl_pick (cache_tree, &key, &value) == 0)
1226                 {
1227                         sfree (key);
1228                         sfree (value);
1229                 }
1230                 avl_destroy (cache_tree);
1231                 cache_tree = NULL;
1232         }
1233
1234         /* TODO: Close `sending_sockets' */
1235
1236         plugin_unregister_config ("network");
1237         plugin_unregister_init ("network");
1238         plugin_unregister_write ("network");
1239         plugin_unregister_shutdown ("network");
1240
1241         return (0);
1242 } /* int network_shutdown */
1243
1244 static int network_init (void)
1245 {
1246         plugin_register_shutdown ("network", network_shutdown);
1247
1248         send_buffer_ptr  = send_buffer;
1249         send_buffer_fill = 0;
1250         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1251         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1252
1253         cache_tree = avl_create ((int (*) (const void *, const void *)) strcmp);
1254
1255         /* setup socket(s) and so on */
1256         if (sending_sockets != NULL)
1257                 plugin_register_write ("network", network_write);
1258
1259         if ((listen_sockets_num != 0) && (listen_thread == 0))
1260         {
1261                 int status;
1262
1263                 status = pthread_create (&listen_thread, NULL /* no attributes */,
1264                                 receive_thread, NULL /* no argument */);
1265
1266                 if (status != 0)
1267                 {
1268                         char errbuf[1024];
1269                         ERROR ("network: pthread_create failed: %s",
1270                                         sstrerror (errno, errbuf,
1271                                                 sizeof (errbuf)));
1272                 }
1273         }
1274         return (0);
1275 } /* int network_init */
1276
1277 void module_register (void)
1278 {
1279         plugin_register_config ("network", network_config,
1280                         config_keys, config_keys_num);
1281         plugin_register_init   ("network", network_init);
1282 }