X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fnetwork.c;h=72e3484ecaa1216f8e35374e2903355412c31fad;hb=e4fe9b4d69c69d922cad1272104fdb8de5c24faf;hp=93d403f9c3b3f760480927bcb8ba1112cd2c8853;hpb=a5177bd853eedbe69d87869316093a0e9526e931;p=collectd.git diff --git a/src/network.c b/src/network.c index 93d403f9..e58d1dc5 100644 --- a/src/network.c +++ b/src/network.c @@ -1,6 +1,6 @@ /** * collectd - src/network.c - * Copyright (C) 2005-2009 Florian octo Forster + * Copyright (C) 2005-2013 Florian octo Forster * Copyright (C) 2009 Aman Gupta * * This program is free software; you can redistribute it and/or modify it @@ -18,10 +18,11 @@ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA * * Authors: - * Florian octo Forster + * Florian octo Forster * Aman Gupta **/ +#define _DEFAULT_SOURCE #define _BSD_SOURCE /* For struct ip_mreq */ #include "collectd.h" @@ -31,6 +32,7 @@ #include "utils_fbhash.h" #include "utils_avltree.h" #include "utils_cache.h" +#include "utils_complain.h" #include "network.h" @@ -57,8 +59,27 @@ #endif #if HAVE_LIBGCRYPT +# include +# if defined __APPLE__ +/* default xcode compiler throws warnings even when deprecated functionality + * is not used. -Werror breaks the build because of erroneous warnings. + * http://stackoverflow.com/questions/10556299/compiler-warnings-with-libgcrypt-v1-5-0/12830209#12830209 + */ +# pragma GCC diagnostic ignored "-Wdeprecated-declarations" +# endif +/* FreeBSD's copy of libgcrypt extends the existing GCRYPT_NO_DEPRECATED + * to properly hide all deprecated functionality. + * http://svnweb.freebsd.org/ports/head/security/libgcrypt/files/patch-src__gcrypt.h.in + */ +# define GCRYPT_NO_DEPRECATED # include +# if defined __APPLE__ +/* Re enable deprecation warnings */ +# pragma GCC diagnostic warning "-Wdeprecated-declarations" +# endif +# if GCRYPT_VERSION_NUMBER < 0x010600 GCRY_THREAD_OPTION_PTHREAD_IMPL; +# endif #endif #ifndef IPV6_ADD_MEMBERSHIP @@ -99,6 +120,8 @@ struct sockent_client gcry_cipher_hd_t cypher; unsigned char password_hash[32]; #endif + cdtime_t next_resolve_reconnect; + cdtime_t resolve_interval; }; struct sockent_server @@ -258,6 +281,7 @@ typedef struct receive_list_entry_s receive_list_entry_t; * Private variables */ static int network_config_ttl = 0; +/* Ethernet - (IPv6 + UDP) = 1500 - (40 + 8) = 1452 */ static size_t network_config_packet_size = 1452; static int network_config_forward = 0; static int network_config_stats = 0; @@ -286,6 +310,7 @@ static pthread_t dispatch_thread_id; static char *send_buffer; static char *send_buffer_ptr; static int send_buffer_fill; +static cdtime_t send_buffer_last_update; static value_list_t send_buffer_vl = VALUE_LIST_STATIC; static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER; @@ -295,14 +320,14 @@ static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER; * example). Only if neither is true, the stats_lock is acquired. The counters * are always read without holding a lock in the hope that writing 8 bytes to * memory is an atomic operation. */ -static uint64_t stats_octets_rx = 0; -static uint64_t stats_octets_tx = 0; -static uint64_t stats_packets_rx = 0; -static uint64_t stats_packets_tx = 0; -static uint64_t stats_values_dispatched = 0; -static uint64_t stats_values_not_dispatched = 0; -static uint64_t stats_values_sent = 0; -static uint64_t stats_values_not_sent = 0; +static derive_t stats_octets_rx = 0; +static derive_t stats_octets_tx = 0; +static derive_t stats_packets_rx = 0; +static derive_t stats_packets_tx = 0; +static derive_t stats_values_dispatched = 0; +static derive_t stats_values_not_dispatched = 0; +static derive_t stats_values_sent = 0; +static derive_t stats_values_not_sent = 0; static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER; /* @@ -319,30 +344,30 @@ static _Bool check_receive_okay (const value_list_t *vl) /* {{{ */ /* This is a value we already sent. Don't allow it to be received again in * order to avoid looping. */ if ((status == 0) && (time_sent >= ((uint64_t) vl->time))) - return (false); + return (0); - return (true); + return (1); } /* }}} _Bool check_receive_okay */ static _Bool check_send_okay (const value_list_t *vl) /* {{{ */ { - _Bool received = false; + _Bool received = 0; int status; if (network_config_forward != 0) - return (true); + return (1); if (vl->meta == NULL) - return (true); + return (1); status = meta_data_get_boolean (vl->meta, "network:received", &received); if (status == -ENOENT) - return (true); + return (1); else if (status != 0) { ERROR ("network plugin: check_send_okay: meta_data_get_boolean failed " "with status %i.", status); - return (true); + return (1); } /* By default, only *send* value lists that were not *received* by the @@ -350,6 +375,43 @@ static _Bool check_send_okay (const value_list_t *vl) /* {{{ */ return (!received); } /* }}} _Bool check_send_okay */ +static _Bool check_notify_received (const notification_t *n) /* {{{ */ +{ + notification_meta_t *ptr; + + for (ptr = n->meta; ptr != NULL; ptr = ptr->next) + if ((strcmp ("network:received", ptr->name) == 0) + && (ptr->type == NM_TYPE_BOOLEAN)) + return ((_Bool) ptr->nm_value.nm_boolean); + + return (0); +} /* }}} _Bool check_notify_received */ + +static _Bool check_send_notify_okay (const notification_t *n) /* {{{ */ +{ + static c_complain_t complain_forwarding = C_COMPLAIN_INIT_STATIC; + _Bool received = 0; + + if (n->meta == NULL) + return (1); + + received = check_notify_received (n); + + if (network_config_forward && received) + { + c_complain_once (LOG_ERR, &complain_forwarding, + "network plugin: A notification has been received via the network " + "and forwarding is enabled. Forwarding of notifications is currently " + "not supported, because there is not loop-deteciton available. " + "Please contact the collectd mailing list if you need this " + "feature."); + } + + /* By default, only *send* value lists that were not *received* by the + * network plugin. */ + return (!received); +} /* }}} _Bool check_send_notify_okay */ + static int network_dispatch_values (value_list_t *vl, /* {{{ */ const char *username) { @@ -383,7 +445,7 @@ static int network_dispatch_values (value_list_t *vl, /* {{{ */ return (-ENOMEM); } - status = meta_data_add_boolean (vl->meta, "network:received", true); + status = meta_data_add_boolean (vl->meta, "network:received", 1); if (status != 0) { ERROR ("network plugin: meta_data_add_boolean failed."); @@ -413,7 +475,53 @@ static int network_dispatch_values (value_list_t *vl, /* {{{ */ return (0); } /* }}} int network_dispatch_values */ +static int network_dispatch_notification (notification_t *n) /* {{{ */ +{ + int status; + + assert (n->meta == NULL); + + status = plugin_notification_meta_add_boolean (n, "network:received", 1); + if (status != 0) + { + ERROR ("network plugin: plugin_notification_meta_add_boolean failed."); + plugin_notification_meta_free (n->meta); + n->meta = NULL; + return (status); + } + + status = plugin_dispatch_notification (n); + + plugin_notification_meta_free (n->meta); + n->meta = NULL; + + return (status); +} /* }}} int network_dispatch_notification */ + #if HAVE_LIBGCRYPT +static void network_init_gcrypt (void) /* {{{ */ +{ + /* http://lists.gnupg.org/pipermail/gcrypt-devel/2003-August/000458.html + * Because you can't know in a library whether another library has + * already initialized the library */ + if (gcry_control (GCRYCTL_ANY_INITIALIZATION_P)) + return; + + /* http://www.gnupg.org/documentation/manuals/gcrypt/Multi_002dThreading.html + * To ensure thread-safety, it's important to set GCRYCTL_SET_THREAD_CBS + * *before* initalizing Libgcrypt with gcry_check_version(), which itself must + * be called before any other gcry_* function. GCRYCTL_ANY_INITIALIZATION_P + * above doesn't count, as it doesn't implicitly initalize Libgcrypt. + * + * tl;dr: keep all these gry_* statements in this exact order please. */ +# if GCRYPT_VERSION_NUMBER < 0x010600 + gcry_control (GCRYCTL_SET_THREAD_CBS, &gcry_threads_pthread); +# endif + gcry_check_version (NULL); + gcry_control (GCRYCTL_INIT_SECMEM, 32768); + gcry_control (GCRYCTL_INITIALIZATION_FINISHED); +} /* }}} void network_init_gcrypt */ + static gcry_cipher_hd_t network_get_aes256_cypher (sockent_t *se, /* {{{ */ const void *iv, size_t iv_size, const char *username) { @@ -599,7 +707,7 @@ static int write_part_number (char **ret_buffer, int *ret_buffer_len, part_header_t pkg_head; uint64_t pkg_value; - + int offset; packet_len = sizeof (pkg_head) + sizeof (pkg_value); @@ -703,7 +811,7 @@ static int parse_part_values (void **ret_buffer, size_t *ret_buffer_len, exp_size = 3 * sizeof (uint16_t) + pkg_numval * (sizeof (uint8_t) + sizeof (value_t)); - if ((buffer_len < 0) || (buffer_len < exp_size)) + if (buffer_len < exp_size) { WARNING ("network plugin: parse_part_values: " "Packet too short: " @@ -787,9 +895,8 @@ static int parse_part_number (void **ret_buffer, size_t *ret_buffer_len, size_t exp_size = 2 * sizeof (uint16_t) + sizeof (uint64_t); uint16_t pkg_length; - uint16_t pkg_type; - if ((buffer_len < 0) || ((size_t) buffer_len < exp_size)) + if (buffer_len < exp_size) { WARNING ("network plugin: parse_part_number: " "Packet too short: " @@ -801,7 +908,7 @@ static int parse_part_number (void **ret_buffer, size_t *ret_buffer_len, memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); - pkg_type = ntohs (tmp16); + /* pkg_type = ntohs (tmp16); */ memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); @@ -818,18 +925,21 @@ static int parse_part_number (void **ret_buffer, size_t *ret_buffer_len, } /* int parse_part_number */ static int parse_part_string (void **ret_buffer, size_t *ret_buffer_len, - char *output, int output_len) + char *output, size_t const output_len) { char *buffer = *ret_buffer; size_t buffer_len = *ret_buffer_len; uint16_t tmp16; - size_t header_size = 2 * sizeof (uint16_t); + size_t const header_size = 2 * sizeof (uint16_t); uint16_t pkg_length; - uint16_t pkg_type; + size_t payload_size; - if ((buffer_len < 0) || (buffer_len < header_size)) + if (output_len <= 0) + return (EINVAL); + + if (buffer_len < header_size) { WARNING ("network plugin: parse_part_string: " "Packet too short: " @@ -841,11 +951,12 @@ static int parse_part_string (void **ret_buffer, size_t *ret_buffer_len, memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); - pkg_type = ntohs (tmp16); + /* pkg_type = ntohs (tmp16); */ memcpy ((void *) &tmp16, buffer, sizeof (tmp16)); buffer += sizeof (tmp16); pkg_length = ntohs (tmp16); + payload_size = ((size_t) pkg_length) - header_size; /* Check that packet fits in the input buffer */ if (pkg_length > buffer_len) @@ -871,22 +982,24 @@ static int parse_part_string (void **ret_buffer, size_t *ret_buffer_len, /* Check that the package data fits into the output buffer. * The previous if-statement ensures that: * `pkg_length > header_size' */ - if ((output_len < 0) - || ((size_t) output_len < ((size_t) pkg_length - header_size))) + if (output_len < payload_size) { WARNING ("network plugin: parse_part_string: " - "Output buffer too small."); + "Buffer too small: " + "Output buffer holds %zu bytes, " + "which is too small to hold the received " + "%zu byte string.", + output_len, payload_size); return (-1); } /* 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; + memcpy ((void *) output, (void *) buffer, payload_size); + buffer += payload_size; /* For some very weird reason '\0' doesn't do the trick on SPARC in * this statement. */ - if (output[output_len - 1] != 0) + if (output[payload_size - 1] != 0) { WARNING ("network plugin: parse_part_string: " "Received string does not end " @@ -917,6 +1030,8 @@ static int parse_packet (sockent_t *se, static int parse_part_sign_sha256 (sockent_t *se, /* {{{ */ void **ret_buffer, size_t *ret_buffer_len, int flags) { + static c_complain_t complain_no_users = C_COMPLAIN_INIT_STATIC; + char *buffer; size_t buffer_len; size_t buffer_offset; @@ -938,8 +1053,9 @@ static int parse_part_sign_sha256 (sockent_t *se, /* {{{ */ if (se->data.server.userdb == NULL) { - NOTICE ("network plugin: Received signed network packet but can't verify " - "it because no user DB has been configured. Will accept it."); + c_complain (LOG_NOTICE, &complain_no_users, + "network plugin: Received signed network packet but can't verify it " + "because no user DB has been configured. Will accept it."); return (0); } @@ -1378,7 +1494,18 @@ static int parse_packet (sockent_t *se, /* {{{ */ if (status == 0) { vl.time = TIME_T_TO_CDTIME_T (tmp); - n.time = TIME_T_TO_CDTIME_T (tmp); + n.time = TIME_T_TO_CDTIME_T (tmp); + } + } + else if (pkg_type == TYPE_TIME_HR) + { + uint64_t tmp = 0; + status = parse_part_number (&buffer, &buffer_size, + &tmp); + if (status == 0) + { + vl.time = (cdtime_t) tmp; + n.time = (cdtime_t) tmp; } } else if (pkg_type == TYPE_INTERVAL) @@ -1387,7 +1514,15 @@ static int parse_packet (sockent_t *se, /* {{{ */ status = parse_part_number (&buffer, &buffer_size, &tmp); if (status == 0) - vl.interval = (int) tmp; + vl.interval = TIME_T_TO_CDTIME_T (tmp); + } + else if (pkg_type == TYPE_INTERVAL_HR) + { + uint64_t tmp = 0; + status = parse_part_number (&buffer, &buffer_size, + &tmp); + if (status == 0) + vl.interval = (cdtime_t) tmp; } else if (pkg_type == TYPE_HOST) { @@ -1462,7 +1597,7 @@ static int parse_packet (sockent_t *se, /* {{{ */ } else { - plugin_dispatch_notification (&n); + network_dispatch_notification (&n); } } else if (pkg_type == TYPE_SEVERITY) @@ -1583,7 +1718,7 @@ static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai) sizeof (network_config_ttl)) != 0) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (ipv4-ttl): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); } @@ -1604,7 +1739,7 @@ static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai) sizeof (network_config_ttl)) != 0) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt(ipv6-ttl): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); @@ -1651,7 +1786,7 @@ static int network_set_interface (const sockent_t *se, const struct addrinfo *ai &mreq, sizeof (mreq)) != 0) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (ipv4-multicast-if): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); } @@ -1670,7 +1805,7 @@ static int network_set_interface (const sockent_t *se, const struct addrinfo *ai sizeof (se->interface)) != 0) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (ipv6-multicast-if): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); @@ -1681,9 +1816,9 @@ static int network_set_interface (const sockent_t *se, const struct addrinfo *ai } /* else: Not a multicast interface. */ -#if defined(HAVE_IF_INDEXTONAME) && HAVE_IF_INDEXTONAME && defined(SO_BINDTODEVICE) if (se->interface != 0) { +#if defined(HAVE_IF_INDEXTONAME) && HAVE_IF_INDEXTONAME && defined(SO_BINDTODEVICE) char interface_name[IFNAMSIZ]; if (if_indextoname (se->interface, interface_name) == NULL) @@ -1696,37 +1831,42 @@ static int network_set_interface (const sockent_t *se, const struct addrinfo *ai sizeof(interface_name)) == -1 ) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (bind-if): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); } - } /* #endif HAVE_IF_INDEXTONAME && SO_BINDTODEVICE */ #else - WARNING ("network plugin: Cannot set the interface on a unicast " + WARNING ("network plugin: Cannot set the interface on a unicast " "socket because " # if !defined(SO_BINDTODEVICE) - "the the \"SO_BINDTODEVICE\" socket option " + "the \"SO_BINDTODEVICE\" socket option " # else "the \"if_indextoname\" function " # endif "is not available on your system."); #endif + } + return (0); } /* }}} network_set_interface */ static int network_bind_socket (int fd, const struct addrinfo *ai, const int interface_idx) { +#if KERNEL_SOLARIS + char loop = 0; +#else int loop = 0; +#endif int yes = 1; /* allow multiple sockets to use the same PORT number */ if (setsockopt (fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) == -1) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (reuseaddr): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); } @@ -1769,7 +1909,7 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const int int &loop, sizeof (loop)) == -1) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (multicast-loop): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); @@ -1779,7 +1919,7 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const int int &mreq, sizeof (mreq)) == -1) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (add-membership): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); @@ -1817,7 +1957,7 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const int int &loop, sizeof (loop)) == -1) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (ipv6-multicast-loop): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); @@ -1827,7 +1967,7 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const int int &mreq, sizeof (mreq)) == -1) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (ipv6-add-membership): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); @@ -1855,7 +1995,7 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const int int sizeof(interface_name)) == -1 ) { char errbuf[1024]; - ERROR ("setsockopt: %s", + ERROR ("network plugin: setsockopt (bind-if): %s", sstrerror (errno, errbuf, sizeof (errbuf))); return (-1); } @@ -1867,14 +2007,19 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const int int /* Initialize a sockent structure. `type' must be either `SOCKENT_TYPE_CLIENT' * or `SOCKENT_TYPE_SERVER' */ -static int sockent_init (sockent_t *se, int type) /* {{{ */ +static sockent_t *sockent_create (int type) /* {{{ */ { - if (se == NULL) - return (-1); + sockent_t *se; + if ((type != SOCKENT_TYPE_CLIENT) && (type != SOCKENT_TYPE_SERVER)) + return (NULL); + + se = malloc (sizeof (*se)); + if (se == NULL) + return (NULL); memset (se, 0, sizeof (*se)); - se->type = SOCKENT_TYPE_CLIENT; + se->type = type; se->node = NULL; se->service = NULL; se->interface = 0; @@ -1882,7 +2027,6 @@ static int sockent_init (sockent_t *se, int type) /* {{{ */ if (type == SOCKENT_TYPE_SERVER) { - se->type = SOCKENT_TYPE_SERVER; se->data.server.fd = NULL; #if HAVE_LIBGCRYPT se->data.server.security_level = SECURITY_LEVEL_NONE; @@ -1895,6 +2039,8 @@ static int sockent_init (sockent_t *se, int type) /* {{{ */ { se->data.client.fd = -1; se->data.client.addr = NULL; + se->data.client.resolve_interval = 0; + se->data.client.next_resolve_reconnect = 0; #if HAVE_LIBGCRYPT se->data.client.security_level = SECURITY_LEVEL_NONE; se->data.client.username = NULL; @@ -1903,28 +2049,18 @@ static int sockent_init (sockent_t *se, int type) /* {{{ */ #endif } - return (0); -} /* }}} int sockent_init */ + return (se); +} /* }}} sockent_t *sockent_create */ -/* Open the file descriptors for a initialized sockent structure. */ -static int sockent_open (sockent_t *se) /* {{{ */ +static int sockent_init_crypto (sockent_t *se) /* {{{ */ { - struct addrinfo ai_hints; - struct addrinfo *ai_list, *ai_ptr; - int ai_return; - - const char *node; - const char *service; - - if (se == NULL) - return (-1); - - /* Set up the security structures. */ #if HAVE_LIBGCRYPT /* {{{ */ if (se->type == SOCKENT_TYPE_CLIENT) { if (se->data.client.security_level > SECURITY_LEVEL_NONE) { + network_init_gcrypt (); + if ((se->data.client.username == NULL) || (se->data.client.password == NULL)) { @@ -1943,6 +2079,8 @@ static int sockent_open (sockent_t *se) /* {{{ */ { if (se->data.server.security_level > SECURITY_LEVEL_NONE) { + network_init_gcrypt (); + if (se->data.server.auth_file == NULL) { ERROR ("network plugin: Server socket with " @@ -1966,13 +2104,150 @@ static int sockent_open (sockent_t *se) /* {{{ */ } #endif /* }}} HAVE_LIBGCRYPT */ + return (0); +} /* }}} int sockent_init_crypto */ + +static int sockent_client_disconnect (sockent_t *se) /* {{{ */ +{ + struct sockent_client *client; + + if ((se == NULL) || (se->type != SOCKENT_TYPE_CLIENT)) + return (EINVAL); + + client = &se->data.client; + if (client->fd >= 0) /* connected */ + { + close (client->fd); + client->fd = -1; + } + + sfree (client->addr); + client->addrlen = 0; + + return (0); +} /* }}} int sockent_client_disconnect */ + +static int sockent_client_connect (sockent_t *se) /* {{{ */ +{ + static c_complain_t complaint = C_COMPLAIN_INIT_STATIC; + + struct sockent_client *client; + struct addrinfo ai_hints; + struct addrinfo *ai_list = NULL, *ai_ptr; + int status; + _Bool reconnect = 0; + cdtime_t now; + + if ((se == NULL) || (se->type != SOCKENT_TYPE_CLIENT)) + return (EINVAL); + + client = &se->data.client; + + now = cdtime (); + if (client->resolve_interval != 0 && client->next_resolve_reconnect < now) { + DEBUG("network plugin: Reconnecting socket, resolve_interval = %lf, next_resolve_reconnect = %lf", + CDTIME_T_TO_DOUBLE(client->resolve_interval), CDTIME_T_TO_DOUBLE(client->next_resolve_reconnect)); + reconnect = 1; + } + + if (client->fd >= 0 && !reconnect) /* already connected and not stale*/ + return (0); + + memset (&ai_hints, 0, sizeof (ai_hints)); +#ifdef AI_ADDRCONFIG + ai_hints.ai_flags |= AI_ADDRCONFIG; +#endif + ai_hints.ai_family = AF_UNSPEC; + ai_hints.ai_socktype = SOCK_DGRAM; + ai_hints.ai_protocol = IPPROTO_UDP; + + status = getaddrinfo (se->node, + (se->service != NULL) ? se->service : NET_DEFAULT_PORT, + &ai_hints, &ai_list); + if (status != 0) + { + c_complain (LOG_ERR, &complaint, + "network plugin: getaddrinfo (%s, %s) failed: %s", + (se->node == NULL) ? "(null)" : se->node, + (se->service == NULL) ? "(null)" : se->service, + gai_strerror (status)); + return (-1); + } + else + { + c_release (LOG_NOTICE, &complaint, + "network plugin: Successfully resolved \"%s\".", + se->node); + } + + for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) + { + if (client->fd >= 0) /* when we reconnect */ + sockent_client_disconnect(se); + + client->fd = socket (ai_ptr->ai_family, + ai_ptr->ai_socktype, + ai_ptr->ai_protocol); + if (client->fd < 0) + { + char errbuf[1024]; + ERROR ("network plugin: socket(2) failed: %s", + sstrerror (errno, errbuf, + sizeof (errbuf))); + continue; + } + + client->addr = malloc (sizeof (*client->addr)); + if (client->addr == NULL) + { + ERROR ("network plugin: malloc failed."); + close (client->fd); + client->fd = -1; + continue; + } + + memset (client->addr, 0, sizeof (*client->addr)); + assert (sizeof (*client->addr) >= ai_ptr->ai_addrlen); + memcpy (client->addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen); + client->addrlen = ai_ptr->ai_addrlen; + + network_set_ttl (se, ai_ptr); + network_set_interface (se, ai_ptr); + + /* We don't open more than one write-socket per + * node/service pair.. */ + break; + } + + freeaddrinfo (ai_list); + if (client->fd < 0) + return (-1); + + if (client->resolve_interval > 0) + client->next_resolve_reconnect = now + client->resolve_interval; + return (0); +} /* }}} int sockent_client_connect */ + +/* Open the file descriptors for a initialized sockent structure. */ +static int sockent_server_listen (sockent_t *se) /* {{{ */ +{ + struct addrinfo ai_hints; + struct addrinfo *ai_list, *ai_ptr; + int status; + + const char *node; + const char *service; + + if (se == NULL) + return (-1); + node = se->node; service = se->service; if (service == NULL) service = NET_DEFAULT_PORT; - DEBUG ("network plugin: sockent_open: node = %s; service = %s;", + DEBUG ("network plugin: sockent_server_listen: node = %s; service = %s;", node, service); memset (&ai_hints, 0, sizeof (ai_hints)); @@ -1987,109 +2262,59 @@ static int sockent_open (sockent_t *se) /* {{{ */ ai_hints.ai_socktype = SOCK_DGRAM; ai_hints.ai_protocol = IPPROTO_UDP; - ai_return = getaddrinfo (node, service, &ai_hints, &ai_list); - if (ai_return != 0) + status = getaddrinfo (node, service, &ai_hints, &ai_list); + if (status != 0) { ERROR ("network plugin: getaddrinfo (%s, %s) failed: %s", (se->node == NULL) ? "(null)" : se->node, (se->service == NULL) ? "(null)" : se->service, - gai_strerror (ai_return)); + gai_strerror (status)); return (-1); } for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) { - int status; + int *tmp; - if (se->type == SOCKENT_TYPE_SERVER) /* {{{ */ + tmp = realloc (se->data.server.fd, + sizeof (*tmp) * (se->data.server.fd_num + 1)); + if (tmp == NULL) { - int *tmp; - - tmp = realloc (se->data.server.fd, - sizeof (*tmp) * (se->data.server.fd_num + 1)); - if (tmp == NULL) - { - ERROR ("network plugin: realloc failed."); - continue; - } - se->data.server.fd = tmp; - tmp = se->data.server.fd + se->data.server.fd_num; - - *tmp = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, - ai_ptr->ai_protocol); - if (*tmp < 0) - { - char errbuf[1024]; - ERROR ("network plugin: socket(2) failed: %s", - sstrerror (errno, errbuf, - sizeof (errbuf))); - continue; - } - - status = network_bind_socket (*tmp, ai_ptr, se->interface); - if (status != 0) - { - close (*tmp); - *tmp = -1; - continue; - } - - se->data.server.fd_num++; + ERROR ("network plugin: realloc failed."); continue; - } /* }}} if (se->type == SOCKENT_TYPE_SERVER) */ - else /* if (se->type == SOCKENT_TYPE_CLIENT) {{{ */ - { - se->data.client.fd = socket (ai_ptr->ai_family, - ai_ptr->ai_socktype, - ai_ptr->ai_protocol); - if (se->data.client.fd < 0) - { - char errbuf[1024]; - ERROR ("network plugin: socket(2) failed: %s", - sstrerror (errno, errbuf, - sizeof (errbuf))); - continue; - } - - se->data.client.addr = malloc (sizeof (*se->data.client.addr)); - if (se->data.client.addr == NULL) - { - ERROR ("network plugin: malloc failed."); - close (se->data.client.fd); - se->data.client.fd = -1; - continue; - } + } + se->data.server.fd = tmp; + tmp = se->data.server.fd + se->data.server.fd_num; - memset (se->data.client.addr, 0, sizeof (*se->data.client.addr)); - assert (sizeof (*se->data.client.addr) >= ai_ptr->ai_addrlen); - memcpy (se->data.client.addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen); - se->data.client.addrlen = ai_ptr->ai_addrlen; + *tmp = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, + ai_ptr->ai_protocol); + if (*tmp < 0) + { + char errbuf[1024]; + ERROR ("network plugin: socket(2) failed: %s", + sstrerror (errno, errbuf, + sizeof (errbuf))); + continue; + } - network_set_ttl (se, ai_ptr); - network_set_interface (se, ai_ptr); + status = network_bind_socket (*tmp, ai_ptr, se->interface); + if (status != 0) + { + close (*tmp); + *tmp = -1; + continue; + } - /* We don't open more than one write-socket per - * node/service pair.. */ - break; - } /* }}} if (se->type == SOCKENT_TYPE_CLIENT) */ + se->data.server.fd_num++; + continue; } /* for (ai_list) */ freeaddrinfo (ai_list); - /* Check if all went well. */ - if (se->type == SOCKENT_TYPE_SERVER) - { - if (se->data.server.fd_num <= 0) - return (-1); - } - else /* if (se->type == SOCKENT_TYPE_CLIENT) */ - { - if (se->data.client.fd < 0) - return (-1); - } - + if (se->data.server.fd_num <= 0) + return (-1); return (0); -} /* }}} int sockent_open */ +} /* }}} int sockent_server_listen */ /* Add a sockent to the global list of sockets */ static int sockent_add (sockent_t *se) /* {{{ */ @@ -2353,30 +2578,37 @@ static void network_init_buffer (void) memset (send_buffer, 0, network_config_packet_size); send_buffer_ptr = send_buffer; send_buffer_fill = 0; + send_buffer_last_update = 0; memset (&send_buffer_vl, 0, sizeof (send_buffer_vl)); } /* int network_init_buffer */ -static void networt_send_buffer_plain (const sockent_t *se, /* {{{ */ +static void networt_send_buffer_plain (sockent_t *se, /* {{{ */ const char *buffer, size_t buffer_size) { int status; while (42) { + status = sockent_client_connect (se); + if (status != 0) + return; + status = sendto (se->data.client.fd, buffer, buffer_size, - /* flags = */ 0, - (struct sockaddr *) se->data.client.addr, - se->data.client.addrlen); - if (status < 0) + /* flags = */ 0, + (struct sockaddr *) se->data.client.addr, + se->data.client.addrlen); + if (status < 0) { char errbuf[1024]; - if (errno == EINTR) + + if ((errno == EINTR) || (errno == EAGAIN)) continue; - ERROR ("network plugin: sendto failed: %s", - sstrerror (errno, errbuf, - sizeof (errbuf))); - break; + + ERROR ("network plugin: sendto failed: %s. Closing sending socket.", + sstrerror (errno, errbuf, sizeof (errbuf))); + sockent_client_disconnect (se); + return; } break; @@ -2389,7 +2621,7 @@ static void networt_send_buffer_plain (const sockent_t *se, /* {{{ */ buffer_offset += (s); \ } while (0) -static void networt_send_buffer_signed (const sockent_t *se, /* {{{ */ +static void networt_send_buffer_signed (sockent_t *se, /* {{{ */ const char *in_buffer, size_t in_buffer_size) { part_signature_sha256_t ps; @@ -2583,16 +2815,15 @@ static int add_to_buffer (char *buffer, int buffer_size, /* {{{ */ if (vl_def->time != vl->time) { - time_t tmp = CDTIME_T_TO_TIME_T (vl->time); - if (write_part_number (&buffer, &buffer_size, TYPE_TIME, - (uint64_t) tmp)) + if (write_part_number (&buffer, &buffer_size, TYPE_TIME_HR, + (uint64_t) vl->time)) return (-1); vl_def->time = vl->time; } if (vl_def->interval != vl->interval) { - if (write_part_number (&buffer, &buffer_size, TYPE_INTERVAL, + if (write_part_number (&buffer, &buffer_size, TYPE_INTERVAL_HR, (uint64_t) vl->interval)) return (-1); vl_def->interval = vl->interval; @@ -2631,7 +2862,7 @@ static int add_to_buffer (char *buffer, int buffer_size, /* {{{ */ return (-1); sstrncpy (vl_def->type_instance, vl->type_instance, sizeof (vl_def->type_instance)); } - + if (write_part_values (&buffer, &buffer_size, ds, vl) != 0) return (-1); @@ -2687,6 +2918,7 @@ static int network_write (const data_set_t *ds, const value_list_t *vl, /* status == bytes added to the buffer */ send_buffer_fill += status; send_buffer_ptr += status; + send_buffer_last_update = cdtime(); stats_values_sent++; } @@ -2776,6 +3008,10 @@ static int network_config_set_ttl (const oconfig_item_t *ci) /* {{{ */ tmp = (int) ci->values[0].value.number; if ((tmp > 0) && (tmp <= 255)) network_config_ttl = tmp; + else { + WARNING ("network plugin: The `TimeToLive' must be between 1 and 255."); + return (-1); + } return (0); } /* }}} int network_config_set_ttl */ @@ -2886,13 +3122,12 @@ static int network_config_add_listen (const oconfig_item_t *ci) /* {{{ */ return (-1); } - se = malloc (sizeof (*se)); + se = sockent_create (SOCKENT_TYPE_SERVER); if (se == NULL) { - ERROR ("network plugin: malloc failed."); + ERROR ("network plugin: sockent_create failed."); return (-1); } - sockent_init (se, SOCKENT_TYPE_SERVER); se->node = strdup (ci->values[0].value.string); if (ci->values_num >= 2) @@ -2932,10 +3167,18 @@ static int network_config_add_listen (const oconfig_item_t *ci) /* {{{ */ } #endif /* HAVE_LIBGCRYPT */ - status = sockent_open (se); + status = sockent_init_crypto (se); + if (status != 0) + { + ERROR ("network plugin: network_config_add_listen: sockent_init_crypto() failed."); + sockent_destroy (se); + return (-1); + } + + status = sockent_server_listen (se); if (status != 0) { - ERROR ("network plugin: network_config_add_listen: sockent_open failed."); + ERROR ("network plugin: network_config_add_server: sockent_server_listen failed."); sockent_destroy (se); return (-1); } @@ -2966,13 +3209,12 @@ static int network_config_add_server (const oconfig_item_t *ci) /* {{{ */ return (-1); } - se = malloc (sizeof (*se)); + se = sockent_create (SOCKENT_TYPE_CLIENT); if (se == NULL) { - ERROR ("network plugin: malloc failed."); + ERROR ("network plugin: sockent_create failed."); return (-1); } - sockent_init (se, SOCKENT_TYPE_CLIENT); se->node = strdup (ci->values[0].value.string); if (ci->values_num >= 2) @@ -2995,6 +3237,8 @@ static int network_config_add_server (const oconfig_item_t *ci) /* {{{ */ if (strcasecmp ("Interface", child->key) == 0) network_config_set_interface (child, &se->interface); + else if (strcasecmp ("ResolveInterval", child->key) == 0) + cf_util_get_cdtime(child, &se->data.client.resolve_interval); else { WARNING ("network plugin: Option `%s' is not allowed here.", @@ -3015,14 +3259,17 @@ static int network_config_add_server (const oconfig_item_t *ci) /* {{{ */ } #endif /* HAVE_LIBGCRYPT */ - status = sockent_open (se); + status = sockent_init_crypto (se); if (status != 0) { - ERROR ("network plugin: network_config_add_server: sockent_open failed."); + ERROR ("network plugin: network_config_add_server: sockent_init_crypto() failed."); sockent_destroy (se); return (-1); } + /* No call to sockent_client_connect() here -- it is called from + * networt_send_buffer_plain(). */ + status = sockent_add (se); if (status != 0) { @@ -3038,6 +3285,14 @@ static int network_config (oconfig_item_t *ci) /* {{{ */ { int i; + /* The options need to be applied first */ + for (i = 0; i < ci->children_num; i++) + { + oconfig_item_t *child = ci->children + i; + if (strcasecmp ("TimeToLive", child->key) == 0) + network_config_set_ttl (child); + } + for (i = 0; i < ci->children_num; i++) { oconfig_item_t *child = ci->children + i; @@ -3046,8 +3301,9 @@ static int network_config (oconfig_item_t *ci) /* {{{ */ network_config_add_listen (child); else if (strcasecmp ("Server", child->key) == 0) network_config_add_server (child); - else if (strcasecmp ("TimeToLive", child->key) == 0) - network_config_set_ttl (child); + else if (strcasecmp ("TimeToLive", child->key) == 0) { + /* Handled earlier */ + } else if (strcasecmp ("MaxPacketSize", child->key) == 0) network_config_set_buffer_size (child); else if (strcasecmp ("Forward", child->key) == 0) @@ -3065,20 +3321,20 @@ static int network_config (oconfig_item_t *ci) /* {{{ */ } /* }}} int network_config */ static int network_notification (const notification_t *n, - user_data_t __attribute__((unused)) *user_data) + user_data_t __attribute__((unused)) *user_data) { char buffer[network_config_packet_size]; char *buffer_ptr = buffer; int buffer_free = sizeof (buffer); int status; - time_t tmp; - memset (buffer, '\0', sizeof (buffer)); + if (!check_send_notify_okay (n)) + return (0); + memset (buffer, 0, sizeof (buffer)); - tmp = CDTIME_T_TO_TIME_T (n->time); - status = write_part_number (&buffer_ptr, &buffer_free, TYPE_TIME, - (uint64_t) tmp); + status = write_part_number (&buffer_ptr, &buffer_free, TYPE_TIME_HR, + (uint64_t) n->time); if (status != 0) return (-1); @@ -3090,7 +3346,7 @@ static int network_notification (const notification_t *n, if (strlen (n->host) > 0) { status = write_part_string (&buffer_ptr, &buffer_free, TYPE_HOST, - n->host, strlen (n->host)); + n->host, strlen (n->host)); if (status != 0) return (-1); } @@ -3098,7 +3354,7 @@ static int network_notification (const notification_t *n, if (strlen (n->plugin) > 0) { status = write_part_string (&buffer_ptr, &buffer_free, TYPE_PLUGIN, - n->plugin, strlen (n->plugin)); + n->plugin, strlen (n->plugin)); if (status != 0) return (-1); } @@ -3106,8 +3362,8 @@ static int network_notification (const notification_t *n, if (strlen (n->plugin_instance) > 0) { status = write_part_string (&buffer_ptr, &buffer_free, - TYPE_PLUGIN_INSTANCE, - n->plugin_instance, strlen (n->plugin_instance)); + TYPE_PLUGIN_INSTANCE, + n->plugin_instance, strlen (n->plugin_instance)); if (status != 0) return (-1); } @@ -3115,7 +3371,7 @@ static int network_notification (const notification_t *n, if (strlen (n->type) > 0) { status = write_part_string (&buffer_ptr, &buffer_free, TYPE_TYPE, - n->type, strlen (n->type)); + n->type, strlen (n->type)); if (status != 0) return (-1); } @@ -3123,7 +3379,7 @@ static int network_notification (const notification_t *n, if (strlen (n->type_instance) > 0) { status = write_part_string (&buffer_ptr, &buffer_free, TYPE_TYPE_INSTANCE, - n->type_instance, strlen (n->type_instance)); + n->type_instance, strlen (n->type_instance)); if (status != 0) return (-1); } @@ -3140,6 +3396,8 @@ static int network_notification (const notification_t *n, static int network_shutdown (void) { + sockent_t *se; + listen_loop++; /* Kill the listening thread */ @@ -3170,7 +3428,9 @@ static int network_shutdown (void) sfree (send_buffer); - /* TODO: Close `sending_sockets' */ + for (se = sending_sockets; se != NULL; se = se->next) + sockent_client_disconnect (se); + sockent_destroy (sending_sockets); plugin_unregister_config ("network"); plugin_unregister_init ("network"); @@ -3182,15 +3442,15 @@ static int network_shutdown (void) static int network_stats_read (void) /* {{{ */ { - uint64_t copy_octets_rx; - uint64_t copy_octets_tx; - uint64_t copy_packets_rx; - uint64_t copy_packets_tx; - uint64_t copy_values_dispatched; - uint64_t copy_values_not_dispatched; - uint64_t copy_values_sent; - uint64_t copy_values_not_sent; - uint64_t copy_receive_list_length; + derive_t copy_octets_rx; + derive_t copy_octets_tx; + derive_t copy_packets_rx; + derive_t copy_packets_tx; + derive_t copy_values_dispatched; + derive_t copy_values_not_dispatched; + derive_t copy_values_sent; + derive_t copy_values_not_sent; + derive_t copy_receive_list_length; value_list_t vl = VALUE_LIST_INIT; value_t values[2]; @@ -3208,19 +3468,18 @@ static int network_stats_read (void) /* {{{ */ vl.values = values; vl.values_len = 2; vl.time = 0; - vl.interval = interval_g; sstrncpy (vl.host, hostname_g, sizeof (vl.host)); sstrncpy (vl.plugin, "network", sizeof (vl.plugin)); /* Octets received / sent */ - vl.values[0].counter = (counter_t) copy_octets_rx; - vl.values[1].counter = (counter_t) copy_octets_tx; + vl.values[0].derive = (derive_t) copy_octets_rx; + vl.values[1].derive = (derive_t) copy_octets_tx; sstrncpy (vl.type, "if_octets", sizeof (vl.type)); plugin_dispatch_values (&vl); /* Packets received / send */ - vl.values[0].counter = (counter_t) copy_packets_rx; - vl.values[1].counter = (counter_t) copy_packets_tx; + vl.values[0].derive = (derive_t) copy_packets_rx; + vl.values[1].derive = (derive_t) copy_packets_tx; sstrncpy (vl.type, "if_packets", sizeof (vl.type)); plugin_dispatch_values (&vl); @@ -3259,18 +3518,16 @@ static int network_stats_read (void) /* {{{ */ static int network_init (void) { - static _Bool have_init = false; + static _Bool have_init = 0; /* Check if we were already initialized. If so, just return - there's * nothing more to do (for now, that is). */ if (have_init) return (0); - have_init = true; + have_init = 1; #if HAVE_LIBGCRYPT - gcry_control (GCRYCTL_SET_THREAD_CBS, &gcry_threads_pthread); - gcry_control (GCRYCTL_INIT_SECMEM, 32768, 0); - gcry_control (GCRYCTL_INITIALIZATION_FINISHED, 0); + network_init_gcrypt (); #endif if (network_config_stats != 0) @@ -3304,7 +3561,7 @@ static int network_init (void) if (dispatch_thread_running == 0) { int status; - status = pthread_create (&dispatch_thread_id, + status = plugin_thread_create (&dispatch_thread_id, NULL /* no attributes */, dispatch_thread, NULL /* no argument */); @@ -3324,7 +3581,7 @@ static int network_init (void) if (receive_thread_running == 0) { int status; - status = pthread_create (&receive_thread_id, + status = plugin_thread_create (&receive_thread_id, NULL /* no attributes */, receive_thread, NULL /* no argument */); @@ -3344,22 +3601,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 __attribute__((unused)) *identifier, - user_data_t __attribute__((unused)) *user_data) +static int network_flush (cdtime_t timeout, + __attribute__((unused)) const char *identifier, + __attribute__((unused)) user_data_t *user_data) { pthread_mutex_lock (&send_buffer_lock); if (send_buffer_fill > 0) - flush_buffer (); - + { + if (timeout > 0) + { + cdtime_t now = cdtime (); + if ((send_buffer_last_update + timeout) > now) + { + pthread_mutex_unlock (&send_buffer_lock); + return (0); + } + } + flush_buffer (); + } pthread_mutex_unlock (&send_buffer_lock); return (0);