将Redis状态视为阀门状态，加入状态监控（主从备份）;feature-reference-count

1 files changed, 118 insertions, 32 deletions

diff --git a/src/pg_valve_main.cpp b/src/pg_valve_main.cpp
index 3e75b90..631c143 100644
--- a/src/pg_valve_main.cpp
+++ b/src/pg_valve_main.cpp
@@ -4,6 +4,7 @@
 #include <sys/time.h>
 #include <sys/stat.h>
 #include <sys/types.h>
+#include <sys/prctl.h>
 #include <errno.h>
 #include <assert.h>
 #include <fcntl.h>
@@ -725,7 +726,7 @@ static int register_field_stat(map<string, configure_table_t *> &tables_one, map
 	return 0;
 }
 
-void* thread_consul_status(void *arg)
+static void* thread_consul_status(void *arg)
 {
 	const char *session_id = (char *)arg;
 	bool retry;
@@ -785,8 +786,82 @@ retry_session:
 	return NULL;
 }
 
-void* thread_asmis_log(void *arg)
+static bool check_redis_server_state(struct maat_table_relation *table_maat_lists, int maat_num)
 {
+	struct timeval tv;
+	redisReply *reply;
+
+	tv.tv_sec = 5;
+	tv.tv_usec = 0;
+	for(int i=0; i<maat_num; i++)
+	{
+		if(table_maat_lists[i].ctx == NULL)
+		{
+			table_maat_lists[i].ctx = redisConnectWithTimeout(table_maat_lists[i].redisip, table_maat_lists[i].redisport, tv);
+			if(table_maat_lists[i].ctx==NULL||table_maat_lists[i].ctx->err)
+			{
+				if(table_maat_lists[i].ctx!=NULL)
+				{
+					redisFree(table_maat_lists[i].ctx);
+					table_maat_lists[i].ctx = NULL;
+				}
+				MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "redis connect %s:%u failed.",
+					table_maat_lists[i].redisip, table_maat_lists[i].redisport);
+				return false;
+			}
+		}
+		reply = (redisReply *)redisCommand(table_maat_lists[i].ctx, "PING");
+		if(NULL == reply)
+		{
+			redisFree(table_maat_lists[i].ctx);
+			table_maat_lists[i].ctx = NULL;
+			MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "redis ping %s:%u failed.",
+				table_maat_lists[i].redisip, table_maat_lists[i].redisport);
+			return false;
+		}
+		freeReplyObject(reply);
+	}
+
+	return true;
+}
+
+struct __maat_redis_list
+{
+	struct maat_table_relation *maat_lists;
+	int num;
+};
+static void* thread_check_redis_server_periodically(void *arg)
+{
+	struct __maat_redis_list *table_maat_lists=(struct __maat_redis_list *)arg;
+	int retry;
+
+	prctl(PR_SET_NAME, "pgvalve_chkredis");
+	while(1)
+	{
+		retry = 0;
+retry_check:
+		if(!check_redis_server_state(table_maat_lists->maat_lists, table_maat_lists->num))
+		{
+			if(retry == 0)
+			{
+				retry = 1;
+				goto retry_check;
+			}
+			else
+			{
+				MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "redis check failed, restart to become slave now.");
+				exit(21);
+			}
+		}
+		sleep(10);
+	}
+
+	return NULL;
+}
+
+static void* thread_asmis_log(void *arg)
+{
+	prctl(PR_SET_NAME, "pgvalve_asmis");
 	while(1)
 	{
 		sleep(g_pgvalve_info.asmis_heart_intvl);
@@ -812,10 +887,49 @@ int main(int argc, char *argv[])
 		MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "read_conf_and_init/init_asmis_log_handle fail.");
 		return -1;
 	}
+	g_pgvalve_info.htable_stat_log = init_and_create_htable(256, 0, NULL);
+	g_pgvalve_info.htable_stat_serv = init_and_create_htable(256, 0, NULL);
+	g_pgvalve_info.hdl_reference = init_and_create_htable(g_pgvalve_info.hash_size*2, 0, htable_destroy_node);
+
+	/*��ȡ�������ò���ʼ������ϵ*/
+	if(load_service_map_info())
+	{
+		MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "load_service_map_info fail.");
+		return -1;
+	}
+	if(load_maat_info(&g_table_maat_lists, &g_table_maat_num))
+	{
+		return -1;
+	}
+	if(setup_sw_tables_relationship(tables, tables_sw, tables_one))
+	{
+		return -2;
+	}
+	if(setup_direct_tables(tables, tables_one))
+	{
+		return -3;
+	}
+	
+	assert(tables.size() == tables_one.size()+tables_sw.size());
+	if(register_field_stat(tables_one, tables_sw, g_pgvalve_info.root_log_dir, g_pgvalve_info.fs_dst_ip, g_pgvalve_info.fs_dst_port))
+	{
+		MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "register_field_stat fail.");
+		return -1;
+	}
 
 	pthread_attr_init(&attr);
 	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);	
 	pthread_create(&thread_desc, &attr, thread_asmis_log, NULL);
+	/*���Redis����˽���״̬��OK�����½���*/
+	while(!check_redis_server_state(g_table_maat_lists, g_table_maat_num))
+	{
+		sleep(5);
+	}
+	MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "check redis server state OK.");
+	struct __maat_redis_list maat_redis_list={g_table_maat_lists, g_table_maat_num};
+	pthread_create(&thread_desc, &attr, thread_check_redis_server_periodically, &maat_redis_list);
+	
+	/*��ȡ����ɫ��ֻ���������·����ã��ұ����·�����*/
 	if(g_pgvalve_info.consul_sw)
 	{
 		time_t wait_seconds=0;
@@ -843,36 +957,7 @@ int main(int argc, char *argv[])
 		exit(20);
 	}
 	
-	g_pgvalve_info.htable_stat_log = init_and_create_htable(256, 0, NULL);
-	g_pgvalve_info.htable_stat_serv = init_and_create_htable(256, 0, NULL);
-	g_pgvalve_info.hdl_reference = init_and_create_htable(g_pgvalve_info.hash_size*2, 0, htable_destroy_node);
-
-	if(load_service_map_info())
-	{
-		MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "load_service_map_info fail.");
-		return -1;
-	}
-	
-	if(load_maat_info(&g_table_maat_lists, &g_table_maat_num))
-	{
-		return -1;
-	}
-	if(setup_sw_tables_relationship(tables, tables_sw, tables_one))
-	{
-		return -2;
-	}
-	if(setup_direct_tables(tables, tables_one))
-	{
-		return -3;
-	}
-	
-	assert(tables.size() == tables_one.size()+tables_sw.size());
-	if(register_field_stat(tables_one, tables_sw, g_pgvalve_info.root_log_dir, g_pgvalve_info.fs_dst_ip, g_pgvalve_info.fs_dst_port))
-	{
-		MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "register_field_stat fail.");
-		return -1;
-	}
-
+	/*��ȡ����ɫ�ɹ��������������û�ȡ���·�*/
 	struct __tables_map_ref table_map = {tables_one, tables_sw};	
 	if(pthread_create(&thread_desc, &attr, thread_dispatch_full_config, (void *)&table_map))
 	{
@@ -917,6 +1002,7 @@ int main(int argc, char *argv[])
 	}
 	MESA_HANDLE_RUNTIME_LOGV2(g_pgvalve_info.log_runtime, RLOG_LV_FATAL, MODULE_NAME, "pangu_valve starts all tables full config over.\n");
 
+	/*��ʱˢ��ͳ������*/
 	time_t now, remain;
 	while(1)
 	{