Track nodes current timeline on the monitor. (#730)

This allows comparing the LSN with the added notion of the current time line for
the nodes, avoiding some advanced data loss hazard if a node makes progress
while being in a network split with the application nodes, and when
rejoining later we should detect if a failover happened on the other nodes
on the system.

Author: DimCitus

src/bin/pg_autoctl/cli_do_monitor.c

@@ -572,6 +572,7 @@ cli_do_monitor_node_active(int argc, char **argv)
 							 keeper.state.current_group,
 							 keeper.state.current_role,
 							 keeper.postgres.pgIsRunning,
+							 keeper.postgres.postgresSetup.control.timeline_id,
 							 keeper.postgres.currentLSN,
 							 keeper.postgres.pgsrSyncState,
 							 &assignedState))

src/bin/pg_autoctl/cli_show.c

@@ -557,6 +557,9 @@ cli_show_local_state()
 					strlcpy(nodeState.node.lsn, "0/0", PG_LSN_MAXLENGTH);
 				}
 
+				nodeState.node.tli =
+					keeper.postgres.postgresSetup.control.timeline_id;
+
 				strlcpy(nodeState.node.lsn,
 						keeper.postgres.currentLSN,
 						PG_LSN_MAXLENGTH);
@@ -569,6 +572,7 @@ cli_show_local_state()
 					/* errors have already been logged, just continue */
 				}
 
+				nodeState.node.tli = config.pgSetup.control.timeline_id;
 				strlcpy(nodeState.node.lsn,
 						config.pgSetup.control.latestCheckpointLSN,
 						PG_LSN_MAXLENGTH);

src/bin/pg_autoctl/defaults.h

@@ -14,7 +14,7 @@
 #define PG_AUTOCTL_STATE_VERSION 1
 
 /* additional version information for printing version on CLI */
-#define PG_AUTOCTL_VERSION "1.6.0.1"
+#define PG_AUTOCTL_VERSION "1.6.0.2"
 
 /* version of the extension that we requite to talk to on the monitor */
 #define PG_AUTOCTL_EXTENSION_VERSION "1.6"

src/bin/pg_autoctl/fsm.c

@@ -428,6 +428,7 @@ keeper_fsm_step(Keeper *keeper)
 							 keeperState->current_group,
 							 keeperState->current_role,
 							 postgres->pgIsRunning,
+							 postgres->postgresSetup.control.timeline_id,
 							 postgres->currentLSN,
 							 postgres->pgsrSyncState,
 							 &assignedState))

src/bin/pg_autoctl/keeper.c

@@ -1608,12 +1608,15 @@ keeper_register_again(Keeper *keeper)
 			return false;
 		}
 
+		int currentTLI = keeper->postgres.postgresSetup.control.timeline_id;
+
 		if (!monitor_node_active(monitor,
 								 config->formation,
 								 assignedState.nodeId,
 								 assignedState.groupId,
 								 assignedState.state,
 								 ReportPgIsRunning(keeper),
+								 currentTLI,
 								 keeper->postgres.currentLSN,
 								 keeper->postgres.pgsrSyncState,
 								 &assignedState))

src/bin/pg_autoctl/keeper_pg_init.c

@@ -551,6 +551,7 @@ wait_until_primary_is_ready(Keeper *keeper,
 							MonitorAssignedState *assignedState)
 {
 	bool pgIsRunning = false;
+	int currentTLI = 1;
 	char currrentLSN[PG_LSN_MAXLENGTH] = "0/0";
 	char *pgsrSyncState = "";
 	int errors = 0, tries = 0;
@@ -593,6 +594,7 @@ wait_until_primary_is_ready(Keeper *keeper,
 								 keeper->state.current_group,
 								 keeper->state.current_role,
 								 pgIsRunning,
+								 currentTLI,
 								 currrentLSN,
 								 pgsrSyncState,
 								 assignedState))
@@ -1068,6 +1070,7 @@ keeper_pg_init_node_active(Keeper *keeper)
 							 keeper->state.current_group,
 							 keeper->state.current_role,
 							 ReportPgIsRunning(keeper),
+							 keeper->postgres.postgresSetup.control.timeline_id,
 							 keeper->postgres.currentLSN,
 							 keeper->postgres.pgsrSyncState,
 							 &assignedState))

src/bin/pg_autoctl/monitor.c

@@ -898,19 +898,20 @@ bool
 monitor_node_active(Monitor *monitor,
 					char *formation, int64_t nodeId,
 					int groupId, NodeState currentState,
-					bool pgIsRunning,
+					bool pgIsRunning, int currentTLI,
 					char *currentLSN, char *pgsrSyncState,
 					MonitorAssignedState *assignedState)
 {
 	PGSQL *pgsql = &monitor->pgsql;
 	const char *sql =
 		"SELECT * FROM pgautofailover.node_active($1, $2, $3, "
-		"$4::pgautofailover.replication_state, $5, $6, $7)";
-	int paramCount = 7;
-	Oid paramTypes[7] = {
-		TEXTOID, INT8OID, INT4OID, TEXTOID, BOOLOID, LSNOID, TEXTOID
+		"$4::pgautofailover.replication_state, $5, $6, $7, $8)";
+	int paramCount = 8;
+	Oid paramTypes[8] = {
+		TEXTOID, INT8OID, INT4OID, TEXTOID,
+		BOOLOID, INT4OID, LSNOID, TEXTOID
 	};
-	const char *paramValues[7];
+	const char *paramValues[8];
 	MonitorAssignedStateParseContext parseContext =
 	{ { 0 }, assignedState, false };
 	const char *nodeStateString = NodeStateToString(currentState);
@@ -920,8 +921,9 @@ monitor_node_active(Monitor *monitor,
 	paramValues[2] = intToString(groupId).strValue;
 	paramValues[3] = nodeStateString;
 	paramValues[4] = pgIsRunning ? "true" : "false";
-	paramValues[5] = currentLSN;
-	paramValues[6] = pgsrSyncState;
+	paramValues[5] = intToString(currentTLI).strValue;
+	paramValues[6] = currentLSN;
+	paramValues[7] = pgsrSyncState;
 
 	if (!pgsql_execute_with_params(pgsql, sql,
 								   paramCount, paramTypes, paramValues,
@@ -1755,7 +1757,7 @@ parseCurrentNodeState(PGresult *result, int rowNumber,
 	int errors = 0;
 
 	/* we don't expect any of the column to be NULL */
-	for (colNumber = 0; colNumber < 12; colNumber++)
+	for (colNumber = 0; colNumber < 13; colNumber++)
 	{
 		if (PQgetisnull(result, rowNumber, 0))
 		{
@@ -1776,8 +1778,9 @@ parseCurrentNodeState(PGresult *result, int rowNumber,
 	 *  7 - OUT assigned_group_state pgautofailover.replication_state,
 	 *  8 - OUT candidate_priority	 int,
 	 *  9 - OUT replication_quorum	 bool,
-	 * 10 - OUT reported_lsn         pg_lsn,
-	 * 11 - OUT health               integer
+	 * 10 - OUT reported_tli         int,
+	 * 11 - OUT reported_lsn         pg_lsn,
+	 * 12 - OUT health               integer
 	 *
 	 * We need the groupId to parse the formation kind into a nodeKind, so we
 	 * begin at column 1 and get back to column 0 later, after column 4.
@@ -1884,11 +1887,18 @@ parseCurrentNodeState(PGresult *result, int rowNumber,
 		nodeState->replicationQuorum = (*value) == 't';
 	}
 
-	/* we trust Postgres pg_lsn data type to fit in our PG_LSN_MAXLENGTH */
 	value = PQgetvalue(result, rowNumber, 10);
-	strlcpy(nodeState->node.lsn, value, PG_LSN_MAXLENGTH);
+	if (!stringToInt(value, &(nodeState->node.tli)))
+	{
+		log_error("Invalid timeline \"%s\" returned by monitor", value);
+		++errors;
+	}
 
+	/* we trust Postgres pg_lsn data type to fit in our PG_LSN_MAXLENGTH */
 	value = PQgetvalue(result, rowNumber, 11);
+	strlcpy(nodeState->node.lsn, value, PG_LSN_MAXLENGTH);
+
+	value = PQgetvalue(result, rowNumber, 12);
 	if (!stringToInt(value, &(nodeState->health)))
 	{
 		log_error("Invalid node health \"%s\" returned by monitor", value);
@@ -1921,9 +1931,9 @@ parseCurrentNodeStateArray(CurrentNodeStateArray *nodesArray, PGresult *result)
 	}
 
 	/* pgautofailover.current_state returns 11 columns */
-	if (PQnfields(result) != 12)
+	if (PQnfields(result) != 13)
 	{
-		log_error("Query returned %d columns, expected 12", PQnfields(result));
+		log_error("Query returned %d columns, expected 13", PQnfields(result));
 		return false;
 	}
 
@@ -3751,6 +3761,7 @@ monitor_check_report_state(void *context, CurrentNodeState *nodeState)
 	char timestring[MAXCTIMESIZE] = { 0 };
 	char hostport[BUFSIZE] = { 0 };
 	char composedId[BUFSIZE] = { 0 };
+	char tliLSN[BUFSIZE] = { 0 };
 
 	/* filter notifications for our own formation */
 	if (strcmp(nodeState->formation, ctx->formation) != 0 ||
@@ -3769,7 +3780,8 @@ monitor_check_report_state(void *context, CurrentNodeState *nodeState)
 								&(nodeState->node),
 								ctx->groupId,
 								hostport,
-								composedId);
+								composedId,
+								tliLSN);
 
 	fformat(stdout, "%8s | %*s | %*s | %*s | %19s | %19s\n",
 			timestring + 11,
@@ -3899,6 +3911,7 @@ monitor_check_node_report_state(void *context, CurrentNodeState *nodeState)
 	char timestring[MAXCTIMESIZE] = { 0 };
 	char hostport[BUFSIZE] = { 0 };
 	char composedId[BUFSIZE] = { 0 };
+	char tliLSN[BUFSIZE] = { 0 };
 
 	/* filter notifications for our own formation */
 	if (strcmp(nodeState->formation, ctx->formation) != 0 ||
@@ -3917,7 +3930,8 @@ monitor_check_node_report_state(void *context, CurrentNodeState *nodeState)
 								&(nodeState->node),
 								ctx->groupId,
 								hostport,
-								composedId);
+								composedId,
+								tliLSN);
 
 	fformat(stdout, "%8s | %*s | %*s | %*s | %19s | %19s\n",
 			timestring + 11,

src/bin/pg_autoctl/monitor.h

@@ -107,7 +107,7 @@ bool monitor_register_node(Monitor *monitor,
 bool monitor_node_active(Monitor *monitor,
 						 char *formation, int64_t nodeId,
 						 int groupId, NodeState currentState,
-						 bool pgIsRunning,
+						 bool pgIsRunning, int currentTLI,
 						 char *currentLSN, char *pgsrSyncState,
 						 MonitorAssignedState *assignedState);
 bool monitor_get_node_replication_settings(Monitor *monitor,

src/bin/pg_autoctl/nodestate_utils.c

@@ -29,7 +29,7 @@ nodestatePrepareHeaders(CurrentNodeStateArray *nodesArray,
 	nodesArray->headers.maxNameSize = 4;  /* "Name" */
 	nodesArray->headers.maxHostSize = 10; /* "Host:Port" */
 	nodesArray->headers.maxNodeSize = 5;  /* "Node" */
-	nodesArray->headers.maxLSNSize = 3;   /* "LSN" */
+	nodesArray->headers.maxLSNSize = 9;   /* "TLI:  LSN" */
 	nodesArray->headers.maxStateSize = MAX_NODE_STATE_LEN;
 	nodesArray->headers.maxHealthSize = strlen("read-write *");
 
@@ -117,38 +117,21 @@ void
 nodestateAdjustHeaders(NodeAddressHeaders *headers,
 					   NodeAddress *node, int groupId)
 {
-	int nameLen = strlen(node->name);
-
-	/* compute strlen of host:port */
-	IntString portString = intToString(node->port);
-	int hostLen =
-		strlen(node->host) + strlen(portString.strValue) + 1;
-
-	/* compute strlen of groupId/nodeId, as in "0/1" */
-	IntString nodeIdString = intToString(node->nodeId);
-	int nodeLen = 0;
-
-	int lsnLen = strlen(node->lsn);
-
-	switch (headers->nodeKind)
-	{
-		case NODE_KIND_STANDALONE:
-		{
-			nodeLen = strlen(nodeIdString.strValue);
-			break;
-		}
-
-		default:
-		{
-			IntString groupIdString = intToString(groupId);
+	char hostport[BUFSIZE] = { 0 };
+	char composedId[BUFSIZE] = { 0 };
+	char tliLSN[BUFSIZE] = { 0 };
 
-			nodeLen =
-				strlen(groupIdString.strValue) +
-				strlen(nodeIdString.strValue) + 1;
+	(void) nodestatePrepareNode(headers,
+								node,
+								groupId,
+								hostport,
+								composedId,
+								tliLSN);
 
-			break;
-		}
-	}
+	int nameLen = strlen(node->name);
+	int hostLen = strlen(hostport);
+	int nodeLen = strlen(composedId);
+	int lsnLen = strlen(tliLSN);
 
 	/*
 	 * In order to have a static nice table output even when using
@@ -178,8 +161,8 @@ nodestateAdjustHeaders(NodeAddressHeaders *headers,
 
 	if (headers->maxLSNSize == 0)
 	{
-		/* Unknown LSN is going to be "0/0" */
-		headers->maxLSNSize = 3;
+		/* Unknown LSN is going to be "  1: 0/0" */
+		headers->maxLSNSize = 9;
 	}
 
 	if (headers->maxHealthSize == 0)
@@ -224,7 +207,7 @@ nodestatePrintHeader(NodeAddressHeaders *headers)
 			headers->maxNameSize, "Name",
 			headers->maxNodeSize, "Node",
 			headers->maxHostSize, "Host:Port",
-			headers->maxLSNSize, "LSN",
+			headers->maxLSNSize, "TLI: LSN",
 			headers->maxHealthSize, "Connection",
 			headers->maxStateSize, "Current State",
 			headers->maxStateSize, "Assigned State");
@@ -250,14 +233,16 @@ nodestatePrintNodeState(NodeAddressHeaders *headers,
 {
 	char hostport[BUFSIZE] = { 0 };
 	char composedId[BUFSIZE] = { 0 };
+	char tliLSN[BUFSIZE] = { 0 };
 	char connection[BUFSIZE] = { 0 };
 	char healthChar = nodestateHealthToChar(nodeState->health);
 
 	(void) nodestatePrepareNode(headers,
 								&(nodeState->node),
 								nodeState->groupId,
 								hostport,
-								composedId);
+								composedId,
+								tliLSN);
 
 	if (healthChar == ' ')
 	{
@@ -273,7 +258,7 @@ nodestatePrintNodeState(NodeAddressHeaders *headers,
 			headers->maxNameSize, nodeState->node.name,
 			headers->maxNodeSize, composedId,
 			headers->maxHostSize, hostport,
-			headers->maxLSNSize, nodeState->node.lsn,
+			headers->maxLSNSize, tliLSN,
 			headers->maxHealthSize, connection,
 			headers->maxStateSize, NodeStateToString(nodeState->reportedState),
 			headers->maxStateSize, NodeStateToString(nodeState->goalState));
@@ -287,9 +272,11 @@ nodestatePrintNodeState(NodeAddressHeaders *headers,
  */
 void
 nodestatePrepareNode(NodeAddressHeaders *headers, NodeAddress *node,
-					 int groupId, char *hostport, char *composedId)
+					 int groupId, char *hostport,
+					 char *composedId, char *tliLSN)
 {
 	sformat(hostport, BUFSIZE, "%s:%d", node->host, node->port);
+	sformat(tliLSN, BUFSIZE, "%3d: %s", node->tli, node->lsn);
 
 	switch (headers->nodeKind)
 	{
@@ -353,6 +340,8 @@ nodestateAsJSON(CurrentNodeState *nodeState, JSON_Value *js)
 	json_object_set_string(jsobj, "assigned_group_state",
 						   NodeStateToString(nodeState->goalState));
 
+	json_object_set_number(jsobj, "timeline", (double) nodeState->node.tli);
+
 	json_object_set_string(jsobj, "Minimum Recovery Ending LSN",
 						   nodeState->node.lsn);
 
@@ -551,14 +540,14 @@ printNodeArray(NodeAddressArray *nodesArray)
 void
 printNodeHeader(NodeAddressHeaders *headers)
 {
-	fformat(stdout, "%*s | %*s | %*s | %18s | %8s\n",
+	fformat(stdout, "%*s | %*s | %*s | %21s | %8s\n",
 			headers->maxNameSize, "Name",
 			headers->maxNodeSize, "Node",
 			headers->maxHostSize, "Host:Port",
-			"LSN",
+			"TLI: LSN",
 			"Primary?");
 
-	fformat(stdout, "%*s-+-%*s-+-%*s-+-%18s-+-%8s\n",
+	fformat(stdout, "%*s-+-%*s-+-%*s-+-%21s-+-%8s\n",
 			headers->maxNameSize, headers->nameSeparatorHeader,
 			headers->maxNodeSize, headers->nodeSeparatorHeader,
 			headers->maxHostSize, headers->hostSeparatorHeader,
@@ -574,13 +563,14 @@ printNodeEntry(NodeAddressHeaders *headers, NodeAddress *node)
 {
 	char hostport[BUFSIZE] = { 0 };
 	char composedId[BUFSIZE] = { 0 };
+	char tliLSN[BUFSIZE] = { 0 };
 
-	(void) nodestatePrepareNode(headers, node, 0, hostport, composedId);
+	(void) nodestatePrepareNode(headers, node, 0, hostport, composedId, tliLSN);
 
-	fformat(stdout, "%*s | %*s | %*s | %18s | %8s\n",
+	fformat(stdout, "%*s | %*s | %*s | %21s | %8s\n",
 			headers->maxNameSize, node->name,
 			headers->maxNodeSize, composedId,
 			headers->maxHostSize, hostport,
-			node->lsn,
+			tliLSN,
 			node->isPrimary ? "yes" : "no");
 }

src/bin/pg_autoctl/nodestate_utils.h

@@ -83,7 +83,9 @@ void nodestatePrintNodeState(NodeAddressHeaders *headers,
 							 CurrentNodeState *nodeState);
 
 void nodestatePrepareNode(NodeAddressHeaders *headers, NodeAddress *node,
-						  int groupId, char *hostport, char *composedId);
+						  int groupId, char *hostport,
+						  char *composedId, char *tliLSN);
+
 void prepareHostNameSeparator(char nameSeparatorHeader[], int size);
 
 bool nodestateAsJSON(CurrentNodeState *nodeState, JSON_Value *js);