src/network.c

   1 /**
   2  * collectd - src/network.c
   3  * Copyright (C) 2005-2008  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 struct receive_list_entry_s
 140 {
 141   char data[BUFF_SIZE];
 142   int  data_len;
 143   struct receive_list_entry_s *next;
 144 };
 145 typedef struct receive_list_entry_s receive_list_entry_t;
 146
 147 /*
 148  * Private variables
 149  */
 150 static const char *config_keys[] =
 151 {
 152         "CacheFlush",
 153         "Listen",
 154         "Server",
 155         "TimeToLive",
 156         "Forward"
 157 };
 158 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
 159
 160 static int network_config_ttl = 0;
 161 static int network_config_forward = 0;
 162
 163 static sockent_t *sending_sockets = NULL;
 164
 165 static receive_list_entry_t *receive_list_head = NULL;
 166 static receive_list_entry_t *receive_list_tail = NULL;
 167 static pthread_mutex_t       receive_list_lock = PTHREAD_MUTEX_INITIALIZER;
 168 static pthread_cond_t        receive_list_cond = PTHREAD_COND_INITIALIZER;
 169
 170 static struct pollfd *listen_sockets = NULL;
 171 static int listen_sockets_num = 0;
 172
 173 static int listen_loop = 0;
 174 static pthread_t receive_thread_id = 0;
 175 static pthread_t dispatch_thread_id = 0;
 176
 177 static char         send_buffer[BUFF_SIZE];
 178 static char        *send_buffer_ptr;
 179 static int          send_buffer_fill;
 180 static value_list_t send_buffer_vl = VALUE_LIST_STATIC;
 181 static char         send_buffer_type[DATA_MAX_NAME_LEN];
 182 static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER;
 183
 184 static c_avl_tree_t      *cache_tree = NULL;
 185 static pthread_mutex_t  cache_lock = PTHREAD_MUTEX_INITIALIZER;
 186 static time_t           cache_flush_last;
 187 static int              cache_flush_interval = 1800;
 188
 189 /*
 190  * Private functions
 191  */
 192 static int cache_flush (void)
 193 {
 194         char **keys = NULL;
 195         int    keys_num = 0;
 196
 197         char **tmp;
 198         int    i;
 199
 200         char   *key;
 201         time_t *value;
 202         c_avl_iterator_t *iter;
 203
 204         time_t curtime = time (NULL);
 205
 206         iter = c_avl_get_iterator (cache_tree);
 207         while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0)
 208         {
 209                 if ((curtime - *value) <= cache_flush_interval)
 210                         continue;
 211                 tmp = (char **) realloc (keys,
 212                                 (keys_num + 1) * sizeof (char *));
 213                 if (tmp == NULL)
 214                 {
 215                         sfree (keys);
 216                         c_avl_iterator_destroy (iter);
 217                         ERROR ("network plugin: cache_flush: realloc"
 218                                         " failed.");
 219                         return (-1);
 220                 }
 221                 keys = tmp;
 222                 keys[keys_num] = key;
 223                 keys_num++;
 224         } /* while (c_avl_iterator_next) */
 225         c_avl_iterator_destroy (iter);
 226
 227         for (i = 0; i < keys_num; i++)
 228         {
 229                 if (c_avl_remove (cache_tree, keys[i], (void *) &key,
 230                                         (void *) &value) != 0)
 231                 {
 232                         WARNING ("network plugin: cache_flush: c_avl_remove"
 233                                         " (%s) failed.", keys[i]);
 234                         continue;
 235                 }
 236
 237                 sfree (key);
 238                 sfree (value);
 239         }
 240
 241         sfree (keys);
 242
 243         DEBUG ("network plugin: cache_flush: Removed %i %s",
 244                         keys_num, (keys_num == 1) ? "entry" : "entries");
 245         cache_flush_last = curtime;
 246         return (0);
 247 } /* int cache_flush */
 248
 249 static int cache_check (const char *type, const value_list_t *vl)
 250 {
 251         char key[1024];
 252         time_t *value = NULL;
 253         int retval = -1;
 254
 255         if (cache_tree == NULL)
 256                 return (-1);
 257
 258         if (format_name (key, sizeof (key), vl->host, vl->plugin,
 259                                 vl->plugin_instance, type, vl->type_instance))
 260                 return (-1);
 261
 262         pthread_mutex_lock (&cache_lock);
 263
 264         if (c_avl_get (cache_tree, key, (void *) &value) == 0)
 265         {
 266                 if (*value < vl->time)
 267                 {
 268                         *value = vl->time;
 269                         retval = 0;
 270                 }
 271                 else
 272                 {
 273                         DEBUG ("network plugin: cache_check: *value = %i >= vl->time = %i",
 274                                         (int) *value, (int) vl->time);
 275                         retval = 1;
 276                 }
 277         }
 278         else
 279         {
 280                 char *key_copy = strdup (key);
 281                 value = malloc (sizeof (time_t));
 282                 if ((key_copy != NULL) && (value != NULL))
 283                 {
 284                         *value = vl->time;
 285                         c_avl_insert (cache_tree, key_copy, value);
 286                         retval = 0;
 287                 }
 288                 else
 289                 {
 290                         sfree (key_copy);
 291                         sfree (value);
 292                 }
 293         }
 294
 295         if ((time (NULL) - cache_flush_last) > cache_flush_interval)
 296                 cache_flush ();
 297
 298         pthread_mutex_unlock (&cache_lock);
 299
 300         DEBUG ("network plugin: cache_check: key = %s; time = %i; retval = %i",
 301                         key, (int) vl->time, retval);
 302
 303         return (retval);
 304 } /* int cache_check */
 305
 306 static int write_part_values (char **ret_buffer, int *ret_buffer_len,
 307                 const data_set_t *ds, const value_list_t *vl)
 308 {
 309         char *packet_ptr;
 310         int packet_len;
 311         int num_values;
 312
 313         part_header_t pkg_ph;
 314         uint16_t      pkg_num_values;
 315         uint8_t      *pkg_values_types;
 316         value_t      *pkg_values;
 317
 318         int offset;
 319         int i;
 320
 321         num_values = vl->values_len;
 322         packet_len = sizeof (part_header_t) + sizeof (uint16_t)
 323                 + (num_values * sizeof (uint8_t))
 324                 + (num_values * sizeof (value_t));
 325
 326         if (*ret_buffer_len < packet_len)
 327                 return (-1);
 328
 329         pkg_values_types = (uint8_t *) malloc (num_values * sizeof (uint8_t));
 330         if (pkg_values_types == NULL)
 331         {
 332                 ERROR ("network plugin: write_part_values: malloc failed.");
 333                 return (-1);
 334         }
 335
 336         pkg_values = (value_t *) malloc (num_values * sizeof (value_t));
 337         if (pkg_values == NULL)
 338         {
 339                 free (pkg_values_types);
 340                 ERROR ("network plugin: write_part_values: malloc failed.");
 341                 return (-1);
 342         }
 343
 344         pkg_ph.type = htons (TYPE_VALUES);
 345         pkg_ph.length = htons (packet_len);
 346
 347         pkg_num_values = htons ((uint16_t) vl->values_len);
 348
 349         for (i = 0; i < num_values; i++)
 350         {
 351                 if (ds->ds[i].type == DS_TYPE_COUNTER)
 352                 {
 353                         pkg_values_types[i] = DS_TYPE_COUNTER;
 354                         pkg_values[i].counter = htonll (vl->values[i].counter);
 355                 }
 356                 else
 357                 {
 358                         pkg_values_types[i] = DS_TYPE_GAUGE;
 359                         pkg_values[i].gauge = htond (vl->values[i].gauge);
 360                 }
 361         }
 362
 363         /*
 364          * Use `memcpy' to write everything to the buffer, because the pointer
 365          * may be unaligned and some architectures, such as SPARC, can't handle
 366          * that.
 367          */
 368         packet_ptr = *ret_buffer;
 369         offset = 0;
 370         memcpy (packet_ptr + offset, &pkg_ph, sizeof (pkg_ph));
 371         offset += sizeof (pkg_ph);
 372         memcpy (packet_ptr + offset, &pkg_num_values, sizeof (pkg_num_values));
 373         offset += sizeof (pkg_num_values);
 374         memcpy (packet_ptr + offset, pkg_values_types, num_values * sizeof (uint8_t));
 375         offset += num_values * sizeof (uint8_t);
 376         memcpy (packet_ptr + offset, pkg_values, num_values * sizeof (value_t));
 377         offset += num_values * sizeof (value_t);
 378
 379         assert (offset == packet_len);
 380
 381         *ret_buffer = packet_ptr + packet_len;
 382         *ret_buffer_len -= packet_len;
 383
 384         free (pkg_values_types);
 385         free (pkg_values);
 386
 387         return (0);
 388 } /* int write_part_values */
 389
 390 static int write_part_number (char **ret_buffer, int *ret_buffer_len,
 391                 int type, uint64_t value)
 392 {
 393         char *packet_ptr;
 394         int packet_len;
 395
 396         part_header_t pkg_head;
 397         uint64_t pkg_value;
 398
 399         int offset;
 400
 401         packet_len = sizeof (pkg_head) + sizeof (pkg_value);
 402
 403         if (*ret_buffer_len < packet_len)
 404                 return (-1);
 405
 406         pkg_head.type = htons (type);
 407         pkg_head.length = htons (packet_len);
 408         pkg_value = htonll (value);
 409
 410         packet_ptr = *ret_buffer;
 411         offset = 0;
 412         memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head));
 413         offset += sizeof (pkg_head);
 414         memcpy (packet_ptr + offset, &pkg_value, sizeof (pkg_value));
 415         offset += sizeof (pkg_value);
 416
 417         assert (offset == packet_len);
 418
 419         *ret_buffer = packet_ptr + packet_len;
 420         *ret_buffer_len -= packet_len;
 421
 422         return (0);
 423 } /* int write_part_number */
 424
 425 static int write_part_string (char **ret_buffer, int *ret_buffer_len,
 426                 int type, const char *str, int str_len)
 427 {
 428         char *packet_ptr;
 429         int packet_len;
 430
 431         part_header_t pkg_head;
 432
 433         int offset;
 434
 435         packet_len = sizeof (pkg_head) + str_len + 1;
 436         if (*ret_buffer_len < packet_len)
 437                 return (-1);
 438
 439         pkg_head.type = htons (type);
 440         pkg_head.length = htons (packet_len);
 441
 442         packet_ptr = *ret_buffer;
 443         offset = 0;
 444         memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head));
 445         offset += sizeof (pkg_head);
 446         memcpy (packet_ptr + offset, str, str_len);
 447         offset += str_len;
 448         memset (packet_ptr + offset, '\0', 1);
 449         offset += 1;
 450
 451         assert (offset == packet_len);
 452
 453         *ret_buffer = packet_ptr + packet_len;
 454         *ret_buffer_len -= packet_len;
 455
 456         return (0);
 457 } /* int write_part_string */
 458
 459 static int parse_part_values (void **ret_buffer, int *ret_buffer_len,
 460                 value_t **ret_values, int *ret_num_values)
 461 {
 462         char *buffer = *ret_buffer;
 463         int   buffer_len = *ret_buffer_len;
 464         part_values_t pv;
 465         int   i;
 466
 467         uint16_t h_length;
 468         uint16_t h_type;
 469         uint16_t h_num;
 470
 471         if (buffer_len < (15))
 472         {
 473                 DEBUG ("network plugin: packet is too short: buffer_len = %i",
 474                                 buffer_len);
 475                 return (-1);
 476         }
 477
 478         pv.head = (part_header_t *) buffer;
 479         h_length = ntohs (pv.head->length);
 480         h_type = ntohs (pv.head->type);
 481
 482         assert (h_type == TYPE_VALUES);
 483
 484         pv.num_values = (uint16_t *) (pv.head + 1);
 485         h_num = ntohs (*pv.num_values);
 486
 487         if (h_num != ((h_length - 6) / 9))
 488         {
 489                 DEBUG ("`length' and `num of values' don't match");
 490                 return (-1);
 491         }
 492
 493         pv.values_types = (uint8_t *) (pv.num_values + 1);
 494         pv.values = (value_t *) (pv.values_types + h_num);
 495
 496         for (i = 0; i < h_num; i++)
 497                 if (pv.values_types[i] == DS_TYPE_COUNTER)
 498                         pv.values[i].counter = ntohll (pv.values[i].counter);
 499                 else
 500                         pv.values[i].gauge = ntohd (pv.values[i].gauge);
 501
 502         *ret_buffer     = (void *) (pv.values + h_num);
 503         *ret_buffer_len = buffer_len - h_length;
 504         *ret_num_values = h_num;
 505         *ret_values     = pv.values;
 506
 507         return (0);
 508 } /* int parse_part_values */
 509
 510 static int parse_part_number (void **ret_buffer, int *ret_buffer_len,
 511                 uint64_t *value)
 512 {
 513         part_number_t pn;
 514         uint16_t len;
 515
 516         pn.head = (part_header_t *) *ret_buffer;
 517         pn.value = (uint64_t *) (pn.head + 1);
 518
 519         len = ntohs (pn.head->length);
 520         if (len != 12)
 521                 return (-1);
 522         if (len > *ret_buffer_len)
 523                 return (-1);
 524         *value = ntohll (*pn.value);
 525
 526         *ret_buffer = (void *) (pn.value + 1);
 527         *ret_buffer_len -= len;
 528
 529         return (0);
 530 } /* int parse_part_number */
 531
 532 static int parse_part_string (void **ret_buffer, int *ret_buffer_len,
 533                 char *output, int output_len)
 534 {
 535         char *buffer = *ret_buffer;
 536         int   buffer_len = *ret_buffer_len;
 537         part_string_t ps;
 538
 539         uint16_t h_length;
 540         uint16_t h_type;
 541
 542         DEBUG ("network plugin: parse_part_string: ret_buffer = %p;"
 543                         " ret_buffer_len = %i; output = %p; output_len = %i;",
 544                         *ret_buffer, *ret_buffer_len,
 545                         (void *) output, output_len);
 546
 547         ps.head = (part_header_t *) buffer;
 548
 549         h_length = ntohs (ps.head->length);
 550         h_type = ntohs (ps.head->type);
 551
 552         DEBUG ("network plugin: parse_part_string: length = %hu; type = %hu;",
 553                         h_length, h_type);
 554
 555         if (buffer_len < h_length)
 556         {
 557                 DEBUG ("packet is too short");
 558                 return (-1);
 559         }
 560         assert ((h_type == TYPE_HOST)
 561                         || (h_type == TYPE_PLUGIN)
 562                         || (h_type == TYPE_PLUGIN_INSTANCE)
 563                         || (h_type == TYPE_TYPE)
 564                         || (h_type == TYPE_TYPE_INSTANCE)
 565                         || (h_type == TYPE_MESSAGE));
 566
 567         ps.value = buffer + 4;
 568         if (ps.value[h_length - 5] != '\0')
 569         {
 570                 DEBUG ("String does not end with a nullbyte");
 571                 return (-1);
 572         }
 573
 574         if (output_len < (h_length - 4))
 575         {
 576                 DEBUG ("output buffer is too small");
 577                 return (-1);
 578         }
 579         strcpy (output, ps.value);
 580
 581         DEBUG ("network plugin: parse_part_string: output = %s", output);
 582
 583         *ret_buffer = (void *) (buffer + h_length);
 584         *ret_buffer_len = buffer_len - h_length;
 585
 586         return (0);
 587 } /* int parse_part_string */
 588
 589 static int parse_packet (void *buffer, int buffer_len)
 590 {
 591         part_header_t *header;
 592         int status;
 593
 594         value_list_t vl = VALUE_LIST_INIT;
 595         char type[DATA_MAX_NAME_LEN];
 596         notification_t n;
 597
 598         DEBUG ("network plugin: parse_packet: buffer = %p; buffer_len = %i;",
 599                         buffer, buffer_len);
 600
 601         memset (&vl, '\0', sizeof (vl));
 602         memset (&type, '\0', sizeof (type));
 603         memset (&n, '\0', sizeof (n));
 604         status = 0;
 605
 606         while ((status == 0) && (0 < buffer_len)
 607                         && ((unsigned int) buffer_len > sizeof (part_header_t)))
 608         {
 609                 header = (part_header_t *) buffer;
 610
 611                 if (ntohs (header->length) > buffer_len)
 612                         break;
 613                 /* Assure that this loop terminates eventually */
 614                 if (ntohs (header->length) < 4)
 615                         break;
 616
 617                 if (ntohs (header->type) == TYPE_VALUES)
 618                 {
 619                         status = parse_part_values (&buffer, &buffer_len,
 620                                         &vl.values, &vl.values_len);
 621
 622                         if (status != 0)
 623                         {
 624                                 DEBUG ("parse_part_values failed.");
 625                                 break;
 626                         }
 627
 628                         if ((vl.time > 0)
 629                                         && (strlen (vl.host) > 0)
 630                                         && (strlen (vl.plugin) > 0)
 631                                         && (strlen (type) > 0)
 632                                         && (cache_check (type, &vl) == 0))
 633                         {
 634                                 DEBUG ("network plugin: parse_packet:"
 635                                                 " dispatching values");
 636                                 plugin_dispatch_values (type, &vl);
 637                         }
 638                         else
 639                         {
 640                                 DEBUG ("network plugin: parse_packet:"
 641                                                 " NOT dispatching values");
 642                         }
 643                 }
 644                 else if (ntohs (header->type) == TYPE_TIME)
 645                 {
 646                         uint64_t tmp = 0;
 647                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 648                         if (status == 0)
 649                         {
 650                                 vl.time = (time_t) tmp;
 651                                 n.time = (time_t) tmp;
 652                         }
 653                 }
 654                 else if (ntohs (header->type) == TYPE_INTERVAL)
 655                 {
 656                         uint64_t tmp = 0;
 657                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 658                         if (status == 0)
 659                                 vl.interval = (int) tmp;
 660                 }
 661                 else if (ntohs (header->type) == TYPE_HOST)
 662                 {
 663                         status = parse_part_string (&buffer, &buffer_len,
 664                                         vl.host, sizeof (vl.host));
 665                         strncpy (n.host, vl.host, sizeof (n.host));
 666                         n.host[sizeof (n.host) - 1] = '\0';
 667                         DEBUG ("network plugin: parse_packet: vl.host = %s",
 668                                         vl.host);
 669                 }
 670                 else if (ntohs (header->type) == TYPE_PLUGIN)
 671                 {
 672                         status = parse_part_string (&buffer, &buffer_len,
 673                                         vl.plugin, sizeof (vl.plugin));
 674                         strncpy (n.plugin, vl.plugin, sizeof (n.plugin));
 675                         n.plugin[sizeof (n.plugin) - 1] = '\0';
 676                         DEBUG ("network plugin: parse_packet: vl.plugin = %s",
 677                                         vl.plugin);
 678                 }
 679                 else if (ntohs (header->type) == TYPE_PLUGIN_INSTANCE)
 680                 {
 681                         status = parse_part_string (&buffer, &buffer_len,
 682                                         vl.plugin_instance,
 683                                         sizeof (vl.plugin_instance));
 684                         strncpy (n.plugin_instance, vl.plugin_instance,
 685                                         sizeof (n.plugin_instance));
 686                         n.plugin_instance[sizeof (n.plugin_instance) - 1] = '\0';
 687                         DEBUG ("network plugin: parse_packet: "
 688                                         "vl.plugin_instance = %s",
 689                                         vl.plugin_instance);
 690                 }
 691                 else if (ntohs (header->type) == TYPE_TYPE)
 692                 {
 693                         status = parse_part_string (&buffer, &buffer_len,
 694                                         type, sizeof (type));
 695                         strncpy (n.type, type, sizeof (n.type));
 696                         n.type[sizeof (n.type) - 1] = '\0';
 697                         DEBUG ("network plugin: parse_packet: type = %s",
 698                                         type);
 699                 }
 700                 else if (ntohs (header->type) == TYPE_TYPE_INSTANCE)
 701                 {
 702                         status = parse_part_string (&buffer, &buffer_len,
 703                                         vl.type_instance,
 704                                         sizeof (vl.type_instance));
 705                         strncpy (n.type_instance, vl.type_instance,
 706                                         sizeof (n.type_instance));
 707                         n.type_instance[sizeof (n.type_instance) - 1] = '\0';
 708                         DEBUG ("network plugin: parse_packet: "
 709                                         "vl.type_instance = %s",
 710                                         vl.type_instance);
 711                 }
 712                 else if (ntohs (header->type) == TYPE_MESSAGE)
 713                 {
 714                         status = parse_part_string (&buffer, &buffer_len,
 715                                         n.message, sizeof (n.message));
 716                         DEBUG ("network plugin: parse_packet: n.message = %s",
 717                                         n.message);
 718
 719                         if ((n.severity != NOTIF_FAILURE)
 720                                         && (n.severity != NOTIF_WARNING)
 721                                         && (n.severity != NOTIF_OKAY))
 722                         {
 723                                 INFO ("network plugin: "
 724                                                 "Ignoring notification with "
 725                                                 "unknown severity %s.",
 726                                                 n.severity);
 727                         }
 728                         else if (n.time <= 0)
 729                         {
 730                                 INFO ("network plugin: "
 731                                                 "Ignoring notification with "
 732                                                 "time == 0.");
 733                         }
 734                         else if (strlen (n.message) <= 0)
 735                         {
 736                                 INFO ("network plugin: "
 737                                                 "Ignoring notification with "
 738                                                 "an empty message.");
 739                         }
 740                         else
 741                         {
 742                                 /*
 743                                  * TODO: Let this do a separate thread so that
 744                                  * no packets are lost if this takes too long.
 745                                  */
 746                                 plugin_dispatch_notification (&n);
 747                         }
 748                 }
 749                 else if (ntohs (header->type) == TYPE_SEVERITY)
 750                 {
 751                         uint64_t tmp = 0;
 752                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 753                         if (status == 0)
 754                                 n.severity = (int) tmp;
 755                 }
 756                 else
 757                 {
 758                         DEBUG ("network plugin: parse_packet: Unknown part"
 759                                         " type: 0x%0hx", ntohs (header->type));
 760                         buffer = ((char *) buffer) + ntohs (header->length);
 761                 }
 762         } /* while (buffer_len > sizeof (part_header_t)) */
 763
 764         return (0);
 765 } /* int parse_packet */
 766
 767 static void free_sockent (sockent_t *se)
 768 {
 769         sockent_t *next;
 770         while (se != NULL)
 771         {
 772                 next = se->next;
 773                 free (se->addr);
 774                 free (se);
 775                 se = next;
 776         }
 777 } /* void free_sockent */
 778
 779 /*
 780  * int network_set_ttl
 781  *
 782  * Set the `IP_MULTICAST_TTL', `IP_TTL', `IPV6_MULTICAST_HOPS' or
 783  * `IPV6_UNICAST_HOPS', depending on which option is applicable.
 784  *
 785  * The `struct addrinfo' is used to destinguish between unicast and multicast
 786  * sockets.
 787  */
 788 static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 789 {
 790         if ((network_config_ttl < 1) || (network_config_ttl > 255))
 791                 return (-1);
 792
 793         DEBUG ("ttl = %i", network_config_ttl);
 794
 795         if (ai->ai_family == AF_INET)
 796         {
 797                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 798                 int optname;
 799
 800                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 801                         optname = IP_MULTICAST_TTL;
 802                 else
 803                         optname = IP_TTL;
 804
 805                 if (setsockopt (se->fd, IPPROTO_IP, optname,
 806                                         &network_config_ttl,
 807                                         sizeof (network_config_ttl)) == -1)
 808                 {
 809                         char errbuf[1024];
 810                         ERROR ("setsockopt: %s",
 811                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 812                         return (-1);
 813                 }
 814         }
 815         else if (ai->ai_family == AF_INET6)
 816         {
 817                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 818                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 819                 int optname;
 820
 821                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 822                         optname = IPV6_MULTICAST_HOPS;
 823                 else
 824                         optname = IPV6_UNICAST_HOPS;
 825
 826                 if (setsockopt (se->fd, IPPROTO_IPV6, optname,
 827                                         &network_config_ttl,
 828                                         sizeof (network_config_ttl)) == -1)
 829                 {
 830                         char errbuf[1024];
 831                         ERROR ("setsockopt: %s",
 832                                         sstrerror (errno, errbuf,
 833                                                 sizeof (errbuf)));
 834                         return (-1);
 835                 }
 836         }
 837
 838         return (0);
 839 } /* int network_set_ttl */
 840
 841 static int network_bind_socket (const sockent_t *se, const struct addrinfo *ai)
 842 {
 843         int loop = 0;
 844         int yes  = 1;
 845
 846         /* allow multiple sockets to use the same PORT number */
 847         if (setsockopt(se->fd, SOL_SOCKET, SO_REUSEADDR,
 848                                 &yes, sizeof(yes)) == -1) {
 849                 char errbuf[1024];
 850                 ERROR ("setsockopt: %s",
 851                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 852                 return (-1);
 853         }
 854
 855         DEBUG ("fd = %i; calling `bind'", se->fd);
 856
 857         if (bind (se->fd, ai->ai_addr, ai->ai_addrlen) == -1)
 858         {
 859                 char errbuf[1024];
 860                 ERROR ("bind: %s",
 861                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 862                 return (-1);
 863         }
 864
 865         if (ai->ai_family == AF_INET)
 866         {
 867                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 868                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 869                 {
 870                         struct ip_mreq mreq;
 871
 872                         DEBUG ("fd = %i; IPv4 multicast address found", se->fd);
 873
 874                         mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
 875                         mreq.imr_interface.s_addr = htonl (INADDR_ANY);
 876
 877                         if (setsockopt (se->fd, IPPROTO_IP, IP_MULTICAST_LOOP,
 878                                                 &loop, sizeof (loop)) == -1)
 879                         {
 880                                 char errbuf[1024];
 881                                 ERROR ("setsockopt: %s",
 882                                                 sstrerror (errno, errbuf,
 883                                                         sizeof (errbuf)));
 884                                 return (-1);
 885                         }
 886
 887                         if (setsockopt (se->fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 888                                                 &mreq, sizeof (mreq)) == -1)
 889                         {
 890                                 char errbuf[1024];
 891                                 ERROR ("setsockopt: %s",
 892                                                 sstrerror (errno, errbuf,
 893                                                         sizeof (errbuf)));
 894                                 return (-1);
 895                         }
 896                 }
 897         }
 898         else if (ai->ai_family == AF_INET6)
 899         {
 900                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 901                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 902                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 903                 {
 904                         struct ipv6_mreq mreq;
 905
 906                         DEBUG ("fd = %i; IPv6 multicast address found", se->fd);
 907
 908                         memcpy (&mreq.ipv6mr_multiaddr,
 909                                         &addr->sin6_addr,
 910                                         sizeof (addr->sin6_addr));
 911
 912                         /* http://developer.apple.com/documentation/Darwin/Reference/ManPages/man4/ip6.4.html
 913                          * ipv6mr_interface may be set to zeroes to
 914                          * choose the default multicast interface or to
 915                          * the index of a particular multicast-capable
 916                          * interface if the host is multihomed.
 917                          * Membership is associ-associated with a
 918                          * single interface; programs running on
 919                          * multihomed hosts may need to join the same
 920                          * group on more than one interface.*/
 921                         mreq.ipv6mr_interface = 0;
 922
 923                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
 924                                                 &loop, sizeof (loop)) == -1)
 925                         {
 926                                 char errbuf[1024];
 927                                 ERROR ("setsockopt: %s",
 928                                                 sstrerror (errno, errbuf,
 929                                                         sizeof (errbuf)));
 930                                 return (-1);
 931                         }
 932
 933                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP,
 934                                                 &mreq, sizeof (mreq)) == -1)
 935                         {
 936                                 char errbuf[1024];
 937                                 ERROR ("setsockopt: %s",
 938                                                 sstrerror (errno, errbuf,
 939                                                         sizeof (errbuf)));
 940                                 return (-1);
 941                         }
 942                 }
 943         }
 944
 945         return (0);
 946 } /* int network_bind_socket */
 947
 948 static sockent_t *network_create_socket (const char *node,
 949                 const char *service,
 950                 int listen)
 951 {
 952         struct addrinfo  ai_hints;
 953         struct addrinfo *ai_list, *ai_ptr;
 954         int              ai_return;
 955
 956         sockent_t *se_head = NULL;
 957         sockent_t *se_tail = NULL;
 958
 959         DEBUG ("node = %s, service = %s", node, service);
 960
 961         memset (&ai_hints, '\0', sizeof (ai_hints));
 962         ai_hints.ai_flags    = 0;
 963 #ifdef AI_PASSIVE
 964         ai_hints.ai_flags |= AI_PASSIVE;
 965 #endif
 966 #ifdef AI_ADDRCONFIG
 967         ai_hints.ai_flags |= AI_ADDRCONFIG;
 968 #endif
 969         ai_hints.ai_family   = AF_UNSPEC;
 970         ai_hints.ai_socktype = SOCK_DGRAM;
 971         ai_hints.ai_protocol = IPPROTO_UDP;
 972
 973         ai_return = getaddrinfo (node, service, &ai_hints, &ai_list);
 974         if (ai_return != 0)
 975         {
 976                 char errbuf[1024];
 977                 ERROR ("getaddrinfo (%s, %s): %s",
 978                                 (node == NULL) ? "(null)" : node,
 979                                 (service == NULL) ? "(null)" : service,
 980                                 (ai_return == EAI_SYSTEM)
 981                                 ? sstrerror (errno, errbuf, sizeof (errbuf))
 982                                 : gai_strerror (ai_return));
 983                 return (NULL);
 984         }
 985
 986         for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 987         {
 988                 sockent_t *se;
 989
 990                 if ((se = (sockent_t *) malloc (sizeof (sockent_t))) == NULL)
 991                 {
 992                         char errbuf[1024];
 993                         ERROR ("malloc: %s",
 994                                         sstrerror (errno, errbuf,
 995                                                 sizeof (errbuf)));
 996                         continue;
 997                 }
 998
 999                 if ((se->addr = (struct sockaddr_storage *) malloc (sizeof (struct sockaddr_storage))) == NULL)
1000                 {
1001                         char errbuf[1024];
1002                         ERROR ("malloc: %s",
1003                                         sstrerror (errno, errbuf,
1004                                                 sizeof (errbuf)));
1005                         free (se);
1006                         continue;
1007                 }
1008
1009                 assert (sizeof (struct sockaddr_storage) >= ai_ptr->ai_addrlen);
1010                 memset (se->addr, '\0', sizeof (struct sockaddr_storage));
1011                 memcpy (se->addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
1012                 se->addrlen = ai_ptr->ai_addrlen;
1013
1014                 se->fd   = socket (ai_ptr->ai_family,
1015                                 ai_ptr->ai_socktype,
1016                                 ai_ptr->ai_protocol);
1017                 se->next = NULL;
1018
1019                 if (se->fd == -1)
1020                 {
1021                         char errbuf[1024];
1022                         ERROR ("socket: %s",
1023                                         sstrerror (errno, errbuf,
1024                                                 sizeof (errbuf)));
1025                         free (se->addr);
1026                         free (se);
1027                         continue;
1028                 }
1029
1030                 if (listen != 0)
1031                 {
1032                         if (network_bind_socket (se, ai_ptr) != 0)
1033                         {
1034                                 close (se->fd);
1035                                 free (se->addr);
1036                                 free (se);
1037                                 continue;
1038                         }
1039                 }
1040                 else /* listen == 0 */
1041                 {
1042                         network_set_ttl (se, ai_ptr);
1043                 }
1044
1045                 if (se_tail == NULL)
1046                 {
1047                         se_head = se;
1048                         se_tail = se;
1049                 }
1050                 else
1051                 {
1052                         se_tail->next = se;
1053                         se_tail = se;
1054                 }
1055
1056                 /* We don't open more than one write-socket per node/service pair.. */
1057                 if (listen == 0)
1058                         break;
1059         }
1060
1061         freeaddrinfo (ai_list);
1062
1063         return (se_head);
1064 } /* sockent_t *network_create_socket */
1065
1066 static sockent_t *network_create_default_socket (int listen)
1067 {
1068         sockent_t *se_ptr  = NULL;
1069         sockent_t *se_head = NULL;
1070         sockent_t *se_tail = NULL;
1071
1072         se_ptr = network_create_socket (NET_DEFAULT_V6_ADDR,
1073                         NET_DEFAULT_PORT, listen);
1074
1075         /* Don't send to the same machine in IPv6 and IPv4 if both are available. */
1076         if ((listen == 0) && (se_ptr != NULL))
1077                 return (se_ptr);
1078
1079         if (se_ptr != NULL)
1080         {
1081                 se_head = se_ptr;
1082                 se_tail = se_ptr;
1083                 while (se_tail->next != NULL)
1084                         se_tail = se_tail->next;
1085         }
1086
1087         se_ptr = network_create_socket (NET_DEFAULT_V4_ADDR, NET_DEFAULT_PORT, listen);
1088
1089         if (se_tail == NULL)
1090                 return (se_ptr);
1091
1092         se_tail->next = se_ptr;
1093         return (se_head);
1094 } /* sockent_t *network_create_default_socket */
1095
1096 static int network_add_listen_socket (const char *node, const char *service)
1097 {
1098         sockent_t *se;
1099         sockent_t *se_ptr;
1100         int se_num = 0;
1101
1102         if (service == NULL)
1103                 service = NET_DEFAULT_PORT;
1104
1105         if (node == NULL)
1106                 se = network_create_default_socket (1 /* listen == true */);
1107         else
1108                 se = network_create_socket (node, service, 1 /* listen == true */);
1109
1110         if (se == NULL)
1111                 return (-1);
1112
1113         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
1114                 se_num++;
1115
1116         listen_sockets = (struct pollfd *) realloc (listen_sockets,
1117                         (listen_sockets_num + se_num)
1118                         * sizeof (struct pollfd));
1119
1120         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
1121         {
1122                 listen_sockets[listen_sockets_num].fd = se_ptr->fd;
1123                 listen_sockets[listen_sockets_num].events = POLLIN | POLLPRI;
1124                 listen_sockets[listen_sockets_num].revents = 0;
1125                 listen_sockets_num++;
1126         } /* for (se) */
1127
1128         free_sockent (se);
1129         return (0);
1130 } /* int network_add_listen_socket */
1131
1132 static int network_add_sending_socket (const char *node, const char *service)
1133 {
1134         sockent_t *se;
1135         sockent_t *se_ptr;
1136
1137         if (service == NULL)
1138                 service = NET_DEFAULT_PORT;
1139
1140         if (node == NULL)
1141                 se = network_create_default_socket (0 /* listen == false */);
1142         else
1143                 se = network_create_socket (node, service, 0 /* listen == false */);
1144
1145         if (se == NULL)
1146                 return (-1);
1147
1148         if (sending_sockets == NULL)
1149         {
1150                 sending_sockets = se;
1151                 return (0);
1152         }
1153
1154         for (se_ptr = sending_sockets; se_ptr->next != NULL; se_ptr = se_ptr->next)
1155                 /* seek end */;
1156
1157         se_ptr->next = se;
1158         return (0);
1159 } /* int network_get_listen_socket */
1160
1161 static void *dispatch_thread (void *arg)
1162 {
1163   while (42)
1164   {
1165     receive_list_entry_t *ent;
1166
1167     /* Lock and wait for more data to come in */
1168     pthread_mutex_lock (&receive_list_lock);
1169     while ((listen_loop == 0)
1170         && (receive_list_head == NULL))
1171       pthread_cond_wait (&receive_list_cond, &receive_list_lock);
1172
1173     /* Remove the head entry and unlock */
1174     ent = receive_list_head;
1175     if (ent != NULL)
1176       receive_list_head = ent->next;
1177     pthread_mutex_unlock (&receive_list_lock);
1178
1179     /* Check whether we are supposed to exit. We do NOT check `listen_loop'
1180      * because we dispatch all missing packets before shutting down. */
1181     if (ent == NULL)
1182       break;
1183
1184     parse_packet (ent->data, ent->data_len);
1185
1186     sfree (ent);
1187   } /* while (42) */
1188
1189   return (NULL);
1190 } /* void *receive_thread */
1191
1192 static int network_receive (void)
1193 {
1194         char buffer[BUFF_SIZE];
1195         int  buffer_len;
1196
1197         int i;
1198         int status;
1199
1200         if (listen_sockets_num == 0)
1201                 network_add_listen_socket (NULL, NULL);
1202
1203         if (listen_sockets_num == 0)
1204         {
1205                 ERROR ("network: Failed to open a listening socket.");
1206                 return (-1);
1207         }
1208
1209         while (listen_loop == 0)
1210         {
1211                 status = poll (listen_sockets, listen_sockets_num, -1);
1212
1213                 if (status <= 0)
1214                 {
1215                         char errbuf[1024];
1216                         if (errno == EINTR)
1217                                 continue;
1218                         ERROR ("poll failed: %s",
1219                                         sstrerror (errno, errbuf, sizeof (errbuf)));
1220                         return (-1);
1221                 }
1222
1223                 for (i = 0; (i < listen_sockets_num) && (status > 0); i++)
1224                 {
1225                         receive_list_entry_t *ent;
1226
1227                         if ((listen_sockets[i].revents & (POLLIN | POLLPRI)) == 0)
1228                                 continue;
1229                         status--;
1230
1231                         buffer_len = recv (listen_sockets[i].fd,
1232                                         buffer, sizeof (buffer),
1233                                         0 /* no flags */);
1234                         if (buffer_len < 0)
1235                         {
1236                                 char errbuf[1024];
1237                                 ERROR ("recv failed: %s",
1238                                                 sstrerror (errno, errbuf,
1239                                                         sizeof (errbuf)));
1240                                 return (-1);
1241                         }
1242
1243                         ent = malloc (sizeof (receive_list_entry_t));
1244                         if (ent == NULL)
1245                         {
1246                                 ERROR ("network plugin: malloc failed.");
1247                                 return (-1);
1248                         }
1249                         memset (ent, '\0', sizeof (receive_list_entry_t));
1250
1251                         /* Hopefully this be optimized out by the compiler. It
1252                          * might help prevent stupid bugs in the future though.
1253                          */
1254                         assert (sizeof (ent->data) == sizeof (buffer));
1255
1256                         memcpy (ent->data, buffer, buffer_len);
1257                         ent->data_len = buffer_len;
1258
1259                         pthread_mutex_lock (&receive_list_lock);
1260                         if (receive_list_head == NULL)
1261                         {
1262                                 receive_list_head = ent;
1263                                 receive_list_tail = ent;
1264                         }
1265                         else
1266                         {
1267                                 receive_list_tail->next = ent;
1268                                 receive_list_tail = ent;
1269                         }
1270                         pthread_cond_signal (&receive_list_cond);
1271                         pthread_mutex_unlock (&receive_list_lock);
1272                 } /* for (listen_sockets) */
1273         } /* while (listen_loop == 0) */
1274
1275         return (0);
1276 }
1277
1278 static void *receive_thread (void *arg)
1279 {
1280         return (network_receive () ? (void *) 1 : (void *) 0);
1281 } /* void *receive_thread */
1282
1283 static void network_send_buffer (const char *buffer, int buffer_len)
1284 {
1285         sockent_t *se;
1286         int status;
1287
1288         DEBUG ("network plugin: network_send_buffer: buffer_len = %i", buffer_len);
1289
1290         for (se = sending_sockets; se != NULL; se = se->next)
1291         {
1292                 while (42)
1293                 {
1294                         status = sendto (se->fd, buffer, buffer_len, 0 /* no flags */,
1295                                         (struct sockaddr *) se->addr, se->addrlen);
1296                         if (status < 0)
1297                         {
1298                                 char errbuf[1024];
1299                                 if (errno == EINTR)
1300                                         continue;
1301                                 ERROR ("network plugin: sendto failed: %s",
1302                                                 sstrerror (errno, errbuf,
1303                                                         sizeof (errbuf)));
1304                                 break;
1305                         }
1306
1307                         break;
1308                 } /* while (42) */
1309         } /* for (sending_sockets) */
1310 } /* void network_send_buffer */
1311
1312 static int add_to_buffer (char *buffer, int buffer_size,
1313                 value_list_t *vl_def, char *type_def,
1314                 const data_set_t *ds, const value_list_t *vl)
1315 {
1316         char *buffer_orig = buffer;
1317
1318         if (strcmp (vl_def->host, vl->host) != 0)
1319         {
1320                 if (write_part_string (&buffer, &buffer_size, TYPE_HOST,
1321                                         vl->host, strlen (vl->host)) != 0)
1322                         return (-1);
1323                 strcpy (vl_def->host, vl->host);
1324         }
1325
1326         if (vl_def->time != vl->time)
1327         {
1328                 if (write_part_number (&buffer, &buffer_size, TYPE_TIME,
1329                                         (uint64_t) vl->time))
1330                         return (-1);
1331                 vl_def->time = vl->time;
1332         }
1333
1334         if (vl_def->interval != vl->interval)
1335         {
1336                 if (write_part_number (&buffer, &buffer_size, TYPE_INTERVAL,
1337                                         (uint64_t) vl->interval))
1338                         return (-1);
1339                 vl_def->interval = vl->interval;
1340         }
1341
1342         if (strcmp (vl_def->plugin, vl->plugin) != 0)
1343         {
1344                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN,
1345                                         vl->plugin, strlen (vl->plugin)) != 0)
1346                         return (-1);
1347                 strcpy (vl_def->plugin, vl->plugin);
1348         }
1349
1350         if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0)
1351         {
1352                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN_INSTANCE,
1353                                         vl->plugin_instance,
1354                                         strlen (vl->plugin_instance)) != 0)
1355                         return (-1);
1356                 strcpy (vl_def->plugin_instance, vl->plugin_instance);
1357         }
1358
1359         if (strcmp (type_def, ds->type) != 0)
1360         {
1361                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE,
1362                                         ds->type, strlen (ds->type)) != 0)
1363                         return (-1);
1364                 strcpy (type_def, ds->type);
1365         }
1366
1367         if (strcmp (vl_def->type_instance, vl->type_instance) != 0)
1368         {
1369                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE_INSTANCE,
1370                                         vl->type_instance,
1371                                         strlen (vl->type_instance)) != 0)
1372                         return (-1);
1373                 strcpy (vl_def->type_instance, vl->type_instance);
1374         }
1375
1376         if (write_part_values (&buffer, &buffer_size, ds, vl) != 0)
1377                 return (-1);
1378
1379         return (buffer - buffer_orig);
1380 } /* int add_to_buffer */
1381
1382 static void flush_buffer (void)
1383 {
1384         DEBUG ("network plugin: flush_buffer: send_buffer_fill = %i",
1385                         send_buffer_fill);
1386
1387         network_send_buffer (send_buffer, send_buffer_fill);
1388         send_buffer_ptr  = send_buffer;
1389         send_buffer_fill = 0;
1390         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1391         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1392 }
1393
1394 static int network_write (const data_set_t *ds, const value_list_t *vl)
1395 {
1396         int status;
1397
1398         /* If the value is already in the cache, we have received it via the
1399          * network. We write it again if forwarding is activated. It's then in
1400          * the cache and should we receive it again we will ignore it. */
1401         status = cache_check (ds->type, vl);
1402         if ((network_config_forward == 0)
1403                         && (status != 0))
1404                 return (0);
1405
1406         pthread_mutex_lock (&send_buffer_lock);
1407
1408         status = add_to_buffer (send_buffer_ptr,
1409                         sizeof (send_buffer) - send_buffer_fill,
1410                         &send_buffer_vl, send_buffer_type,
1411                         ds, vl);
1412         if (status >= 0)
1413         {
1414                 /* status == bytes added to the buffer */
1415                 send_buffer_fill += status;
1416                 send_buffer_ptr  += status;
1417         }
1418         else
1419         {
1420                 flush_buffer ();
1421
1422                 status = add_to_buffer (send_buffer_ptr,
1423                                 sizeof (send_buffer) - send_buffer_fill,
1424                                 &send_buffer_vl, send_buffer_type,
1425                                 ds, vl);
1426
1427                 if (status >= 0)
1428                 {
1429                         send_buffer_fill += status;
1430                         send_buffer_ptr  += status;
1431                 }
1432         }
1433
1434         if (status < 0)
1435         {
1436                 ERROR ("network plugin: Unable to append to the "
1437                                 "buffer for some weird reason");
1438         }
1439         else if ((sizeof (send_buffer) - send_buffer_fill) < 15)
1440         {
1441                 flush_buffer ();
1442         }
1443
1444         pthread_mutex_unlock (&send_buffer_lock);
1445
1446         return ((status < 0) ? -1 : 0);
1447 } /* int network_write */
1448
1449 static int network_config (const char *key, const char *val)
1450 {
1451         char *node;
1452         char *service;
1453
1454         char *fields[3];
1455         int   fields_num;
1456
1457         if ((strcasecmp ("Listen", key) == 0)
1458                         || (strcasecmp ("Server", key) == 0))
1459         {
1460                 char *val_cpy = strdup (val);
1461                 if (val_cpy == NULL)
1462                         return (1);
1463
1464                 service = NET_DEFAULT_PORT;
1465                 fields_num = strsplit (val_cpy, fields, 3);
1466                 if ((fields_num != 1)
1467                                 && (fields_num != 2))
1468                         return (1);
1469                 else if (fields_num == 2)
1470                 {
1471                         if ((service = strchr (fields[1], '.')) != NULL)
1472                                 *service = '\0';
1473                         service = fields[1];
1474                 }
1475                 node = fields[0];
1476
1477                 if (strcasecmp ("Listen", key) == 0)
1478                         network_add_listen_socket (node, service);
1479                 else
1480                         network_add_sending_socket (node, service);
1481         }
1482         else if (strcasecmp ("TimeToLive", key) == 0)
1483         {
1484                 int tmp = atoi (val);
1485                 if ((tmp > 0) && (tmp < 256))
1486                         network_config_ttl = tmp;
1487                 else
1488                         return (1);
1489         }
1490         else if (strcasecmp ("Forward", key) == 0)
1491         {
1492                 if ((strcasecmp ("true", val) == 0)
1493                                 || (strcasecmp ("yes", val) == 0)
1494                                 || (strcasecmp ("on", val) == 0))
1495                         network_config_forward = 1;
1496                 else
1497                         network_config_forward = 0;
1498         }
1499         else if (strcasecmp ("CacheFlush", key) == 0)
1500         {
1501                 int tmp = atoi (val);
1502                 if (tmp > 0)
1503                         cache_flush_interval = tmp;
1504                 else return (1);
1505         }
1506         else
1507         {
1508                 return (-1);
1509         }
1510         return (0);
1511 } /* int network_config */
1512
1513 static int network_notification (const notification_t *n)
1514 {
1515   char  buffer[BUFF_SIZE];
1516   char *buffer_ptr = buffer;
1517   int   buffer_free = sizeof (buffer);
1518   int   status;
1519
1520   memset (buffer, '\0', sizeof (buffer));
1521
1522
1523   status = write_part_number (&buffer_ptr, &buffer_free, TYPE_TIME,
1524       (uint64_t) n->time);
1525   if (status != 0)
1526     return (-1);
1527
1528   status = write_part_number (&buffer_ptr, &buffer_free, TYPE_SEVERITY,
1529       (uint64_t) n->severity);
1530   if (status != 0)
1531     return (-1);
1532
1533   if (strlen (n->host) > 0)
1534   {
1535     status = write_part_string (&buffer_ptr, &buffer_free, TYPE_HOST,
1536         n->host, strlen (n->host));
1537     if (status != 0)
1538       return (-1);
1539   }
1540
1541   if (strlen (n->plugin) > 0)
1542   {
1543     status = write_part_string (&buffer_ptr, &buffer_free, TYPE_PLUGIN,
1544         n->plugin, strlen (n->plugin));
1545     if (status != 0)
1546       return (-1);
1547   }
1548
1549   if (strlen (n->plugin_instance) > 0)
1550   {
1551     status = write_part_string (&buffer_ptr, &buffer_free,
1552         TYPE_PLUGIN_INSTANCE,
1553         n->plugin_instance, strlen (n->plugin_instance));
1554     if (status != 0)
1555       return (-1);
1556   }
1557
1558   if (strlen (n->type) > 0)
1559   {
1560     status = write_part_string (&buffer_ptr, &buffer_free, TYPE_TYPE,
1561         n->type, strlen (n->type));
1562     if (status != 0)
1563       return (-1);
1564   }
1565
1566   if (strlen (n->type_instance) > 0)
1567   {
1568     status = write_part_string (&buffer_ptr, &buffer_free, TYPE_TYPE_INSTANCE,
1569         n->type_instance, strlen (n->type_instance));
1570     if (status != 0)
1571       return (-1);
1572   }
1573
1574   status = write_part_string (&buffer_ptr, &buffer_free, TYPE_MESSAGE,
1575       n->message, strlen (n->message));
1576   if (status != 0)
1577     return (-1);
1578
1579   network_send_buffer (buffer, sizeof (buffer) - buffer_free);
1580
1581   return (0);
1582 } /* int network_notification */
1583
1584 static int network_shutdown (void)
1585 {
1586         listen_loop++;
1587
1588         /* Kill the listening thread */
1589         if (receive_thread_id != (pthread_t) 0)
1590         {
1591                 pthread_kill (receive_thread_id, SIGTERM);
1592                 pthread_join (receive_thread_id, NULL /* no return value */);
1593                 receive_thread_id = (pthread_t) 0;
1594         }
1595
1596         /* Shutdown the dispatching thread */
1597         if (dispatch_thread_id != (pthread_t) 0)
1598                 pthread_cond_broadcast (&receive_list_cond);
1599
1600         if (send_buffer_fill > 0)
1601                 flush_buffer ();
1602
1603         if (cache_tree != NULL)
1604         {
1605                 void *key;
1606                 void *value;
1607
1608                 while (c_avl_pick (cache_tree, &key, &value) == 0)
1609                 {
1610                         sfree (key);
1611                         sfree (value);
1612                 }
1613                 c_avl_destroy (cache_tree);
1614                 cache_tree = NULL;
1615         }
1616
1617         /* TODO: Close `sending_sockets' */
1618
1619         plugin_unregister_config ("network");
1620         plugin_unregister_init ("network");
1621         plugin_unregister_write ("network");
1622         plugin_unregister_shutdown ("network");
1623
1624         return (0);
1625 } /* int network_shutdown */
1626
1627 static int network_init (void)
1628 {
1629         plugin_register_shutdown ("network", network_shutdown);
1630
1631         send_buffer_ptr  = send_buffer;
1632         send_buffer_fill = 0;
1633         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1634         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1635
1636         cache_tree = c_avl_create ((int (*) (const void *, const void *)) strcmp);
1637         cache_flush_last = time (NULL);
1638
1639         /* setup socket(s) and so on */
1640         if (sending_sockets != NULL)
1641         {
1642                 plugin_register_write ("network", network_write);
1643                 plugin_register_notification ("network", network_notification);
1644         }
1645
1646         if ((listen_sockets_num != 0) && (receive_thread_id == 0))
1647         {
1648                 int status;
1649
1650                 status = pthread_create (&dispatch_thread_id,
1651                                 NULL /* no attributes */,
1652                                 dispatch_thread,
1653                                 NULL /* no argument */);
1654                 if (status != 0)
1655                 {
1656                         char errbuf[1024];
1657                         ERROR ("network: pthread_create failed: %s",
1658                                         sstrerror (errno, errbuf,
1659                                                 sizeof (errbuf)));
1660                 }
1661
1662                 status = pthread_create (&receive_thread_id,
1663                                 NULL /* no attributes */,
1664                                 receive_thread,
1665                                 NULL /* no argument */);
1666                 if (status != 0)
1667                 {
1668                         char errbuf[1024];
1669                         ERROR ("network: pthread_create failed: %s",
1670                                         sstrerror (errno, errbuf,
1671                                                 sizeof (errbuf)));
1672                 }
1673         }
1674         return (0);
1675 } /* int network_init */
1676
1677 static int network_flush (int timeout)
1678 {
1679         pthread_mutex_lock (&send_buffer_lock);
1680
1681         if (((time (NULL) - cache_flush_last) >= timeout)
1682                         && (send_buffer_fill > 0))
1683         {
1684                 flush_buffer ();
1685         }
1686
1687         pthread_mutex_unlock (&send_buffer_lock);
1688
1689         return (0);
1690 } /* int network_flush */
1691
1692 void module_register (void)
1693 {
1694         plugin_register_config ("network", network_config,
1695                         config_keys, config_keys_num);
1696         plugin_register_init   ("network", network_init);
1697         plugin_register_flush   ("network", network_flush);
1698 } /* void module_register */