Merge branch 'master' into issue_79

Author: gsmolk

src/pg_probackup.c

@@ -314,7 +314,7 @@ main(int argc, char *argv[])
 				uint32 agent_version = parse_program_version(remote_agent);
 				elog(agent_version < AGENT_PROTOCOL_VERSION ? ERROR : WARNING,
 					 "Agent version %s doesn't match master pg_probackup version %s",
-					 remote_agent, PROGRAM_VERSION);
+					 PROGRAM_VERSION, remote_agent);
 			}
 			fio_communicate(STDIN_FILENO, STDOUT_FILENO);
 			return 0;

src/utils/file.c

@@ -20,6 +20,7 @@ static __thread unsigned long fio_fdset = 0;
 static __thread void* fio_stdin_buffer;
 static __thread int fio_stdout = 0;
 static __thread int fio_stdin = 0;
+static __thread int fio_stderr = 0;
 
 fio_location MyLocation;
 
@@ -38,10 +39,29 @@ typedef struct
 #define fio_fileno(f) (((size_t)f - 1) | FIO_PIPE_MARKER)
 
 /* Use specified file descriptors as stdin/stdout for FIO functions */
-void fio_redirect(int in, int out)
+void fio_redirect(int in, int out, int err)
 {
 	fio_stdin = in;
 	fio_stdout = out;
+	fio_stderr = err;
+}
+
+void fio_error(int rc, int size, char const* file, int line)
+{
+	if (remote_agent)
+	{
+		fprintf(stderr, "%s:%d: proceeds %d bytes instead of %d: %s\n", file, line, rc, size, rc >= 0 ? "end of data" :  strerror(errno));
+		exit(EXIT_FAILURE);
+	}
+	else
+	{
+		char buf[PRINTF_BUF_SIZE];
+		int err_size = read(fio_stderr, buf, sizeof(buf));
+		if (err_size > 0)
+			elog(ERROR, "Agent error: %s", buf);
+		else
+			elog(ERROR, "Communication error: %s", rc >= 0 ? "end of data" :  strerror(errno));
+	}
 }
 
 /* Check if file descriptor is local or remote (created by FIO) */

src/utils/file.h

@@ -49,7 +49,7 @@ typedef enum
 #define PAGE_CHECKSUM_MISMATCH (-256)
 
 #define SYS_CHECK(cmd) do if ((cmd) < 0) { fprintf(stderr, "%s:%d: (%s) %s\n", __FILE__, __LINE__, #cmd, strerror(errno)); exit(EXIT_FAILURE); } while (0)
-#define IO_CHECK(cmd, size) do { int _rc = (cmd); if (_rc != (size)) { if (remote_agent) { fprintf(stderr, "%s:%d: proceeds %d bytes instead of %d: %s\n", __FILE__, __LINE__, _rc, (int)(size), _rc >= 0 ? "end of data" :  strerror(errno)); exit(EXIT_FAILURE); } else elog(ERROR, "Communication error: %s", _rc >= 0 ? "end of data" :  strerror(errno)); } } while (0)
+#define IO_CHECK(cmd, size) do { int _rc = (cmd); if (_rc != (size)) fio_error(_rc, size, __FILE__, __LINE__); } while (0)
 
 typedef struct
 {
@@ -64,7 +64,7 @@ extern fio_location MyLocation;
 /* Check if FILE handle is local or remote (created by FIO) */
 #define fio_is_remote_file(file) ((size_t)(file) <= FIO_FDMAX)
 
-extern void    fio_redirect(int in, int out);
+extern void    fio_redirect(int in, int out, int err);
 extern void    fio_communicate(int in, int out);
 
 extern FILE*   fio_fopen(char const* name, char const* mode, fio_location location);
@@ -77,6 +77,7 @@ extern int     fio_fseek(FILE* f, off_t offs);
 extern int     fio_ftruncate(FILE* f, off_t size);
 extern int     fio_fclose(FILE* f);
 extern int     fio_ffstat(FILE* f, struct stat* st);
+extern void    fio_error(int rc, int size, char const* file, int line);
 
 struct pgFile;
 extern  int    fio_send_pages(FILE* in, FILE* out, struct pgFile *file, XLogRecPtr horizonLsn, 

src/utils/remote.c

@@ -110,6 +110,7 @@ bool launch_agent(void)
 	int ssh_argc;
 	int outfd[2];
 	int infd[2];
+	int errfd[2];
 
 	ssh_argc = 0;
 #ifdef WIN32
@@ -195,20 +196,25 @@ bool launch_agent(void)
 #else
 	SYS_CHECK(pipe(infd));
 	SYS_CHECK(pipe(outfd));
+	SYS_CHECK(pipe(errfd));
 
 	SYS_CHECK(child_pid = fork());
 
 	if (child_pid == 0) { /* child */
 		SYS_CHECK(close(STDIN_FILENO));
 		SYS_CHECK(close(STDOUT_FILENO));
+		SYS_CHECK(close(STDERR_FILENO));
 
 		SYS_CHECK(dup2(outfd[0], STDIN_FILENO));
 		SYS_CHECK(dup2(infd[1],  STDOUT_FILENO));
+		SYS_CHECK(dup2(errfd[1], STDERR_FILENO));
 
 		SYS_CHECK(close(infd[0]));
 		SYS_CHECK(close(infd[1]));
 		SYS_CHECK(close(outfd[0]));
 		SYS_CHECK(close(outfd[1]));
+		SYS_CHECK(close(errfd[0]));
+		SYS_CHECK(close(errfd[1]));
 
 		if (execvp(ssh_argv[0], ssh_argv) < 0)
 			return false;
@@ -217,9 +223,10 @@ bool launch_agent(void)
 		elog(LOG, "Spawn agent %d version %s", child_pid, PROGRAM_VERSION);
 		SYS_CHECK(close(infd[1]));  /* These are being used by the child */
 		SYS_CHECK(close(outfd[0]));
+		SYS_CHECK(close(errfd[1]));
 		/*atexit(kill_child);*/
 
-		fio_redirect(infd[0], outfd[1]); /* write to stdout */
+		fio_redirect(infd[0], outfd[1], errfd[0]); /* write to stdout */
 	}
 	return true;
 }

tests/pgpro560.py

@@ -3,6 +3,7 @@
 from .helpers.ptrack_helpers import ProbackupTest, ProbackupException
 from datetime import datetime, timedelta
 import subprocess
+from time import sleep
 
 
 module_name = 'pgpro560'
@@ -98,6 +99,8 @@ def test_pgpro560_systemid_mismatch(self):
                     '\n Unexpected Error Message: {0}\n CMD: {1}'.format(
                         repr(e.message), self.cmd))
 
+        sleep(1)
+
         try:
             self.backup_node(
                 backup_dir, 'node1', node2,

tests/validate.py

@@ -3499,7 +3499,7 @@ def test_validate_target_lsn(self):
 
     # @unittest.expectedFailure
     # @unittest.skip("skip")
-    def test_recovery_target_backup_victim(self):
+    def test_recovery_target_time_backup_victim(self):
         """
         Check that for validation to recovery target
         probackup chooses valid backup
@@ -3555,6 +3555,61 @@ def test_recovery_target_backup_victim(self):
             backup_dir, 'node',
             options=['--recovery-target-time={0}'.format(target_time)])
 
+    # @unittest.expectedFailure
+    # @unittest.skip("skip")
+    def test_recovery_target_lsn_backup_victim(self):
+        """
+        Check that for validation to recovery target
+        probackup chooses valid backup
+        https://github.com/postgrespro/pg_probackup/issues/104
+        """
+        fname = self.id().split('.')[3]
+        backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
+        node = self.make_simple_node(
+            base_dir=os.path.join(module_name, fname, 'node'),
+            set_replication=True,
+            initdb_params=['--data-checksums'])
+
+        self.init_pb(backup_dir)
+        self.add_instance(backup_dir, 'node', node)
+        self.set_archiving(backup_dir, 'node', node)
+        node.slow_start()
+
+        # FULL backup
+        self.backup_node(backup_dir, 'node', node)
+
+        node.safe_psql(
+            "postgres",
+            "create table t_heap as select 1 as id, md5(i::text) as text, "
+            "md5(repeat(i::text,10))::tsvector as tsvector "
+            "from generate_series(0,10000) i")
+
+        node.safe_psql(
+            "postgres",
+            "create table t_heap1 as select 1 as id, md5(i::text) as text, "
+            "md5(repeat(i::text,10))::tsvector as tsvector "
+            "from generate_series(0,100) i")
+
+        gdb = self.backup_node(backup_dir, 'node', node, gdb=True)
+
+        gdb.set_breakpoint('pg_stop_backup')
+        gdb.run_until_break()
+        gdb.remove_all_breakpoints()
+        gdb._execute('signal SIGINT')
+        gdb.continue_execution_until_error()
+
+        backup_id = self.show_pb(backup_dir, 'node')[1]['id']
+
+        self.assertEqual(
+            'ERROR',
+            self.show_pb(backup_dir, 'node', backup_id)['status'],
+            'Backup STATUS should be "ERROR"')
+
+        target_lsn = self.show_pb(backup_dir, 'node', backup_id)['start-lsn']
+
+        self.validate_pb(
+            backup_dir, 'node',
+            options=['--recovery-target-lsn={0}'.format(target_lsn)])
 
 # validate empty backup list
 # page from future during validate