X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fnetwork.c;h=34f89d96a39898cc4e27eb98024245fd686d1a65;hb=0b6de87f3d08326ee3200db8e16f3738df027ac0;hp=29e6708a0744682de69958c260b4f2b4c4ddd6ad;hpb=f9e4634b168f40a704d8b516f4d702e835f01588;p=collectd.git diff --git a/src/network.c b/src/network.c index 29e6708a..34f89d96 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 @@ -178,12 +178,11 @@ 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_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 time_t cache_flush_last = 0; static int cache_flush_interval = 1800; /* @@ -246,7 +245,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; @@ -256,7 +255,7 @@ 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); @@ -425,33 +424,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) { - char *packet_ptr; - int packet_len; + char *buffer; + int buffer_len; - part_header_t pkg_head; + uint16_t pkg_type; + uint16_t pkg_length; int offset; - packet_len = sizeof (pkg_head) + str_len + 1; - if (*ret_buffer_len < packet_len) + buffer_len = 2 * sizeof (uint16_t) + str_len + 1; + if (*ret_buffer_len < buffer_len) return (-1); - pkg_head.type = htons (type); - pkg_head.length = htons (packet_len); + pkg_type = htons (type); + pkg_length = htons (buffer_len); - packet_ptr = *ret_buffer; + buffer = *ret_buffer; offset = 0; - memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head)); - offset += sizeof (pkg_head); - memcpy (packet_ptr + offset, str, str_len); + 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 (packet_ptr + offset, '\0', 1); + memset (buffer + offset, '\0', 1); offset += 1; - assert (offset == packet_len); + assert (offset == buffer_len); - *ret_buffer = packet_ptr + packet_len; - *ret_buffer_len -= packet_len; + *ret_buffer = buffer + buffer_len; + *ret_buffer_len -= buffer_len; return (0); } /* int write_part_string */ @@ -482,11 +484,11 @@ static int parse_part_values (void **ret_buffer, int *ret_buffer_len, memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); - pkg_length = ntohs (tmp16); + pkg_type = ntohs (tmp16); memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); - pkg_type = ntohs (tmp16); + pkg_length = ntohs (tmp16); memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); @@ -572,11 +574,11 @@ static int parse_part_number (void **ret_buffer, int *ret_buffer_len, memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); - pkg_length = ntohs (tmp16); + pkg_type = ntohs (tmp16); memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); - pkg_type = ntohs (tmp16); + pkg_length = ntohs (tmp16); memcpy ((void *) &tmp64, buffer, sizeof (tmp64)); buffer += sizeof (tmp64); @@ -612,11 +614,11 @@ static int parse_part_string (void **ret_buffer, int *ret_buffer_len, memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); - pkg_length = ntohs (tmp16); + pkg_type = ntohs (tmp16); memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); - pkg_type = ntohs (tmp16); + pkg_length = ntohs (tmp16); /* Check that packet fits in the input buffer */ if (pkg_length > buffer_len) @@ -654,7 +656,9 @@ static int parse_part_string (void **ret_buffer, int *ret_buffer_len, memcpy ((void *) output, (void *) buffer, output_len); buffer += output_len; - if (output[output_len - 1] != '\0'); + /* 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 " @@ -673,26 +677,27 @@ static int parse_packet (void *buffer, int buffer_len) 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))) { uint16_t pkg_length; uint16_t pkg_type; - memcpy ((void *) &pkg_length, - (void *) buffer, - sizeof (pkg_length)); memcpy ((void *) &pkg_type, - (void *) buffer + sizeof (pkg_length), + (void *) buffer, sizeof (pkg_type)); + memcpy ((void *) &pkg_length, + (void *) (buffer + sizeof (pkg_type)), + sizeof (pkg_length)); pkg_length = ntohs (pkg_length); pkg_type = ntohs (pkg_type); @@ -714,10 +719,10 @@ static int parse_packet (void *buffer, int buffer_len) 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)) { - plugin_dispatch_values (type, &vl); + plugin_dispatch_values (&vl); } else { @@ -730,14 +735,19 @@ static int parse_packet (void *buffer, int buffer_len) 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 (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; } @@ -745,31 +755,85 @@ static int parse_packet (void *buffer, int buffer_len) { 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 (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)); - DEBUG ("network plugin: parse_packet: vl.plugin_instance = %s", vl.plugin_instance); + vl.plugin_instance, + sizeof (vl.plugin_instance)); + if (status == 0) + sstrncpy (n.plugin_instance, + vl.plugin_instance, + sizeof (n.plugin_instance)); } else if (pkg_type == TYPE_TYPE) { status = parse_part_string (&buffer, &buffer_len, - type, sizeof (type)); - DEBUG ("network plugin: parse_packet: type = %s", type); + vl.type, sizeof (vl.type)); + if (status == 0) + sstrncpy (n.type, vl.type, sizeof (n.type)); } else if (pkg_type == TYPE_TYPE_INSTANCE) { 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); + vl.type_instance, + sizeof (vl.type_instance)); + if (status == 0) + sstrncpy (n.type_instance, vl.type_instance, + sizeof (n.type_instance)); + } + else if (pkg_type == TYPE_MESSAGE) + { + status = parse_part_string (&buffer, &buffer_len, + 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 { @@ -859,6 +923,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); @@ -1205,6 +1279,9 @@ static int network_receive (void) int i; int status; + receive_list_entry_t *private_list_head; + receive_list_entry_t *private_list_tail; + if (listen_sockets_num == 0) network_add_listen_socket (NULL, NULL); @@ -1214,6 +1291,9 @@ static int network_receive (void) return (-1); } + private_list_head = NULL; + private_list_tail = NULL; + while (listen_loop == 0) { status = poll (listen_sockets, listen_sockets_num, -1); @@ -1254,7 +1334,8 @@ static int network_receive (void) ERROR ("network plugin: malloc failed."); return (-1); } - memset (ent, '\0', sizeof (receive_list_entry_t)); + memset (ent, 0, sizeof (receive_list_entry_t)); + ent->next = NULL; /* Hopefully this be optimized out by the compiler. It * might help prevent stupid bugs in the future though. @@ -1264,22 +1345,49 @@ static int network_receive (void) 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; - } + if (private_list_head == NULL) + private_list_head = ent; else + private_list_tail->next = ent; + private_list_tail = ent; + + /* Do not block here. Blocking here has led to + * insufficient performance in the past. */ + if (pthread_mutex_trylock (&receive_list_lock) == 0) { - receive_list_tail->next = ent; - receive_list_tail = ent; + if (receive_list_head == NULL) + receive_list_head = private_list_head; + else + receive_list_tail->next = private_list_head; + receive_list_tail = private_list_tail; + + private_list_head = NULL; + private_list_tail = NULL; + + pthread_cond_signal (&receive_list_cond); + pthread_mutex_unlock (&receive_list_lock); } - pthread_cond_signal (&receive_list_cond); - pthread_mutex_unlock (&receive_list_lock); } /* for (listen_sockets) */ } /* while (listen_loop == 0) */ + /* Make sure everything is dispatched before exiting. */ + if (private_list_head != NULL) + { + pthread_mutex_lock (&receive_list_lock); + + if (receive_list_head == NULL) + receive_list_head = private_list_head; + else + receive_list_tail->next = private_list_head; + receive_list_tail = private_list_tail; + + private_list_head = NULL; + private_list_tail = NULL; + + pthread_cond_signal (&receive_list_cond); + pthread_mutex_unlock (&receive_list_lock); + } + return (0); } @@ -1318,7 +1426,7 @@ static void network_send_buffer (const char *buffer, int buffer_len) } /* void network_send_buffer */ static int add_to_buffer (char *buffer, int buffer_size, - value_list_t *vl_def, char *type_def, + value_list_t *vl_def, const data_set_t *ds, const value_list_t *vl) { char *buffer_orig = buffer; @@ -1328,7 +1436,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) @@ -1352,7 +1460,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) @@ -1361,15 +1469,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 (vl_def->type, 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 (vl_def->type, ds->type, sizeof (vl_def->type)); } if (strcmp (vl_def->type_instance, vl->type_instance) != 0) @@ -1378,7 +1486,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) @@ -1395,8 +1503,7 @@ static void flush_buffer (void) network_send_buffer (send_buffer, send_buffer_fill); send_buffer_ptr = send_buffer; send_buffer_fill = 0; - memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl)); - memset (send_buffer_type, '\0', sizeof (send_buffer_type)); + memset (&send_buffer_vl, 0, sizeof (send_buffer_vl)); } static int network_write (const data_set_t *ds, const value_list_t *vl) @@ -1406,7 +1513,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); @@ -1415,7 +1522,7 @@ static int network_write (const data_set_t *ds, const value_list_t *vl) status = add_to_buffer (send_buffer_ptr, sizeof (send_buffer) - send_buffer_fill, - &send_buffer_vl, send_buffer_type, + &send_buffer_vl, ds, vl); if (status >= 0) { @@ -1429,7 +1536,7 @@ static int network_write (const data_set_t *ds, const value_list_t *vl) status = add_to_buffer (send_buffer_ptr, sizeof (send_buffer) - send_buffer_fill, - &send_buffer_vl, send_buffer_type, + &send_buffer_vl, ds, vl); if (status >= 0) @@ -1473,7 +1580,10 @@ static int network_config (const char *key, const char *val) fields_num = strsplit (val_cpy, fields, 3); if ((fields_num != 1) && (fields_num != 2)) + { + sfree (val_cpy); return (1); + } else if (fields_num == 2) { if ((service = strchr (fields[1], '.')) != NULL) @@ -1486,6 +1596,8 @@ static int network_config (const char *key, const char *val) network_add_listen_socket (node, service); else network_add_sending_socket (node, service); + + sfree (val_cpy); } else if (strcasecmp ("TimeToLive", key) == 0) { @@ -1518,6 +1630,77 @@ 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++; @@ -1558,24 +1741,34 @@ 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; send_buffer_fill = 0; - memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl)); - memset (send_buffer_type, '\0', sizeof (send_buffer_type)); + memset (&send_buffer_vl, 0, sizeof (send_buffer_vl)); 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) && (receive_thread_id == 0)) { @@ -1608,9 +1801,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 */