Track connected Postgres server version and Citus extension version

src/bin/common/pgsql.c

@@ -50,6 +50,7 @@ static bool pgsql_alter_system_set(PGSQL *pgsql, GUC setting);
 static bool pgsql_get_current_setting(PGSQL *pgsql, char *settingName,
 									  char **currentValue);
 static void parsePgMetadata(void *ctx, PGresult *result);
+static void parsePgVersion(void *ctx, PGresult *result);
 static void parsePgReachedTargetLSN(void *ctx, PGresult *result);
 static void parseReplicationSlotMaintain(void *ctx, PGresult *result);
 static void parsePgReachedTargetLSN(void *ctx, PGresult *result);
@@ -2778,6 +2779,118 @@ parsePgMetadata(void *ctx, PGresult *result)
 }
 
 
+/*
+ * PgVersionInfo carries the parsed result of pgsql_get_postgres_version's
+ * query. Deliberately not PostgresVersionInfo (primary_standby.h): that
+ * struct also carries needsReport, a keeper-side bookkeeping field this
+ * generic pgsql.c layer has no business setting.
+ */
+typedef struct PgVersionInfo
+{
+	char sqlstate[6];
+	bool parsedOk;
+	int versionNum;
+	char version[NAMEDATALEN];
+	char versionString[BUFSIZE];
+	char citusVersion[NAMEDATALEN];
+} PgVersionInfo;
+
+
+/*
+ * pgsql_get_postgres_version fetches the connected server's own version
+ * (server_version_num/server_version/version()) and, when installed, the
+ * Citus extension's version. Unlike pgsql_get_postgres_metadata, none of
+ * this can change without a Postgres restart, so callers should fetch it
+ * once per restart rather than on every periodic report.
+ */
+bool
+pgsql_get_postgres_version(PGSQL *pgsql,
+						   int *versionNum,
+						   char *version,
+						   char *versionString,
+						   char *citusVersion)
+{
+	PgVersionInfo context = { 0 };
+
+	char *sql =
+		"select current_setting('server_version_num')::int,"
+		" current_setting('server_version'),"
+		" version(),"
+		" (select extversion from pg_extension where extname = 'citus')";
+
+	if (!pgsql_execute_with_params(pgsql, sql, 0, NULL, NULL,
+								   &context, &parsePgVersion))
+	{
+		/* errors have been logged already */
+		return false;
+	}
+
+	if (!context.parsedOk)
+	{
+		log_error("Failed to parse the Postgres version");
+		return false;
+	}
+
+	*versionNum = context.versionNum;
+	strlcpy(version, context.version, NAMEDATALEN);
+	strlcpy(versionString, context.versionString, BUFSIZE);
+	strlcpy(citusVersion, context.citusVersion, NAMEDATALEN);
+
+	pgsql_finish(pgsql);
+
+	return true;
+}
+
+
+/*
+ * parsePgVersion parses the result from pgsql_get_postgres_version's query:
+ * server_version_num, server_version, version(), and Citus's extversion
+ * (NULL when Citus isn't installed).
+ */
+static void
+parsePgVersion(void *ctx, PGresult *result)
+{
+	PgVersionInfo *context = (PgVersionInfo *) ctx;
+	char *value;
+
+	if (PQnfields(result) != 4)
+	{
+		log_error("Query returned %d columns, expected 4", PQnfields(result));
+		context->parsedOk = false;
+		return;
+	}
+
+	if (PQntuples(result) != 1)
+	{
+		log_error("Query returned %d rows, expected 1", PQntuples(result));
+		context->parsedOk = false;
+		return;
+	}
+
+	value = PQgetvalue(result, 0, 0);
+	if (!stringToInt(value, &(context->versionNum)))
+	{
+		log_error("Failed to parse server_version_num \"%s\"", value);
+		context->parsedOk = false;
+		return;
+	}
+
+	strlcpy(context->version, PQgetvalue(result, 0, 1), NAMEDATALEN);
+	strlcpy(context->versionString, PQgetvalue(result, 0, 2), BUFSIZE);
+
+	if (PQgetisnull(result, 0, 3))
+	{
+		context->citusVersion[0] = '\0';
+	}
+	else
+	{
+		strlcpy(context->citusVersion, PQgetvalue(result, 0, 3), NAMEDATALEN);
+	}
+
+	context->parsedOk = true;
+}
+
+
 typedef struct PgReachedTargetLSN
 {
 	char sqlstate[6];

src/bin/common/pgsql.h

@@ -389,6 +389,12 @@ bool pgsql_get_postgres_metadata(PGSQL *pgsql,
 								 char *pgsrSyncState, char *currentLSN,
 								 PostgresControlData *control);
 
+bool pgsql_get_postgres_version(PGSQL *pgsql,
+								int *versionNum,
+								char *version,
+								char *versionString,
+								char *citusVersion);
+
 bool pgsql_one_slot_has_reached_target_lsn(PGSQL *pgsql,
 										   char *targetLSN,
 										   char *currentLSN,

src/bin/pg_autoctl/keeper.c

@@ -435,6 +435,27 @@ keeper_update_pg_state(Keeper *keeper, int logLevel)
 
 	bool pgIsNotRunningIsOk = true;
 
+	/*
+	 * true iff *this function* already determined Postgres was running as
+	 * of its own previous call. Compared against the freshly determined
+	 * postgres->pgIsRunning further down to detect a not-running ->
+	 * running edge -- the trigger for re-fetching version info that can't
+	 * change without a Postgres restart (see PostgresVersionInfo).
+	 *
+	 * Deliberately NOT postgres->pgIsRunning's own previous value: that
+	 * field is also written directly by fsm_transition.c/primary_standby.c
+	 * during FSM transitions (e.g. the initial init -> single transition),
+	 * outside of this function entirely. Comparing against those writes
+	 * would make the very first, most important edge -- Postgres coming up
+	 * for the first time after `pg_autoctl create`/`run` -- invisible,
+	 * since pgIsRunning could already read true by the time this function
+	 * next runs. postgres->pgVersion.lastKnownRunning is written only
+	 * here, immediately below, so it always reflects this function's own
+	 * last determination, regardless of what else touches pgIsRunning in
+	 * between two of its calls.
+	 */
+	bool wasRunning = postgres->pgVersion.lastKnownRunning;
+
 	log_debug("Update local PostgreSQL state");
 
 	/* reinitialize the replication state values each time we update */
@@ -513,6 +534,34 @@ keeper_update_pg_state(Keeper *keeper, int logLevel)
 		keeperState->pg_control_version = pgSetup->control.pg_control_version;
 		keeperState->catalog_version_no = pgSetup->control.catalog_version_no;
 		keeperState->system_identifier = pgSetup->control.system_identifier;
+
+		/*
+		 * Postgres just transitioned from not-running to running: fetch its
+		 * version (and Citus's, if installed) once now, rather than on
+		 * every periodic report -- neither can change without a Postgres
+		 * restart. needsReport is cleared by keeper_node_active() once it's
+		 * been successfully sent to the monitor; a failed fetch here simply
+		 * leaves it unset, to be retried on the next such edge (i.e. the
+		 * next Postgres restart) rather than every tick.
+		 */
+		if (!wasRunning && postgres->pgIsRunning)
+		{
+			PostgresVersionInfo *pgVersion = &(postgres->pgVersion);
+
+			if (pgsql_get_postgres_version(pgsql,
+										   &(pgVersion->versionNum),
+										   pgVersion->version,
+										   pgVersion->versionString,
+										   pgVersion->citusVersion))
+			{
+				pgVersion->needsReport = true;
+			}
+			else
+			{
+				log_level(logLevel,
+						  "Failed to fetch Postgres/Citus version info");
+			}
+		}
 	}
 	else
 	{
@@ -543,6 +592,14 @@ keeper_update_pg_state(Keeper *keeper, int logLevel)
 		}
 	}
 
+	/*
+	 * Sync our own edge-tracker to what this function just determined,
+	 * regardless of which branch above ran -- see the comment on
+	 * wasRunning above for why this is deliberately not the same as
+	 * postgres->pgIsRunning's own (multiply-written) value.
+	 */
+	postgres->pgVersion.lastKnownRunning = postgres->pgIsRunning;
+
 	/*
 	 * In some states, PostgreSQL isn't expected to be running, or not expected
 	 * to have a streaming replication to monitor at all.
@@ -1305,16 +1362,42 @@ keeper_node_active(Keeper *keeper, bool doInit,
 	/*
 	 * Report the current state to the monitor and get the assigned state.
 	 */
-	return monitor_node_active(monitor,
-							   config->formation,
-							   keeperState->current_node_id,
-							   keeperState->current_group,
-							   keeperState->current_role,
-							   reportPgIsRunning,
-							   postgres->postgresSetup.control.timeline_id,
-							   postgres->currentLSN,
-							   postgres->pgsrSyncState,
-							   assignedState);
+	bool nodeActiveReturned =
+		monitor_node_active(monitor,
+							config->formation,
+							keeperState->current_node_id,
+							keeperState->current_group,
+							keeperState->current_role,
+							reportPgIsRunning,
+							postgres->postgresSetup.control.timeline_id,
+							postgres->currentLSN,
+							postgres->pgsrSyncState,
+							assignedState);
+
+	/*
+	 * Postgres/Citus version info doesn't change without a Postgres
+	 * restart, so it's reported separately from the per-tick node_active
+	 * call above, only when keeper_update_pg_state() detected a fresh
+	 * not-running -> running edge. Only attempt it once node_active itself
+	 * succeeded, so we know the monitor connection is good; leave
+	 * needsReport set on failure so the next tick retries.
+	 */
+	if (nodeActiveReturned && postgres->pgVersion.needsReport)
+	{
+		if (monitor_report_postgres_version(monitor,
+											keeperState->current_node_id,
+											&(postgres->pgVersion)))
+		{
+			postgres->pgVersion.needsReport = false;
+		}
+		else
+		{
+			log_warn("Failed to report Postgres/Citus version to the "
+					 "monitor, will retry");
+		}
+	}
+
+	return nodeActiveReturned;
 }
 
 

src/bin/pg_autoctl/monitor.c

@@ -1150,6 +1150,51 @@ monitor_set_node_region(Monitor *monitor,
 }
 
 
+/*
+ * monitor_report_postgres_version reports the connected Postgres server's
+ * own version and, when installed, the Citus extension's version, for the
+ * given node. Called once per Postgres restart (see
+ * keeper_update_pg_state()'s pgIsRunning edge detection), never on every
+ * periodic report -- neither piece of information can change without a
+ * Postgres restart. citusVersion may legitimately be an empty string
+ * (Citus not installed on this node), reported as SQL NULL.
+ */
+bool
+monitor_report_postgres_version(Monitor *monitor, int64_t nodeId,
+								PostgresVersionInfo *pgVersion)
+{
+	PGSQL *pgsql = &monitor->pgsql;
+	const char *sql =
+		"SELECT pgautofailover.report_postgres_version($1, $2, $3, $4, $5)";
+
+	int paramCount = 5;
+	Oid paramTypes[5] = { INT8OID, INT4OID, TEXTOID, TEXTOID, TEXTOID };
+	const char *paramValues[5];
+
+	IntString nodeIdString = intToString(nodeId);
+	IntString versionNumString = intToString(pgVersion->versionNum);
+
+	paramValues[0] = nodeIdString.strValue;
+	paramValues[1] = versionNumString.strValue;
+	paramValues[2] = pgVersion->version;
+	paramValues[3] = pgVersion->versionString;
+	paramValues[4] = IS_EMPTY_STRING_BUFFER(pgVersion->citusVersion)
+					 ? NULL
+					 : pgVersion->citusVersion;
+
+	if (!pgsql_execute_with_params(pgsql, sql, paramCount, paramTypes,
+								   paramValues, NULL, NULL))
+	{
+		log_error("Failed to report Postgres/Citus version for node %" PRId64,
+				  nodeId);
+
+		return false;
+	}
+
+	return true;
+}
+
+
 /*
  * monitor_get_node_region retrieves the region label of a node from the
  * monitor.

src/bin/pg_autoctl/monitor.h

@@ -163,6 +163,8 @@ bool monitor_set_node_region(Monitor *monitor,
 bool monitor_get_node_region(Monitor *monitor,
 							 char *name,
 							 char *region, size_t size);
+bool monitor_report_postgres_version(Monitor *monitor, int64_t nodeId,
+									 PostgresVersionInfo *pgVersion);
 bool monitor_get_formation_number_sync_standbys(Monitor *monitor, char *formation,
 												int *numberSyncStandbys);
 bool monitor_set_formation_number_sync_standbys(Monitor *monitor, char *formation,

src/bin/pg_autoctl/primary_standby.h

@@ -25,6 +25,37 @@ typedef struct LocalExpectedPostgresStatus
 } LocalExpectedPostgresStatus;
 
 
+/*
+ * PostgresVersionInfo holds the connected Postgres server's own version
+ * (server_version_num/server_version/version()) and, when installed, the
+ * Citus extension's version. Unlike pgsrSyncState/currentLSN, none of this
+ * can change without a Postgres restart, so it's fetched once per restart
+ * (see keeper_update_pg_state()'s pgIsRunning edge detection) rather than
+ * on every periodic report. needsReport stays true from the moment it's
+ * fetched until the next successful monitor_report_postgres_version() call.
+ */
+typedef struct PostgresVersionInfo
+{
+	int versionNum;                 /* server_version_num */
+	char version[NAMEDATALEN];      /* server_version, e.g. "16.3" */
+	char versionString[BUFSIZE];    /* version(), full descriptive string */
+	char citusVersion[NAMEDATALEN]; /* citus extversion; "" when not installed */
+	bool needsReport;
+
+	/*
+	 * lastKnownRunning is keeper_update_pg_state()'s own private edge
+	 * tracker, exclusively written by that function -- deliberately NOT
+	 * the same thing as LocalPostgresServer.pgIsRunning above, which other
+	 * code (fsm_transition.c, primary_standby.c) also writes directly
+	 * during FSM transitions. Comparing against a value only this
+	 * function ever sets is what makes a not-running -> running edge
+	 * reliably observable here, regardless of what else touches
+	 * pgIsRunning in between two of this function's own calls.
+	 */
+	bool lastKnownRunning;
+} PostgresVersionInfo;
+
+
 /*
  * LocalPostgresServer represents a local postgres database cluster that
  * we can manage via a SQL connection and operations on the database
@@ -47,6 +78,7 @@ typedef struct LocalPostgresServer
 	LocalExpectedPostgresStatus expectedPgStatus;
 	char standbyTargetLSN[PG_LSN_MAXLENGTH];
 	char synchronousStandbyNames[BUFSIZE];
+	PostgresVersionInfo pgVersion;
 } LocalPostgresServer;