X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fnetwork.c;h=c347552e904a981b0588028093ed759b10847d61;hb=c88e459eb3823b766fc3a9ed37f25bc1f630146c;hp=b5d8efb4de62d65e3ec5c01e7a4018ac048087b5;hpb=6f2fea73761343ab03d6baebb2c32907e4e0ae3d;p=collectd.git diff --git a/src/network.c b/src/network.c index b5d8efb4..c347552e 100644 --- a/src/network.c +++ b/src/network.c @@ -23,6 +23,7 @@ #include "plugin.h" #include "common.h" #include "configfile.h" +#include "utils_avltree.h" #include "network.h" @@ -140,14 +141,16 @@ typedef struct part_values_s part_values_t; */ static const char *config_keys[] = { + "CacheFlush", "Listen", "Server", "TimeToLive", - NULL + "Forward" }; -static int config_keys_num = 3; +static int config_keys_num = STATIC_ARRAY_SIZE (config_keys); static int network_config_ttl = 0; +static int network_config_forward = 0; static sockent_t *sending_sockets = NULL; @@ -159,13 +162,132 @@ static int listen_loop = 0; static char send_buffer[BUFF_SIZE]; static char *send_buffer_ptr; static int send_buffer_fill; -static value_list_t send_buffer_vl = VALUE_LIST_INIT; +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 c_avl_tree_t *cache_tree = NULL; +static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; +static time_t cache_flush_last; +static int cache_flush_interval = 1800; + /* * Private functions */ +static int cache_flush (void) +{ + char **keys = NULL; + int keys_num = 0; + + char **tmp; + int i; + + char *key; + time_t *value; + c_avl_iterator_t *iter; + + time_t curtime = time (NULL); + + 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; + tmp = (char **) realloc (keys, + (keys_num + 1) * sizeof (char *)); + if (tmp == NULL) + { + sfree (keys); + c_avl_iterator_destroy (iter); + ERROR ("network plugin: cache_flush: realloc" + " failed."); + return (-1); + } + keys = tmp; + keys[keys_num] = key; + keys_num++; + } /* while (c_avl_iterator_next) */ + c_avl_iterator_destroy (iter); + + for (i = 0; i < keys_num; i++) + { + if (c_avl_remove (cache_tree, keys[i], (void *) &key, + (void *) &value) != 0) + { + WARNING ("network plugin: cache_flush: c_avl_remove" + " (%s) failed.", keys[i]); + continue; + } + + sfree (key); + sfree (value); + } + + sfree (keys); + + DEBUG ("network plugin: cache_flush: Removed %i %s", + keys_num, (keys_num == 1) ? "entry" : "entries"); + cache_flush_last = curtime; + return (0); +} /* int cache_flush */ + +static int cache_check (const char *type, const value_list_t *vl) +{ + char key[1024]; + time_t *value = NULL; + int retval = -1; + + if (cache_tree == NULL) + return (-1); + + if (format_name (key, sizeof (key), vl->host, vl->plugin, + vl->plugin_instance, type, vl->type_instance)) + return (-1); + + pthread_mutex_lock (&cache_lock); + + if (c_avl_get (cache_tree, key, (void *) &value) == 0) + { + if (*value < vl->time) + { + *value = vl->time; + retval = 0; + } + else + { + DEBUG ("network plugin: cache_check: *value = %i >= vl->time = %i", + (int) *value, (int) vl->time); + retval = 1; + } + } + else + { + char *key_copy = strdup (key); + value = malloc (sizeof (time_t)); + if ((key_copy != NULL) && (value != NULL)) + { + *value = vl->time; + c_avl_insert (cache_tree, key_copy, value); + retval = 0; + } + else + { + sfree (key_copy); + sfree (value); + } + } + + if ((time (NULL) - cache_flush_last) > cache_flush_interval) + cache_flush (); + + pthread_mutex_unlock (&cache_lock); + + DEBUG ("network plugin: cache_check: key = %s; time = %i; retval = %i", + key, (int) vl->time, retval); + + return (retval); +} /* int cache_check */ + static int write_part_values (char **ret_buffer, int *ret_buffer_len, const data_set_t *ds, const value_list_t *vl) { @@ -263,7 +385,8 @@ static int parse_part_values (void **ret_buffer, int *ret_buffer_len, if (buffer_len < (15)) { - DEBUG ("packet is too short: buffer_len = %i", buffer_len); + DEBUG ("network plugin: packet is too short: buffer_len = %i", + buffer_len); return (-1); } @@ -383,7 +506,8 @@ static int parse_packet (void *buffer, int buffer_len) value_list_t vl = VALUE_LIST_INIT; char type[DATA_MAX_NAME_LEN]; - DEBUG ("buffer = %p; buffer_len = %i;", buffer, buffer_len); + DEBUG ("network plugin: parse_packet: buffer = %p; buffer_len = %i;", + buffer, buffer_len); memset (&vl, '\0', sizeof (vl)); memset (&type, '\0', sizeof (type)); @@ -413,14 +537,17 @@ static int parse_packet (void *buffer, int buffer_len) if ((vl.time > 0) && (strlen (vl.host) > 0) && (strlen (vl.plugin) > 0) - && (strlen (type) > 0)) + && (strlen (type) > 0) + && (cache_check (type, &vl) == 0)) { - DEBUG ("dispatching values"); + DEBUG ("network plugin: parse_packet:" + " dispatching values"); plugin_dispatch_values (type, &vl); } else { - DEBUG ("NOT dispatching values"); + DEBUG ("network plugin: parse_packet:" + " NOT dispatching values"); } } else if (ntohs (header->type) == TYPE_TIME) @@ -430,6 +557,13 @@ static int parse_packet (void *buffer, int buffer_len) if (status == 0) vl.time = (time_t) tmp; } + else if (ntohs (header->type) == TYPE_INTERVAL) + { + uint64_t tmp = 0; + status = parse_part_number (&buffer, &buffer_len, &tmp); + if (status == 0) + vl.interval = (int) tmp; + } else if (ntohs (header->type) == TYPE_HOST) { status = parse_part_string (&buffer, &buffer_len, @@ -458,11 +592,12 @@ static int parse_packet (void *buffer, int buffer_len) { status = parse_part_string (&buffer, &buffer_len, vl.type_instance, sizeof (vl.type_instance)); - DEBUG ("network type: parse_packet: vl.type_instance = %s", vl.type_instance); + DEBUG ("network plugin: parse_packet: vl.type_instance = %s", vl.type_instance); } else { - DEBUG ("Unknown part type: 0x%0hx", ntohs (header->type)); + DEBUG ("network plugin: parse_packet: Unknown part" + " type: 0x%0hx", ntohs (header->type)); buffer = ((char *) buffer) + ntohs (header->length); } } /* while (buffer_len > sizeof (part_header_t)) */ @@ -546,7 +681,17 @@ 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 = 1; + 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); @@ -854,7 +999,7 @@ static int network_add_sending_socket (const char *node, const char *service) return (0); } /* int network_get_listen_socket */ -int network_receive (void) +static int network_receive (void) { char buffer[BUFF_SIZE]; int buffer_len; @@ -920,7 +1065,7 @@ static void network_send_buffer (const char *buffer, int buffer_len) sockent_t *se; int status; - DEBUG ("buffer_len = %i", buffer_len); + DEBUG ("network plugin: network_send_buffer: buffer_len = %i", buffer_len); for (se = sending_sockets; se != NULL; se = se->next) { @@ -956,7 +1101,6 @@ static int add_to_buffer (char *buffer, int buffer_size, vl->host, strlen (vl->host)) != 0) return (-1); strcpy (vl_def->host, vl->host); - DEBUG ("host = %s", vl->host); } if (vl_def->time != vl->time) @@ -965,7 +1109,14 @@ static int add_to_buffer (char *buffer, int buffer_size, (uint64_t) vl->time)) return (-1); vl_def->time = vl->time; - DEBUG ("time = %u", (unsigned int) vl->time); + } + + if (vl_def->interval != vl->interval) + { + if (write_part_number (&buffer, &buffer_size, TYPE_INTERVAL, + (uint64_t) vl->interval)) + return (-1); + vl_def->interval = vl->interval; } if (strcmp (vl_def->plugin, vl->plugin) != 0) @@ -974,7 +1125,6 @@ static int add_to_buffer (char *buffer, int buffer_size, vl->plugin, strlen (vl->plugin)) != 0) return (-1); strcpy (vl_def->plugin, vl->plugin); - DEBUG ("plugin = %s", vl->plugin); } if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0) @@ -984,7 +1134,6 @@ static int add_to_buffer (char *buffer, int buffer_size, strlen (vl->plugin_instance)) != 0) return (-1); strcpy (vl_def->plugin_instance, vl->plugin_instance); - DEBUG ("plugin_instance = %s", vl->plugin_instance); } if (strcmp (type_def, ds->type) != 0) @@ -993,7 +1142,6 @@ static int add_to_buffer (char *buffer, int buffer_size, ds->type, strlen (ds->type)) != 0) return (-1); strcpy (type_def, ds->type); - DEBUG ("type = %s", ds->type); } if (strcmp (vl_def->type_instance, vl->type_instance) != 0) @@ -1003,7 +1151,6 @@ static int add_to_buffer (char *buffer, int buffer_size, strlen (vl->type_instance)) != 0) return (-1); strcpy (vl_def->type_instance, vl->type_instance); - DEBUG ("type_instance = %s", vl->type_instance); } if (write_part_values (&buffer, &buffer_size, ds, vl) != 0) @@ -1014,6 +1161,9 @@ static int add_to_buffer (char *buffer, int buffer_size, static void flush_buffer (void) { + DEBUG ("network plugin: flush_buffer: send_buffer_fill = %i", + send_buffer_fill); + network_send_buffer (send_buffer, send_buffer_fill); send_buffer_ptr = send_buffer; send_buffer_fill = 0; @@ -1025,6 +1175,14 @@ static int network_write (const data_set_t *ds, const value_list_t *vl) { int status; + /* 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); + if ((network_config_forward == 0) + && (status != 0)) + return (0); + pthread_mutex_lock (&send_buffer_lock); status = add_to_buffer (send_buffer_ptr, @@ -1089,7 +1247,11 @@ static int network_config (const char *key, const char *val) && (fields_num != 2)) return (1); else if (fields_num == 2) + { + if ((service = strchr (fields[1], '.')) != NULL) + *service = '\0'; service = fields[1]; + } node = fields[0]; if (strcasecmp ("Listen", key) == 0) @@ -1105,26 +1267,66 @@ static int network_config (const char *key, const char *val) else return (1); } + else if (strcasecmp ("Forward", key) == 0) + { + if ((strcasecmp ("true", val) == 0) + || (strcasecmp ("yes", val) == 0) + || (strcasecmp ("on", val) == 0)) + network_config_forward = 1; + else + network_config_forward = 0; + } + else if (strcasecmp ("CacheFlush", key) == 0) + { + int tmp = atoi (val); + if (tmp > 0) + cache_flush_interval = tmp; + else return (1); + } else { return (-1); } return (0); -} +} /* int network_config */ static int network_shutdown (void) { - DEBUG ("Shutting down."); - listen_loop++; - pthread_kill (listen_thread, SIGTERM); - pthread_join (listen_thread, NULL /* no return value */); + if (listen_thread != (pthread_t) 0) + { + pthread_kill (listen_thread, SIGTERM); + pthread_join (listen_thread, NULL /* no return value */); + listen_thread = (pthread_t) 0; + } - listen_thread = 0; + if (send_buffer_fill > 0) + flush_buffer (); + + if (cache_tree != NULL) + { + void *key; + void *value; + + while (c_avl_pick (cache_tree, &key, &value) == 0) + { + sfree (key); + sfree (value); + } + c_avl_destroy (cache_tree); + cache_tree = NULL; + } + + /* TODO: Close `sending_sockets' */ + + plugin_unregister_config ("network"); + plugin_unregister_init ("network"); + plugin_unregister_write ("network"); + plugin_unregister_shutdown ("network"); return (0); -} +} /* int network_shutdown */ static int network_init (void) { @@ -1135,6 +1337,9 @@ 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 = 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); @@ -1162,4 +1367,4 @@ void module_register (void) plugin_register_config ("network", network_config, config_keys, config_keys_num); plugin_register_init ("network", network_init); -} +} /* void module_register */