Add status change time

knizhnik · knizhnik · commit 4760ea1e8931 · 2016-04-11T20:32:43.000+03:00
diff --git a/contrib/mmts/multimaster--1.0.sql b/contrib/mmts/multimaster--1.0.sql
@@ -24,7 +24,7 @@ AS 'MODULE_PATHNAME','mtm_get_snapshot'
 LANGUAGE C;
 
 
-CREATE TYPE mtm.node_state AS (id integer, disabled bool, disconnected bool, catchUp bool, slotLag bigint, avgTransDelay bigint, connStr text);
+CREATE TYPE mtm.node_state AS (id integer, disabled bool, disconnected bool, catchUp bool, slotLag bigint, avgTransDelay bigint, lastStatusChange timestamp, connStr text);
 
 CREATE FUNCTION mtm.get_nodes_state() RETURNS SETOF mtm.node_state
 AS 'MODULE_PATHNAME','mtm_get_nodes_state'
diff --git a/contrib/mmts/multimaster.c b/contrib/mmts/multimaster.c
@@ -179,6 +179,7 @@ int   MtmConnectAttempts;
 int   MtmConnectTimeout;
 int   MtmKeepaliveTimeout;
 int   MtmReconnectAttempts;
+int   MtmNodeDisableDelay;
 bool  MtmUseRaftable;
 MtmConnectionInfo* MtmConnections;
 
@@ -993,6 +994,7 @@ void MtmRecoveryCompleted(void)
 	MtmLock(LW_EXCLUSIVE);
 	Mtm->recoverySlot = 0;
 	BIT_CLEAR(Mtm->disabledNodeMask, MtmNodeId-1);
+	Mtm->nodes[MtmNodeId-1].lastStatusChangeTime = time(NULL);
 	/* Mode will be changed to online once all locagical reciever are connected */
 	MtmSwitchClusterMode(MTM_CONNECTED);
 	MtmUnlock();
@@ -1081,6 +1083,7 @@ bool MtmRecoveryCaughtUp(int nodeId, XLogRecPtr slotLSN)
 				/* We are lucky: caugth-up without locking cluster! */
 			}
 			BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
+			Mtm->nodes[nodeId-1].lastStatusChangeTime = time(NULL);
 			Mtm->nNodes += 1;
 			caughtUp = true;
 		} else if (!BIT_CHECK(Mtm->nodeLockerMask, nodeId-1)
@@ -1223,13 +1226,15 @@ bool MtmRefreshClusterStatus(bool nowait)
 			if (mask & 1) { 
 				Mtm->nNodes -= 1;
 				BIT_SET(Mtm->disabledNodeMask, i);
+				Mtm->nodes[i].lastStatusChangeTime = time(NULL);
 			}
 		}
 		mask = clique & Mtm->disabledNodeMask; /* new enabled nodes mask */		
 		for (i = 0; mask != 0; i++, mask >>= 1) {
 			if (mask & 1) { 
 				Mtm->nNodes += 1;
 				BIT_CLEAR(Mtm->disabledNodeMask, i);
+				Mtm->nodes[i].lastStatusChangeTime = time(NULL);
 			}
 		}
 		MtmCheckQuorum();
@@ -1269,6 +1274,11 @@ void MtmOnNodeDisconnect(int nodeId)
 { 
 	MtmTransState *ts;
 
+	if (Mtm->nodes[nodeId-1].lastStatusChangeTime + MtmNodeDisableDelay > time(NULL)) { 
+		/* Avoid false detection of node failure and prevent node status blinking */
+		return;
+	}
+
 	BIT_SET(Mtm->connectivityMask, nodeId-1);
 	BIT_SET(Mtm->reconnectMask, nodeId-1);
 	RaftableSet(psprintf("node-mask-%d", MtmNodeId), &Mtm->connectivityMask, sizeof Mtm->connectivityMask, false);
@@ -1279,6 +1289,7 @@ void MtmOnNodeDisconnect(int nodeId)
 	if (!MtmRefreshClusterStatus(false)) { 
 		MtmLock(LW_EXCLUSIVE);
 		if (!BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) { 
+			Mtm->nodes[nodeId-1].lastStatusChangeTime = time(NULL);
 			BIT_SET(Mtm->disabledNodeMask, nodeId-1);
 			Mtm->nNodes -= 1;
 			MtmCheckQuorum();
@@ -1446,6 +1457,7 @@ static void MtmInitialize()
 		for (i = 0; i < MtmNodes; i++) {
 			Mtm->nodes[i].oldestSnapshot = 0;
 			Mtm->nodes[i].transDelay = 0;
+			Mtm->nodes[i].lastStatusChangeTime = time(NULL);
 			Mtm->nodes[i].con = MtmConnections[i];
 		}
 		PGSemaphoreCreate(&Mtm->votingSemaphore);
@@ -1566,10 +1578,25 @@ _PG_init(void)
 	if (!process_shared_preload_libraries_in_progress)
 		return;
 
+	DefineCustomIntVariable(
+		"multimaster.node_disable_delay",
+		"Minamal amount of time (sec) between node status change",
+		"This delay is used to avoid false detection of node failure and to prevent blinking of node status node",
+		&MtmNodeDisableDelay,
+		1,
+		1,
+		INT_MAX,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomIntVariable(
 		"multimaster.min_recovery_lag",
 		"Minamal lag of WAL-sender performing recovery after which cluster is locked until recovery is completed",
-		"When wal-sender almost catch-up WAL current position we need to stop 'Achilles tortile compeition' and "
+		"When wal-sender almost catch-up WAL current position we need to stop 'Achilles tortile competition' and "
 		"temporary stop commit of new transactions until node will be completely repared",
 		&MtmMinRecoveryLag,
 		100000,
@@ -1891,6 +1918,7 @@ void MtmDropNode(int nodeId, bool dropSlot)
 		{ 
 			elog(ERROR, "NodeID %d is out of range [1,%d]", nodeId, Mtm->nNodes);
 		}
+		Mtm->nodes[nodeId-1].lastStatusChangeTime = time(NULL);
 		BIT_SET(Mtm->disabledNodeMask, nodeId-1);
 		Mtm->nNodes -= 1;
 		MtmCheckQuorum();
@@ -1941,13 +1969,15 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 	if (MtmIsRecoverySession) {
 		MTM_LOG1("%d: Node %d start recovery of node %d", MyProcPid, MtmNodeId, MtmReplicationNodeId);
 		if (!BIT_CHECK(Mtm->disabledNodeMask,  MtmReplicationNodeId-1)) {
+			Mtm->nodes[MtmReplicationNodeId-1].lastStatusChangeTime = time(NULL);
 			BIT_SET(Mtm->disabledNodeMask,  MtmReplicationNodeId-1);
 			Mtm->nNodes -= 1;			
 			MtmCheckQuorum();
 		}
 	} else if (BIT_CHECK(Mtm->disabledNodeMask,  MtmReplicationNodeId-1)) {
 		if (recoveryCompleted) { 
 			MTM_LOG1("Node %d consider that recovery of node %d is completed: start normal replication", MtmNodeId, MtmReplicationNodeId); 
+			Mtm->nodes[MtmReplicationNodeId-1].lastStatusChangeTime = time(NULL);
 			BIT_CLEAR(Mtm->disabledNodeMask,  MtmReplicationNodeId-1);
 			Mtm->nNodes += 1;
 			MtmCheckQuorum();
@@ -2058,8 +2088,8 @@ typedef struct
 	int       nodeId;
 	char*     connStrPtr;
 	TupleDesc desc;
-    Datum     values[7];
-    bool      nulls[7];
+    Datum     values[8];
+    bool      nulls[8];
 } MtmGetNodeStateCtx;
 
 Datum
@@ -2096,11 +2126,12 @@ mtm_get_nodes_state(PG_FUNCTION_ARGS)
 	usrfctx->values[4] = Int64GetDatum(lag);
 	usrfctx->nulls[4] = lag < 0;
 	usrfctx->values[5] = Int64GetDatum(Mtm->transCount ? Mtm->nodes[usrfctx->nodeId-1].transDelay/Mtm->transCount : 0);
+	usrfctx->values[6] = TimestampTzGetDatum(time_t_to_timestamptz(Mtm->nodes[usrfctx->nodeId-1].lastStatusChangeTime));
 	p = strchr(usrfctx->connStrPtr, ',');
 	if (p != NULL) { 
 		*p++ = '\0';
 	}
-	usrfctx->values[6] = CStringGetTextDatum(usrfctx->connStrPtr);
+	usrfctx->values[7] = CStringGetTextDatum(usrfctx->connStrPtr);
 	usrfctx->connStrPtr = p;
 	usrfctx->nodeId += 1;
 
diff --git a/contrib/mmts/multimaster.h b/contrib/mmts/multimaster.h
@@ -119,6 +119,7 @@ typedef struct
 {
 	MtmConnectionInfo con;
 	time_t transDelay;
+	time_t lastStatusChangeTime;
 	csn_t  oldestSnapshot; /* Oldest snapshot used by active transactions at this node */
 } MtmNodeInfo;
 
@@ -170,7 +171,6 @@ typedef struct
     MtmTransState** transListTail;     /* Tail of L1 list of all finished transactionds, used to append new elements.
 								  		  This list is expected to be in CSN ascending order, by strict order may be violated */
 	uint64 transCount;                 /* Counter of transactions perfromed by this node */	
-	time_t nodeTransDelay[MAX_NODES];  /* Time of waiting transaction acknowledgment from node */
     BgwPool pool;                      /* Pool of background workers for applying logical replication patches */
 	MtmNodeInfo nodes[1];              /* [MtmNodes]: per-node data */ 
 } MtmState;
@@ -190,6 +190,7 @@ extern int   MtmConnectAttempts;
 extern int   MtmConnectTimeout;
 extern int   MtmReconnectAttempts;
 extern int   MtmKeepaliveTimeout;
+extern int   MtmNodeDisableDelay;
 extern HTAB* MtmXid2State;
 
 extern MtmConnectionInfo* MtmConnections;
diff --git a/contrib/mmts/t/001_basic_recovery.pl b/contrib/mmts/t/001_basic_recovery.pl
@@ -115,8 +115,8 @@ sub PostgresNode::inet_connstr {
 
 diag("starting node 2");
 $nodes[2]->start;
-diag("sleeping 30");
-sleep(30); # XXX: here we can poll
+diag("sleeping 10");
+sleep(10); # XXX: here we can poll
 diag("inserting 3");
 $nodes[0]->psql('postgres', "insert into t values(3, 30);");
 diag("selecting");
