- [DEV-115] enhanced sender to use only one TCP connection (Alexei)

git-svn-id: svn://svn.zabbix.com/trunk@5352 97f52cf1-0a1b-0410-bd0e-c28be96e8082

7 files changed, 303 insertions, 69 deletions

diff --git a/ChangeLog b/ChangeLog
index 50737921..d48202ef 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,5 +1,6 @@
 Changes for 1.5:
 
+ - [DEV-115] enhanced sender to use only one TCP connection (Alexei)
  - [DEV-112] added themes option in frontend (Artem)
  - [DEV-114] special processing of simple SNMP OIDs like ifDescr, ifInOctets, etc (Alexei)
  - [DEV-114] monitoring of SNMP objects having flexible Index (Alexei)
diff --git a/include/common.h b/include/common.h
index 61f56ce2..d79f5fd7 100644
--- a/include/common.h
+++ b/include/common.h
@@ -482,7 +482,7 @@ typedef enum
 
 #define	AGENT_TIMEOUT	3
 
-#define	SENDER_TIMEOUT		5
+#define	SENDER_TIMEOUT		60
 #define	TRAPPER_TIMEOUT		5
 #define	SNMPTRAPPER_TIMEOUT	5
 
diff --git a/include/zbxjson.h b/include/zbxjson.h
index a10f918e..7538b601 100644
--- a/include/zbxjson.h
+++ b/include/zbxjson.h
@@ -22,6 +22,19 @@
 
 #include <stdarg.h>
 
+#define	ZBX_PROTO_TAG_DATA	"data"
+#define	ZBX_PROTO_TAG_HOST	"host"
+#define	ZBX_PROTO_TAG_INFO	"info"
+#define	ZBX_PROTO_TAG_KEY	"key"
+#define	ZBX_PROTO_TAG_REQUEST	"request"
+#define	ZBX_PROTO_TAG_RESPONSE	"response"
+#define	ZBX_PROTO_TAG_VALUE	"value"
+
+#define	ZBX_PROTO_VALUE_FAILED	"failed"
+#define	ZBX_PROTO_VALUE_SUCCESS	"success"
+
+#define	ZBX_PROTO_VALUE_SENDER_DATA	"sender data"
+
 typedef enum
 {
 	ZBX_JSON_TYPE_UNKNOWN = 0,
diff --git a/src/zabbix_sender/Makefile.am b/src/zabbix_sender/Makefile.am
index 63f65bf0..8cce8667 100644
--- a/src/zabbix_sender/Makefile.am
+++ b/src/zabbix_sender/Makefile.am
@@ -5,9 +5,10 @@ sbin_PROGRAMS = zabbix_sender
 zabbix_sender_SOURCES = zabbix_sender.c
 
 zabbix_sender_LDADD = \
-	../libs/zbxcommon/libzbxcommon.a \
-	../libs/zbxlog/libzbxlog.a \
-	../libs/zbxsys/libzbxsys.a \
-	../libs/zbxcrypto/libzbxcrypto.a \
-	../libs/zbxconf/libzbxconf.a \
-	../libs/zbxcomms/libzbxcomms.a
+	$(top_srcdir)/src/libs/zbxcommon/libzbxcommon.a \
+	$(top_srcdir)/src/libs/zbxlog/libzbxlog.a \
+	$(top_srcdir)/src/libs/zbxsys/libzbxsys.a \
+	$(top_srcdir)/src/libs/zbxcrypto/libzbxcrypto.a \
+	$(top_srcdir)/src/libs/zbxconf/libzbxconf.a \
+	$(top_srcdir)/src/libs/zbxjson/libzbxjson.a \
+	$(top_srcdir)/src/libs/zbxcomms/libzbxcomms.a
diff --git a/src/zabbix_sender/zabbix_sender.c b/src/zabbix_sender/zabbix_sender.c
index 8e65928c..d7735c6c 100644
--- a/src/zabbix_sender/zabbix_sender.c
+++ b/src/zabbix_sender/zabbix_sender.c
@@ -24,12 +24,13 @@
 #include "cfg.h"
 #include "log.h"
 #include "zbxgetopt.h"
+#include "zbxjson.h"
 
 char *progname = NULL;
 
 char title_message[] = "ZABBIX send";
 
-char usage_message[] = "[-Vhv] {[-zps] -ko | -i <file>} [-c <file>]";
+char usage_message[] = "[-Vhv] {[-zps] -ko | [-p] -z<ZABBIX server> -i <file>} [-c <file>]";
 
 #ifdef HAVE_GETOPT_LONG
 char *help_message[] = {
@@ -135,11 +136,63 @@ typedef struct zbx_active_metric_type
 {
 	char*	server;
 	unsigned short	port;
-	char*	hostname;
-	char*	key;
-	char*	key_value;
+	struct zbx_json json;
 } ZBX_THREAD_SENDVAL_ARGS;
 
+/******************************************************************************
+ *                                                                            *
+ * Function: check_response                                                   *
+ *                                                                            *
+ * Purpose: Check if json response is SUCCEED                                 *
+ *                                                                            *
+ * Parameters: result SUCCEED or FAIL                                         *
+ *                                                                            *
+ * Return value:  SUCCEED - processed successfully                            *
+ *                FAIL - an error occured                                     *
+ *                                                                            *
+ * Author: Alexei Vladishev                                                   *
+ *                                                                            *
+ * Comments:                                                                  *
+ *                                                                            *
+ ******************************************************************************/
+static int	check_response(char *response)
+{
+	struct 		zbx_json_parse jp;
+	const char 	*p;
+	char		value[MAX_STRING_LEN];
+	char		info[MAX_STRING_LEN];
+
+	int	ret = SUCCEED;
+
+	ret = zbx_json_open(response, &jp);
+
+	if(SUCCEED == ret)
+	{
+		if (NULL == (p = zbx_json_pair_by_name(&jp, ZBX_PROTO_TAG_RESPONSE))
+				|| NULL == zbx_json_decodevalue(p, value, sizeof(value)))
+		{
+			ret = FAIL;
+		}
+	}
+
+	if(SUCCEED == ret)
+	{
+		if(strcmp(value, ZBX_PROTO_VALUE_SUCCESS) != 0)
+		{
+			ret = FAIL;
+		}
+	}
+
+	if (NULL != (p = zbx_json_pair_by_name(&jp, ZBX_PROTO_TAG_INFO))
+			&& NULL != zbx_json_decodevalue(p, info, sizeof(info)))
+	{
+		printf("Info from server: \"%s\"\n",
+			info);
+	}
+
+	return ret;
+}
+
 static ZBX_THREAD_ENTRY(send_value, args)
 {
 	ZBX_THREAD_SENDVAL_ARGS *sentdval_args;
@@ -156,14 +209,6 @@ static ZBX_THREAD_ENTRY(send_value, args)
 
 	sentdval_args = ((ZBX_THREAD_SENDVAL_ARGS *)args);
 
-	zabbix_log( LOG_LEVEL_DEBUG, "Send to: '%s:%i' As: '%s' Key: '%s' Value: '%s'", 
-		sentdval_args->server,
-		sentdval_args->port,
-		sentdval_args->hostname,
-		sentdval_args->key,
-		sentdval_args->key_value
-		);
-
 #if !defined(_WINDOWS)
 	signal( SIGINT,  send_signal_handler );
 	signal( SIGTERM, send_signal_handler );
@@ -172,20 +217,22 @@ static ZBX_THREAD_ENTRY(send_value, args)
 #endif /* NOT _WINDOWS */
 	
 	if (SUCCEED == (tcp_ret = zbx_tcp_connect(&sock, sentdval_args->server, sentdval_args->port, SENDER_TIMEOUT))) {
-		tosend = comms_create_request(sentdval_args->hostname, sentdval_args->key, sentdval_args->key_value,
+/*		tosend = comms_create_request(sentdval_args->server, sentdval_args->server, sentdval_args->server,
 			NULL, NULL, NULL, NULL);
 
-		zabbix_log( LOG_LEVEL_DEBUG, "Send data: '%s'", tosend);
+		zabbix_log( LOG_LEVEL_DEBUG, "Send data: '%s'", tosend);*/
 
-		tcp_ret = zbx_tcp_send(&sock, tosend);
+/*		tcp_ret = zbx_tcp_send(&sock, tosend); */
+		tcp_ret = zbx_tcp_send(&sock, sentdval_args->json.buffer);
 
-		zbx_free(tosend);
+/*		zbx_free(tosend);*/
 
 		if( SUCCEED == tcp_ret )
 		{
 			if( SUCCEED == (tcp_ret = zbx_tcp_recv(&sock, &answer)) )
 			{
-				if( !answer || strcmp(answer,"OK") )
+				zabbix_log( LOG_LEVEL_DEBUG, "Answer [%s]", answer);
+				if( !answer || check_response(answer) != SUCCEED )
 				{
 					zabbix_log( LOG_LEVEL_WARNING, "Incorrect answer from server [%s]", answer);
 				}
@@ -305,6 +352,12 @@ static zbx_task_t parse_commandline(int argc, char **argv)
 				break;
 		}
 
+	if(ZABBIX_SERVER == NULL)
+	{
+		usage();
+		exit(FAIL);
+	}
+
 	return task;
 }
 
@@ -313,8 +366,10 @@ int main(int argc, char **argv)
 	FILE	*in;
 
 	char	in_line[MAX_STRING_LEN],
-		*str_port,
-		*s;
+		*s,
+		*hostname,
+		*key,
+		*key_value;
 
 	int	task = ZBX_TASK_START,
 		total_count = 0,
@@ -337,51 +392,38 @@ int main(int argc, char **argv)
 		{
 			zabbix_log( LOG_LEVEL_WARNING, "Cannot open [%s] [%s]", INPUT_FILE, strerror(errno));
 			return FAIL;
-		}	
+		}
+
+		zbx_json_init(&sentdval_args.json, 8*1024);
+		zbx_json_addstring(&sentdval_args.json, ZBX_PROTO_TAG_REQUEST, ZBX_PROTO_VALUE_SENDER_DATA, ZBX_JSON_TYPE_STRING);
+		zbx_json_addarray(&sentdval_args.json, ZBX_PROTO_TAG_DATA);
 
 		while(fgets(in_line, sizeof(in_line), in) != NULL)
 		{ /* <zabbix_server> <hostname> <port> <key> <value> */
 			total_count++; /* also used as inputline */
-	
-			sentdval_args.server = in_line;
-
-			if( !(sentdval_args.hostname = strchr(sentdval_args.server, ' ')) )
-			{
-				zabbix_log( LOG_LEVEL_WARNING, "[line %i] 'Server' required", total_count);
-				continue;
-			}
-
-			*sentdval_args.hostname = '\0';
-			sentdval_args.hostname++;
-
-			if( !(str_port = strchr(sentdval_args.hostname, ' ')) )
-			{
-				zabbix_log( LOG_LEVEL_WARNING, "[line %i] 'Server port' required", total_count);
-				continue;
-			}
 
-			*str_port = '\0';
-			str_port++;
+	
+			hostname = in_line;
 
-			if( !(sentdval_args.key = strchr(str_port, ' ')) )
+			if( !(key = strchr(hostname, ' ')) )
 			{
 				zabbix_log( LOG_LEVEL_WARNING, "[line %i] 'Key' required", total_count);
 				continue;
 			}
 
-			*sentdval_args.key = '\0';
-			sentdval_args.key++;
+			*key = '\0';
+			key++;
 
-			if( !(sentdval_args.key_value = strchr(sentdval_args.key, ' ')) )
+			if( !(key_value = strchr(key, ' ')) )
 			{
 				zabbix_log( LOG_LEVEL_WARNING, "[line %i] 'Key value' required", total_count);
 				continue;
 			}
 
-			*sentdval_args.key_value = '\0';
-			sentdval_args.key_value++;
+			*key_value = '\0';
+			key_value++;
 
-			for(s = sentdval_args.key_value; s && *s; s++)
+			for(s = key_value; s && *s; s++)
 			{
 				if(*s == '\r' || *s == '\n' )
 				{
@@ -391,14 +433,12 @@ int main(int argc, char **argv)
 			}
 
 			if( ZABBIX_SERVER )		sentdval_args.server	= ZABBIX_SERVER;
-			if( ZABBIX_HOSTNAME )		sentdval_args.hostname	= ZABBIX_HOSTNAME;
-			if( ZABBIX_KEY )		sentdval_args.key	= ZABBIX_KEY;
-			if( ZABBIX_KEY_VALUE )		sentdval_args.key_value	= ZABBIX_KEY_VALUE;
+			if( ZABBIX_HOSTNAME )		hostname	= ZABBIX_HOSTNAME;
+			if( ZABBIX_KEY )		key		= ZABBIX_KEY;
+			if( ZABBIX_KEY_VALUE )		key_value	= ZABBIX_KEY_VALUE;
 
 			if( ZABBIX_SERVER_PORT )
 				sentdval_args.port	= ZABBIX_SERVER_PORT;
-			else
-				sentdval_args.port	= (unsigned short)atoi(str_port);
 
 			if( MIN_ZABBIX_PORT > sentdval_args.port /* || sentdval_args.port > MAX_ZABBIX_PORT (MAX_ZABBIX_PORT == max unsigned short) */)
 			{
@@ -406,7 +446,17 @@ int main(int argc, char **argv)
 					total_count, sentdval_args.port, MIN_ZABBIX_PORT, MAX_ZABBIX_PORT);
 				continue;
 			}
-			
+			zbx_json_addobject(&sentdval_args.json, NULL);
+			zbx_json_addstring(&sentdval_args.json, ZBX_PROTO_TAG_HOST, hostname, ZBX_JSON_TYPE_STRING);
+			zbx_json_addstring(&sentdval_args.json, ZBX_PROTO_TAG_KEY, key, ZBX_JSON_TYPE_STRING);
+			zbx_json_addstring(&sentdval_args.json, ZBX_PROTO_TAG_VALUE, key_value, ZBX_JSON_TYPE_STRING);
+			zbx_json_close(&sentdval_args.json);
+
+			succeed_count++;
+		}
+
+		if(succeed_count > 0)
+		{
 			if( SUCCEED == zbx_thread_wait(
 				zbx_thread_start(
 					send_value,
@@ -414,11 +464,17 @@ int main(int argc, char **argv)
 					)
 				)
 			)
-				succeed_count++;
-
+			{
+				printf("sent: %i; skipped: %i; total: %i\n", succeed_count, (total_count - succeed_count), total_count);
+			}
+			else
+			{
+				printf("Sending failed. Use option -vv for more detailed output.\n");
+			}
 		}
 
 		fclose(in);
+		zbx_json_free(&sentdval_args.json);
 	}
 	else
 	{
@@ -442,9 +498,21 @@ int main(int argc, char **argv)
 
 			sentdval_args.server	= ZABBIX_SERVER;
 			sentdval_args.port	= ZABBIX_SERVER_PORT;
-			sentdval_args.hostname	= ZABBIX_HOSTNAME;
-			sentdval_args.key	= ZABBIX_KEY;
-			sentdval_args.key_value	= ZABBIX_KEY_VALUE;
+			hostname	= ZABBIX_HOSTNAME;
+			key		= ZABBIX_KEY;
+			key_value	= ZABBIX_KEY_VALUE;
+
+			zbx_json_init(&sentdval_args.json, 8*1024);
+			zbx_json_addstring(&sentdval_args.json, ZBX_PROTO_TAG_REQUEST, ZBX_PROTO_VALUE_SENDER_DATA, ZBX_JSON_TYPE_STRING);
+			zbx_json_addarray(&sentdval_args.json, ZBX_PROTO_TAG_DATA);
+
+			zbx_json_addobject(&sentdval_args.json, NULL);
+			zbx_json_addstring(&sentdval_args.json, ZBX_PROTO_TAG_HOST, hostname, ZBX_JSON_TYPE_STRING);
+			zbx_json_addstring(&sentdval_args.json, ZBX_PROTO_TAG_KEY, key, ZBX_JSON_TYPE_STRING);
+			zbx_json_addstring(&sentdval_args.json, ZBX_PROTO_TAG_VALUE, key_value, ZBX_JSON_TYPE_STRING);
+			zbx_json_close(&sentdval_args.json);
+
+			succeed_count++;
 
 			if( SUCCEED == zbx_thread_wait(
 				zbx_thread_start(
@@ -453,12 +521,18 @@ int main(int argc, char **argv)
 					)
 				)
 			)
-				succeed_count++;
+			{
+				printf("sent: %i; skipped: %i; total: %i\n", succeed_count, (total_count - succeed_count), total_count);
+			}
+			else
+			{
+				printf("Sending failed. Use option -vv for more detailed outputs.\n");
+			}
+			zbx_json_free(&sentdval_args.json);
 		}
 		while(0); /* try block simulation */
 	}
 
-	printf("sent: %i; failed: %i; total: %i\n", succeed_count, (total_count - succeed_count), total_count);
 
 	zabbix_close_log();
 
diff --git a/src/zabbix_server/poller/checks_snmp.c b/src/zabbix_server/poller/checks_snmp.c
index c8ceadb9..b1a0b937 100644
--- a/src/zabbix_server/poller/checks_snmp.c
+++ b/src/zabbix_server/poller/checks_snmp.c
@@ -55,7 +55,6 @@ static int snmp_get_index (DB_ITEM * item, char *OID, char *value, int *idx)
 	int status;
 	int running;
 
-	unsigned char *ip;
 	int ret = NOTSUPPORTED;
 
 	zabbix_log (LOG_LEVEL_DEBUG, "In %s(oid:%s)",
diff --git a/src/zabbix_server/trapper/trapper.c b/src/zabbix_server/trapper/trapper.c
index bd120183..49bd1e5f 100644
--- a/src/zabbix_server/trapper/trapper.c
+++ b/src/zabbix_server/trapper/trapper.c
@@ -43,6 +43,145 @@
 
 static zbx_process_t	zbx_process;
 
+/******************************************************************************
+ *                                                                            *
+ * Function: send_result                                                      *
+ *                                                                            *
+ * Purpose: send json SUCCEED or FAIL to socket along with an info message    *
+ *                                                                            *
+ * Parameters: result SUCCEED or FAIL                                         *
+ *                                                                            *
+ * Return value:  SUCCEED - processed successfully                            *
+ *                FAIL - an error occured                                     *
+ *                                                                            *
+ * Author: Alexei Vladishev                                                   *
+ *                                                                            *
+ * Comments:                                                                  *
+ *                                                                            *
+ ******************************************************************************/
+static int	send_result(zbx_sock_t *sock, int result, char *info)
+{
+	int	ret = SUCCEED;
+	struct	zbx_json json;
+
+	zbx_json_init(&json, 1024);
+
+	zbx_json_addstring(&json, ZBX_PROTO_TAG_RESPONSE, SUCCEED == result ? ZBX_PROTO_VALUE_SUCCESS : ZBX_PROTO_VALUE_FAILED, ZBX_JSON_TYPE_STRING);
+	if(info != NULL && info[0]!='\0')
+	{
+		zbx_json_addstring(&json, ZBX_PROTO_TAG_INFO, info, ZBX_JSON_TYPE_STRING);
+	}
+	ret = zbx_tcp_send(sock, json.buffer);
+
+	zbx_json_free(&json);
+
+	return ret;
+}
+
+/******************************************************************************
+ *                                                                            *
+ * Function: process_new_values                                               *
+ *                                                                            *
+ * Purpose: process values sent by active agents and senders                  *
+ *                                                                            *
+ * Parameters:                                                                *
+ *                                                                            *
+ * Return value:  SUCCEED - processed successfully                            *
+ *                FAIL - an error occured                                     *
+ *                                                                            *
+ * Author: Alexei Vladishev                                                   *
+ *                                                                            *
+ * Comments:                                                                  *
+ *                                                                            *
+ ******************************************************************************/
+static int	process_new_values(zbx_sock_t *sock, struct zbx_json_parse *json)
+{
+	struct zbx_json_parse   jp_data, jp_row;
+	const char		*p, *pf;
+	char			host[MAX_STRING_LEN], key[MAX_STRING_LEN], value[MAX_STRING_LEN];
+	char			info[MAX_STRING_LEN];
+	int			ret = SUCCEED;
+	int			processed_ok = 0, processed_fail = 0;
+
+	double	sec;
+
+
+	zabbix_log(LOG_LEVEL_DEBUG, "In process_new_values(json:%.*s)",
+				json->end - json->start + 1,
+				json->start);
+
+	sec = zbx_time();
+
+/* {"request":"ZBX_SENDER_DATA","data":[{"key":"system.cpu.num",...,...},{...},...]} 
+ *                                     ^
+ */	if (NULL == (p = zbx_json_pair_by_name(json, ZBX_PROTO_TAG_DATA)))
+	{
+		zabbix_log(LOG_LEVEL_WARNING, "Can't find \"data\" pair");
+		ret = FAIL;
+	}
+
+	if(SUCCEED == ret)
+	{
+/* {"request":"ZBX_SENDER_DATA","data":[{"key":"system.cpu.num",...,...},{...},...]} 
+ *                                     ^------------------------------------------^
+ */		if (FAIL == (ret = zbx_json_brackets_open(p, &jp_data)))
+			zabbix_log(LOG_LEVEL_WARNING, "Can't proceed jason request. %s",
+					zbx_json_strerror());
+	}
+
+/* {"request":"ZBX_SENDER_DATA","data":[{"key":"system.cpu.num",...,...},{...},...]} 
+ *                                      ^
+ */	p = NULL;
+	while (SUCCEED == ret && NULL != (p = zbx_json_next(&jp_data, p)))
+	{
+/* {"request":"ZBX_SENDER_DATA","data":[{"key":"system.cpu.num",...,...},{...},...]} 
+ *                                      ^------------------------------^
+ */ 		if (FAIL == (ret = zbx_json_brackets_open(p, &jp_row)))
+			break;
+
+/*		zabbix_log(LOG_LEVEL_WARNING, "Next \"%.*s\"",
+				jp_row.end - jp_row.start + 1,
+				jp_row.start);*/
+
+		if (NULL == (pf = zbx_json_pair_by_name(&jp_row, ZBX_PROTO_TAG_HOST)) ||
+				NULL == (pf = zbx_json_decodevalue(pf, host, sizeof(host))))
+			continue;
+
+		if (NULL == (pf = zbx_json_pair_by_name(&jp_row, ZBX_PROTO_TAG_KEY)) ||
+				NULL == (pf = zbx_json_decodevalue(pf, key, sizeof(key))))
+			continue;
+
+		if (NULL == (pf = zbx_json_pair_by_name(&jp_row, ZBX_PROTO_TAG_VALUE)) ||
+				NULL == (pf = zbx_json_decodevalue(pf, value, sizeof(value))))
+			continue;
+
+		DBbegin();
+		if(SUCCEED == process_data(sock,host,key,value,NULL,NULL,NULL,NULL))
+		{
+			processed_ok++;
+		}
+		else
+		{
+			processed_fail++;
+		}
+		DBcommit();
+	}
+
+	zbx_snprintf(info,sizeof(info),"Processed %d Failed %d Total %d Seconds spent " ZBX_FS_DBL,
+			processed_ok,
+			processed_fail,
+			processed_ok+processed_fail,
+			(double)(zbx_time()-sec));
+
+	if(send_result(sock, ret, info) != SUCCEED)
+	{
+		zabbix_log( LOG_LEVEL_WARNING, "Error sending result back");
+		zabbix_syslog("Trapper: error sending result back");
+	}
+
+	return ret;
+}
+
 static int	process_trap(zbx_sock_t	*sock,char *s, int max_len)
 {
 	char	*line,*host;
@@ -137,10 +276,17 @@ static int	process_trap(zbx_sock_t	*sock,char *s, int max_len)
 		/* JSON protocol? */
 		else if (SUCCEED == zbx_json_open(s, &jp))
 		{
-			if (NULL != (p = zbx_json_pair_by_name(&jp, "request"))
-					&& NULL != zbx_json_decodevalue(p, value, sizeof(value))) {
+			if (NULL != (p = zbx_json_pair_by_name(&jp, ZBX_PROTO_TAG_REQUEST))
+					&& NULL != zbx_json_decodevalue(p, value, sizeof(value)))
+			{
 				if (0 == strcmp(value, "ZBX_PROXY_CONFIG") && zbx_process == ZBX_PROCESS_SERVER)
+				{
 					send_proxyconfig(sock, &jp);
+				}
+				else if (0 == strcmp(value, ZBX_PROTO_VALUE_SENDER_DATA) && zbx_process == ZBX_PROCESS_SERVER)
+				{
+					ret = process_new_values(sock, &jp);
+				}
 			}
 			return ret;
 		}