- [DEV-109] Server changes

   process_new_value


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

7 files changed, 111 insertions, 59 deletions

diff --git a/src/libs/zbxserver/functions.c b/src/libs/zbxserver/functions.c
index c3f5744c..868cec24 100644
--- a/src/libs/zbxserver/functions.c
+++ b/src/libs/zbxserver/functions.c
@@ -544,15 +544,11 @@ static void	update_item(DB_ITEM *item, AGENT_RESULT *value, time_t now)
  * Comments: for trapper poller process                                       *
  *                                                                            *
  ******************************************************************************/
-void	process_new_value(DB_ITEM *item, AGENT_RESULT *value)
+void	process_new_value(DB_ITEM *item, AGENT_RESULT *value, time_t now)
 {
-	time_t 	now;
-
 	zabbix_log( LOG_LEVEL_DEBUG, "In process_new_value(%s)",
 		item->key);
 
-	now = time(NULL);
-
 	if( ITEM_MULTIPLIER_USE == item->multiplier )
 	{
 		if( ITEM_VALUE_TYPE_FLOAT == item->value_type )
@@ -582,7 +578,7 @@ void	process_new_value(DB_ITEM *item, AGENT_RESULT *value)
 
 	add_history(item, value, now);
 	update_item(item, value, now);
-	update_functions( item );
+	update_functions(item);
 }
 
 /******************************************************************************
@@ -736,15 +732,11 @@ static void	proxy_update_item(DB_ITEM *item, AGENT_RESULT *value, time_t now)
  * Comments: for trapper poller process                                       *
  *                                                                            *
  ******************************************************************************/
-void	proxy_process_new_value(DB_ITEM *item, AGENT_RESULT *value)
+void	proxy_process_new_value(DB_ITEM *item, AGENT_RESULT *value, time_t now)
 {
-	time_t 	now;
-
 	zabbix_log( LOG_LEVEL_DEBUG, "In proxy_process_new_value(%s)",
 		item->key);
 
-	now = time(NULL);
-
 	proxy_add_history(item, value, now);
 	proxy_update_item(item, value, now);
 }
diff --git a/src/zabbix_proxy/zlog.c b/src/zabbix_proxy/zlog.c
index 4eeccfc4..51570a02 100644
--- a/src/zabbix_proxy/zlog.c
+++ b/src/zabbix_proxy/zlog.c
@@ -56,9 +56,9 @@ void __zbx_zabbix_syslog(const char *fmt, ...)
 
 	DB_ITEM		item;
 	DB_RESULT	result;
-	DB_ROW	row;
-
+	DB_ROW		row;
 	AGENT_RESULT	agent;
+	time_t		now;
 
 	zabbix_log(LOG_LEVEL_DEBUG, "In zabbix_log()");
 
@@ -71,6 +71,8 @@ void __zbx_zabbix_syslog(const char *fmt, ...)
 		ITEM_VALUE_TYPE_STR,
 		DBnode_local("h.hostid"));
 
+	now = time(NULL);
+
 	while((row=DBfetch(result)))
 	{
 		DBget_item_from_db(&item,row);
@@ -82,7 +84,7 @@ void __zbx_zabbix_syslog(const char *fmt, ...)
 
 		init_result(&agent);
 		SET_STR_RESULT(&agent, strdup(value_str));
-		process_new_value(&item,&agent);
+		process_new_value(&item, &agent, now);
 		free_result(&agent);
 
 		update_triggers(item.itemid);
diff --git a/src/zabbix_server/httppoller/httptest.c b/src/zabbix_server/httppoller/httptest.c
index d4fe457e..388f70ef 100644
--- a/src/zabbix_server/httppoller/httptest.c
+++ b/src/zabbix_server/httppoller/httptest.c
@@ -57,6 +57,7 @@ static int process_value(zbx_uint64_t itemid, AGENT_RESULT *value)
 	DB_RESULT	result;
 	DB_ROW		row;
 	DB_ITEM		item;
+	time_t		now;
 
 	INIT_CHECK_MEMORY();
 
@@ -81,14 +82,16 @@ static int process_value(zbx_uint64_t itemid, AGENT_RESULT *value)
 
 	DBget_item_from_db(&item,row);
 
+	now = time(NULL);
+
 	DBbegin();
 	switch (zbx_process) {
 	case ZBX_PROCESS_SERVER:
-		process_new_value(&item, value);
+		process_new_value(&item, value, now);
 		update_triggers(item.itemid);
 		break;
 	case ZBX_PROCESS_PROXY:
-		proxy_process_new_value(&item, value);
+		proxy_process_new_value(&item, value, now);
 		break;
 	}
 	DBcommit();
diff --git a/src/zabbix_server/pinger/pinger.c b/src/zabbix_server/pinger/pinger.c
index 9b3189e2..c8585209 100644
--- a/src/zabbix_server/pinger/pinger.c
+++ b/src/zabbix_server/pinger/pinger.c
@@ -59,7 +59,7 @@ static int		pinger_num;
  * Comments: can be done in process_data()                                    *
  *                                                                            *
  ******************************************************************************/
-static int process_value(char *key, ZBX_FPING_HOST *host, AGENT_RESULT *value)
+static int process_value(char *key, ZBX_FPING_HOST *host, AGENT_RESULT *value, time_t now)
 {
 	DB_RESULT	result;
 	DB_ROW		row;
@@ -90,11 +90,11 @@ static int process_value(char *key, ZBX_FPING_HOST *host, AGENT_RESULT *value)
 		DBbegin();
 		switch (zbx_process) {
 		case ZBX_PROCESS_SERVER:
-			process_new_value(&item, value);
+			process_new_value(&item, value, now);
 			update_triggers(item.itemid);
 			break;
 		case ZBX_PROCESS_PROXY:
-			proxy_process_new_value(&item, value);
+			proxy_process_new_value(&item, value, now);
 			break;
 		}
 		DBcommit();
@@ -221,6 +221,7 @@ static int do_ping(ZBX_FPING_HOST *hosts, int hosts_count)
 	char		*c;
 	ZBX_FPING_HOST	*host;
 	AGENT_RESULT	value;
+	time_t		now;
 
 	zabbix_log(LOG_LEVEL_DEBUG, "In do_ping() [hosts_count:%d]",
 			hosts_count);
@@ -297,6 +298,8 @@ static int do_ping(ZBX_FPING_HOST *hosts, int hosts_count)
 
 	unlink(filename);
 
+	now = time(NULL);
+
 	for (i = 0; i < hosts_count; i++) {
 		zabbix_log(LOG_LEVEL_DEBUG, "Host [%s] alive [%d]",
 				hosts[i].useip ? hosts[i].ip : hosts[i].dns,
@@ -304,12 +307,12 @@ static int do_ping(ZBX_FPING_HOST *hosts, int hosts_count)
 
 		init_result(&value);
 		SET_UI64_RESULT(&value, hosts[i].alive);
-		process_value(SERVER_ICMPPING_KEY, &hosts[i], &value);
+		process_value(SERVER_ICMPPING_KEY, &hosts[i], &value, now);
 		free_result(&value);
 				
 		init_result(&value);
 		SET_DBL_RESULT(&value, hosts[i].mseconds/1000);
-		process_value(SERVER_ICMPPINGSEC_KEY, &hosts[i], &value);
+		process_value(SERVER_ICMPPINGSEC_KEY, &hosts[i], &value, now);
 		free_result(&value);
 	}
 
diff --git a/src/zabbix_server/poller/poller.c b/src/zabbix_server/poller/poller.c
index eced40a3..0ff0451b 100644
--- a/src/zabbix_server/poller/poller.c
+++ b/src/zabbix_server/poller/poller.c
@@ -193,7 +193,7 @@ static int get_minnextcheck(int now)
 }
 
 /* Update special host's item - "status" */
-static void update_key_status(zbx_uint64_t hostid,int host_status)
+static void update_key_status(zbx_uint64_t hostid, int host_status, time_t now)
 {
 /*	char		value_str[MAX_STRING_LEN];*/
 	AGENT_RESULT	agent;
@@ -227,11 +227,11 @@ static void update_key_status(zbx_uint64_t hostid,int host_status)
 
 			switch (zbx_process) {
 			case ZBX_PROCESS_SERVER:
-				process_new_value(&item, &agent);
+				process_new_value(&item, &agent, now);
 				update_triggers(item.itemid);
 				break;
 			case ZBX_PROCESS_PROXY:
-				proxy_process_new_value(&item, &agent);
+				proxy_process_new_value(&item, &agent, now);
 				break;
 			}
 
@@ -254,7 +254,7 @@ static void enable_host(DB_ITEM *item, time_t now, char *error)
 	switch (zbx_process) {
 	case ZBX_PROCESS_SERVER:
 		DBupdate_host_availability(item->hostid, HOST_AVAILABLE_TRUE, now, error);
-		update_key_status(item->hostid, HOST_STATUS_MONITORED); /* 0 */
+		update_key_status(item->hostid, HOST_STATUS_MONITORED, now); /* 0 */
 		break;
 	case ZBX_PROCESS_PROXY:
 		DBproxy_update_host_availability(item->hostid, HOST_AVAILABLE_TRUE, now);
@@ -278,7 +278,7 @@ static void disable_host(DB_ITEM *item, time_t now, char *error)
 	switch (zbx_process) {
 	case ZBX_PROCESS_SERVER:
 		DBupdate_host_availability(item->hostid, HOST_AVAILABLE_FALSE, now, error);
-		update_key_status(item->hostid, HOST_AVAILABLE_FALSE); /* 2 */
+		update_key_status(item->hostid, HOST_AVAILABLE_FALSE, now); /* 2 */
 		break;
 	case ZBX_PROCESS_PROXY:
 		DBproxy_update_host_availability(item->hostid, HOST_AVAILABLE_FALSE, now);
@@ -314,7 +314,7 @@ int get_values(void)
 	DB_ROW	row;
 	DB_ROW	row2;
 
-	int		now;
+	time_t		now;
 	int		delay;
 	int		res;
 	DB_ITEM		item;
@@ -430,14 +430,16 @@ int get_values(void)
 
 		DBbegin();
 		
+		now = time(NULL);
+
 		if(res == SUCCEED )
 		{
 			switch (zbx_process) {
 			case ZBX_PROCESS_SERVER:
-				process_new_value(&item, &agent);
+				process_new_value(&item, &agent, now);
 				break;
 			case ZBX_PROCESS_PROXY:
-				proxy_process_new_value(&item, &agent);
+				proxy_process_new_value(&item, &agent, now);
 				break;
 			}
 
@@ -462,7 +464,6 @@ int get_values(void)
 		}
 		else if(res == NOTSUPPORTED || res == AGENT_ERROR)
 		{
-			now = time(NULL);
 			if(item.status == ITEM_STATUS_NOTSUPPORTED)
 			{
 				/* It is not correct */
@@ -500,8 +501,6 @@ int get_values(void)
 		}
 		else if(res == NETWORK_ERROR)
 		{
-			now = time(NULL);
-
 			/* First error */
 			if(item.host_errors_from==0)
 			{
diff --git a/src/zabbix_server/trapper/trapper.c b/src/zabbix_server/trapper/trapper.c
index 3e84d983..d6401448 100644
--- a/src/zabbix_server/trapper/trapper.c
+++ b/src/zabbix_server/trapper/trapper.c
@@ -135,14 +135,15 @@ static void	calc_timestamp(char *line,int *timestamp, char *format)
  * Comments: for trapper server process                                       *
  *                                                                            *
  ******************************************************************************/
-static int	process_data(zbx_sock_t *sock,char *server,char *key,char *value,char *lastlogsize, char *timestamp,
-				char *source, char *severity)
+static int	process_data(zbx_sock_t *sock, zbx_uint64_t proxyid, time_t now, char *server, char *key, char *value,
+				char *lastlogsize, char *timestamp, char *source, char *severity)
 {
 	AGENT_RESULT	agent;
 	DB_RESULT	result;
 	DB_ROW		row;
 	DB_ITEM		item;
 	char		server_esc[MAX_STRING_LEN], key_esc[MAX_STRING_LEN];
+	char		item_types[32];
 
 	zabbix_log(LOG_LEVEL_DEBUG, "In process_data([%s],[%s],[%s],[%s])",
 			server,
@@ -153,15 +154,32 @@ static int	process_data(zbx_sock_t *sock,char *server,char *key,char *value,char
 	DBescape_string(server, server_esc, MAX_STRING_LEN);
 	DBescape_string(key, key_esc, MAX_STRING_LEN);
 
-	result = DBselect("select %s where h.status=%d and h.hostid=i.hostid and h.host='%s' and h.proxyid=0"
-			" and i.key_='%s' and i.status in (%d,%d) and i.type in (%d,%d)" DB_NODE,
+	if (proxyid == 0) {
+		zbx_snprintf(item_types, sizeof(item_types), "%d,%d",
+				ITEM_TYPE_TRAPPER,
+				ITEM_TYPE_ZABBIX_ACTIVE);
+	} else {
+		zbx_snprintf(item_types, sizeof(item_types), "%d,%d,%d,%d,%d,%d,%d,%d,%d",
+				ITEM_TYPE_ZABBIX,
+				ITEM_TYPE_SNMPv1,
+				ITEM_TYPE_TRAPPER,
+				ITEM_TYPE_SIMPLE,
+				ITEM_TYPE_SNMPv2c,
+				ITEM_TYPE_SNMPv3,
+				ITEM_TYPE_ZABBIX_ACTIVE,
+				ITEM_TYPE_HTTPTEST,
+				ITEM_TYPE_EXTERNAL);
+	}
+
+	result = DBselect("select %s where h.status=%d and h.hostid=i.hostid and h.host='%s' and h.proxyid=" ZBX_FS_UI64
+			" and i.key_='%s' and i.status in (%d,%d) and i.type in (%s)" DB_NODE,
 			ZBX_SQL_ITEM_SELECT,
 			HOST_STATUS_MONITORED,
 			server_esc,
+			proxyid,
 			key_esc,
 			ITEM_STATUS_ACTIVE, ITEM_STATUS_NOTSUPPORTED,
-			ITEM_TYPE_TRAPPER,
-			ITEM_TYPE_ZABBIX_ACTIVE,
+			item_types,
 			DBnode_local("h.hostid"));
 
 	if (NULL == (row = DBfetch(result))) {
@@ -214,8 +232,8 @@ static int	process_data(zbx_sock_t *sock,char *server,char *key,char *value,char
 				item.key,
 				item.host_name);
 		zabbix_syslog("Active parameter [%s] is not supported by agent on host [%s]",
-			item.key,
-					item.host_name);
+				item.key,
+				item.host_name);
 		DBupdate_item_status_to_notsupported(item.itemid, "Not supported by ZABBIX agent");
 	} else {
 		if (0 == strncmp(item.key, "log[", 4) || 0 == strncmp(item.key, "eventlog[", 9)) {
@@ -235,7 +253,7 @@ static int	process_data(zbx_sock_t *sock,char *server,char *key,char *value,char
 		init_result(&agent);
 
 		if( SUCCEED == set_result_type(&agent, item.value_type, value)) {
-			process_new_value(&item, &agent);
+			process_new_value(&item, &agent, now);
 			update_triggers(item.itemid);
 		} else {
 			zabbix_log( LOG_LEVEL_WARNING, "Type of received value [%s] is not suitable for [%s@%s]",
@@ -309,13 +327,16 @@ 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;
-	char			host[MAX_HOST_HOST_LEN], key[MAX_ITEM_KEY_LEN], value[MAX_STRING_LEN];
-	char			info[MAX_STRING_LEN];
+	char			proxy[PROXY_NAME_LEN_MAX], host[HOST_HOST_LEN_MAX], key[ITEM_KEY_LEN_MAX],
+				value[MAX_STRING_LEN], info[MAX_STRING_LEN], lastlogsize[MAX_STRING_LEN],
+				timestamp[MAX_STRING_LEN], source[MAX_STRING_LEN], severity[MAX_STRING_LEN];
 	int			ret = SUCCEED;
 	int			processed_ok = 0, processed_fail = 0;
-
-	double	sec;
-
+	DB_RESULT		result;
+	DB_ROW			row;
+	double			sec;
+	zbx_uint64_t		proxyid = 0;
+	time_t			now = 0;
 
 	zabbix_log(LOG_LEVEL_DEBUG, "In process_new_values(json:%.*s)",
 				json->end - json->start + 1,
@@ -323,6 +344,21 @@ static int	process_new_values(zbx_sock_t *sock, struct zbx_json_parse *json)
 
 	sec = zbx_time();
 
+	if (SUCCEED == zbx_json_value_by_name(json, ZBX_PROTO_TAG_PROXY, proxy, sizeof(proxy))) {
+		result = DBselect("select proxyid from proxies where name='%s'",
+				proxy);
+
+		if (NULL != (row = DBfetch(result)) && FAIL == DBis_null(row[0]))
+			proxyid = zbx_atoui64(row[0]);
+		DBfree_result(result);
+
+	}
+
+	if (SUCCEED == zbx_json_value_by_name(json, ZBX_PROTO_TAG_CLOCK, timestamp, sizeof(timestamp)))
+		now = atoi(timestamp);
+
+zabbix_log(LOG_LEVEL_DEBUG, "In process_new_values(proxyid:" ZBX_FS_UI64 ", clock:%d)", proxyid, now);
+
 /* {"request":"ZBX_SENDER_DATA","data":[{"key":"system.cpu.num",...,...},{...},...]} 
  *                                     ^
  */	if (NULL == (p = zbx_json_pair_by_name(json, ZBX_PROTO_TAG_DATA)))
@@ -350,27 +386,42 @@ static int	process_new_values(zbx_sock_t *sock, struct zbx_json_parse *json)
  */ 		if (FAIL == (ret = zbx_json_brackets_open(p, &jp_row)))
 			break;
 
-/*		zabbix_log(LOG_LEVEL_WARNING, "Next \"%.*s\"",
+		zabbix_log(LOG_LEVEL_WARNING, "Next \"%.*s\"",
 				jp_row.end - jp_row.start + 1,
-				jp_row.start);*/
-
-		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_HOST, host, sizeof(host)))
+				jp_row.start);
+
+		*host = '\0';
+		*key = '\0';
+		*value = '\0';
+		*lastlogsize = '\0';
+		*timestamp = '\0';
+		*source = '\0';
+		*severity = '\0';
+		
+		if (FAIL == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_HOST, host, sizeof(host)))
 			continue;
 
-		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_KEY, key, sizeof(key)))
+		if (FAIL == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_KEY, key, sizeof(key)))
 			continue;
 
-		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_VALUE, value, sizeof(value)))
+		if (FAIL == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_VALUE, value, sizeof(value)))
 			continue;
 
+		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_CLOCK, timestamp, sizeof(timestamp)))
+			now = time(NULL) - (now - atoi(timestamp));
+
+		zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_VALUE, source, sizeof(severity));
+
+		zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_VALUE, severity, sizeof(severity));
+
 		DBbegin();
-		if(SUCCEED == process_data(sock,host,key,value,NULL,NULL,NULL,NULL))
+		if(SUCCEED == process_data(sock, proxyid, now, host, key, value, lastlogsize, timestamp, source, severity))
 		{
-			processed_ok++;
+			processed_ok ++;
 		}
 		else
 		{
-			processed_fail++;
+			processed_fail ++;
 		}
 		DBcommit();
 	}
@@ -539,7 +590,7 @@ static int	process_trap(zbx_sock_t	*sock,char *s, int max_len)
 		zabbix_log( LOG_LEVEL_DEBUG, "Value [%s]", value_string);
 
 		DBbegin();
-		ret=process_data(sock,server,key,value_string,lastlogsize,timestamp,source,severity);
+		ret=process_data(sock, 0, time(NULL), server, key, value_string, lastlogsize, timestamp, source, severity);
 		DBcommit();
 		
 		if( zbx_tcp_send_raw(sock, SUCCEED == ret ? "OK" : "NOT OK") != SUCCEED)
diff --git a/src/zabbix_server/zlog.c b/src/zabbix_server/zlog.c
index 4284c0cc..e44d10ce 100644
--- a/src/zabbix_server/zlog.c
+++ b/src/zabbix_server/zlog.c
@@ -57,9 +57,9 @@ void __zbx_zabbix_syslog(const char *fmt, ...)
 
 	DB_ITEM		item;
 	DB_RESULT	result;
-	DB_ROW	row;
-
+	DB_ROW		row;
 	AGENT_RESULT	agent;
+	time_t		now;
 
 	zabbix_log(LOG_LEVEL_DEBUG, "In zabbix_log()");
 
@@ -72,6 +72,8 @@ void __zbx_zabbix_syslog(const char *fmt, ...)
 		ITEM_VALUE_TYPE_STR,
 		DBnode_local("h.hostid"));
 
+	now = time(NULL);
+
 	while((row=DBfetch(result)))
 	{
 		DBget_item_from_db(&item,row);
@@ -83,7 +85,7 @@ void __zbx_zabbix_syslog(const char *fmt, ...)
 
 		init_result(&agent);
 		SET_STR_RESULT(&agent, strdup(value_str));
-		process_new_value(&item,&agent);
+		process_new_value(&item, &agent, now);
 		free_result(&agent);
 
 		update_triggers(item.itemid);