Merge branch 'forwarding_udp' into devel

Author: Lukas955

base/src/storage/forwarding/configuration.c

@@ -49,10 +49,14 @@
 
 /** Default destination port                 */
 #define DEF_PORT "4739"
-/** Default retry interval (seconds)         */
-#define DEF_RETRY_INT (5)
+/** Default transport protocol               */
+#define DEF_PROTO IPPROTO_TCP
+/** Default re-connection period (in milliseconds)*/
+#define DEF_RECONN_PERIOD (1000)
 /** Default maximal packet size              */
 #define DEF_PACKET_SIZE (4096)
+/** Default template refresh timeout         */
+#define DEF_TEMPLATE_REFRESH (300U)
 
 static const char *msg_module = "forwarding(config)";
 
@@ -134,6 +138,26 @@ static enum DIST_MODE config_parse_distr(const char *str)
 	}
 }
 
+/**
+ * \brief Convert string to transport protocol
+ * \param[in] str String
+ * \return Return either parsed value or default protocol
+ */
+static int config_parse_proto(const char *str)
+{
+	if (!str) {
+		return DEF_PROTO;
+	}
+
+	if (!strcasecmp(str, "tcp")) {
+		return IPPROTO_TCP;
+	} else if (!strcasecmp(str, "udp")) {
+		return IPPROTO_UDP;
+	} else {
+		return DEF_PROTO;
+	}
+}
+
 /**
  * \brief Parse only default values from the plugin configuration
  * \param[in,out] ctx Parser context
@@ -145,12 +169,21 @@ static int config_parse_def_values(struct parser_context *ctx)
 	xmlNode *cur = ctx->node->children;
 	struct plugin_config *cfg = ctx->cfg;
 
+	// Set default protocol
+	cfg->def_proto = DEF_PROTO;
+
 	while (cur) {
 		if (!xmlStrcasecmp(cur->name, (const xmlChar *) "defaultPort")) {
 			// Default port
 			free(cfg->def_port);
 			cfg->def_port = (char *) xmlNodeListGetString(doc,
 				cur->xmlChildrenNode, 1);
+		} else if (!xmlStrcasecmp(cur->name, (const xmlChar *) "defaultProtocol")) {
+			// Default protocol
+			char *str_proto = (char *) xmlNodeListGetString(doc,
+				cur->xmlChildrenNode, 1);
+			cfg->def_proto = config_parse_proto(str_proto);
+			free(str_proto);
 		} else {
 			// Other nodes -> skip
 		}
@@ -195,6 +228,7 @@ static fwd_sender_t *config_parse_destination(struct parser_context *ctx)
 {
 	xmlChar *str_ip = NULL;
 	xmlChar *str_port = NULL;
+	xmlChar *str_proto = NULL;
 	xmlNode *cur = ctx->node;
 
 	// Find all related XML nodes & parse them
@@ -213,6 +247,12 @@ static fwd_sender_t *config_parse_destination(struct parser_context *ctx)
 				xmlFree(str_port);
 			}
 			str_port = xmlNodeListGetString(ctx->doc, cur->xmlChildrenNode, 1);
+		} else if (!xmlStrcasecmp(cur->name, (const xmlChar *) "protocol")) {
+			// Destination protocol
+			if (str_proto) {
+				xmlFree(str_proto);
+			}
+			str_proto = xmlNodeListGetString(ctx->doc, cur->xmlChildrenNode, 1);
 		} else {
 			// Other unknown nodes
 			MSG_WARNING(msg_module, "Unknown node '%s' in 'destination' node "
@@ -226,15 +266,21 @@ static fwd_sender_t *config_parse_destination(struct parser_context *ctx)
 	const char *dst_ip = (const char *) str_ip;
 	const char *dst_port = (str_port != NULL)
 			? (const char *) str_port : config_def_port(ctx->cfg);
+	int proto = ctx->cfg->def_proto;
+	if (str_proto != NULL) {
+		proto = config_parse_proto((const char *) str_proto);
+	}
 
 	fwd_sender_t *new_sender;
-	new_sender = sender_create(dst_ip, dst_port);
+	new_sender = sender_create(dst_ip, dst_port, proto);
 
 	// Clean up
 	if (str_ip)
 		xmlFree(str_ip);
 	if (str_port)
 		xmlFree(str_port);
+	if (str_proto)
+		xmlFree(str_proto);
 
 	return new_sender;
 }
@@ -260,6 +306,8 @@ static int config_parse_xml(struct parser_context *ctx)
 	while (cur && !failed) {
 		if (!xmlStrcasecmp(cur->name, (const xmlChar *) "defaultPort")) {
 			// Default values were already processed -> skip
+		} else if (!xmlStrcasecmp(cur->name, (const xmlChar *) "defaultProtocol")) {
+			// Default values were already processed -> skip
 		} else if (!xmlStrcasecmp(cur->name, (const xmlChar *) "fileFormat")) {
 			// Useless node -> skip
 		} else if (cur->type == XML_COMMENT_NODE) {
@@ -302,7 +350,7 @@ static int config_parse_xml(struct parser_context *ctx)
 			aux_str = xmlNodeListGetString(doc, cur->xmlChildrenNode, 1);
 			if (config_parse_int((char *) aux_str, &result)) {
 				// Conversion failed
-				MSG_ERROR(msg_module, "Failed to parse 'reconnectionPeriod' "
+				MSG_ERROR(msg_module, "Failed to parse the 'reconnectionPeriod' "
 					"node.");
 				failed = true;
 			} else if (result <= 0) {
@@ -313,6 +361,24 @@ static int config_parse_xml(struct parser_context *ctx)
 			} else {
 				ctx->cfg->reconn_period = result;
 			}
+		} else if (!xmlStrcasecmp(cur->name, (const xmlChar *) "udpTemplateRefreshTimeout")) {
+			// Refresh timeout for UDP (options) templates
+			int result;
+			aux_str = xmlNodeListGetString(doc, cur->xmlChildrenNode, 1);
+			if (config_parse_int((char *) aux_str, &result)) {
+				// Conversion failed
+				MSG_ERROR(msg_module, "Failed to parse the "
+					"'udpTemplateRefreshTimeout' node.");
+				failed = true;
+			} else if (result <= 0) {
+				// Out of range
+				MSG_ERROR(msg_module, "Template refresh timeout cannot be zero "
+					"or negative.");
+				failed = true;
+			} else {
+				// Success (now it is safe to convert int to unsigned int)
+				ctx->cfg->udp_refresh_timeout = (unsigned int) result;
+			}
 		} else {
 			// Other unknown nodes
 			MSG_WARNING(msg_module, "Unknown node '%s' skipped.", cur->name);
@@ -374,7 +440,8 @@ struct plugin_config *config_parse(const char *cfg_string)
 	// Set default values
 	config->mode = DIST_ALL;
 	config->packet_size = DEF_PACKET_SIZE;
-	config->reconn_period = 1000; // milliseconds
+	config->reconn_period = DEF_RECONN_PERIOD; // milliseconds
+	config->udp_refresh_timeout = DEF_TEMPLATE_REFRESH; // seconds
 
 	config->builder_all = bldr_create();
 	config->builder_tmplt = bldr_create();

base/src/storage/forwarding/configuration.h

@@ -58,9 +58,12 @@
  */
 struct plugin_config {
 	char *def_port;             /**< Default port                            */
+	int def_proto;              /**< Default protocol                        */
 	enum DIST_MODE mode;        /**< Distribution mode                       */
 	uint16_t packet_size;       /**< Maximal size per generated packet       */
 	int reconn_period;          /**< Reconnection period (in milliseconds)   */
+	unsigned int udp_refresh_timeout; /**< UDP template refresh timeout
+	                                    * (in seconds)                       */
 
 	fwd_dest_t *dest_mgr;       /**< Destination manager                     */
 

base/src/storage/forwarding/destination.c

@@ -134,6 +134,15 @@ struct tmplts_for_reconnected {
 	uint32_t cnt;
 };
 
+/**
+ * \brief Auxiliary structure for checking timeout of UDP templates
+ */
+struct udp_template_timeout {
+	time_t now;             /**< Current time */
+	unsigned int timeout;   /**< Template timeout (in seconds) */
+};
+
+
 /**
  * \brief Callback function prototype
  * \param[in,out] sndr Destination
@@ -506,6 +515,28 @@ static bool aux_dummy_true(struct dst_client *client, void *data)
 	return true;
 }
 
+/**
+ * \brief Test whether UDP client has expired templates
+ * \param[in] client Destination
+ * \param[in] data   UDP template timeout structure
+ * \return When the client expired returns true and updates expiration time.
+ *   Otherwise returns false.
+ */
+static bool aux_udp_expired(struct dst_client *client, void *data)
+{
+	struct udp_template_timeout *params = (struct udp_template_timeout *) data;
+	if (sender_get_proto(client->sender) != IPPROTO_UDP) {
+		return false;
+	}
+
+	if (sender_get_tmpl_time(client->sender) + params->timeout > params->now) {
+		return false;
+	}
+
+	sender_set_tmpl_time(client->sender, params->now);
+	return true;
+}
+
 /**
  * \brief Thread for reconnection of disconnected destinations
  * \param[in,out] arg Configuration of Destination Manager
@@ -811,6 +842,18 @@ static struct tmplts_per_odid *dest_templates_prepare(
 	return result;
 }
 
+// Check UDP connections for expired templates and move them to ready state
+void dest_check_expired_udp(fwd_dest_t *dst_mgr, unsigned int timeout)
+{
+	struct udp_template_timeout params;
+	params.now = time(NULL);
+	params.timeout = timeout;
+
+	pthread_mutex_lock(&dst_mgr->group_mtx);
+	group_move(dst_mgr->conn, dst_mgr->ready, &aux_udp_expired, &params);
+	pthread_mutex_unlock(&dst_mgr->group_mtx);
+}
+
 // Check and move reconnected destinations to connected destinations
 void dest_check_reconnected(fwd_dest_t *dst_mgr)
 {

base/src/storage/forwarding/destination.h

@@ -113,6 +113,14 @@ int dest_connector_start(fwd_dest_t *dst_mgr, int period);
  */
 int dest_connector_stop(fwd_dest_t *dst_mgr);
 
+/**
+ * \brief Check UDP connections for expired templates and move them to ready
+ *   state
+ * \param[in,out] dst_mgr Destination manager
+ * \param[in]     timeout Template refresh timeout
+ */
+void dest_check_expired_udp(fwd_dest_t *dst_mgr, unsigned int timeout);
+
 /**
  * \brief Check and move reconnected destinations to connected destinations
  * \warning This functions must be called only by the thread that use

base/src/storage/forwarding/forwarding.c

@@ -488,6 +488,9 @@ int store_packet (void *config, const struct ipfix_message *ipfix_msg,
 	} else
 	*/
 
+	// Check for UDP connections with expired templates
+	dest_check_expired_udp(cfg->dest_mgr, cfg->udp_refresh_timeout);
+
 	// Add reconnected clients
 	dest_check_reconnected(cfg->dest_mgr);
 

base/src/storage/forwarding/ipfixcol-forwarding-output.dbk

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
-<refentry 
-		xmlns="http://docbook.org/ns/docbook" 
-		xmlns:xlink="http://www.w3.org/1999/xlink" 
+<refentry
+		xmlns="http://docbook.org/ns/docbook"
+		xmlns:xlink="http://www.w3.org/1999/xlink"
 		xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 		xsi:schemaLocation="http://www.w3.org/1999/xlink http://docbook.org/xml/5.0/xsd/xlink.xsd
 			http://docbook.org/ns/docbook http://docbook.org/xml/5.0/xsd/docbook.xsd"
@@ -35,14 +35,14 @@
 		<refname>ipfixcol-forwarding-output</refname>
 		<refpurpose>Forwarding output plugin for IPFIXcol.</refpurpose>
 	</refnamediv>
-	
+
 	<refsect1>
 		<title>Description</title>
 		<simpara>
 		The <command>ipfixcol-forwarding-output.so</command> is output plugin for IPFIXcol (IPFIX collector).
 		</simpara>
 		<simpara>
-		The plugin distributes IPFIX packets over the network to one or more destinations using TCP protocol and non-blocking sockets. As a destination can be used another instance of IPFIXcol or any other collector. Every packet can be distributed to all destinations or forwarded to one of destinations using Round Robin distribution model. 
+		The plugin distributes IPFIX packets over the network to one or more destinations using TCP protocol and non-blocking sockets. The plugins also supports UDP protocol transfer although this options is only experimental. When it is possible, always prefer TCP over UDP. As a destination can be used another instance of IPFIXcol or any other collector. Every packet can be distributed to all destinations or forwarded to one of destinations using Round Robin distribution model.
 		</simpara>
 		<simpara>
 		The plugin preserves Observation Domain ID (ODID) of all packets. If more (independent) metering processes (i.e. sources of IPFIX packets) use the same ODID, the plugin remap identification numbers of templates of packets to prevent misinterpretation of IPFIX records. It is very <emphasis>important</emphasis> to avoid using different types and configurations of flow sampling by the metering processes as the packets are mixed. (Flow sampling is not recommended).
@@ -68,9 +68,9 @@
 	]]>
 		</programlisting>
 		<para></para>
-		
-		<simpara>The collector must be configured to use forwarding output plugin in startup.xml configuration (<filename>/etc/ipfixcol/startup.xml</filename>). 
-		The configuration specifies which plugins are used by the collector to process data and provides configuration for the plugins themselves. 
+
+		<simpara>The collector must be configured to use forwarding output plugin in startup.xml configuration (<filename>/etc/ipfixcol/startup.xml</filename>).
+		The configuration specifies which plugins are used by the collector to process data and provides configuration for the plugins themselves.
 		</simpara>
 		<simpara><filename>startup.xml</filename> forwarding example</simpara>
 		<programlisting>
@@ -82,6 +82,7 @@
 
 			<distribution>RoundRobin</distribution>
 			<defaultPort>4739</defaultPort>
+			<defaultProtocol>tcp</defaultProtocol>
 
 			<destination>
 				<ip>192.168.0.1</ip>
@@ -92,6 +93,7 @@
 			<destination>
 				<ip>192.168.0.3</ip>
 				<port>4740</port>
+				<protocol>udp</protocol>
 			</destination>
 		</fileWriter>
 	</destination>
@@ -118,6 +120,25 @@
 					</simpara>
 				</listitem>
 			</varlistentry>
+			<varlistentry>
+				<term>
+					<command>defaultProtocol</command>
+				</term>
+				<listitem>
+					<simpara>Default transport protocol. It can be overriden by <command>protocol</command> tag in <command>destination</command>. Allowed values are <emphasis>tcp</emphasis> and <emphasis>udp</emphasis>. TCP protocol is the default when this configuration is missing.
+					</simpara>
+				</listitem>
+			</varlistentry>
+
+			<varlistentry>
+				<term>
+					<command>udpTemplateRefreshTimeout</command>
+				</term>
+				<listitem>
+					<simpara>Template refresh timeout for UDP destinations (in seconds). This value defines how often the plugin retransmits all templates to the destinations. [default == 300]
+					</simpara>
+				</listitem>
+			</varlistentry>
 
 			<varlistentry>
 				<term>
@@ -140,6 +161,13 @@
 						<simpara>Destination port number</simpara>
 					</varlistentry>
 
+					<varlistentry>
+						<term>
+							<command>protocol</command>
+						</term>
+						<simpara>Transport protocol. Allowed values are same as for <command>defaultProtocol</command></simpara>
+					</varlistentry>
+
 				</listitem>
 			</varlistentry>
 			<varlistentry>