X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fnetwork.c;h=39069aa751a551eef139b92e9ecd467bb8e5b27a;hb=92445ff3363441d0f515de4a3ab92a504cfc0366;hp=c81d7110664da7d5217d07cc4e3076fc55eec640;hpb=09c6d16ffba1b1b15e50579bbcf663f05f630932;p=collectd.git diff --git a/src/network.c b/src/network.c index c81d7110..39069aa7 100644 --- a/src/network.c +++ b/src/network.c @@ -1,6 +1,6 @@ /** * collectd - src/network.c - * Copyright (C) 2005-2007 Florian octo Forster + * Copyright (C) 2005-2008 Florian octo Forster * * This program is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License as published by the @@ -136,6 +136,14 @@ struct part_values_s }; typedef struct part_values_s part_values_t; +struct receive_list_entry_s +{ + char data[BUFF_SIZE]; + int data_len; + struct receive_list_entry_s *next; +}; +typedef struct receive_list_entry_s receive_list_entry_t; + /* * Private variables */ @@ -154,10 +162,17 @@ static int network_config_forward = 0; static sockent_t *sending_sockets = NULL; +static receive_list_entry_t *receive_list_head = NULL; +static receive_list_entry_t *receive_list_tail = NULL; +static pthread_mutex_t receive_list_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t receive_list_cond = PTHREAD_COND_INITIALIZER; + static struct pollfd *listen_sockets = NULL; static int listen_sockets_num = 0; -static pthread_t listen_thread = 0; + static int listen_loop = 0; +static pthread_t receive_thread_id = 0; +static pthread_t dispatch_thread_id = 0; static char send_buffer[BUFF_SIZE]; static char *send_buffer_ptr; @@ -166,9 +181,9 @@ static value_list_t send_buffer_vl = VALUE_LIST_STATIC; static char send_buffer_type[DATA_MAX_NAME_LEN]; static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER; -static avl_tree_t *cache_tree = NULL; +static c_avl_tree_t *cache_tree = NULL; static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; -static time_t cache_flush_last; +static time_t cache_flush_last = 0; static int cache_flush_interval = 1800; /* @@ -184,12 +199,12 @@ static int cache_flush (void) char *key; time_t *value; - avl_iterator_t *iter; + c_avl_iterator_t *iter; time_t curtime = time (NULL); - iter = avl_get_iterator (cache_tree); - while (avl_iterator_next (iter, (void *) &key, (void *) &value) == 0) + iter = c_avl_get_iterator (cache_tree); + while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0) { if ((curtime - *value) <= cache_flush_interval) continue; @@ -198,7 +213,7 @@ static int cache_flush (void) if (tmp == NULL) { sfree (keys); - avl_iterator_destroy (iter); + c_avl_iterator_destroy (iter); ERROR ("network plugin: cache_flush: realloc" " failed."); return (-1); @@ -206,15 +221,15 @@ static int cache_flush (void) keys = tmp; keys[keys_num] = key; keys_num++; - } /* while (avl_iterator_next) */ - avl_iterator_destroy (iter); + } /* while (c_avl_iterator_next) */ + c_avl_iterator_destroy (iter); for (i = 0; i < keys_num; i++) { - if (avl_remove (cache_tree, keys[i], (void *) &key, + if (c_avl_remove (cache_tree, keys[i], (void *) &key, (void *) &value) != 0) { - WARNING ("network plugin: cache_flush: avl_remove" + WARNING ("network plugin: cache_flush: c_avl_remove" " (%s) failed.", keys[i]); continue; } @@ -231,7 +246,7 @@ static int cache_flush (void) return (0); } /* int cache_flush */ -static int cache_check (const char *type, const value_list_t *vl) +static int cache_check (const value_list_t *vl) { char key[1024]; time_t *value = NULL; @@ -241,12 +256,12 @@ static int cache_check (const char *type, const value_list_t *vl) return (-1); if (format_name (key, sizeof (key), vl->host, vl->plugin, - vl->plugin_instance, type, vl->type_instance)) + vl->plugin_instance, vl->type, vl->type_instance)) return (-1); pthread_mutex_lock (&cache_lock); - if (avl_get (cache_tree, key, (void *) &value) == 0) + if (c_avl_get (cache_tree, key, (void *) &value) == 0) { if (*value < vl->time) { @@ -267,7 +282,7 @@ static int cache_check (const char *type, const value_list_t *vl) if ((key_copy != NULL) && (value != NULL)) { *value = vl->time; - avl_insert (cache_tree, key_copy, value); + c_avl_insert (cache_tree, key_copy, value); retval = 0; } else @@ -291,37 +306,83 @@ static int cache_check (const char *type, const value_list_t *vl) static int write_part_values (char **ret_buffer, int *ret_buffer_len, const data_set_t *ds, const value_list_t *vl) { - part_values_t pv; + char *packet_ptr; + int packet_len; + int num_values; + + part_header_t pkg_ph; + uint16_t pkg_num_values; + uint8_t *pkg_values_types; + value_t *pkg_values; + + int offset; int i; - i = 6 + (9 * vl->values_len); - if (*ret_buffer_len < i) + num_values = vl->values_len; + packet_len = sizeof (part_header_t) + sizeof (uint16_t) + + (num_values * sizeof (uint8_t)) + + (num_values * sizeof (value_t)); + + if (*ret_buffer_len < packet_len) + return (-1); + + pkg_values_types = (uint8_t *) malloc (num_values * sizeof (uint8_t)); + if (pkg_values_types == NULL) + { + ERROR ("network plugin: write_part_values: malloc failed."); return (-1); - *ret_buffer_len -= i; + } - pv.head = (part_header_t *) *ret_buffer; - pv.num_values = (uint16_t *) (pv.head + 1); - pv.values_types = (uint8_t *) (pv.num_values + 1); - pv.values = (value_t *) (pv.values_types + vl->values_len); - *ret_buffer = (void *) (pv.values + vl->values_len); + pkg_values = (value_t *) malloc (num_values * sizeof (value_t)); + if (pkg_values == NULL) + { + free (pkg_values_types); + ERROR ("network plugin: write_part_values: malloc failed."); + return (-1); + } - pv.head->type = htons (TYPE_VALUES); - pv.head->length = htons (6 + (9 * vl->values_len)); - *pv.num_values = htons ((uint16_t) vl->values_len); - - for (i = 0; i < vl->values_len; i++) + pkg_ph.type = htons (TYPE_VALUES); + pkg_ph.length = htons (packet_len); + + pkg_num_values = htons ((uint16_t) vl->values_len); + + for (i = 0; i < num_values; i++) { if (ds->ds[i].type == DS_TYPE_COUNTER) { - pv.values_types[i] = DS_TYPE_COUNTER; - pv.values[i].counter = htonll (vl->values[i].counter); + pkg_values_types[i] = DS_TYPE_COUNTER; + pkg_values[i].counter = htonll (vl->values[i].counter); } else { - pv.values_types[i] = DS_TYPE_GAUGE; - pv.values[i].gauge = vl->values[i].gauge; + pkg_values_types[i] = DS_TYPE_GAUGE; + pkg_values[i].gauge = htond (vl->values[i].gauge); } - } /* for (values) */ + } + + /* + * Use `memcpy' to write everything to the buffer, because the pointer + * may be unaligned and some architectures, such as SPARC, can't handle + * that. + */ + packet_ptr = *ret_buffer; + offset = 0; + memcpy (packet_ptr + offset, &pkg_ph, sizeof (pkg_ph)); + offset += sizeof (pkg_ph); + memcpy (packet_ptr + offset, &pkg_num_values, sizeof (pkg_num_values)); + offset += sizeof (pkg_num_values); + memcpy (packet_ptr + offset, pkg_values_types, num_values * sizeof (uint8_t)); + offset += num_values * sizeof (uint8_t); + memcpy (packet_ptr + offset, pkg_values, num_values * sizeof (value_t)); + offset += num_values * sizeof (value_t); + + assert (offset == packet_len); + + *ret_buffer = packet_ptr + packet_len; + *ret_buffer_len -= packet_len; + + free (pkg_values_types); + free (pkg_values); return (0); } /* int write_part_values */ @@ -329,20 +390,34 @@ static int write_part_values (char **ret_buffer, int *ret_buffer_len, static int write_part_number (char **ret_buffer, int *ret_buffer_len, int type, uint64_t value) { - part_number_t pn; + char *packet_ptr; + int packet_len; + + part_header_t pkg_head; + uint64_t pkg_value; + + int offset; + + packet_len = sizeof (pkg_head) + sizeof (pkg_value); - if (*ret_buffer_len < 12) + if (*ret_buffer_len < packet_len) return (-1); - pn.head = (part_header_t *) *ret_buffer; - pn.value = (uint64_t *) (pn.head + 1); + pkg_head.type = htons (type); + pkg_head.length = htons (packet_len); + pkg_value = htonll (value); - pn.head->type = htons (type); - pn.head->length = htons (12); - *pn.value = htonll (value); + packet_ptr = *ret_buffer; + offset = 0; + memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head)); + offset += sizeof (pkg_head); + memcpy (packet_ptr + offset, &pkg_value, sizeof (pkg_value)); + offset += sizeof (pkg_value); - *ret_buffer = (char *) (pn.value + 1); - *ret_buffer_len -= 12; + assert (offset == packet_len); + + *ret_buffer = packet_ptr + packet_len; + *ret_buffer_len -= packet_len; return (0); } /* int write_part_number */ @@ -350,23 +425,36 @@ static int write_part_number (char **ret_buffer, int *ret_buffer_len, static int write_part_string (char **ret_buffer, int *ret_buffer_len, int type, const char *str, int str_len) { - part_string_t ps; - int len; + char *buffer; + int buffer_len; + + uint16_t pkg_type; + uint16_t pkg_length; - len = 4 + str_len + 1; - if (*ret_buffer_len < len) + int offset; + + buffer_len = 2 * sizeof (uint16_t) + str_len + 1; + if (*ret_buffer_len < buffer_len) return (-1); - *ret_buffer_len -= len; - ps.head = (part_header_t *) *ret_buffer; - ps.value = (char *) (ps.head + 1); + pkg_type = htons (type); + pkg_length = htons (buffer_len); + + buffer = *ret_buffer; + offset = 0; + memcpy (buffer + offset, (void *) &pkg_type, sizeof (pkg_type)); + offset += sizeof (pkg_type); + memcpy (buffer + offset, (void *) &pkg_length, sizeof (pkg_length)); + offset += sizeof (pkg_length); + memcpy (buffer + offset, str, str_len); + offset += str_len; + memset (buffer + offset, '\0', 1); + offset += 1; - ps.head->type = htons ((uint16_t) type); - ps.head->length = htons ((uint16_t) str_len + 5); - if (str_len > 0) - memcpy (ps.value, str, str_len); - ps.value[str_len] = '\0'; - *ret_buffer = (void *) (ps.value + (str_len + 1)); + assert (offset == buffer_len); + + *ret_buffer = buffer + buffer_len; + *ret_buffer_len -= buffer_len; return (0); } /* int write_part_string */ @@ -376,12 +464,17 @@ static int parse_part_values (void **ret_buffer, int *ret_buffer_len, { char *buffer = *ret_buffer; int buffer_len = *ret_buffer_len; - part_values_t pv; + + uint16_t tmp16; + size_t exp_size; int i; - uint16_t h_length; - uint16_t h_type; - uint16_t h_num; + uint16_t pkg_length; + uint16_t pkg_type; + uint16_t pkg_numval; + + uint8_t *pkg_types; + value_t *pkg_values; if (buffer_len < (15)) { @@ -390,32 +483,69 @@ static int parse_part_values (void **ret_buffer, int *ret_buffer_len, return (-1); } - pv.head = (part_header_t *) buffer; - h_length = ntohs (pv.head->length); - h_type = ntohs (pv.head->type); + memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); + buffer += sizeof (tmp16); + pkg_type = ntohs (tmp16); - assert (h_type == TYPE_VALUES); + memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); + buffer += sizeof (tmp16); + pkg_length = ntohs (tmp16); - pv.num_values = (uint16_t *) (pv.head + 1); - h_num = ntohs (*pv.num_values); + memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); + buffer += sizeof (tmp16); + pkg_numval = ntohs (tmp16); - if (h_num != ((h_length - 6) / 9)) + assert (pkg_type == TYPE_VALUES); + + exp_size = 3 * sizeof (uint16_t) + + pkg_numval * (sizeof (uint8_t) + sizeof (value_t)); + if (buffer_len < exp_size) { - DEBUG ("`length' and `num of values' don't match"); + WARNING ("network plugin: parse_part_values: " + "Packet too short: " + "Chunk of size %u expected, " + "but buffer has only %i bytes left.", + (unsigned int) exp_size, buffer_len); return (-1); } - pv.values_types = (uint8_t *) (pv.num_values + 1); - pv.values = (value_t *) (pv.values_types + h_num); + if (pkg_length != exp_size) + { + WARNING ("network plugin: parse_part_values: " + "Length and number of values " + "in the packet don't match."); + return (-1); + } - for (i = 0; i < h_num; i++) - if (pv.values_types[i] == DS_TYPE_COUNTER) - pv.values[i].counter = ntohll (pv.values[i].counter); + pkg_types = (uint8_t *) malloc (pkg_numval * sizeof (uint8_t)); + pkg_values = (value_t *) malloc (pkg_numval * sizeof (value_t)); + if ((pkg_types == NULL) || (pkg_values == NULL)) + { + sfree (pkg_types); + sfree (pkg_values); + ERROR ("network plugin: parse_part_values: malloc failed."); + return (-1); + } + + memcpy ((void *) pkg_types, (void *) buffer, pkg_numval * sizeof (uint8_t)); + buffer += pkg_numval * sizeof (uint8_t); + memcpy ((void *) pkg_values, (void *) buffer, pkg_numval * sizeof (value_t)); + buffer += pkg_numval * sizeof (value_t); + + for (i = 0; i < pkg_numval; i++) + { + if (pkg_types[i] == DS_TYPE_COUNTER) + pkg_values[i].counter = ntohll (pkg_values[i].counter); + else if (pkg_types[i] == DS_TYPE_GAUGE) + pkg_values[i].gauge = ntohd (pkg_values[i].gauge); + } - *ret_buffer = (void *) (pv.values + h_num); - *ret_buffer_len = buffer_len - h_length; - *ret_num_values = h_num; - *ret_values = pv.values; + *ret_buffer = buffer; + *ret_buffer_len = buffer_len - pkg_length; + *ret_num_values = pkg_numval; + *ret_values = pkg_values; + + sfree (pkg_types); return (0); } /* int parse_part_values */ @@ -423,21 +553,40 @@ static int parse_part_values (void **ret_buffer, int *ret_buffer_len, static int parse_part_number (void **ret_buffer, int *ret_buffer_len, uint64_t *value) { - part_number_t pn; - uint16_t len; + char *buffer = *ret_buffer; + int buffer_len = *ret_buffer_len; - pn.head = (part_header_t *) *ret_buffer; - pn.value = (uint64_t *) (pn.head + 1); + uint16_t tmp16; + uint64_t tmp64; + size_t exp_size = 2 * sizeof (uint16_t) + sizeof (uint64_t); - len = ntohs (pn.head->length); - if (len != 12) - return (-1); - if (len > *ret_buffer_len) + uint16_t pkg_length; + uint16_t pkg_type; + + if (buffer_len < exp_size) + { + WARNING ("network plugin: parse_part_number: " + "Packet too short: " + "Chunk of size %u expected, " + "but buffer has only %i bytes left.", + (unsigned int) exp_size, buffer_len); return (-1); - *value = ntohll (*pn.value); + } + + memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); + buffer += sizeof (tmp16); + pkg_type = ntohs (tmp16); + + memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); + buffer += sizeof (tmp16); + pkg_length = ntohs (tmp16); + + memcpy ((void *) &tmp64, buffer, sizeof (tmp64)); + buffer += sizeof (tmp64); + *value = ntohll (tmp64); - *ret_buffer = (void *) (pn.value + 1); - *ret_buffer_len -= len; + *ret_buffer = buffer; + *ret_buffer_len = buffer_len - pkg_length; return (0); } /* int parse_part_number */ @@ -447,162 +596,255 @@ static int parse_part_string (void **ret_buffer, int *ret_buffer_len, { char *buffer = *ret_buffer; int buffer_len = *ret_buffer_len; - part_string_t ps; - uint16_t h_length; - uint16_t h_type; + uint16_t tmp16; + size_t header_size = 2 * sizeof (uint16_t); - DEBUG ("network plugin: parse_part_string: ret_buffer = %p;" - " ret_buffer_len = %i; output = %p; output_len = %i;", - *ret_buffer, *ret_buffer_len, - (void *) output, output_len); + uint16_t pkg_length; + uint16_t pkg_type; - ps.head = (part_header_t *) buffer; + if (buffer_len < header_size) + { + WARNING ("network plugin: parse_part_string: " + "Packet too short: " + "Chunk of at least size %u expected, " + "but buffer has only %i bytes left.", + (unsigned int) header_size, buffer_len); + return (-1); + } - h_length = ntohs (ps.head->length); - h_type = ntohs (ps.head->type); + memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); + buffer += sizeof (tmp16); + pkg_type = ntohs (tmp16); - DEBUG ("network plugin: parse_part_string: length = %hu; type = %hu;", - h_length, h_type); + memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); + buffer += sizeof (tmp16); + pkg_length = ntohs (tmp16); - if (buffer_len < h_length) + /* Check that packet fits in the input buffer */ + if (pkg_length > buffer_len) { - DEBUG ("packet is too short"); + WARNING ("network plugin: parse_part_string: " + "Packet too big: " + "Chunk of size %hu received, " + "but buffer has only %i bytes left.", + pkg_length, buffer_len); return (-1); } - assert ((h_type == TYPE_HOST) - || (h_type == TYPE_PLUGIN) - || (h_type == TYPE_PLUGIN_INSTANCE) - || (h_type == TYPE_TYPE) - || (h_type == TYPE_TYPE_INSTANCE)); - - ps.value = buffer + 4; - if (ps.value[h_length - 5] != '\0') + + /* Check that pkg_length is in the valid range */ + if (pkg_length <= header_size) { - DEBUG ("String does not end with a nullbyte"); + WARNING ("network plugin: parse_part_string: " + "Packet too short: " + "Header claims this packet is only %hu " + "bytes long.", pkg_length); return (-1); } - if (output_len < (h_length - 4)) + /* Check that the package data fits into the output buffer. + * The previous if-statement ensures that: + * `pkg_length > header_size' */ + if ((pkg_length - header_size) > output_len) { - DEBUG ("output buffer is too small"); + WARNING ("network plugin: parse_part_string: " + "Output buffer too small."); return (-1); } - strcpy (output, ps.value); - DEBUG ("network plugin: parse_part_string: output = %s", output); + /* All sanity checks successfull, let's copy the data over */ + output_len = pkg_length - header_size; + memcpy ((void *) output, (void *) buffer, output_len); + buffer += output_len; - *ret_buffer = (void *) (buffer + h_length); - *ret_buffer_len = buffer_len - h_length; + /* For some very weird reason '\0' doesn't do the trick on SPARC in + * this statement. */ + if (output[output_len - 1] != 0) + { + WARNING ("network plugin: parse_part_string: " + "Received string does not end " + "with a NULL-byte."); + return (-1); + } + + *ret_buffer = buffer; + *ret_buffer_len = buffer_len - pkg_length; return (0); } /* int parse_part_string */ static int parse_packet (void *buffer, int buffer_len) { - part_header_t *header; int status; value_list_t vl = VALUE_LIST_INIT; - char type[DATA_MAX_NAME_LEN]; + notification_t n; DEBUG ("network plugin: parse_packet: buffer = %p; buffer_len = %i;", buffer, buffer_len); memset (&vl, '\0', sizeof (vl)); - memset (&type, '\0', sizeof (type)); + memset (&n, '\0', sizeof (n)); status = 0; - while ((status == 0) && (buffer_len > sizeof (part_header_t))) + while ((status == 0) && (0 < buffer_len) + && ((unsigned int) buffer_len > sizeof (part_header_t))) { - header = (part_header_t *) buffer; + uint16_t pkg_length; + uint16_t pkg_type; + + memcpy ((void *) &pkg_type, + (void *) buffer, + sizeof (pkg_type)); + memcpy ((void *) &pkg_length, + (void *) (buffer + sizeof (pkg_type)), + sizeof (pkg_length)); - if (ntohs (header->length) > buffer_len) + pkg_length = ntohs (pkg_length); + pkg_type = ntohs (pkg_type); + + if (pkg_length > buffer_len) break; - /* Assure that this loop terminates eventually */ - if (ntohs (header->length) < 4) + /* Ensure that this loop terminates eventually */ + if (pkg_length < (2 * sizeof (uint16_t))) break; - if (ntohs (header->type) == TYPE_VALUES) + if (pkg_type == TYPE_VALUES) { status = parse_part_values (&buffer, &buffer_len, &vl.values, &vl.values_len); if (status != 0) - { - DEBUG ("parse_part_values failed."); break; - } if ((vl.time > 0) && (strlen (vl.host) > 0) && (strlen (vl.plugin) > 0) - && (strlen (type) > 0) - && (cache_check (type, &vl) == 0)) + && (strlen (vl.type) > 0) + && (cache_check (&vl) == 0)) { - DEBUG ("network plugin: parse_packet:" - " dispatching values"); - plugin_dispatch_values (type, &vl); + plugin_dispatch_values (&vl); } else { DEBUG ("network plugin: parse_packet:" " NOT dispatching values"); } + + sfree (vl.values); } - else if (ntohs (header->type) == TYPE_TIME) + else if (pkg_type == TYPE_TIME) { uint64_t tmp = 0; - status = parse_part_number (&buffer, &buffer_len, &tmp); + status = parse_part_number (&buffer, &buffer_len, + &tmp); if (status == 0) + { vl.time = (time_t) tmp; + n.time = (time_t) tmp; + } } - else if (ntohs (header->type) == TYPE_INTERVAL) + else if (pkg_type == TYPE_INTERVAL) { uint64_t tmp = 0; - status = parse_part_number (&buffer, &buffer_len, &tmp); + status = parse_part_number (&buffer, &buffer_len, + &tmp); if (status == 0) vl.interval = (int) tmp; } - else if (ntohs (header->type) == TYPE_HOST) + else if (pkg_type == TYPE_HOST) { status = parse_part_string (&buffer, &buffer_len, vl.host, sizeof (vl.host)); - DEBUG ("network plugin: parse_packet: vl.host = %s", vl.host); + if (status == 0) + sstrncpy (n.host, vl.host, sizeof (n.host)); } - else if (ntohs (header->type) == TYPE_PLUGIN) + else if (pkg_type == TYPE_PLUGIN) { status = parse_part_string (&buffer, &buffer_len, vl.plugin, sizeof (vl.plugin)); - DEBUG ("network plugin: parse_packet: vl.plugin = %s", vl.plugin); + if (status == 0) + sstrncpy (n.plugin, vl.plugin, + sizeof (n.plugin)); + } + else if (pkg_type == TYPE_PLUGIN_INSTANCE) + { + status = parse_part_string (&buffer, &buffer_len, + vl.plugin_instance, + sizeof (vl.plugin_instance)); + if (status == 0) + sstrncpy (n.plugin_instance, + vl.plugin_instance, + sizeof (n.plugin_instance)); } - else if (ntohs (header->type) == TYPE_PLUGIN_INSTANCE) + else if (pkg_type == TYPE_TYPE) { status = parse_part_string (&buffer, &buffer_len, - vl.plugin_instance, sizeof (vl.plugin_instance)); - DEBUG ("network plugin: parse_packet: vl.plugin_instance = %s", vl.plugin_instance); + vl.type, sizeof (vl.type)); + if (status == 0) + sstrncpy (n.type, vl.type, sizeof (n.type)); } - else if (ntohs (header->type) == TYPE_TYPE) + else if (pkg_type == TYPE_TYPE_INSTANCE) { status = parse_part_string (&buffer, &buffer_len, - type, sizeof (type)); - DEBUG ("network plugin: parse_packet: type = %s", type); + vl.type_instance, + sizeof (vl.type_instance)); + if (status == 0) + sstrncpy (n.type_instance, vl.type_instance, + sizeof (n.type_instance)); } - else if (ntohs (header->type) == TYPE_TYPE_INSTANCE) + else if (pkg_type == TYPE_MESSAGE) { status = parse_part_string (&buffer, &buffer_len, - vl.type_instance, sizeof (vl.type_instance)); - DEBUG ("network plugin: parse_packet: vl.type_instance = %s", vl.type_instance); + n.message, sizeof (n.message)); + + if (status != 0) + { + /* do nothing */ + } + else if ((n.severity != NOTIF_FAILURE) + && (n.severity != NOTIF_WARNING) + && (n.severity != NOTIF_OKAY)) + { + INFO ("network plugin: " + "Ignoring notification with " + "unknown severity %i.", + n.severity); + } + else if (n.time <= 0) + { + INFO ("network plugin: " + "Ignoring notification with " + "time == 0."); + } + else if (strlen (n.message) <= 0) + { + INFO ("network plugin: " + "Ignoring notification with " + "an empty message."); + } + else + { + plugin_dispatch_notification (&n); + } + } + else if (pkg_type == TYPE_SEVERITY) + { + uint64_t tmp = 0; + status = parse_part_number (&buffer, &buffer_len, + &tmp); + if (status == 0) + n.severity = (int) tmp; } else { DEBUG ("network plugin: parse_packet: Unknown part" - " type: 0x%0hx", ntohs (header->type)); - buffer = ((char *) buffer) + ntohs (header->length); + " type: 0x%04hx", pkg_type); + buffer = ((char *) buffer) + pkg_length; } } /* while (buffer_len > sizeof (part_header_t)) */ - return (0); + return (status); } /* int parse_packet */ static void free_sockent (sockent_t *se) @@ -682,6 +924,16 @@ static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai) static int network_bind_socket (const sockent_t *se, const struct addrinfo *ai) { int loop = 0; + int yes = 1; + + /* allow multiple sockets to use the same PORT number */ + if (setsockopt(se->fd, SOL_SOCKET, SO_REUSEADDR, + &yes, sizeof(yes)) == -1) { + char errbuf[1024]; + ERROR ("setsockopt: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return (-1); + } DEBUG ("fd = %i; calling `bind'", se->fd); @@ -989,7 +1241,38 @@ static int network_add_sending_socket (const char *node, const char *service) return (0); } /* int network_get_listen_socket */ -int network_receive (void) +static void *dispatch_thread (void *arg) +{ + while (42) + { + receive_list_entry_t *ent; + + /* Lock and wait for more data to come in */ + pthread_mutex_lock (&receive_list_lock); + while ((listen_loop == 0) + && (receive_list_head == NULL)) + pthread_cond_wait (&receive_list_cond, &receive_list_lock); + + /* Remove the head entry and unlock */ + ent = receive_list_head; + if (ent != NULL) + receive_list_head = ent->next; + pthread_mutex_unlock (&receive_list_lock); + + /* Check whether we are supposed to exit. We do NOT check `listen_loop' + * because we dispatch all missing packets before shutting down. */ + if (ent == NULL) + break; + + parse_packet (ent->data, ent->data_len); + + sfree (ent); + } /* while (42) */ + + return (NULL); +} /* void *receive_thread */ + +static int network_receive (void) { char buffer[BUFF_SIZE]; int buffer_len; @@ -1022,6 +1305,8 @@ int network_receive (void) for (i = 0; (i < listen_sockets_num) && (status > 0); i++) { + receive_list_entry_t *ent; + if ((listen_sockets[i].revents & (POLLIN | POLLPRI)) == 0) continue; status--; @@ -1038,7 +1323,35 @@ int network_receive (void) return (-1); } - parse_packet (buffer, buffer_len); + ent = malloc (sizeof (receive_list_entry_t)); + if (ent == NULL) + { + ERROR ("network plugin: malloc failed."); + return (-1); + } + memset (ent, '\0', sizeof (receive_list_entry_t)); + + /* Hopefully this be optimized out by the compiler. It + * might help prevent stupid bugs in the future though. + */ + assert (sizeof (ent->data) == sizeof (buffer)); + + memcpy (ent->data, buffer, buffer_len); + ent->data_len = buffer_len; + + pthread_mutex_lock (&receive_list_lock); + if (receive_list_head == NULL) + { + receive_list_head = ent; + receive_list_tail = ent; + } + else + { + receive_list_tail->next = ent; + receive_list_tail = ent; + } + pthread_cond_signal (&receive_list_cond); + pthread_mutex_unlock (&receive_list_lock); } /* for (listen_sockets) */ } /* while (listen_loop == 0) */ @@ -1090,7 +1403,7 @@ static int add_to_buffer (char *buffer, int buffer_size, if (write_part_string (&buffer, &buffer_size, TYPE_HOST, vl->host, strlen (vl->host)) != 0) return (-1); - strcpy (vl_def->host, vl->host); + sstrncpy (vl_def->host, vl->host, sizeof (vl_def->host)); } if (vl_def->time != vl->time) @@ -1107,8 +1420,6 @@ static int add_to_buffer (char *buffer, int buffer_size, (uint64_t) vl->interval)) return (-1); vl_def->interval = vl->interval; - DEBUG ("network plugin: add_to_buffer: interval = %i", - (int) vl->interval); } if (strcmp (vl_def->plugin, vl->plugin) != 0) @@ -1116,7 +1427,7 @@ static int add_to_buffer (char *buffer, int buffer_size, if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN, vl->plugin, strlen (vl->plugin)) != 0) return (-1); - strcpy (vl_def->plugin, vl->plugin); + sstrncpy (vl_def->plugin, vl->plugin, sizeof (vl_def->plugin)); } if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0) @@ -1125,15 +1436,15 @@ static int add_to_buffer (char *buffer, int buffer_size, vl->plugin_instance, strlen (vl->plugin_instance)) != 0) return (-1); - strcpy (vl_def->plugin_instance, vl->plugin_instance); + sstrncpy (vl_def->plugin_instance, vl->plugin_instance, sizeof (vl_def->plugin_instance)); } - if (strcmp (type_def, ds->type) != 0) + if (strcmp (type_def, vl->type) != 0) { if (write_part_string (&buffer, &buffer_size, TYPE_TYPE, - ds->type, strlen (ds->type)) != 0) + vl->type, strlen (vl->type)) != 0) return (-1); - strcpy (type_def, ds->type); + sstrncpy (type_def, vl->type, sizeof (type_def)); } if (strcmp (vl_def->type_instance, vl->type_instance) != 0) @@ -1142,7 +1453,7 @@ static int add_to_buffer (char *buffer, int buffer_size, vl->type_instance, strlen (vl->type_instance)) != 0) return (-1); - strcpy (vl_def->type_instance, vl->type_instance); + sstrncpy (vl_def->type_instance, vl->type_instance, sizeof (vl_def->type_instance)); } if (write_part_values (&buffer, &buffer_size, ds, vl) != 0) @@ -1170,7 +1481,7 @@ static int network_write (const data_set_t *ds, const value_list_t *vl) /* If the value is already in the cache, we have received it via the * network. We write it again if forwarding is activated. It's then in * the cache and should we receive it again we will ignore it. */ - status = cache_check (ds->type, vl); + status = cache_check (vl); if ((network_config_forward == 0) && (status != 0)) return (0); @@ -1282,17 +1593,93 @@ static int network_config (const char *key, const char *val) return (0); } /* int network_config */ +static int network_notification (const notification_t *n) +{ + char buffer[BUFF_SIZE]; + char *buffer_ptr = buffer; + int buffer_free = sizeof (buffer); + int status; + + memset (buffer, '\0', sizeof (buffer)); + + + status = write_part_number (&buffer_ptr, &buffer_free, TYPE_TIME, + (uint64_t) n->time); + if (status != 0) + return (-1); + + status = write_part_number (&buffer_ptr, &buffer_free, TYPE_SEVERITY, + (uint64_t) n->severity); + if (status != 0) + return (-1); + + if (strlen (n->host) > 0) + { + status = write_part_string (&buffer_ptr, &buffer_free, TYPE_HOST, + n->host, strlen (n->host)); + if (status != 0) + return (-1); + } + + if (strlen (n->plugin) > 0) + { + status = write_part_string (&buffer_ptr, &buffer_free, TYPE_PLUGIN, + n->plugin, strlen (n->plugin)); + if (status != 0) + return (-1); + } + + if (strlen (n->plugin_instance) > 0) + { + status = write_part_string (&buffer_ptr, &buffer_free, + TYPE_PLUGIN_INSTANCE, + n->plugin_instance, strlen (n->plugin_instance)); + if (status != 0) + return (-1); + } + + if (strlen (n->type) > 0) + { + status = write_part_string (&buffer_ptr, &buffer_free, TYPE_TYPE, + n->type, strlen (n->type)); + if (status != 0) + return (-1); + } + + if (strlen (n->type_instance) > 0) + { + status = write_part_string (&buffer_ptr, &buffer_free, TYPE_TYPE_INSTANCE, + n->type_instance, strlen (n->type_instance)); + if (status != 0) + return (-1); + } + + status = write_part_string (&buffer_ptr, &buffer_free, TYPE_MESSAGE, + n->message, strlen (n->message)); + if (status != 0) + return (-1); + + network_send_buffer (buffer, sizeof (buffer) - buffer_free); + + return (0); +} /* int network_notification */ + static int network_shutdown (void) { listen_loop++; - if (listen_thread != (pthread_t) 0) + /* Kill the listening thread */ + if (receive_thread_id != (pthread_t) 0) { - pthread_kill (listen_thread, SIGTERM); - pthread_join (listen_thread, NULL /* no return value */); - listen_thread = (pthread_t) 0; + pthread_kill (receive_thread_id, SIGTERM); + pthread_join (receive_thread_id, NULL /* no return value */); + receive_thread_id = (pthread_t) 0; } + /* Shutdown the dispatching thread */ + if (dispatch_thread_id != (pthread_t) 0) + pthread_cond_broadcast (&receive_list_cond); + if (send_buffer_fill > 0) flush_buffer (); @@ -1301,12 +1688,12 @@ static int network_shutdown (void) void *key; void *value; - while (avl_pick (cache_tree, &key, &value) == 0) + while (c_avl_pick (cache_tree, &key, &value) == 0) { sfree (key); sfree (value); } - avl_destroy (cache_tree); + c_avl_destroy (cache_tree); cache_tree = NULL; } @@ -1317,11 +1704,19 @@ static int network_shutdown (void) plugin_unregister_write ("network"); plugin_unregister_shutdown ("network"); + /* Let the init function do it's move again ;) */ + cache_flush_last = 0; + return (0); } /* int network_shutdown */ static int network_init (void) { + /* Check if we were already initialized. If so, just return - there's + * nothing more to do (for now, that is). */ + if (cache_flush_last != 0) + return (0); + plugin_register_shutdown ("network", network_shutdown); send_buffer_ptr = send_buffer; @@ -1329,20 +1724,36 @@ static int network_init (void) memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl)); memset (send_buffer_type, '\0', sizeof (send_buffer_type)); - cache_tree = avl_create ((int (*) (const void *, const void *)) strcmp); + cache_tree = c_avl_create ((int (*) (const void *, const void *)) strcmp); cache_flush_last = time (NULL); /* setup socket(s) and so on */ if (sending_sockets != NULL) + { plugin_register_write ("network", network_write); + plugin_register_notification ("network", network_notification); + } - if ((listen_sockets_num != 0) && (listen_thread == 0)) + if ((listen_sockets_num != 0) && (receive_thread_id == 0)) { int status; - status = pthread_create (&listen_thread, NULL /* no attributes */, - receive_thread, NULL /* no argument */); + status = pthread_create (&dispatch_thread_id, + NULL /* no attributes */, + dispatch_thread, + NULL /* no argument */); + if (status != 0) + { + char errbuf[1024]; + ERROR ("network: pthread_create failed: %s", + sstrerror (errno, errbuf, + sizeof (errbuf))); + } + status = pthread_create (&receive_thread_id, + NULL /* no attributes */, + receive_thread, + NULL /* no argument */); if (status != 0) { char errbuf[1024]; @@ -1354,9 +1765,32 @@ static int network_init (void) return (0); } /* int network_init */ +/* + * The flush option of the network plugin cannot flush individual identifiers. + * All the values are added to a buffer and sent when the buffer is full, the + * requested value may or may not be in there, it's not worth finding out. We + * just send the buffer if `flush' is called - if the requested value was in + * there, good. If not, well, then there is nothing to flush.. -octo + */ +static int network_flush (int timeout, const char *identifier) +{ + pthread_mutex_lock (&send_buffer_lock); + + if (((time (NULL) - cache_flush_last) >= timeout) + && (send_buffer_fill > 0)) + { + flush_buffer (); + } + + pthread_mutex_unlock (&send_buffer_lock); + + return (0); +} /* int network_flush */ + void module_register (void) { plugin_register_config ("network", network_config, config_keys, config_keys_num); plugin_register_init ("network", network_init); + plugin_register_flush ("network", network_flush); } /* void module_register */