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 ("buffer = %p; buffer_len = %i;", buffer, buffer_len);
 448
 449         memset (&vl, '\0', sizeof (vl));
 450         memset (&type, '\0', sizeof (type));
 451         status = 0;
 452
 453         while ((status == 0) && (buffer_len > sizeof (part_header_t)))
 454         {
 455                 header = (part_header_t *) buffer;
 456
 457                 if (ntohs (header->length) > buffer_len)
 458                         break;
 459                 /* Assure that this loop terminates eventually */
 460                 if (ntohs (header->length) < 4)
 461                         break;
 462
 463                 if (ntohs (header->type) == TYPE_VALUES)
 464                 {
 465                         status = parse_part_values (&buffer, &buffer_len,
 466                                         &vl.values, &vl.values_len);
 467
 468                         if (status != 0)
 469                         {
 470                                 DEBUG ("parse_part_values failed.");
 471                                 break;
 472                         }
 473
 474                         if ((vl.time > 0)
 475                                         && (strlen (vl.host) > 0)
 476                                         && (strlen (vl.plugin) > 0)
 477                                         && (strlen (type) > 0)
 478                                         && (cache_check (type, &vl) == 0))
 479                         {
 480                                 DEBUG ("dispatching values");
 481                                 plugin_dispatch_values (type, &vl);
 482                         }
 483                         else
 484                         {
 485                                 DEBUG ("NOT dispatching values");
 486                         }
 487                 }
 488                 else if (ntohs (header->type) == TYPE_TIME)
 489                 {
 490                         uint64_t tmp = 0;
 491                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 492                         if (status == 0)
 493                                 vl.time = (time_t) tmp;
 494                 }
 495                 else if (ntohs (header->type) == TYPE_HOST)
 496                 {
 497                         status = parse_part_string (&buffer, &buffer_len,
 498                                         vl.host, sizeof (vl.host));
 499                         DEBUG ("network plugin: parse_packet: vl.host = %s", vl.host);
 500                 }
 501                 else if (ntohs (header->type) == TYPE_PLUGIN)
 502                 {
 503                         status = parse_part_string (&buffer, &buffer_len,
 504                                         vl.plugin, sizeof (vl.plugin));
 505                         DEBUG ("network plugin: parse_packet: vl.plugin = %s", vl.plugin);
 506                 }
 507                 else if (ntohs (header->type) == TYPE_PLUGIN_INSTANCE)
 508                 {
 509                         status = parse_part_string (&buffer, &buffer_len,
 510                                         vl.plugin_instance, sizeof (vl.plugin_instance));
 511                         DEBUG ("network plugin: parse_packet: vl.plugin_instance = %s", vl.plugin_instance);
 512                 }
 513                 else if (ntohs (header->type) == TYPE_TYPE)
 514                 {
 515                         status = parse_part_string (&buffer, &buffer_len,
 516                                         type, sizeof (type));
 517                         DEBUG ("network plugin: parse_packet: type = %s", type);
 518                 }
 519                 else if (ntohs (header->type) == TYPE_TYPE_INSTANCE)
 520                 {
 521                         status = parse_part_string (&buffer, &buffer_len,
 522                                         vl.type_instance, sizeof (vl.type_instance));
 523                         DEBUG ("network plugin: parse_packet: vl.type_instance = %s", vl.type_instance);
 524                 }
 525                 else
 526                 {
 527                         DEBUG ("Unknown part type: 0x%0hx", ntohs (header->type));
 528                         buffer = ((char *) buffer) + ntohs (header->length);
 529                 }
 530         } /* while (buffer_len > sizeof (part_header_t)) */
 531
 532         return (0);
 533 } /* int parse_packet */
 534
 535 static void free_sockent (sockent_t *se)
 536 {
 537         sockent_t *next;
 538         while (se != NULL)
 539         {
 540                 next = se->next;
 541                 free (se->addr);
 542                 free (se);
 543                 se = next;
 544         }
 545 } /* void free_sockent */
 546
 547 /*
 548  * int network_set_ttl
 549  *
 550  * Set the `IP_MULTICAST_TTL', `IP_TTL', `IPV6_MULTICAST_HOPS' or
 551  * `IPV6_UNICAST_HOPS', depending on which option is applicable.
 552  *
 553  * The `struct addrinfo' is used to destinguish between unicast and multicast
 554  * sockets.
 555  */
 556 static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 557 {
 558         if ((network_config_ttl < 1) || (network_config_ttl > 255))
 559                 return (-1);
 560
 561         DEBUG ("ttl = %i", network_config_ttl);
 562
 563         if (ai->ai_family == AF_INET)
 564         {
 565                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 566                 int optname;
 567
 568                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 569                         optname = IP_MULTICAST_TTL;
 570                 else
 571                         optname = IP_TTL;
 572
 573                 if (setsockopt (se->fd, IPPROTO_IP, optname,
 574                                         &network_config_ttl,
 575                                         sizeof (network_config_ttl)) == -1)
 576                 {
 577                         char errbuf[1024];
 578                         ERROR ("setsockopt: %s",
 579                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 580                         return (-1);
 581                 }
 582         }
 583         else if (ai->ai_family == AF_INET6)
 584         {
 585                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 586                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 587                 int optname;
 588
 589                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 590                         optname = IPV6_MULTICAST_HOPS;
 591                 else
 592                         optname = IPV6_UNICAST_HOPS;
 593
 594                 if (setsockopt (se->fd, IPPROTO_IPV6, optname,
 595                                         &network_config_ttl,
 596                                         sizeof (network_config_ttl)) == -1)
 597                 {
 598                         char errbuf[1024];
 599                         ERROR ("setsockopt: %s",
 600                                         sstrerror (errno, errbuf,
 601                                                 sizeof (errbuf)));
 602                         return (-1);
 603                 }
 604         }
 605
 606         return (0);
 607 } /* int network_set_ttl */
 608
 609 static int network_bind_socket (const sockent_t *se, const struct addrinfo *ai)
 610 {
 611         int loop = 1;
 612
 613         DEBUG ("fd = %i; calling `bind'", se->fd);
 614
 615         if (bind (se->fd, ai->ai_addr, ai->ai_addrlen) == -1)
 616         {
 617                 char errbuf[1024];
 618                 ERROR ("bind: %s",
 619                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 620                 return (-1);
 621         }
 622
 623         if (ai->ai_family == AF_INET)
 624         {
 625                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 626                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 627                 {
 628                         struct ip_mreq mreq;
 629
 630                         DEBUG ("fd = %i; IPv4 multicast address found", se->fd);
 631
 632                         mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
 633                         mreq.imr_interface.s_addr = htonl (INADDR_ANY);
 634
 635                         if (setsockopt (se->fd, IPPROTO_IP, IP_MULTICAST_LOOP,
 636                                                 &loop, sizeof (loop)) == -1)
 637                         {
 638                                 char errbuf[1024];
 639                                 ERROR ("setsockopt: %s",
 640                                                 sstrerror (errno, errbuf,
 641                                                         sizeof (errbuf)));
 642                                 return (-1);
 643                         }
 644
 645                         if (setsockopt (se->fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 646                                                 &mreq, sizeof (mreq)) == -1)
 647                         {
 648                                 char errbuf[1024];
 649                                 ERROR ("setsockopt: %s",
 650                                                 sstrerror (errno, errbuf,
 651                                                         sizeof (errbuf)));
 652                                 return (-1);
 653                         }
 654                 }
 655         }
 656         else if (ai->ai_family == AF_INET6)
 657         {
 658                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 659                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 660                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 661                 {
 662                         struct ipv6_mreq mreq;
 663
 664                         DEBUG ("fd = %i; IPv6 multicast address found", se->fd);
 665
 666                         memcpy (&mreq.ipv6mr_multiaddr,
 667                                         &addr->sin6_addr,
 668                                         sizeof (addr->sin6_addr));
 669
 670                         /* http://developer.apple.com/documentation/Darwin/Reference/ManPages/man4/ip6.4.html
 671                          * ipv6mr_interface may be set to zeroes to
 672                          * choose the default multicast interface or to
 673                          * the index of a particular multicast-capable
 674                          * interface if the host is multihomed.
 675                          * Membership is associ-associated with a
 676                          * single interface; programs running on
 677                          * multihomed hosts may need to join the same
 678                          * group on more than one interface.*/
 679                         mreq.ipv6mr_interface = 0;
 680
 681                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
 682                                                 &loop, sizeof (loop)) == -1)
 683                         {
 684                                 char errbuf[1024];
 685                                 ERROR ("setsockopt: %s",
 686                                                 sstrerror (errno, errbuf,
 687                                                         sizeof (errbuf)));
 688                                 return (-1);
 689                         }
 690
 691                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP,
 692                                                 &mreq, sizeof (mreq)) == -1)
 693                         {
 694                                 char errbuf[1024];
 695                                 ERROR ("setsockopt: %s",
 696                                                 sstrerror (errno, errbuf,
 697                                                         sizeof (errbuf)));
 698                                 return (-1);
 699                         }
 700                 }
 701         }
 702
 703         return (0);
 704 } /* int network_bind_socket */
 705
 706 static sockent_t *network_create_socket (const char *node,
 707                 const char *service,
 708                 int listen)
 709 {
 710         struct addrinfo  ai_hints;
 711         struct addrinfo *ai_list, *ai_ptr;
 712         int              ai_return;
 713
 714         sockent_t *se_head = NULL;
 715         sockent_t *se_tail = NULL;
 716
 717         DEBUG ("node = %s, service = %s", node, service);
 718
 719         memset (&ai_hints, '\0', sizeof (ai_hints));
 720         ai_hints.ai_flags    = 0;
 721 #ifdef AI_PASSIVE
 722         ai_hints.ai_flags |= AI_PASSIVE;
 723 #endif
 724 #ifdef AI_ADDRCONFIG
 725         ai_hints.ai_flags |= AI_ADDRCONFIG;
 726 #endif
 727         ai_hints.ai_family   = AF_UNSPEC;
 728         ai_hints.ai_socktype = SOCK_DGRAM;
 729         ai_hints.ai_protocol = IPPROTO_UDP;
 730
 731         ai_return = getaddrinfo (node, service, &ai_hints, &ai_list);
 732         if (ai_return != 0)
 733         {
 734                 char errbuf[1024];
 735                 ERROR ("getaddrinfo (%s, %s): %s",
 736                                 (node == NULL) ? "(null)" : node,
 737                                 (service == NULL) ? "(null)" : service,
 738                                 (ai_return == EAI_SYSTEM)
 739                                 ? sstrerror (errno, errbuf, sizeof (errbuf))
 740                                 : gai_strerror (ai_return));
 741                 return (NULL);
 742         }
 743
 744         for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 745         {
 746                 sockent_t *se;
 747
 748                 if ((se = (sockent_t *) malloc (sizeof (sockent_t))) == NULL)
 749                 {
 750                         char errbuf[1024];
 751                         ERROR ("malloc: %s",
 752                                         sstrerror (errno, errbuf,
 753                                                 sizeof (errbuf)));
 754                         continue;
 755                 }
 756
 757                 if ((se->addr = (struct sockaddr_storage *) malloc (sizeof (struct sockaddr_storage))) == NULL)
 758                 {
 759                         char errbuf[1024];
 760                         ERROR ("malloc: %s",
 761                                         sstrerror (errno, errbuf,
 762                                                 sizeof (errbuf)));
 763                         free (se);
 764                         continue;
 765                 }
 766
 767                 assert (sizeof (struct sockaddr_storage) >= ai_ptr->ai_addrlen);
 768                 memset (se->addr, '\0', sizeof (struct sockaddr_storage));
 769                 memcpy (se->addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 770                 se->addrlen = ai_ptr->ai_addrlen;
 771
 772                 se->fd   = socket (ai_ptr->ai_family,
 773                                 ai_ptr->ai_socktype,
 774                                 ai_ptr->ai_protocol);
 775                 se->next = NULL;
 776
 777                 if (se->fd == -1)
 778                 {
 779                         char errbuf[1024];
 780                         ERROR ("socket: %s",
 781                                         sstrerror (errno, errbuf,
 782                                                 sizeof (errbuf)));
 783                         free (se->addr);
 784                         free (se);
 785                         continue;
 786                 }
 787
 788                 if (listen != 0)
 789                 {
 790                         if (network_bind_socket (se, ai_ptr) != 0)
 791                         {
 792                                 close (se->fd);
 793                                 free (se->addr);
 794                                 free (se);
 795                                 continue;
 796                         }
 797                 }
 798                 else /* listen == 0 */
 799                 {
 800                         network_set_ttl (se, ai_ptr);
 801                 }
 802
 803                 if (se_tail == NULL)
 804                 {
 805                         se_head = se;
 806                         se_tail = se;
 807                 }
 808                 else
 809                 {
 810                         se_tail->next = se;
 811                         se_tail = se;
 812                 }
 813
 814                 /* We don't open more than one write-socket per node/service pair.. */
 815                 if (listen == 0)
 816                         break;
 817         }
 818
 819         freeaddrinfo (ai_list);
 820
 821         return (se_head);
 822 } /* sockent_t *network_create_socket */
 823
 824 static sockent_t *network_create_default_socket (int listen)
 825 {
 826         sockent_t *se_ptr  = NULL;
 827         sockent_t *se_head = NULL;
 828         sockent_t *se_tail = NULL;
 829
 830         se_ptr = network_create_socket (NET_DEFAULT_V6_ADDR,
 831                         NET_DEFAULT_PORT, listen);
 832
 833         /* Don't send to the same machine in IPv6 and IPv4 if both are available. */
 834         if ((listen == 0) && (se_ptr != NULL))
 835                 return (se_ptr);
 836
 837         if (se_ptr != NULL)
 838         {
 839                 se_head = se_ptr;
 840                 se_tail = se_ptr;
 841                 while (se_tail->next != NULL)
 842                         se_tail = se_tail->next;
 843         }
 844
 845         se_ptr = network_create_socket (NET_DEFAULT_V4_ADDR, NET_DEFAULT_PORT, listen);
 846
 847         if (se_tail == NULL)
 848                 return (se_ptr);
 849
 850         se_tail->next = se_ptr;
 851         return (se_head);
 852 } /* sockent_t *network_create_default_socket */
 853
 854 static int network_add_listen_socket (const char *node, const char *service)
 855 {
 856         sockent_t *se;
 857         sockent_t *se_ptr;
 858         int se_num = 0;
 859
 860         if (service == NULL)
 861                 service = NET_DEFAULT_PORT;
 862
 863         if (node == NULL)
 864                 se = network_create_default_socket (1 /* listen == true */);
 865         else
 866                 se = network_create_socket (node, service, 1 /* listen == true */);
 867
 868         if (se == NULL)
 869                 return (-1);
 870
 871         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
 872                 se_num++;
 873
 874         listen_sockets = (struct pollfd *) realloc (listen_sockets,
 875                         (listen_sockets_num + se_num)
 876                         * sizeof (struct pollfd));
 877
 878         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
 879         {
 880                 listen_sockets[listen_sockets_num].fd = se_ptr->fd;
 881                 listen_sockets[listen_sockets_num].events = POLLIN | POLLPRI;
 882                 listen_sockets[listen_sockets_num].revents = 0;
 883                 listen_sockets_num++;
 884         } /* for (se) */
 885
 886         free_sockent (se);
 887         return (0);
 888 } /* int network_add_listen_socket */
 889
 890 static int network_add_sending_socket (const char *node, const char *service)
 891 {
 892         sockent_t *se;
 893         sockent_t *se_ptr;
 894
 895         if (service == NULL)
 896                 service = NET_DEFAULT_PORT;
 897
 898         if (node == NULL)
 899                 se = network_create_default_socket (0 /* listen == false */);
 900         else
 901                 se = network_create_socket (node, service, 0 /* listen == false */);
 902
 903         if (se == NULL)
 904                 return (-1);
 905
 906         if (sending_sockets == NULL)
 907         {
 908                 sending_sockets = se;
 909                 return (0);
 910         }
 911
 912         for (se_ptr = sending_sockets; se_ptr->next != NULL; se_ptr = se_ptr->next)
 913                 /* seek end */;
 914
 915         se_ptr->next = se;
 916         return (0);
 917 } /* int network_get_listen_socket */
 918
 919 int network_receive (void)
 920 {
 921         char buffer[BUFF_SIZE];
 922         int  buffer_len;
 923
 924         int i;
 925         int status;
 926
 927         if (listen_sockets_num == 0)
 928                 network_add_listen_socket (NULL, NULL);
 929
 930         if (listen_sockets_num == 0)
 931         {
 932                 ERROR ("network: Failed to open a listening socket.");
 933                 return (-1);
 934         }
 935
 936         while (listen_loop == 0)
 937         {
 938                 status = poll (listen_sockets, listen_sockets_num, -1);
 939
 940                 if (status <= 0)
 941                 {
 942                         char errbuf[1024];
 943                         if (errno == EINTR)
 944                                 continue;
 945                         ERROR ("poll failed: %s",
 946                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 947                         return (-1);
 948                 }
 949
 950                 for (i = 0; (i < listen_sockets_num) && (status > 0); i++)
 951                 {
 952                         if ((listen_sockets[i].revents & (POLLIN | POLLPRI)) == 0)
 953                                 continue;
 954                         status--;
 955
 956                         buffer_len = recv (listen_sockets[i].fd,
 957                                         buffer, sizeof (buffer),
 958                                         0 /* no flags */);
 959                         if (buffer_len < 0)
 960                         {
 961                                 char errbuf[1024];
 962                                 ERROR ("recv failed: %s",
 963                                                 sstrerror (errno, errbuf,
 964                                                         sizeof (errbuf)));
 965                                 return (-1);
 966                         }
 967
 968                         parse_packet (buffer, buffer_len);
 969                 } /* for (listen_sockets) */
 970         } /* while (listen_loop == 0) */
 971
 972         return (0);
 973 }
 974
 975 static void *receive_thread (void *arg)
 976 {
 977         return (network_receive () ? (void *) 1 : (void *) 0);
 978 } /* void *receive_thread */
 979
 980 static void network_send_buffer (const char *buffer, int buffer_len)
 981 {
 982         sockent_t *se;
 983         int status;
 984
 985         DEBUG ("buffer_len = %i", buffer_len);
 986
 987         for (se = sending_sockets; se != NULL; se = se->next)
 988         {
 989                 while (42)
 990                 {
 991                         status = sendto (se->fd, buffer, buffer_len, 0 /* no flags */,
 992                                         (struct sockaddr *) se->addr, se->addrlen);
 993                         if (status < 0)
 994                         {
 995                                 char errbuf[1024];
 996                                 if (errno == EINTR)
 997                                         continue;
 998                                 ERROR ("network plugin: sendto failed: %s",
 999                                                 sstrerror (errno, errbuf,
1000                                                         sizeof (errbuf)));
1001                                 break;
1002                         }
1003
1004                         break;
1005                 } /* while (42) */
1006         } /* for (sending_sockets) */
1007 } /* void network_send_buffer */
1008
1009 static int add_to_buffer (char *buffer, int buffer_size,
1010                 value_list_t *vl_def, char *type_def,
1011                 const data_set_t *ds, const value_list_t *vl)
1012 {
1013         char *buffer_orig = buffer;
1014
1015         if (strcmp (vl_def->host, vl->host) != 0)
1016         {
1017                 if (write_part_string (&buffer, &buffer_size, TYPE_HOST,
1018                                         vl->host, strlen (vl->host)) != 0)
1019                         return (-1);
1020                 strcpy (vl_def->host, vl->host);
1021                 DEBUG ("network plugin: add_to_buffer: host = %s", vl->host);
1022         }
1023
1024         if (vl_def->time != vl->time)
1025         {
1026                 if (write_part_number (&buffer, &buffer_size, TYPE_TIME,
1027                                         (uint64_t) vl->time))
1028                         return (-1);
1029                 vl_def->time = vl->time;
1030                 DEBUG ("network plugin: add_to_buffer: time = %u",
1031                                 (unsigned int) vl->time);
1032         }
1033
1034         if (strcmp (vl_def->plugin, vl->plugin) != 0)
1035         {
1036                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN,
1037                                         vl->plugin, strlen (vl->plugin)) != 0)
1038                         return (-1);
1039                 strcpy (vl_def->plugin, vl->plugin);
1040                 DEBUG ("network plugin: add_to_buffer: plugin = %s",
1041                                 vl->plugin);
1042         }
1043
1044         if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0)
1045         {
1046                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN_INSTANCE,
1047                                         vl->plugin_instance,
1048                                         strlen (vl->plugin_instance)) != 0)
1049                         return (-1);
1050                 strcpy (vl_def->plugin_instance, vl->plugin_instance);
1051                 DEBUG ("network plugin: add_to_buffer: plugin_instance = %s",
1052                                 vl->plugin_instance);
1053         }
1054
1055         if (strcmp (type_def, ds->type) != 0)
1056         {
1057                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE,
1058                                         ds->type, strlen (ds->type)) != 0)
1059                         return (-1);
1060                 strcpy (type_def, ds->type);
1061                 DEBUG ("network plugin: add_to_buffer: type = %s", ds->type);
1062         }
1063
1064         if (strcmp (vl_def->type_instance, vl->type_instance) != 0)
1065         {
1066                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE_INSTANCE,
1067                                         vl->type_instance,
1068                                         strlen (vl->type_instance)) != 0)
1069                         return (-1);
1070                 strcpy (vl_def->type_instance, vl->type_instance);
1071                 DEBUG ("network plugin: add_to_buffer: type_instance = %s",
1072                                 vl->type_instance);
1073         }
1074
1075         if (write_part_values (&buffer, &buffer_size, ds, vl) != 0)
1076                 return (-1);
1077
1078         return (buffer - buffer_orig);
1079 } /* int add_to_buffer */
1080
1081 static void flush_buffer (void)
1082 {
1083         network_send_buffer (send_buffer, send_buffer_fill);
1084         send_buffer_ptr  = send_buffer;
1085         send_buffer_fill = 0;
1086         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1087         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1088 }
1089
1090 static int network_write (const data_set_t *ds, const value_list_t *vl)
1091 {
1092         int status;
1093
1094         /* If the value is already in the cache, we have received it via the
1095          * network. We write it again if forwarding is activated. It's then in
1096          * the cache and should we receive it again we will ignore it. */
1097         status = cache_check (ds->type, vl);
1098         if ((network_config_forward == 0)
1099                         && (status != 0))
1100                 return (0);
1101
1102         pthread_mutex_lock (&send_buffer_lock);
1103
1104         status = add_to_buffer (send_buffer_ptr,
1105                         sizeof (send_buffer) - send_buffer_fill,
1106                         &send_buffer_vl, send_buffer_type,
1107                         ds, vl);
1108         if (status >= 0)
1109         {
1110                 /* status == bytes added to the buffer */
1111                 send_buffer_fill += status;
1112                 send_buffer_ptr  += status;
1113         }
1114         else
1115         {
1116                 flush_buffer ();
1117
1118                 status = add_to_buffer (send_buffer_ptr,
1119                                 sizeof (send_buffer) - send_buffer_fill,
1120                                 &send_buffer_vl, send_buffer_type,
1121                                 ds, vl);
1122
1123                 if (status >= 0)
1124                 {
1125                         send_buffer_fill += status;
1126                         send_buffer_ptr  += status;
1127                 }
1128         }
1129
1130         if (status < 0)
1131         {
1132                 ERROR ("network plugin: Unable to append to the "
1133                                 "buffer for some weird reason");
1134         }
1135         else if ((sizeof (send_buffer) - send_buffer_fill) < 15)
1136         {
1137                 flush_buffer ();
1138         }
1139
1140         pthread_mutex_unlock (&send_buffer_lock);
1141
1142         return ((status < 0) ? -1 : 0);
1143 } /* int network_write */
1144
1145 static int network_config (const char *key, const char *val)
1146 {
1147         char *node;
1148         char *service;
1149
1150         char *fields[3];
1151         int   fields_num;
1152
1153         if ((strcasecmp ("Listen", key) == 0)
1154                         || (strcasecmp ("Server", key) == 0))
1155         {
1156                 char *val_cpy = strdup (val);
1157                 if (val_cpy == NULL)
1158                         return (1);
1159
1160                 service = NET_DEFAULT_PORT;
1161                 fields_num = strsplit (val_cpy, fields, 3);
1162                 if ((fields_num != 1)
1163                                 && (fields_num != 2))
1164                         return (1);
1165                 else if (fields_num == 2)
1166                         service = fields[1];
1167                 node = fields[0];
1168
1169                 if (strcasecmp ("Listen", key) == 0)
1170                         network_add_listen_socket (node, service);
1171                 else
1172                         network_add_sending_socket (node, service);
1173         }
1174         else if (strcasecmp ("TimeToLive", key) == 0)
1175         {
1176                 int tmp = atoi (val);
1177                 if ((tmp > 0) && (tmp < 256))
1178                         network_config_ttl = tmp;
1179                 else
1180                         return (1);
1181         }
1182         else if (strcasecmp ("Forward", key) == 0)
1183         {
1184                 if ((strcasecmp ("true", value) == 0)
1185                                 || (strcasecmp ("yes", value) == 0)
1186                                 || (strcasecmp ("on", value) == 0))
1187                         network_config_forward = 1;
1188                 else
1189                         network_config_forward = 0;
1190         }
1191         else
1192         {
1193                 return (-1);
1194         }
1195         return (0);
1196 }
1197
1198 static int network_shutdown (void)
1199 {
1200         DEBUG ("Shutting down.");
1201
1202         listen_loop++;
1203
1204         if (listen_thread != (pthread_t) 0)
1205         {
1206                 pthread_kill (listen_thread, SIGTERM);
1207                 pthread_join (listen_thread, NULL /* no return value */);
1208                 listen_thread = (pthread_t) 0;
1209         }
1210
1211         listen_thread = 0;
1212
1213         if (cache_tree != NULL)
1214         {
1215                 void *key;
1216                 void *value;
1217
1218                 while (avl_pick (cache_tree, &key, &value) == 0)
1219                 {
1220                         sfree (key);
1221                         sfree (value);
1222                 }
1223                 avl_destroy (cache_tree);
1224                 cache_tree = NULL;
1225         }
1226
1227         /* TODO: Close `sending_sockets' */
1228
1229         plugin_unregister_config ("network");
1230         plugin_unregister_init ("network");
1231         plugin_unregister_write ("network");
1232         plugin_unregister_shutdown ("network");
1233
1234         return (0);
1235 } /* int network_shutdown */
1236
1237 static int network_init (void)
1238 {
1239         plugin_register_shutdown ("network", network_shutdown);
1240
1241         send_buffer_ptr  = send_buffer;
1242         send_buffer_fill = 0;
1243         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1244         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1245
1246         cache_tree = avl_create ((int (*) (const void *, const void *)) strcmp);
1247
1248         /* setup socket(s) and so on */
1249         if (sending_sockets != NULL)
1250                 plugin_register_write ("network", network_write);
1251
1252         if ((listen_sockets_num != 0) && (listen_thread == 0))
1253         {
1254                 int status;
1255
1256                 status = pthread_create (&listen_thread, NULL /* no attributes */,
1257                                 receive_thread, NULL /* no argument */);
1258
1259                 if (status != 0)
1260                 {
1261                         char errbuf[1024];
1262                         ERROR ("network: pthread_create failed: %s",
1263                                         sstrerror (errno, errbuf,
1264                                                 sizeof (errbuf)));
1265                 }
1266         }
1267         return (0);
1268 } /* int network_init */
1269
1270 void module_register (void)
1271 {
1272         plugin_register_config ("network", network_config,
1273                         config_keys, config_keys_num);
1274         plugin_register_init   ("network", network_init);
1275 }