REL_2_5-PBCKP-236

src/pg_probackup.h

@@ -341,8 +341,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"
@@ -881,6 +881,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 */
@@ -3314,9 +3319,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,104 @@ 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 *_1C = "1C";
+	static char *vanilla = "vanilla";
+	static char *std = "standard";
+	static char *ent = "enterprise";
+	static char *std_cert = "standard-certified";
+	static char *ent_cert = "enterprise-certified";
+
+#ifdef PGPRO_EDITION
+	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 }
+
+/*
+ * 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)
+{
+	struct { const char* name; const char* strval; int intval; } 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; i+=2)
+	{
+		if (compatibility_params[i].strval != NULL)
+			result_size += snprintf(compatibility_buf + result_size, compatibility_buf_size - result_size,
+									"%s=%s/n",
+									compatibility_params[i].name,
+									compatibility_params[i].strval);
+		else
+			result_size += snprintf(compatibility_buf + result_size, compatibility_buf_size - result_size,
+									"%s=%d/n",
+									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 +349,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

@@ -10,6 +10,72 @@
 
 class CompatibilityTest(ProbackupTest, unittest.TestCase):
 
+    def setUp(self):
+        self.fname = self.id().split('.')[3]
+
+    # @unittest.expectedFailure
+    # @unittest.skip("skip")
+    def test_catchup_with_different_remote_major_pg(self):
+        """
+        Decription in jira issue PBCKP-236
+        This test requires builds both PGPROEE11 and PGPROEE9_6
+
+        prerequisites:
+        - git tag for PBCKP 2.5.1
+        - master probackup build should be inside PGPROEE11
+        - agent probackup build is inside PGPROEE9_6
+
+        calling probackup PGPROEE9_6 agent from PGPROEE11 probackup master for DELTA backup causes the PBCKP-236 problem
+
+        please correct path for agent's pg_path_remote_version = '/home/avaness/postgres/postgres.build.ee.9.6/bin/'
+        """
+
+        self.verbose = True
+        self.remote = True
+        # please use your own local path
+        pg_path_remote_version = '/home/avaness/postgres/postgres.build.clean/bin'
+
+        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 = {}
+        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=' + pg_path_remote_version]
+            )
+
+        # Clean after yourself
+        self.del_test_dir(module_name, self.fname)
+
     # @unittest.expectedFailure
     # @unittest.skip("skip")
     def test_backward_compatibility_page(self):