X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fnetwork.c;h=cf2a01799cbb1723c710357be176810aac053fb6;hb=d561a8387f633491bdcd7fe6964f0b63ae694af5;hp=f42693e1c494f90359b1877c4148c8e4a5fef45e;hpb=66eb6c7178d61098864aa1dd3fb5e10c2afb67ee;p=collectd.git diff --git a/src/network.c b/src/network.c index f42693e1..cf2a0179 100644 --- a/src/network.c +++ b/src/network.c @@ -1,11 +1,10 @@ /** * collectd - src/network.c - * Copyright (C) 2006 Florian octo Forster + * Copyright (C) 2005-2007 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 - * Free Software Foundation; either version 2 of the License, or (at your - * option) any later version. + * Free Software Foundation; only version 2 of the License is applicable. * * This program is distributed in the hope that it will be useful, but * WITHOUT ANY WARRANTY; without even the implied warranty of @@ -20,47 +19,49 @@ * Florian octo Forster **/ -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include "network.h" +#include "collectd.h" +#include "plugin.h" #include "common.h" +#include "configfile.h" +#include "utils_debug.h" -/* - * From RFC2365: - * - * The IPv4 Organization Local Scope -- 239.192.0.0/14 - * - * 239.192.0.0/14 is defined to be the IPv4 Organization Local Scope, and is - * the space from which an organization should allocate sub-ranges when - * defining scopes for private use. - * - * Port 25826 is not assigned as of 2005-09-12 - */ +#include "network.h" -#define IPV4_MCAST_GROUP "239.192.74.66" -#define UDP_PORT 25826 +#if HAVE_PTHREAD_H +# include +#endif +#if HAVE_SYS_SOCKET_H +# include +#endif +#if HAVE_NETDB_H +# include +#endif +#if HAVE_NETINET_IN_H +# include +#endif +#if HAVE_ARPA_INET_H +# include +#endif +#if HAVE_POLL_H +# include +#endif /* 1500 - 40 - 8 = Ethernet packet - IPv6 header - UDP header */ -#define BUFF_SIZE 1452 +/* #define BUFF_SIZE 1452 */ -#define BUFF_SIZE 4096 +#ifndef IPV6_ADD_MEMBERSHIP +# ifdef IPV6_JOIN_GROUP +# define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP +# else +# error "Neither IP_ADD_MEMBERSHIP nor IPV6_JOIN_GROUP is defined" +# endif +#endif /* !IP_ADD_MEMBERSHIP */ -#ifdef HAVE_LIBRRD -extern int operating_mode; -#else -static int operating_mode = MODE_CLIENT; -#endif +#define BUFF_SIZE 1024 +/* + * Private data types + */ typedef struct sockent { int fd; @@ -69,13 +70,472 @@ typedef struct sockent struct sockent *next; } sockent_t; -static sockent_t *socklist_head = NULL; +/* 1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3 + * 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + * +-------+-----------------------+-------------------------------+ + * ! Ver. ! ! Length ! + * +-------+-----------------------+-------------------------------+ + */ +struct part_header_s +{ + uint16_t type; + uint16_t length; +}; +typedef struct part_header_s part_header_t; + +/* 1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3 + * 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + * +-------------------------------+-------------------------------+ + * ! Type ! Length ! + * +-------------------------------+-------------------------------+ + * : (Length - 4) Bytes : + * +---------------------------------------------------------------+ + */ +struct part_string_s +{ + part_header_t *head; + char *value; +}; +typedef struct part_string_s part_string_t; + +/* 1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3 + * 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + * +-------------------------------+-------------------------------+ + * ! Type ! Length ! + * +-------------------------------+-------------------------------+ + * : (Length - 4 == 2 || 4 || 8) Bytes : + * +---------------------------------------------------------------+ + */ +struct part_number_s +{ + part_header_t *head; + uint64_t *value; +}; +typedef struct part_number_s part_number_t; + +/* 1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3 + * 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 + * +-------------------------------+-------------------------------+ + * ! Type ! Length ! + * +-------------------------------+---------------+---------------+ + * ! Num of values ! Type0 ! Type1 ! + * +-------------------------------+---------------+---------------+ + * ! Value0 ! + * ! ! + * +---------------------------------------------------------------+ + * ! Value1 ! + * ! ! + * +---------------------------------------------------------------+ + */ +struct part_values_s +{ + part_header_t *head; + uint16_t *num_values; + uint8_t *values_types; + value_t *values; +}; +typedef struct part_values_s part_values_t; + +/* + * Private variables + */ +static const char *config_keys[] = +{ + "Listen", + "Server", + "TimeToLive", + NULL +}; +static int config_keys_num = 3; + +static int network_config_ttl = 0; + +static sockent_t *sending_sockets = NULL; + +static struct pollfd *listen_sockets = NULL; +static int listen_sockets_num = 0; +static pthread_t listen_thread = 0; +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 char send_buffer_type[DATA_MAX_NAME_LEN]; + +/* + * Private functions + */ +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; + int i; + + i = 6 + (9 * vl->values_len); + if (*ret_buffer_len < i) + 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); + + 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++) + { + if (ds->ds[i].type == DS_TYPE_COUNTER) + { + pv.values_types[i] = DS_TYPE_COUNTER; + pv.values[i].counter = htonll (vl->values[i].counter); + } + else + { + pv.values_types[i] = DS_TYPE_GAUGE; + pv.values[i].gauge = vl->values[i].gauge; + } + } /* for (values) */ + + return (0); +} /* int write_part_values */ + +static int write_part_number (char **ret_buffer, int *ret_buffer_len, + int type, uint64_t value) +{ + part_number_t pn; + + if (*ret_buffer_len < 12) + return (-1); + + pn.head = (part_header_t *) *ret_buffer; + pn.value = (uint64_t *) (pn.head + 1); + + pn.head->type = htons (type); + pn.head->length = htons (12); + *pn.value = htonll (value); + + *ret_buffer = (char *) (pn.value + 1); + *ret_buffer_len -= 12; + + return (0); +} /* int write_part_number */ + +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; + + len = 4 + str_len + 1; + if (*ret_buffer_len < len) + return (-1); + *ret_buffer_len -= len; + + ps.head = (part_header_t *) *ret_buffer; + ps.value = (char *) (ps.head + 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)); + + return (0); +} /* int write_part_string */ + +static int parse_part_values (void **ret_buffer, int *ret_buffer_len, + value_t **ret_values, int *ret_num_values) +{ + char *buffer = *ret_buffer; + int buffer_len = *ret_buffer_len; + part_values_t pv; + int i; + + uint16_t h_length; + uint16_t h_type; + uint16_t h_num; + + if (buffer_len < (15)) + { + DBG ("packet is too short: buffer_len = %i", buffer_len); + return (-1); + } + + pv.head = (part_header_t *) buffer; + h_length = ntohs (pv.head->length); + h_type = ntohs (pv.head->type); + + assert (h_type == TYPE_VALUES); + + pv.num_values = (uint16_t *) (pv.head + 1); + h_num = ntohs (*pv.num_values); + + if (h_num != ((h_length - 6) / 9)) + { + DBG ("`length' and `num of values' don't match"); + return (-1); + } + + pv.values_types = (uint8_t *) (pv.num_values + 1); + pv.values = (value_t *) (pv.values_types + h_num); + + for (i = 0; i < h_num; i++) + if (pv.values_types[i] == DS_TYPE_COUNTER) + pv.values[i].counter = ntohll (pv.values[i].counter); + + *ret_buffer = (void *) (pv.values + h_num); + *ret_buffer_len = buffer_len - h_length; + *ret_num_values = h_num; + *ret_values = pv.values; + + return (0); +} /* int parse_part_values */ + +static int parse_part_number (void **ret_buffer, int *ret_buffer_len, + uint64_t *value) +{ + part_number_t pn; + uint16_t len; + + pn.head = (part_header_t *) *ret_buffer; + pn.value = (uint64_t *) (pn.head + 1); + + len = ntohs (pn.head->length); + if (len != 12) + return (-1); + if (len > *ret_buffer_len) + return (-1); + *value = ntohll (*pn.value); + + *ret_buffer = (void *) (pn.value + 1); + *ret_buffer_len -= len; + + return (0); +} /* int parse_part_number */ + +static int parse_part_string (void **ret_buffer, int *ret_buffer_len, + char *output, int output_len) +{ + char *buffer = *ret_buffer; + int buffer_len = *ret_buffer_len; + part_string_t ps; + + uint16_t h_length; + uint16_t h_type; + + DBG ("ret_buffer = %p; ret_buffer_len = %i; output = %p; output_len = %i;", + *ret_buffer, *ret_buffer_len, + (void *) output, output_len); + + ps.head = (part_header_t *) buffer; + + h_length = ntohs (ps.head->length); + h_type = ntohs (ps.head->type); + + DBG ("length = %hu; type = %hu;", h_length, h_type); + + if (buffer_len < h_length) + { + DBG ("packet is too short"); + 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') + { + DBG ("String does not end with a nullbyte"); + return (-1); + } + + if (output_len < (h_length - 4)) + { + DBG ("output buffer is too small"); + return (-1); + } + strcpy (output, ps.value); + + DBG ("output = %s", output); + + *ret_buffer = (void *) (buffer + h_length); + *ret_buffer_len = buffer_len - h_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]; + + DBG ("buffer = %p; buffer_len = %i;", buffer, buffer_len); + + memset (&vl, '\0', sizeof (vl)); + memset (&type, '\0', sizeof (type)); + status = 0; + + while ((status == 0) && (buffer_len > sizeof (part_header_t))) + { + header = (part_header_t *) buffer; + + if (ntohs (header->length) > buffer_len) + break; + + if (header->type == htons (TYPE_VALUES)) + { + status = parse_part_values (&buffer, &buffer_len, + &vl.values, &vl.values_len); + + if (status != 0) + { + DBG ("parse_part_values failed."); + break; + } + + if ((vl.time > 0) + && (strlen (vl.host) > 0) + && (strlen (vl.plugin) > 0) + && (strlen (type) > 0)) + { + DBG ("dispatching values"); + plugin_dispatch_values (type, &vl); + } + else + { + DBG ("NOT dispatching values"); + } + } + else if (header->type == ntohs (TYPE_TIME)) + { + uint64_t tmp = 0; + status = parse_part_number (&buffer, &buffer_len, &tmp); + if (status == 0) + vl.time = (time_t) tmp; + } + else if (header->type == ntohs (TYPE_HOST)) + { + status = parse_part_string (&buffer, &buffer_len, + vl.host, sizeof (vl.host)); + } + else if (header->type == ntohs (TYPE_PLUGIN)) + { + status = parse_part_string (&buffer, &buffer_len, + vl.plugin, sizeof (vl.plugin)); + } + else if (header->type == ntohs (TYPE_PLUGIN_INSTANCE)) + { + status = parse_part_string (&buffer, &buffer_len, + vl.plugin_instance, sizeof (vl.plugin_instance)); + } + else if (header->type == ntohs (TYPE_TYPE)) + { + status = parse_part_string (&buffer, &buffer_len, + type, sizeof (type)); + } + else if (header->type == ntohs (TYPE_TYPE_INSTANCE)) + { + status = parse_part_string (&buffer, &buffer_len, + vl.type_instance, sizeof (vl.type_instance)); + } + else + { + DBG ("Unknown part type: 0x%0hx", header->type); + buffer = ((char *) buffer) + header->length; + } + } /* while (buffer_len > sizeof (part_header_t)) */ + + return (0); +} /* int parse_packet */ + +static void free_sockent (sockent_t *se) +{ + sockent_t *next; + while (se != NULL) + { + next = se->next; + free (se->addr); + free (se); + se = next; + } +} /* void free_sockent */ + +/* + * int network_set_ttl + * + * Set the `IP_MULTICAST_TTL', `IP_TTL', `IPV6_MULTICAST_HOPS' or + * `IPV6_UNICAST_HOPS', depending on which option is applicable. + * + * The `struct addrinfo' is used to destinguish between unicast and multicast + * sockets. + */ +static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai) +{ + if ((network_config_ttl < 1) || (network_config_ttl > 255)) + return (-1); + + DBG ("ttl = %i", network_config_ttl); + + if (ai->ai_family == AF_INET) + { + struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr; + int optname; + + if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr))) + optname = IP_MULTICAST_TTL; + else + optname = IP_TTL; + + if (setsockopt (se->fd, IPPROTO_IP, optname, + &network_config_ttl, + sizeof (network_config_ttl)) == -1) + { + syslog (LOG_ERR, "setsockopt: %s", strerror (errno)); + return (-1); + } + } + else if (ai->ai_family == AF_INET6) + { + /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */ + struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr; + int optname; + + if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr)) + optname = IPV6_MULTICAST_HOPS; + else + optname = IPV6_UNICAST_HOPS; -static int network_bind_socket (int fd, const struct addrinfo *ai, const sockent_t *se) + if (setsockopt (se->fd, IPPROTO_IPV6, optname, + &network_config_ttl, + sizeof (network_config_ttl)) == -1) + { + syslog (LOG_ERR, "setsockopt: %s", strerror (errno)); + return (-1); + } + } + + return (0); +} /* int network_set_ttl */ + +static int network_bind_socket (const sockent_t *se, const struct addrinfo *ai) { int loop = 1; - if (bind (fd, ai->ai_addr, ai->ai_addrlen) == -1) + DBG ("fd = %i; calling `bind'", se->fd); + + if (bind (se->fd, ai->ai_addr, ai->ai_addrlen) == -1) { syslog (LOG_ERR, "bind: %s", strerror (errno)); return (-1); @@ -88,6 +548,8 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const sockent { struct ip_mreq mreq; + DBG ("fd = %i; IPv4 multicast address found", se->fd); + mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr; mreq.imr_interface.s_addr = htonl (INADDR_ANY); @@ -114,6 +576,8 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const sockent { struct ipv6_mreq mreq; + DBG ("fd = %i; IPv6 multicast address found", se->fd); + memcpy (&mreq.ipv6mr_multiaddr, &addr->sin6_addr, sizeof (addr->sin6_addr)); @@ -127,7 +591,7 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const sockent * single interface; programs running on * multihomed hosts may need to join the same * group on more than one interface.*/ - mreq6.ipv6mr_interface = 0; + mreq.ipv6mr_interface = 0; if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP, &loop, sizeof (loop)) == -1) @@ -146,40 +610,43 @@ static int network_bind_socket (int fd, const struct addrinfo *ai, const sockent } return (0); -} +} /* int network_bind_socket */ -int network_create_socket (const char *node, const char *service) +static sockent_t *network_create_socket (const char *node, + const char *service, + int listen) { - sockent_t *socklist_tail; - struct addrinfo ai_hints; struct addrinfo *ai_list, *ai_ptr; int ai_return; - int num_added = 0; + sockent_t *se_head = NULL; + sockent_t *se_tail = NULL; DBG ("node = %s, service = %s", node, service); - if (operating_mode == MODE_LOCAL) - return (-1); - - socklist_tail = socklist_head; - while ((socklist_tail != NULL) && (socklist_tail->next != NULL)) - socklist_tail = socklist_tail->next; - memset (&ai_hints, '\0', sizeof (ai_hints)); - ai_hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG; - ai_hints.ai_family = PF_UNSPEC; + ai_hints.ai_flags = 0; +#ifdef AI_PASSIVE + ai_hints.ai_flags |= AI_PASSIVE; +#endif +#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; /* XXX is this right here?!? */ + ai_hints.ai_protocol = IPPROTO_UDP; - if ((ai_return = getaddrinfo (node, service, &ai_hints, &ai_list)) != 0) + ai_return = getaddrinfo (node, service, &ai_hints, &ai_list); + if (ai_return != 0) { syslog (LOG_ERR, "getaddrinfo (%s, %s): %s", - node == NULL ? "(null)" : node, - service == NULL ? "(null)" : service, - ai_return == EAI_SYSTEM ? strerror (errno) : gai_strerror (ai_return)); - return (-1); + (node == NULL) ? "(null)" : node, + (service == NULL) ? "(null)" : service, + (ai_return == EAI_SYSTEM) + ? strerror (errno) + : gai_strerror (ai_return)); + return (NULL); } for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next) @@ -199,12 +666,14 @@ int network_create_socket (const char *node, const char *service) continue; } - assert (sizeof (struct sockaddr_storage) >= ai_ptr->addrlen); + assert (sizeof (struct sockaddr_storage) >= ai_ptr->ai_addrlen); memset (se->addr, '\0', sizeof (struct sockaddr_storage)); - memcpy (se->addr, ai_ptr->ai_addr, ai_ptr->addrlen); - se->addrlen = ai_ptr->addrlen; + memcpy (se->addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen); + se->addrlen = ai_ptr->ai_addrlen; - se->fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol); + se->fd = socket (ai_ptr->ai_family, + ai_ptr->ai_socktype, + ai_ptr->ai_protocol); se->next = NULL; if (se->fd == -1) @@ -215,222 +684,427 @@ int network_create_socket (const char *node, const char *service) continue; } - if (operating_mode == MODE_SERVER) - if (network_bind_socket (se->fd, ai_ptr, se->addr) != 0) + if (listen != 0) + { + if (network_bind_socket (se, ai_ptr) != 0) { + close (se->fd); free (se->addr); free (se); continue; } + } + else /* listen == 0 */ + { + network_set_ttl (se, ai_ptr); + } - if (socklist_tail == NULL) + if (se_tail == NULL) { - socklist_head = socklist_tail = se; + se_head = se; + se_tail = se; } else { - socklist_tail->next = se; - socklist_tail = se; + se_tail->next = se; + se_tail = se; } - num_added++; - /* We don't open more than one write-socket per node/service pair.. */ - if (operating_mode == MODE_CLIENT) + if (listen == 0) break; } freeaddrinfo (ai_list); - return (num_added); -} + return (se_head); +} /* sockent_t *network_create_socket */ -static int network_get_listen_socket (void) +static sockent_t *network_create_default_socket (int listen) { - int fd; - int max_fd; + sockent_t *se_ptr = NULL; + sockent_t *se_head = NULL; + sockent_t *se_tail = NULL; + + se_ptr = network_create_socket (NET_DEFAULT_V6_ADDR, + NET_DEFAULT_PORT, listen); + + /* Don't send to the same machine in IPv6 and IPv4 if both are available. */ + if ((listen == 0) && (se_ptr != NULL)) + return (se_ptr); + + if (se_ptr != NULL) + { + se_head = se_ptr; + se_tail = se_ptr; + while (se_tail->next != NULL) + se_tail = se_tail->next; + } + + se_ptr = network_create_socket (NET_DEFAULT_V4_ADDR, NET_DEFAULT_PORT, listen); - fd_set readfds; + if (se_tail == NULL) + return (se_ptr); + + se_tail->next = se_ptr; + return (se_head); +} /* sockent_t *network_create_default_socket */ + +static int network_add_listen_socket (const char *node, const char *service) +{ sockent_t *se; + sockent_t *se_ptr; + int se_num = 0; - while (1) + if (service == NULL) + service = NET_DEFAULT_PORT; + + if (node == NULL) + se = network_create_default_socket (1 /* listen == true */); + else + se = network_create_socket (node, service, 1 /* listen == true */); + + if (se == NULL) + return (-1); + + for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next) + se_num++; + + listen_sockets = (struct pollfd *) realloc (listen_sockets, + (listen_sockets_num + se_num) + * sizeof (struct pollfd)); + + for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next) { - FD_ZERO (&readfds); - max_fd = -1; - for (se = socklist_head; se != NULL; se = se->next) - { - FD_SET (se->fd, &readfds); - if (se->fd >= max_fd) - max_fd = se->fd + 1; - } + listen_sockets[listen_sockets_num].fd = se_ptr->fd; + listen_sockets[listen_sockets_num].events = POLLIN | POLLPRI; + listen_sockets[listen_sockets_num].revents = 0; + listen_sockets_num++; + } /* for (se) */ - if (max_fd == -1) - { - syslog (LOG_WARNING, "No listen sockets found!"); - return (-1); - } + free_sockent (se); + return (0); +} /* int network_add_listen_socket */ + +static int network_add_sending_socket (const char *node, const char *service) +{ + sockent_t *se; + sockent_t *se_ptr; - status = select (max_fd, &readfds, NULL, NULL, NULL); + if (service == NULL) + service = NET_DEFAULT_PORT; - if ((status == -1) && (errno == EINTR)) - continue; - else if (status == -1) + if (node == NULL) + se = network_create_default_socket (0 /* listen == false */); + else + se = network_create_socket (node, service, 0 /* listen == false */); + + if (se == NULL) + return (-1); + + if (sending_sockets == NULL) + { + sending_sockets = se; + return (0); + } + + for (se_ptr = sending_sockets; se_ptr->next != NULL; se_ptr = se_ptr->next) + /* seek end */; + + se_ptr->next = se; + return (0); +} /* int network_get_listen_socket */ + +int network_receive (void) +{ + char buffer[BUFF_SIZE]; + int buffer_len; + + int i; + int status; + + if (listen_sockets_num == 0) + network_add_listen_socket (NULL, NULL); + + if (listen_sockets_num == 0) + { + syslog (LOG_ERR, "network: Failed to open a listening socket."); + return (-1); + } + + while (listen_loop == 0) + { + status = poll (listen_sockets, listen_sockets_num, -1); + + if (status <= 0) { - syslog (LOG_ERR, "select: %s", strerror (errno)); + if (errno == EINTR) + continue; + syslog (LOG_ERR, "poll failed: %s", + strerror (errno)); return (-1); } - else - break; - } /* while (true) */ - fd = -1; - for (se = socklist_head; se != NULL; se = se->next) - if (FD_ISSET (se->fd, &readfds)) + for (i = 0; (i < listen_sockets_num) && (status > 0); i++) { - fd = se->fd; - break; - } + if ((listen_sockets[i].revents & (POLLIN | POLLPRI)) == 0) + continue; + status--; - if (fd == -1) - syslog (LOG_WARNING, "No socket ready..?"); + buffer_len = recv (listen_sockets[i].fd, + buffer, sizeof (buffer), + 0 /* no flags */); + if (buffer_len < 0) + { + syslog (LOG_ERR, "recv failed: %s", strerror (errno)); + return (-1); + } + + parse_packet (buffer, buffer_len); + } /* for (listen_sockets) */ + } /* while (listen_loop == 0) */ - DBG ("fd = %i", fd); - return (fd); + return (0); } -int network_receive (char **host, char **type, char **inst, char **value) +static void *receive_thread (void *arg) { - int fd; - char buffer[BUFF_SIZE]; + return (network_receive () ? (void *) 1 : (void *) 0); +} /* void *receive_thread */ - struct sockaddr_storage addr; +static void network_send_buffer (const char *buffer, int buffer_len) +{ + sockent_t *se; int status; - char *fields[4]; - - assert (operating_mode == MODE_SERVER); + DBG ("buffer_len = %i", buffer_len); - *host = NULL; - *type = NULL; - *inst = NULL; - *value = NULL; + for (se = sending_sockets; se != NULL; se = se->next) + { + while (42) + { + status = sendto (se->fd, buffer, buffer_len, 0 /* no flags */, + (struct sockaddr *) se->addr, se->addrlen); + if (status < 0) + { + if (errno == EINTR) + continue; + syslog (LOG_ERR, "network plugin: sendto failed: %s", + strerror (errno)); + break; + } - if ((fd = network_get_listen_socket ()) < 0) - return (-1); + break; + } /* while (42) */ + } /* for (sending_sockets) */ +} /* void network_send_buffer */ - if (recvfrom (fd, buffer, BUFF_SIZE, 0, (struct sockaddr *) &addr, sizeof (addr)) == -1) +static int add_to_buffer (char *buffer, int buffer_size, + value_list_t *vl_def, char *type_def, + const data_set_t *ds, const value_list_t *vl) +{ + if (strcmp (vl_def->host, vl->host) != 0) { - syslog (LOG_ERR, "recvfrom: %s", strerror (errno)); - return (-1); + if (write_part_string (&buffer, &buffer_size, TYPE_HOST, + vl->host, strlen (vl->host)) != 0) + return (-1); + strcpy (vl_def->host, vl->host); + DBG ("host = %s", vl->host); } - if ((*host = (char *) malloc (BUFF_SIZE)) == NULL) + if (vl_def->time != vl->time) { - syslog (LOG_EMERG, "malloc: %s", strerror (errno)); - return (-1); + if (write_part_number (&buffer, &buffer_size, TYPE_TIME, + (uint64_t) vl->time)) + return (-1); + vl_def->time = vl->time; + DBG ("time = %u", (unsigned int) vl->time); } - status = getnameinfo ((struct sockaddr *) &addr, sizeof (addr), - *host, BUFF_SIZE, NULL, 0, 0); - if (status != 0) + if (strcmp (vl_def->plugin, vl->plugin) != 0) { - free (*host); *host = NULL; - syslog (LOG_ERR, "getnameinfo: %s", - status == EAI_SYSTEM ? strerror (errno) : gai_strerror (status)); - return (-1); + if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN, + vl->plugin, strlen (vl->plugin)) != 0) + return (-1); + strcpy (vl_def->plugin, vl->plugin); + DBG ("plugin = %s", vl->plugin); } - if (strsplit (buffer, fields, 4) != 3) + if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0) { - syslog (LOG_WARNING, "Invalid message from `%s'", *host); - free (*host); *host = NULL; - return (-1); + if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN_INSTANCE, + vl->plugin_instance, + strlen (vl->plugin_instance)) != 0) + return (-1); + strcpy (vl_def->plugin_instance, vl->plugin_instance); + DBG ("plugin_instance = %s", vl->plugin_instance); } - if ((*type = strdup (fields[0])) == NULL) + if (strcmp (type_def, ds->type) != 0) { - syslog (LOG_EMERG, "strdup: %s", strerror ()); - free (*host); *host = NULL; - return (-1); + if (write_part_string (&buffer, &buffer_size, TYPE_TYPE, + ds->type, strlen (ds->type)) != 0) + return (-1); + strcpy (type_def, ds->type); + DBG ("type = %s", ds->type); } - if ((*inst = strdup (fields[1])) == NULL) + if (strcmp (vl_def->type_instance, vl->type_instance) != 0) { - syslog (LOG_EMERG, "strdup: %s", strerror ()); - free (*host); *host = NULL; - free (*type); *type = NULL; + if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN_INSTANCE, + vl->type_instance, + strlen (vl->type_instance)) != 0) + return (-1); + strcpy (vl_def->type_instance, vl->type_instance); + DBG ("type_instance = %s", vl->type_instance); + } + + if (write_part_values (&buffer, &buffer_size, ds, vl) != 0) return (-1); + + return (buffer_size); +} /* int add_to_buffer */ + +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)); +} + +static int network_write (const data_set_t *ds, const value_list_t *vl) +{ + int status; + /* TODO: lock buffer */ + status = add_to_buffer (send_buffer_ptr, + sizeof (send_buffer) - send_buffer_fill, + &send_buffer_vl, send_buffer_type, + ds, vl); + if (status >= 0) + { + send_buffer_fill += status; + send_buffer_ptr += status; + } + else + { + flush_buffer (); + + status = add_to_buffer (send_buffer_ptr, + sizeof (send_buffer) - send_buffer_fill, + &send_buffer_vl, send_buffer_type, + ds, vl); + + if (status >= 0) + { + send_buffer_fill += status; + send_buffer_ptr += status; + } } - if ((*value = strdup (fields[2])) == NULL) + if (status < 0) { - syslog (LOG_EMERG, "strdup: %s", strerror ()); - free (*host); *host = NULL; - free (*type); *type = NULL; - free (*inst); *inst = NULL; - return (-1); + syslog (LOG_ERR, "network plugin: Unable to append to the " + "buffer for some weird reason"); + } + else if ((sizeof (send_buffer) - send_buffer_fill) < 15) + { + flush_buffer (); } + /* TODO: unlock buffer */ - DBG ("host = %s, type = %s, inst = %s, value = %s", - *host, *type, *inst, *value); + return ((status < 0) ? -1 : 0); +} /* int network_write */ + +static int network_config (const char *key, const char *val) +{ + char *node; + char *service; + char *fields[3]; + int fields_num; + + if ((strcasecmp ("Listen", key) == 0) + || (strcasecmp ("Server", key) == 0)) + { + char *val_cpy = strdup (val); + if (val_cpy == NULL) + return (1); + + service = NET_DEFAULT_PORT; + fields_num = strsplit (val_cpy, fields, 3); + if ((fields_num != 1) + && (fields_num != 2)) + return (1); + else if (fields_num == 2) + service = fields[1]; + node = fields[0]; + + if (strcasecmp ("Listen", key) == 0) + network_add_listen_socket (node, service); + else + network_add_sending_socket (node, service); + } + else if (strcasecmp ("TimeToLive", key) == 0) + { + int tmp = atoi (val); + if ((tmp > 0) && (tmp < 256)) + network_config_ttl = tmp; + else + return (1); + } + else + { + return (-1); + } return (0); } -int network_send (char *type, char *inst, char *value) +static int network_shutdown (void) { - char buf[BUFF_SIZE]; - int buflen; + DBG ("Shutting down."); - sockent_t *se; + listen_loop++; - int ret; - int status; + pthread_kill (listen_thread, SIGTERM); + pthread_join (listen_thread, NULL /* no return value */); - DBG ("type = %s, inst = %s, value = %s", type, inst, value); + listen_thread = 0; - assert (operating_mode == MODE_CLIENT); + return (0); +} - buflen = snprintf (buf, BUFF_SIZE, "%s %s %s", type, inst, value); - if ((buflen >= BUFF_SIZE) || (buflen < 1)) - { - syslog (LOG_WARNING, "network_send: snprintf failed.."); - return (-1); - } - buf[buflen] = '\0'; - buflen++; +static int network_init (void) +{ + 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)); - ret = 0; - for (se = socklist_head; se != NULL; se = se->next) + /* setup socket(s) and so on */ + if (sending_sockets != NULL) + plugin_register_write ("network", network_write); + + if ((listen_sockets_num != 0) && (listen_thread == 0)) { - DBG ("fd = %i", se->fd); + int status; - while (1) - { - status = sendto (se->fd, buf, buflen, 0, - (struct sockaddr *) se->addr, se->addrlen); + status = pthread_create (&listen_thread, NULL /* no attributes */, + receive_thread, NULL /* no argument */); - if (status == -1) - { - if (errno == EINTR) - { - DBG ("sendto was interrupted"); - continue; - } - else - { - syslog (LOG_ERR, "sendto: %s", strerror (errno)); - break; - } - } - else if (ret >= 0) - ret++; - break; - } + if (status != 0) + syslog (LOG_ERR, "network: pthread_create failed: %s", + strerror (errno)); } + return (0); +} /* int network_init */ - if (ret == 0) - syslog (LOG_WARNING, "Message wasn't sent to anybody.."); - - return (ret); +void module_register (void) +{ + plugin_register_config ("network", network_config, + config_keys, config_keys_num); + plugin_register_init ("network", network_init); }