More changes for distributed monitoring.

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

3 files changed, 196 insertions, 11 deletions

diff --git a/src/libs/zbxdbhigh/db.c b/src/libs/zbxdbhigh/db.c
index c4dda150..657834ff 100644
--- a/src/libs/zbxdbhigh/db.c
+++ b/src/libs/zbxdbhigh/db.c
@@ -1341,14 +1341,14 @@ int	DBadd_history(zbx_uint64_t itemid, double value, int clock)
 	DBexecute("insert into history (clock,itemid,value) values (%d," ZBX_FS_UI64 ",%f)",
 		clock,itemid,value);
 
+	DBadd_trend(itemid, value, clock);
+
 	if(CONFIG_MASTER_NODEID>=0)
 	{
 		DBexecute("insert into history_sync (nodeid,clock,itemid,value) values (%d,%d," ZBX_FS_UI64 ",%f)",
 			get_nodeid_by_id(itemid),clock,itemid,value);
 	}
 
-	DBadd_trend(itemid, value, clock);
-
 	return SUCCEED;
 }
 
@@ -1361,6 +1361,12 @@ int	DBadd_history_uint(zbx_uint64_t itemid, zbx_uint64_t value, int clock)
 
 	DBadd_trend(itemid, (double)value, clock);
 
+	if(CONFIG_MASTER_NODEID>=0)
+	{
+		DBexecute("insert into history_uint_sync (nodeid,clock,itemid,value) values (%d,%d," ZBX_FS_UI64 "," ZBX_FS_UI64 ")",
+			get_nodeid_by_id(itemid),clock,itemid,value);
+	}
+
 	return SUCCEED;
 }
 
@@ -1374,6 +1380,12 @@ int	DBadd_history_str(zbx_uint64_t itemid, char *value, int clock)
 	DBexecute("insert into history_str (clock,itemid,value) values (%d," ZBX_FS_UI64 ",'%s')",
 		clock,itemid,value_esc);
 
+	if(CONFIG_MASTER_NODEID>=0)
+	{
+		DBexecute("insert into history_str_sync (nodeid,clock,itemid,value) values (%d,%d," ZBX_FS_UI64 ",'%s')",
+			get_nodeid_by_id(itemid),clock,itemid,value_esc);
+	}
+
 	return SUCCEED;
 }
 
diff --git a/src/zabbix_server/nodewatcher/history.c b/src/zabbix_server/nodewatcher/history.c
index 53828ba4..69dcfaa4 100644
--- a/src/zabbix_server/nodewatcher/history.c
+++ b/src/zabbix_server/nodewatcher/history.c
@@ -56,6 +56,148 @@ extern	int	CONFIG_NODEID;
 
 /******************************************************************************
  *                                                                            *
+ * Function: process_node_history_str                                         *
+ *                                                                            *
+ * Purpose: process new history_str data                                      *
+ *                                                                            *
+ * Parameters:                                                                *
+ *                                                                            *
+ * Return value: SUCCESS - processed succesfully                              * 
+ *               FAIL - an error occured                                      *
+ *                                                                            *
+ * Author: Alexei Vladishev                                                   *
+ *                                                                            *
+ * Comments:                                                                  *
+ *                                                                            *
+ ******************************************************************************/
+static int process_node_history_str(int nodeid, int master_nodeid)
+{
+	DB_RESULT	result;
+	DB_ROW		row;
+	char		*data;
+	char		tmp[MAX_STRING_LEN];
+	char		sql[MAX_STRING_LEN];
+	int		found = 0;
+
+	zbx_uint64_t	id;
+	zbx_uint64_t	itemid;
+
+#define DATA_MAX	1024*1024
+
+//	zabbix_log( LOG_LEVEL_WARNING, "In process_node(local:%d, event_lastid:" ZBX_FS_UI64 ")",nodeid, event_lastid);
+	/* Begin work */
+
+	data = malloc(DATA_MAX);
+	memset(data,0,DATA_MAX);
+
+	zbx_snprintf(tmp,sizeof(tmp),"History|%d|%d\n", CONFIG_NODEID, nodeid);
+	strncat(data,tmp,DATA_MAX);
+
+	zbx_snprintf(sql,sizeof(sql),"select id,itemid,clock,value from history_str_sync where nodeid=%d order by id", nodeid);
+
+	result = DBselectN(sql, 100000);
+	while((row=DBfetch(result)))
+	{
+		ZBX_STR2UINT64(id,row[0])
+		ZBX_STR2UINT64(itemid,row[1])
+//		zabbix_log( LOG_LEVEL_WARNING, "Processing itemid " ZBX_FS_UI64, itemid);
+		found = 1;
+		zbx_snprintf(tmp,sizeof(tmp),"%d|%s|%s|%s\n", ZBX_TABLE_HISTORY_STR,row[1],row[2],row[3]);
+		strncat(data,tmp,DATA_MAX);
+	}
+	if(found == 1)
+	{
+//		zabbix_log( LOG_LEVEL_WARNING, "Sending [%s]",data);
+		if(send_to_node(master_nodeid, nodeid, data) == SUCCEED)
+		{
+//			zabbix_log( LOG_LEVEL_WARNING, "Updating nodes.history_lastid");
+			DBexecute("update nodes set history_str_lastid=" ZBX_FS_UI64 " where nodeid=%d", id, nodeid);
+			DBexecute("delete from history_str_sync where nodeid=%d and id<=" ZBX_FS_UI64, nodeid, id);
+		}
+		else
+		{
+			zabbix_log( LOG_LEVEL_WARNING, "Not updating nodes.history_str_lastid");
+		}
+	}
+	DBfree_result(result);
+	free(data);
+
+	return SUCCEED;
+}
+
+/******************************************************************************
+ *                                                                            *
+ * Function: process_node_history_uint                                        *
+ *                                                                            *
+ * Purpose: process new history_uint data                                     *
+ *                                                                            *
+ * Parameters:                                                                *
+ *                                                                            *
+ * Return value: SUCCESS - processed succesfully                              * 
+ *               FAIL - an error occured                                      *
+ *                                                                            *
+ * Author: Alexei Vladishev                                                   *
+ *                                                                            *
+ * Comments:                                                                  *
+ *                                                                            *
+ ******************************************************************************/
+static int process_node_history_uint(int nodeid, int master_nodeid)
+{
+	DB_RESULT	result;
+	DB_ROW		row;
+	char		*data;
+	char		tmp[MAX_STRING_LEN];
+	char		sql[MAX_STRING_LEN];
+	int		found = 0;
+
+	zbx_uint64_t	id;
+	zbx_uint64_t	itemid;
+
+#define DATA_MAX	1024*1024
+
+//	zabbix_log( LOG_LEVEL_WARNING, "In process_node(local:%d, event_lastid:" ZBX_FS_UI64 ")",nodeid, event_lastid);
+	/* Begin work */
+
+	data = malloc(DATA_MAX);
+	memset(data,0,DATA_MAX);
+
+	zbx_snprintf(tmp,sizeof(tmp),"History|%d|%d\n", CONFIG_NODEID, nodeid);
+	strncat(data,tmp,DATA_MAX);
+
+	zbx_snprintf(sql,sizeof(sql),"select id,itemid,clock,value from history_uint_sync where nodeid=%d order by id", nodeid);
+
+	result = DBselectN(sql, 100000);
+	while((row=DBfetch(result)))
+	{
+		ZBX_STR2UINT64(id,row[0])
+		ZBX_STR2UINT64(itemid,row[1])
+//		zabbix_log( LOG_LEVEL_WARNING, "Processing itemid " ZBX_FS_UI64, itemid);
+		found = 1;
+		zbx_snprintf(tmp,sizeof(tmp),"%d|%s|%s|%s\n", ZBX_TABLE_HISTORY_UINT,row[1],row[2],row[3]);
+		strncat(data,tmp,DATA_MAX);
+	}
+	if(found == 1)
+	{
+//		zabbix_log( LOG_LEVEL_WARNING, "Sending [%s]",data);
+		if(send_to_node(master_nodeid, nodeid, data) == SUCCEED)
+		{
+//			zabbix_log( LOG_LEVEL_WARNING, "Updating nodes.history_lastid");
+			DBexecute("update nodes set history_uint_lastid=" ZBX_FS_UI64 " where nodeid=%d", id, nodeid);
+			DBexecute("delete from history_uint_sync where nodeid=%d and id<=" ZBX_FS_UI64, nodeid, id);
+		}
+		else
+		{
+			zabbix_log( LOG_LEVEL_WARNING, "Not updating nodes.history_uint_lastid");
+		}
+	}
+	DBfree_result(result);
+	free(data);
+
+	return SUCCEED;
+}
+
+/******************************************************************************
+ *                                                                            *
  * Function: process_node_history                                             *
  *                                                                            *
  * Purpose: process new history data                                          *
@@ -102,7 +244,7 @@ static int process_node_history(int nodeid, int master_nodeid)
 		ZBX_STR2UINT64(itemid,row[1])
 //		zabbix_log( LOG_LEVEL_WARNING, "Processing itemid " ZBX_FS_UI64, itemid);
 		found = 1;
-		zbx_snprintf(tmp,sizeof(tmp),"%s|%s|%s\n", row[1],row[2],row[3]);
+		zbx_snprintf(tmp,sizeof(tmp),"%d|%s|%s|%s\n", ZBX_TABLE_HISTORY,row[1],row[2],row[3]);
 		strncat(data,tmp,DATA_MAX);
 	}
 	if(found == 1)
@@ -146,9 +288,9 @@ static void process_node(int nodeid, int master_nodeid)
 //	zabbix_log( LOG_LEVEL_WARNING, "In process_node(local:%d, master_nodeid:" ZBX_FS_UI64 ")",nodeid, master_nodeid);
 
 	process_node_history(nodeid, master_nodeid);
-/*	process_node_history_uint(nodeid, master_nodeid);
+	process_node_history_uint(nodeid, master_nodeid);
 	process_node_history_str(nodeid, master_nodeid);
-	process_node_history_log(nodeid, master_nodeid);
+/*	process_node_history_log(nodeid, master_nodeid);
 	process_node_trends(nodeid, master_nodeid);*/
 }
 
diff --git a/src/zabbix_server/trapper/nodehistory.c b/src/zabbix_server/trapper/nodehistory.c
index 2fdfa700..57560bbc 100644
--- a/src/zabbix_server/trapper/nodehistory.c
+++ b/src/zabbix_server/trapper/nodehistory.c
@@ -64,20 +64,51 @@ extern int     process_event(DB_EVENT *event);
 static int	process_record(int nodeid, char *record)
 {
 	char	tmp[MAX_STRING_LEN];
+	int		table;
 	zbx_uint64_t	itemid;
 	int		timestamp;
 	double		value;
+	zbx_uint64_t	value_uint;
+	int		res = FAIL;
 
 	zabbix_log( LOG_LEVEL_DEBUG, "In process_record [%s]", record);
 
 	zbx_get_field(record,tmp,0,'|');
-	sscanf(tmp,ZBX_FS_UI64,&itemid);
-	zbx_get_field(record,tmp,1,'|');
-	timestamp=atoi(tmp);
-	zbx_get_field(record,tmp,2,'|');
-	value=atof(tmp);
+	table=atoi(tmp);
+	if(table == ZBX_TABLE_HISTORY)
+	{
+		zbx_get_field(record,tmp,1,'|');
+		sscanf(tmp,ZBX_FS_UI64,&itemid);
+		zbx_get_field(record,tmp,2,'|');
+		timestamp=atoi(tmp);
+		zbx_get_field(record,tmp,3,'|');
+		value=atof(tmp);
+
+		res =  DBadd_history(itemid, value, timestamp);
+	}
+	else if(table == ZBX_TABLE_HISTORY_UINT)
+	{
+		zbx_get_field(record,tmp,1,'|');
+		sscanf(tmp,ZBX_FS_UI64,&itemid);
+		zbx_get_field(record,tmp,2,'|');
+		timestamp=atoi(tmp);
+		zbx_get_field(record,tmp,3,'|');
+		sscanf(tmp,ZBX_FS_UI64,&value_uint);
+
+		res =  DBadd_history_uint(itemid, value_uint, timestamp);
+	}
+	else if(table == ZBX_TABLE_HISTORY_STR)
+	{
+		zbx_get_field(record,tmp,1,'|');
+		sscanf(tmp,ZBX_FS_UI64,&itemid);
+		zbx_get_field(record,tmp,2,'|');
+		timestamp=atoi(tmp);
+		zbx_get_field(record,tmp,3,'|');
+
+		res =  DBadd_history_str(itemid, tmp, timestamp);
+	}
 
-	return DBadd_history(itemid, value, clock);
+	return res;
 }
 
 /******************************************************************************