Merge pull request #531 from postgrespro/REL_2_5-PBCKP-236

[PBCKP-236] merge into REL_2_5

Author: avaness

src/pg_probackup.h

@@ -348,8 +348,8 @@ typedef enum ShowFormat
 #define PROGRAM_VERSION	"2.5.8"
 
 /* update when remote agent API or behaviour changes */
-#define AGENT_PROTOCOL_VERSION 20501
-#define AGENT_PROTOCOL_VERSION_STR "2.5.1"
+#define AGENT_PROTOCOL_VERSION 20509
+#define AGENT_PROTOCOL_VERSION_STR "2.5.9"
 
 /* update only when changing storage format */
 #define STORAGE_FORMAT_VERSION "2.4.4"
@@ -888,6 +888,11 @@ extern bool tliIsPartOfHistory(const parray *timelines, TimeLineID tli);
 extern DestDirIncrCompatibility check_incremental_compatibility(const char *pgdata, uint64 system_identifier,
 																IncrRestoreMode incremental_mode);
 
+/* in remote.c */
+extern void check_remote_agent_compatibility(int agent_version,
+											 char *compatibility_str, size_t compatibility_str_max_size);
+extern size_t prepare_compatibility_str(char* compatibility_buf, size_t compatibility_buf_size);
+
 /* in merge.c */
 extern void do_merge(InstanceState *instanceState, time_t backup_id, bool no_validate, bool no_sync);
 extern void merge_backups(pgBackup *backup, pgBackup *next_backup);

src/utils/file.c

@@ -269,17 +269,22 @@ fio_write_all(int fd, void const* buf, size_t size)
 }
 
 /* Get version of remote agent */
-int
-fio_get_agent_version(void)
+void
+fio_get_agent_version(int* protocol, char* payload_buf, size_t payload_buf_size)
 {
 	fio_header hdr;
 	hdr.cop = FIO_AGENT_VERSION;
 	hdr.size = 0;
 
 	IO_CHECK(fio_write_all(fio_stdout, &hdr, sizeof(hdr)), sizeof(hdr));
 	IO_CHECK(fio_read_all(fio_stdin, &hdr, sizeof(hdr)), sizeof(hdr));
+	if (hdr.size > payload_buf_size)
+	{
+		elog(ERROR, "Corrupted remote compatibility protocol: insufficient payload_buf_size=%zu", payload_buf_size);
+	}
 
-	return hdr.arg;
+	*protocol = hdr.arg;
+	IO_CHECK(fio_read_all(fio_stdin, payload_buf, hdr.size), hdr.size);
 }
 
 /* Open input stream. Remote file is fetched to the in-memory buffer and then accessed through Linux fmemopen */
@@ -3322,9 +3327,16 @@ fio_communicate(int in, int out)
 				IO_CHECK(fio_write_all(out, buf, hdr.size),  hdr.size);
 			break;
 		  case FIO_AGENT_VERSION:
-			hdr.arg = AGENT_PROTOCOL_VERSION;
-			IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
-			break;
+			{
+				size_t payload_size = prepare_compatibility_str(buf, buf_size);
+
+				hdr.arg = AGENT_PROTOCOL_VERSION;
+				hdr.size = payload_size;
+
+				IO_CHECK(fio_write_all(out, &hdr, sizeof(hdr)), sizeof(hdr));
+				IO_CHECK(fio_write_all(out, buf, payload_size), payload_size);
+				break;
+			}
 		  case FIO_STAT: /* Get information about file with specified path */
 			hdr.size = sizeof(st);
 			rc = hdr.arg ? stat(buf, &st) : lstat(buf, &st);

src/utils/file.h

@@ -91,7 +91,7 @@ extern fio_location MyLocation;
 extern void    fio_redirect(int in, int out, int err);
 extern void    fio_communicate(int in, int out);
 
-extern int     fio_get_agent_version(void);
+extern void    fio_get_agent_version(int* protocol, char* payload_buf, size_t payload_buf_size);
 extern FILE*   fio_fopen(char const* name, char const* mode, fio_location location);
 extern size_t  fio_fwrite(FILE* f, void const* buf, size_t size);
 extern ssize_t fio_fwrite_async_compressed(FILE* f, void const* buf, size_t size, int compress_alg);

src/utils/remote.c

@@ -238,10 +238,113 @@ bool launch_agent(void)
 		fio_redirect(infd[0], outfd[1], errfd[0]); /* write to stdout */
 	}
 
-	/* Make sure that remote agent has the same version
-	 * TODO: we must also check PG version and fork edition
-	 */
-	agent_version = fio_get_agent_version();
+
+	/* Make sure that remote agent has the same version, fork and other features to be binary compatible */
+	{
+		char payload_buf[1024];
+		fio_get_agent_version(&agent_version, payload_buf, sizeof payload_buf);
+		check_remote_agent_compatibility(agent_version, payload_buf, sizeof payload_buf);
+	}
+
+	return true;
+}
+
+#ifdef PGPRO_EDITION
+/* PGPRO 10-13 checks to be "(certified)", with exceptional case PGPRO_11 conforming to "(standard certified)" */
+static bool check_certified()
+{
+	return strstr(PGPRO_VERSION_STR, "(certified)") ||
+		   strstr(PGPRO_VERSION_STR, "(standard certified)");
+}
+#endif
+
+static char* extract_pg_edition_str()
+{
+	static char *vanilla = "vanilla";
+#ifdef PGPRO_EDITION
+	static char *_1C = "1C";
+	static char *std = "standard";
+	static char *ent = "enterprise";
+	static char *std_cert = "standard-certified";
+	static char *ent_cert = "enterprise-certified";
+
+	if (strcmp(PGPRO_EDITION, _1C) == 0)
+		return vanilla;
+
+	if (PG_VERSION_NUM < 100000)
+		return PGPRO_EDITION;
+
+	/* these "certified" checks are applicable to PGPRO from 10 up to 12 versions.
+	 * 13+ certified versions are compatible to non-certified ones */
+	if (PG_VERSION_NUM < 130000 && check_certified())
+	{
+		if (strcmp(PGPRO_EDITION, std) == 0)
+			return std_cert;
+		else if (strcmp(PGPRO_EDITION, ent) == 0)
+			return ent_cert;
+		else
+			Assert("Bad #define PGPRO_EDITION value" == 0);
+	}
+
+	return PGPRO_EDITION;
+#else
+	return vanilla;
+#endif
+}
+
+#define COMPATIBILITY_VAL_STR(macro) { #macro, macro, 0 }
+#define COMPATIBILITY_VAL_INT(macro) { #macro, NULL, macro }
+
+#define COMPATIBILITY_VAL_SEPARATOR "="
+#define COMPATIBILITY_LINE_SEPARATOR "\n"
+
+/*
+ * Compose compatibility string to be sent by pg_probackup agent
+ * through ssh and to be verified by pg_probackup peer.
+ * Compatibility string contains postgres essential vars as strings
+ * in format "var_name" + COMPATIBILITY_VAL_SEPARATOR + "var_value" + COMPATIBILITY_LINE_SEPARATOR
+ */
+size_t prepare_compatibility_str(char* compatibility_buf, size_t compatibility_buf_size)
+{
+	typedef struct compatibility_param_tag {
+		const char* name;
+		const char* strval;
+		int intval;
+	} compatibility_param;
+
+	compatibility_param compatibility_params[] = {
+		COMPATIBILITY_VAL_STR(PG_MAJORVERSION),
+		{ "edition", extract_pg_edition_str(), 0 },
+		COMPATIBILITY_VAL_INT(SIZEOF_VOID_P),
+	};
+
+	size_t result_size = 0;
+	*compatibility_buf = '\0';
+
+	for (int i = 0; i < (sizeof compatibility_params / sizeof(compatibility_param)); i++)
+	{
+		if (compatibility_params[i].strval != NULL)
+			result_size += snprintf(compatibility_buf + result_size, compatibility_buf_size - result_size,
+									"%s" COMPATIBILITY_VAL_SEPARATOR "%s" COMPATIBILITY_LINE_SEPARATOR,
+									compatibility_params[i].name,
+									compatibility_params[i].strval);
+		else
+			result_size += snprintf(compatibility_buf + result_size, compatibility_buf_size - result_size,
+									"%s" COMPATIBILITY_VAL_SEPARATOR "%d" COMPATIBILITY_LINE_SEPARATOR,
+									compatibility_params[i].name,
+									compatibility_params[i].intval);
+		Assert(result_size < compatibility_buf_size);
+	}
+	return result_size + 1;
+}
+
+/*
+ * Check incoming remote agent's compatibility params for equality to local ones.
+ */
+void check_remote_agent_compatibility(int agent_version, char *compatibility_str, size_t compatibility_str_max_size)
+{
+	elog(LOG, "Agent version=%d\n", agent_version);
+
 	if (agent_version != AGENT_PROTOCOL_VERSION)
 	{
 		char agent_version_str[1024];
@@ -255,5 +358,21 @@ bool launch_agent(void)
 			agent_version_str, AGENT_PROTOCOL_VERSION_STR);
 	}
 
-	return true;
+	/* checking compatibility params */
+	if (strnlen(compatibility_str, compatibility_str_max_size) == compatibility_str_max_size)
+	{
+		elog(ERROR, "Corrupted remote compatibility protocol: compatibility string has no terminating \\0");
+	}
+
+	elog(LOG, "Agent compatibility params:\n%s", compatibility_str);
+
+	{
+		char buf[1024];
+
+		prepare_compatibility_str(buf, sizeof buf);
+		if(strcmp(compatibility_str, buf))
+		{
+			elog(ERROR, "Incompatible remote agent params, expected:\n%s, actual:\n:%s", buf, compatibility_str);
+		}
+	}
 }

tests/compatibility.py

@@ -8,8 +8,89 @@
 module_name = 'compatibility'
 
 
+def check_manual_tests_enabled():
+    return 'PGPROBACKUP_MANUAL' in os.environ and os.environ['PGPROBACKUP_MANUAL'] == 'ON'
+
+
+def check_ssh_agent_path_exists():
+    return 'PGPROBACKUP_SSH_AGENT_PATH' in os.environ
+
+
 class CompatibilityTest(ProbackupTest, unittest.TestCase):
 
+    def setUp(self):
+        self.fname = self.id().split('.')[3]
+
+    # @unittest.expectedFailure
+    @unittest.skipUnless(check_manual_tests_enabled(), 'skip manual test')
+    @unittest.skipUnless(check_ssh_agent_path_exists(), 'skip no ssh agent path exist')
+    # @unittest.skip("skip")
+    def test_catchup_with_different_remote_major_pg(self):
+        """
+        Decription in jira issue PBCKP-236
+        This test exposures ticket error using pg_probackup builds for both PGPROEE11 and PGPROEE9_6
+
+        Prerequisites:
+        - pg_probackup git tag for PBCKP 2.5.1
+        - master pg_probackup build should be made for PGPROEE11
+        - agent pg_probackup build should be made for PGPROEE9_6
+
+        Calling probackup PGPROEE9_6 pg_probackup agent from PGPROEE11 pg_probackup master for DELTA backup causes
+        the PBCKP-236 problem
+
+        Please give env variables PROBACKUP_MANUAL=ON;PGPROBACKUP_SSH_AGENT_PATH=<pg_probackup_ssh_agent_path>
+        for the test
+
+        Please make path for agent's pgprobackup_ssh_agent_path = '/home/avaness/postgres/postgres.build.ee.9.6/bin/'
+        without pg_probackup executable
+        """
+
+        self.verbose = True
+        self.remote = True
+        # please use your own local path like
+        # pgprobackup_ssh_agent_path = '/home/avaness/postgres/postgres.build.clean/bin/'
+        pgprobackup_ssh_agent_path = os.environ['PGPROBACKUP_SSH_AGENT_PATH']
+
+        src_pg = self.make_simple_node(
+            base_dir=os.path.join(module_name, self.fname, 'src'),
+            set_replication=True,
+            )
+        src_pg.slow_start()
+        src_pg.safe_psql(
+            "postgres",
+            "CREATE TABLE ultimate_question AS SELECT 42 AS answer")
+
+        # do full catchup
+        dst_pg = self.make_empty_node(os.path.join(module_name, self.fname, 'dst'))
+        self.catchup_node(
+            backup_mode='FULL',
+            source_pgdata=src_pg.data_dir,
+            destination_node=dst_pg,
+            options=['-d', 'postgres', '-p', str(src_pg.port), '--stream']
+            )
+
+        dst_options = {'port': str(dst_pg.port)}
+        self.set_auto_conf(dst_pg, dst_options)
+        dst_pg.slow_start()
+        dst_pg.stop()
+
+        src_pg.safe_psql(
+            "postgres",
+            "CREATE TABLE ultimate_question2 AS SELECT 42 AS answer")
+
+        # do delta catchup with remote pg_probackup agent with another postgres major version
+        # this DELTA backup should fail without PBCKP-236 patch.
+        self.catchup_node(
+            backup_mode='DELTA',
+            source_pgdata=src_pg.data_dir,
+            destination_node=dst_pg,
+            # here's substitution of --remoge-path pg_probackup agent compiled with another postgres version
+            options=['-d', 'postgres', '-p', str(src_pg.port), '--stream', '--remote-path=' + pgprobackup_ssh_agent_path]
+            )
+
+        # Clean after yourself
+        self.del_test_dir(module_name, self.fname)
+
     # @unittest.expectedFailure
     # @unittest.skip("skip")
     def test_backward_compatibility_page(self):