- [DEV-137] Misc bugs and improvements

   [added timediff for processing nextchecks for proxy items]


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

6 files changed, 77 insertions, 77 deletions

diff --git a/include/dbcache.h b/include/dbcache.h
index a0dee28f..693c7390 100644
--- a/include/dbcache.h
+++ b/include/dbcache.h
@@ -100,7 +100,7 @@ void	init_database_cache(zbx_process_t p);
 void	free_database_cache(void);
 
 void	DCinit_nextchecks();
-void	DCadd_nextcheck(DB_ITEM *item, time_t now, const char *error_msg);
+void	DCadd_nextcheck(DB_ITEM *item, time_t now, time_t timediff, const char *error_msg);
 void	DCflush_nextchecks();
 
 #endif
diff --git a/src/libs/zbxdbcache/nextchecks.c b/src/libs/zbxdbcache/nextchecks.c
index 935f2cd9..ba389598 100644
--- a/src/libs/zbxdbcache/nextchecks.c
+++ b/src/libs/zbxdbcache/nextchecks.c
@@ -95,7 +95,7 @@ static int	DCget_nextcheck_nearestindex(time_t clock)
  * Comments:                                                                  *
  *                                                                            *
  ******************************************************************************/
-void	DCadd_nextcheck(DB_ITEM *item, time_t now, const char *error_msg)
+void	DCadd_nextcheck(DB_ITEM *item, time_t now, time_t timediff, const char *error_msg)
 {
 	int	index, nextcheck;
 
@@ -107,7 +107,7 @@ void	DCadd_nextcheck(DB_ITEM *item, time_t now, const char *error_msg)
 		nextchecks = zbx_realloc(nextchecks, nextcheck_allocated * sizeof(ZBX_DC_NEXTCHECK));
 	}
 
-	nextcheck = calculate_item_nextcheck(item->itemid, item->type, item->delay, item->delay_flex, now);
+	nextcheck = calculate_item_nextcheck(item->itemid, item->type, item->delay, item->delay_flex, now - timediff) + timediff;
 	index = DCget_nextcheck_nearestindex(nextcheck);
 
 	memmove(&nextchecks[index + 1], &nextchecks[index], sizeof(ZBX_DC_NEXTCHECK) * (nextcheck_num - index));
diff --git a/src/zabbix_proxy/datasender/datasender.c b/src/zabbix_proxy/datasender/datasender.c
index 0e836259..64506048 100644
--- a/src/zabbix_proxy/datasender/datasender.c
+++ b/src/zabbix_proxy/datasender/datasender.c
@@ -314,9 +314,6 @@ static int	history_sender(struct zbx_json *j)
 
 	zabbix_log(LOG_LEVEL_DEBUG, "In history_sender()");
 
-	if (FAIL == connect_to_server(&sock, 600))	/* alarm !!! */
-		return FAIL;
-
 	zbx_json_clean(j);
 	zbx_json_addstring(j, ZBX_PROTO_TAG_REQUEST, ZBX_PROTO_VALUE_HISTORY_DATA, ZBX_JSON_TYPE_STRING);
 	zbx_json_addstring(j, ZBX_PROTO_TAG_HOST, CONFIG_HOSTNAME, ZBX_JSON_TYPE_STRING);
@@ -336,14 +333,20 @@ static int	history_sender(struct zbx_json *j)
 
 	zbx_json_adduint64(j, ZBX_PROTO_TAG_CLOCK, (int)time(NULL));
 
-	if (SUCCEED == put_data_to_server(&sock, j)) {
-		DBbegin();
-		for (i = 0; i < li_no; i++)
-			set_lastid(li[i].ht, li[i].lastid);
-		DBcommit();
-	}
+	if (records)
+	{
+		if (FAIL == connect_to_server(&sock, 600))	/* alarm !!! */
+			return FAIL;
+
+		if (SUCCEED == put_data_to_server(&sock, j)) {
+			DBbegin();
+			for (i = 0; i < li_no; i++)
+				set_lastid(li[i].ht, li[i].lastid);
+			DBcommit();
+		}
 
-	disconnect_server(&sock);
+		disconnect_server(&sock);
+	}
 
 	return records;
 }
@@ -356,7 +359,7 @@ static int	history_sender(struct zbx_json *j)
  *                                                                            *
  * Parameters:                                                                *
  *                                                                            *
- * Return value:                                                              * 
+ * Return value: number of records or FAIL if server is not accessible        * 
  *                                                                            *
  * Author: Aleksander Vladishev                                               *
  *                                                                            *
@@ -373,9 +376,6 @@ static int	dhistory_sender(struct zbx_json *j)
 
 	zabbix_log(LOG_LEVEL_DEBUG, "In dhistory_sender()");
 
-	if (FAIL == connect_to_server(&sock, 600))	/* alarm !!! */
-		return FAIL;
-
 	zbx_json_clean(j);
 	zbx_json_addstring(j, ZBX_PROTO_TAG_REQUEST, ZBX_PROTO_VALUE_DISCOVERY_DATA, ZBX_JSON_TYPE_STRING);
 	zbx_json_addstring(j, ZBX_PROTO_TAG_HOST, CONFIG_HOSTNAME, ZBX_JSON_TYPE_STRING);
@@ -395,14 +395,20 @@ static int	dhistory_sender(struct zbx_json *j)
 
 	zbx_json_adduint64(j, ZBX_PROTO_TAG_CLOCK, (int)time(NULL));
 
-	if (SUCCEED == put_data_to_server(&sock, j)) {
-		DBbegin();
-		for (i = 0; i < li_no; i++)
-			set_lastid(li[i].ht, li[i].lastid);
-		DBcommit();
-	}
+	if (records)
+	{
+		if (FAIL == connect_to_server(&sock, 600))	/* alarm !!! */
+			return FAIL;
 
-	disconnect_server(&sock);
+		if (SUCCEED == put_data_to_server(&sock, j)) {
+			DBbegin();
+			for (i = 0; i < li_no; i++)
+				set_lastid(li[i].ht, li[i].lastid);
+			DBcommit();
+		}
+
+		disconnect_server(&sock);
+	}
 
 	return records;
 }
@@ -426,7 +432,7 @@ int	main_datasender_loop()
 {
 	struct sigaction	phan;
 	int			now, sleeptime,
-				records;
+				records, r;
 	double			sec;
 	struct zbx_json		j;
 
@@ -450,8 +456,11 @@ int	main_datasender_loop()
 		zbx_setproctitle("data sender [sending data]");
 
 		records = 0;
-		records += history_sender(&j);
-		records += dhistory_sender(&j);
+		if (FAIL != (r = history_sender(&j)))
+			records += r;
+				
+		if (FAIL != (r = dhistory_sender(&j)))
+			records += r;
 
 		zabbix_log(LOG_LEVEL_DEBUG, "Datasender spent " ZBX_FS_DBL " seconds while processing %3d values.",
 				zbx_time() - sec,
diff --git a/src/zabbix_server/pinger/pinger.c b/src/zabbix_server/pinger/pinger.c
index 97de1fa2..efb5e618 100644
--- a/src/zabbix_server/pinger/pinger.c
+++ b/src/zabbix_server/pinger/pinger.c
@@ -118,7 +118,7 @@ static void process_value(char *key, ZBX_FPING_HOST *host, zbx_uint64_t *value_u
 				proxy_process_new_value(&item, &value, now);
 				break;
 			}
-			DCadd_nextcheck(&item, now, NULL);
+			DCadd_nextcheck(&item, now, 0, NULL);
 		}
 
 		free_result(&value);
diff --git a/src/zabbix_server/poller/poller.c b/src/zabbix_server/poller/poller.c
index dbabb5b2..084c7914 100644
--- a/src/zabbix_server/poller/poller.c
+++ b/src/zabbix_server/poller/poller.c
@@ -455,7 +455,7 @@ static int get_values(int now)
 					proxy_process_new_value(&item, &agent, now);
 					break;
 				}
-				DCadd_nextcheck(&item, now, NULL);
+				DCadd_nextcheck(&item, now, 0, NULL);
 			}
 		}
 		else if (res == NOTSUPPORTED || res == AGENT_ERROR)
@@ -480,7 +480,7 @@ static int get_values(int now)
 				DBcommit();
 			}
 			else
-				DCadd_nextcheck(&item, now, agent.msg);
+				DCadd_nextcheck(&item, now, 0, agent.msg);
 
 			if (HOST_AVAILABLE_TRUE != item.host_available) {
 				DBbegin();
diff --git a/src/zabbix_server/trapper/trapper.c b/src/zabbix_server/trapper/trapper.c
index dc4d47bd..182e01ea 100644
--- a/src/zabbix_server/trapper/trapper.c
+++ b/src/zabbix_server/trapper/trapper.c
@@ -137,7 +137,8 @@ static void	calc_timestamp(char *line,int *timestamp, char *format)
  * Comments: for trapper server process                                       *
  *                                                                            *
  ******************************************************************************/
-static void	process_mass_data(zbx_sock_t *sock, zbx_uint64_t proxy_hostid, AGENT_VALUE *values, int value_num, int *processed)
+static void	process_mass_data(zbx_sock_t *sock, zbx_uint64_t proxy_hostid, AGENT_VALUE *values, int value_num,
+		int *processed, time_t proxy_timediff)
 {
 	AGENT_RESULT	agent;
 	DB_RESULT	result;
@@ -203,9 +204,6 @@ static void	process_mass_data(zbx_sock_t *sock, zbx_uint64_t proxy_hostid, AGENT
 
 	result = DBselect("%s", sql);
 
-	if (0 != CONFIG_DBSYNCER_FORKS)
-		DCinit_nextchecks();
-
 	while (NULL != (row = DBfetch(result))) {
 		DBget_item_from_db(&item, row);
 
@@ -229,13 +227,10 @@ static void	process_mass_data(zbx_sock_t *sock, zbx_uint64_t proxy_hostid, AGENT
 							item.key,
 							item.host_name);
 
-					if (0 == CONFIG_DBSYNCER_FORKS)
-						DBupdate_item_status_to_notsupported(&item, values[i].clock, "Not supported by ZABBIX agent");
-					else
-						DCadd_nextcheck(&item, values[i].clock, "Not supported by ZABBIX agent");
-
-					(*processed)++;
+					DCadd_nextcheck(&item, values[i].clock, proxy_timediff, "Not supported by ZABBIX agent");
 
+					if (NULL != processed)
+						(*processed)++;
 				}
 				else
 				{
@@ -282,7 +277,8 @@ static void	process_mass_data(zbx_sock_t *sock, zbx_uint64_t proxy_hostid, AGENT
 								break;
 							}
 						}
-						(*processed)++;
+						if (NULL != processed)
+							(*processed)++;
 
 						/* only for screen Administration|Queue */
 						if (0 != proxy_hostid && item.type != ITEM_TYPE_TRAPPER &&
@@ -292,7 +288,7 @@ static void	process_mass_data(zbx_sock_t *sock, zbx_uint64_t proxy_hostid, AGENT
 								0 != strcmp(item.key, SERVER_ICMPPING_KEY) &&
 								0 != strcmp(item.key, SERVER_ICMPPINGSEC_KEY) &&
 								0 != strcmp(item.key, SERVER_ZABBIXLOG_KEY))
-							DCadd_nextcheck(&item, values[i].clock, NULL);
+							DCadd_nextcheck(&item, values[i].clock, proxy_timediff, NULL);
 					}
 					else
 					{
@@ -313,9 +309,6 @@ static void	process_mass_data(zbx_sock_t *sock, zbx_uint64_t proxy_hostid, AGENT
 	DBfree_result(result);
 
 	DCflush_nextchecks();
-
-	if (0 != CONFIG_DBSYNCER_FORKS)
-		DCflush_nextchecks();
 }
 
 /******************************************************************************
@@ -384,17 +377,14 @@ static int	process_new_values(zbx_sock_t *sock, struct zbx_json_parse *jp, const
 {
 	struct zbx_json_parse   jp_data, jp_row;
 	const char		*p;
-	char			/*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],
-				clock[MAX_STRING_LEN], */tmp[MAX_STRING_LEN];
+	char			info[MAX_STRING_LEN], tmp[MAX_STRING_LEN];
 	int			ret = SUCCEED;
 	int			processed = 0, processed_fail = 0;
 	double			sec;
-	time_t			now, hosttime = 0/*, itemtime*/;
+	time_t			now, proxy_timediff = 0;
 
 #define VALUES_MAX	256
-	static AGENT_VALUE	*values = NULL;
+	static AGENT_VALUE	*values = NULL, *av;
 	int			value_num = 0;
 
 	zabbix_log(LOG_LEVEL_DEBUG, "In process_new_values()");
@@ -406,7 +396,7 @@ static int	process_new_values(zbx_sock_t *sock, struct zbx_json_parse *jp, const
 		values = zbx_malloc(values, VALUES_MAX * sizeof(AGENT_VALUE));
 
 	if (SUCCEED == zbx_json_value_by_name(jp, ZBX_PROTO_TAG_CLOCK, tmp, sizeof(tmp)))
-		hosttime = atoi(tmp);
+		proxy_timediff = now - atoi(tmp);
 
 /* {"request":"ZBX_SENDER_DATA","data":[{"key":"system.cpu.num",...,...},{...},...]} 
  *                                     ^
@@ -439,62 +429,53 @@ static int	process_new_values(zbx_sock_t *sock, struct zbx_json_parse *jp, const
 				jp_row.end - jp_row.start + 1,
 				jp_row.start);*/
 
-		memset(&values[value_num], 0, sizeof(AGENT_VALUE));
+		av = &values[value_num];
 
-		values[value_num].clock = now;
+		memset(av, 0, sizeof(AGENT_VALUE));
 
-		if (hosttime && SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_CLOCK, tmp, sizeof(tmp)))
-			values[value_num].clock -= hosttime - atoi(tmp);
+		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_CLOCK, tmp, sizeof(tmp)))
+			av->clock = atoi(tmp) + proxy_timediff;
+		else
+			av->clock = now;
 
-		if (FAIL == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_HOST, values[value_num].host_name, sizeof(values[value_num].host_name)))
+		if (FAIL == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_HOST, av->host_name, sizeof(av->host_name)))
 			continue;
 
-		if (FAIL == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_KEY, values[value_num].key, sizeof(values[value_num].key)))
+		if (FAIL == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_KEY, av->key, sizeof(av->key)))
 			continue;
 
 		if (FAIL == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_VALUE, tmp, sizeof(tmp)))
 			continue;
 
-		values[value_num].value = strdup(tmp);
+		av->value = strdup(tmp);
 
 		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_LOGLASTSIZE, tmp, sizeof(tmp))) 
-			values[value_num].lastlogsize = atoi(tmp);
+			av->lastlogsize = atoi(tmp);
 
 		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_LOGTIMESTAMP, tmp, sizeof(tmp)))
-			values[value_num].timestamp = atoi(tmp);
+			av->timestamp = atoi(tmp);
 
 		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_LOGSOURCE, tmp, sizeof(tmp)))
-			values[value_num].source = strdup(tmp);
+			av->source = strdup(tmp);
 
 		if (SUCCEED == zbx_json_value_by_name(&jp_row, ZBX_PROTO_TAG_LOGSEVERITY, tmp, sizeof(tmp)))
-			values[value_num].severity = atoi(tmp);
+			av->severity = atoi(tmp);
 
 		value_num ++;
 
 		if (value_num == VALUES_MAX) {
 			DBbegin();
-			process_mass_data(sock, proxy_hostid, values, value_num, &processed);
+			process_mass_data(sock, proxy_hostid, values, value_num, &processed, proxy_timediff);
 			DBcommit();
 
 			clean_agent_values(values, value_num);
 			value_num = 0;
 		}
-
-/*		DBbegin();
-		if(SUCCEED == process_data(sock, proxy_hostid, itemtime, host, key, value, lastlogsize, timestamp, source, severity))
-		{
-			processed_ok ++;
-		}
-		else
-		{
-			processed_fail ++;
-		}
-		DBcommit();*/
 	}
 
 	if (value_num > 0) {
 		DBbegin();
-		process_mass_data(sock, proxy_hostid, values, value_num, &processed);
+		process_mass_data(sock, proxy_hostid, values, value_num, &processed, proxy_timediff);
 		DBcommit();
 	}
 
@@ -598,6 +579,7 @@ static int	process_trap(zbx_sock_t	*sock, char *s, int max_len)
 
 	struct 		zbx_json_parse jp;
 	char		value[MAX_STRING_LEN];
+	AGENT_VALUE	av;
 
 	zbx_rtrim(s, " \r\n\0");
 
@@ -750,8 +732,17 @@ static int	process_trap(zbx_sock_t	*sock, char *s, int max_len)
 		}
 		zabbix_log( LOG_LEVEL_DEBUG, "Value [%s]", value_string);
 
+		av.clock = time(NULL);
+		zbx_strlcpy(av.host_name, server, sizeof(av.host_name));
+		zbx_strlcpy(av.key, key, sizeof(av.key));
+		av.value = value_string;
+		av.lastlogsize = atoi(lastlogsize);
+		av.timestamp = atoi(timestamp);
+		av.source = source;
+		av.severity = atoi(severity);
+
 		DBbegin();
-/*		ret=process_data(sock, 0, time(NULL), server, key, value_string, lastlogsize, timestamp, source, severity);*/
+		process_mass_data(sock, 0, &av, 1, NULL, 0);
 		DBcommit();
 		
 		if( zbx_tcp_send_raw(sock, SUCCEED == ret ? "OK" : "NOT OK") != SUCCEED)