[Issue #310] Detect timeline switch via repprotocol

Author: gsmolk

src/backup.c

@@ -165,8 +165,10 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool
 						"trying to look up on previous timelines",
 						current.tli);
 
-			/* TODO: use read_timeline_history */
-			tli_list = catalog_get_timelines(&instance_config);
+			tli_list = get_history_streaming(&instance_config.conn_opt, current.tli, backup_list);
+			if (!tli_list)
+				/* fallback to using archive */
+				tli_list = catalog_get_timelines(&instance_config);
 
 			if (parray_num(tli_list) == 0)
 				elog(WARNING, "Cannot find valid backup on previous timelines, "

src/catalog.c

@@ -48,7 +48,7 @@ typedef struct LockInfo
 	bool exclusive;
 } LockInfo;
 
-static timelineInfo *
+timelineInfo *
 timelineInfoNew(TimeLineID tli)
 {
 	timelineInfo *tlinfo = (timelineInfo *) pgut_malloc(sizeof(timelineInfo));
@@ -74,7 +74,8 @@ timelineInfoFree(void *tliInfo)
 
 	if (tli->backups)
 	{
-		parray_walk(tli->backups, pgBackupFree);
+		/* backups themselves should freed separately  */
+//		parray_walk(tli->backups, pgBackupFree);
 		parray_free(tli->backups);
 	}
 

src/pg_probackup.h

@@ -891,6 +891,8 @@ extern int validate_one_page(Page page, BlockNumber absolute_blkno,
 							 uint32 checksum_version);
 extern bool validate_tablespace_map(pgBackup *backup);
 
+extern parray* get_history_streaming(ConnectionOptions *conn_opt, TimeLineID tli, parray *backup_list);
+
 /* return codes for validate_one_page */
 /* TODO: use enum */
 #define PAGE_IS_VALID (-1)
@@ -920,6 +922,7 @@ extern pgBackup *catalog_get_last_data_backup(parray *backup_list,
 extern pgBackup *get_multi_timeline_parent(parray *backup_list, parray *tli_list,
 	                      TimeLineID current_tli, time_t current_start_time,
 						  InstanceConfig *instance);
+extern timelineInfo *timelineInfoNew(TimeLineID tli);
 extern void timelineInfoFree(void *tliInfo);
 extern parray *catalog_get_timelines(InstanceConfig *instance);
 extern void do_set_backup(const char *instance_name, time_t backup_id,

src/stream.c

@@ -10,6 +10,7 @@
 #include "pg_probackup.h"
 #include "receivelog.h"
 #include "streamutil.h"
+#include "access/timeline.h"
 
 #include <time.h>
 #include <unistd.h>
@@ -69,6 +70,7 @@ static void add_walsegment_to_filelist(parray *filelist, uint32 timeline,
                                        uint32 xlog_seg_size);
 static void add_history_file_to_filelist(parray *filelist, uint32 timeline,
 										 char *basedir);
+static parray* parse_tli_history_buffer(char *history, TimeLineID tli);
 
 /*
  * Run IDENTIFY_SYSTEM through a given connection and
@@ -353,6 +355,204 @@ stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
 
 /* --- External API --- */
 
+/*
+ * Maybe add a StreamOptions struct ?
+ * Backup conn only needed to calculate stream_stop_timeout. Think about refactoring it.
+ */
+parray*
+get_history_streaming(ConnectionOptions *conn_opt, TimeLineID tli, parray *backup_list)
+{
+	PGresult     *res;
+	PGconn	     *conn;
+	char         *history;
+	char          query[128];
+	parray	     *result = NULL;
+	parray       *tli_list = NULL;
+	timelineInfo *tlinfo = NULL;
+	int           i,j;
+
+	snprintf(query, sizeof(query), "TIMELINE_HISTORY %u", tli);
+
+	/*
+	 * Connect in replication mode to the server.
+	 */
+	conn = pgut_connect_replication(conn_opt->pghost,
+									conn_opt->pgport,
+									conn_opt->pgdatabase,
+									conn_opt->pguser,
+									false);
+
+	if (!conn)
+		return NULL;
+
+	res = PQexec(conn, query);
+	PQfinish(conn);
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		elog(WARNING, "Could not send replication command \"%s\": %s",
+					query, PQresultErrorMessage(res));
+		PQclear(res);
+		return NULL;
+	}
+
+	/*
+	 * The response to TIMELINE_HISTORY is a single row result set
+	 * with two fields: filename and content
+	 */
+
+	if (PQnfields(res) != 2 || PQntuples(res) != 1)
+	{
+		elog(WARNING, "Unexpected response to TIMELINE_HISTORY command: "
+				"got %d rows and %d fields, expected %d rows and %d fields",
+				PQntuples(res), PQnfields(res), 1, 2);
+		PQclear(res);
+		return NULL;
+	}
+
+	history = pgut_strdup(PQgetvalue(res, 0, 1));
+	result = parse_tli_history_buffer(history, tli);
+
+	/* some cleanup */
+	pg_free(history);
+	PQclear(res);
+
+	if (result)
+		tlinfo = timelineInfoNew(tli);
+	else
+		return NULL;
+
+	/* transform TimeLineHistoryEntry into timelineInfo */
+	for (i = parray_num(result) -1; i >= 0; i--)
+	{
+		TimeLineHistoryEntry *tln = (TimeLineHistoryEntry *) parray_get(result, i);
+
+		tlinfo->parent_tli = tln->tli;
+		tlinfo->switchpoint = tln->end;
+
+		if (!tli_list)
+			tli_list = parray_new();
+
+		parray_append(tli_list, tlinfo);
+
+		/* Next tli */
+		tlinfo = timelineInfoNew(tln->tli);
+
+		/* oldest tli */
+		if (i == 0)
+		{
+			tlinfo->tli = tln->tli;
+			tlinfo->parent_tli = 0;
+			tlinfo->switchpoint = 0;
+			parray_append(tli_list, tlinfo);
+		}
+	}
+
+	/* link parent to child */
+	for (i = 0; i < parray_num(tli_list); i++)
+	{
+		timelineInfo *tlinfo = (timelineInfo *) parray_get(tli_list, i);
+
+		for (j = 0; j < parray_num(tli_list); j++)
+		{
+			timelineInfo *tlinfo_parent = (timelineInfo *) parray_get(tli_list, j);
+
+			if (tlinfo->parent_tli == tlinfo_parent->tli)
+			{
+				tlinfo->parent_link = tlinfo_parent;
+				break;
+			}
+		}
+	}
+
+	/* add backups to each timeline info */
+	for (i = 0; i < parray_num(tli_list); i++)
+	{
+		timelineInfo *tlinfo = parray_get(tli_list, i);
+		for (j = 0; j < parray_num(backup_list); j++)
+		{
+			pgBackup *backup = parray_get(backup_list, j);
+			if (tlinfo->tli == backup->tli)
+			{
+				if (tlinfo->backups == NULL)
+					tlinfo->backups = parray_new();
+				parray_append(tlinfo->backups, backup);
+			}
+		}
+	}
+
+	/* cleanup */
+	parray_walk(result, pg_free);
+	pg_free(result);
+
+	return tli_list;
+}
+
+parray*
+parse_tli_history_buffer(char *history, TimeLineID tli)
+{
+	char   *curLine = history;
+	TimeLineHistoryEntry *entry;
+	TimeLineHistoryEntry *last_timeline = NULL;
+	parray *result = NULL;
+
+	/* Parse timeline history buffer string by string */
+	while (curLine)
+	{
+		char    tempStr[1024];
+		char   *nextLine = strchr(curLine, '\n');
+		int     curLineLen = nextLine ? (nextLine-curLine) : strlen(curLine);
+
+		memcpy(tempStr, curLine, curLineLen);
+		tempStr[curLineLen] = '\0';  // NUL-terminate!
+		curLine = nextLine ? (nextLine+1) : NULL;
+
+		if (curLineLen > 0)
+		{
+			char	   *ptr;
+			TimeLineID	tli;
+			uint32		switchpoint_hi;
+			uint32		switchpoint_lo;
+			int			nfields;
+
+			for (ptr = tempStr; *ptr; ptr++)
+			{
+				if (!isspace((unsigned char) *ptr))
+					break;
+			}
+			if (*ptr == '\0' || *ptr == '#')
+				continue;
+
+			nfields = sscanf(tempStr, "%u\t%X/%X", &tli, &switchpoint_hi, &switchpoint_lo);
+
+			if (nfields < 1)
+			{
+				/* expect a numeric timeline ID as first field of line */
+				elog(ERROR, "Syntax error in timeline history: \"%s\". Expected a numeric timeline ID.", tempStr);
+			}
+			if (nfields != 3)
+				elog(ERROR, "Syntax error in timeline history: \"%s\". Expected a transaction log switchpoint location.", tempStr);
+
+			if (last_timeline && tli <= last_timeline->tli)
+				elog(ERROR, "Timeline IDs must be in increasing sequence: \"%s\"", tempStr);
+
+			entry = pgut_new(TimeLineHistoryEntry);
+			entry->tli = tli;
+			entry->end = ((uint64) switchpoint_hi << 32) | switchpoint_lo;
+
+			last_timeline = entry;
+			/* Build list with newest item first */
+			if (!result)
+				result = parray_new();
+			parray_append(result, entry);
+
+			/* we ignore the remainder of each line */
+		}
+	}
+
+	return result;
+}
+
 /*
  * Maybe add a StreamOptions struct ?
  * Backup conn only needed to calculate stream_stop_timeout. Think about refactoring it.
@@ -374,7 +574,8 @@ start_WAL_streaming(PGconn *backup_conn, char *stream_dst_path, ConnectionOption
 	stream_thread_arg.conn = pgut_connect_replication(conn_opt->pghost,
 													  conn_opt->pgport,
 													  conn_opt->pgdatabase,
-													  conn_opt->pguser);
+													  conn_opt->pguser,
+													  true);
 	/* sanity check*/
 	IdentifySystem(&stream_thread_arg);
 

src/utils/pgut.c

@@ -270,7 +270,8 @@ pgut_connect(const char *host, const char *port,
 
 PGconn *
 pgut_connect_replication(const char *host, const char *port,
-						 const char *dbname, const char *username)
+						 const char *dbname, const char *username,
+						 bool strict)
 {
 	PGconn	   *tmpconn;
 	int			argcount = 7;	/* dbname, replication, fallback_app_name,
@@ -356,7 +357,7 @@ pgut_connect_replication(const char *host, const char *port,
 			continue;
 		}
 
-		elog(ERROR, "could not connect to database %s: %s",
+		elog(strict ? ERROR : WARNING, "could not connect to database %s: %s",
 			 dbname, PQerrorMessage(tmpconn));
 		PQfinish(tmpconn);
 		free(values);

src/utils/pgut.h

@@ -40,8 +40,8 @@ extern char *pgut_get_conninfo_string(PGconn *conn);
 extern PGconn *pgut_connect(const char *host, const char *port,
 							const char *dbname, const char *username);
 extern PGconn *pgut_connect_replication(const char *host, const char *port,
-										const char *dbname,
-										const char *username);
+										const char *dbname, const char *username,
+										bool strict);
 extern void pgut_disconnect(PGconn *conn);
 extern void pgut_disconnect_callback(bool fatal, void *userdata);
 extern PGresult *pgut_execute(PGconn* conn, const char *query, int nParams,