Start Maintenance Daemon for Main DB at the server start. (#7254)

DESCRIPTION: This change starts a maintenance deamon at the time of
server start if there is a designated main database.

This is the code flow:

1. User designates a main database:
   `ALTER SYSTEM SET citus.main_db =  "myadmindb";`

2. When postmaster starts, in _PG_Init, citus calls 
    `InitializeMaintenanceDaemonForMainDb`
  
This function registers a background worker to run
`CitusMaintenanceDaemonMain `with `databaseOid = 0 `

3. `CitusMaintenanceDaemonMain ` takes some special actions when
databaseOid is 0:
     - Gets the citus.main_db  value.
     - Connects to the  citus.main_db
     - Now the `MyDatabaseId `is available, creates a hash entry for it.
     - Then follows the same control flow as for a regular db,

src/backend/distributed/shared_library_init.c

@@ -481,6 +481,7 @@ _PG_init(void)
 #endif
 
 	InitializeMaintenanceDaemon();
+	InitializeMaintenanceDaemonForMainDb();
 
 	/* initialize coordinated transaction management */
 	InitializeTransactionManagement();
@@ -1820,6 +1821,16 @@ RegisterCitusConfigVariables(void)
 		GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE | GUC_UNIT_MS,
 		NULL, NULL, NULL);
 
+	DefineCustomStringVariable(
+		"citus.main_db",
+		gettext_noop("Which database is designated as the main_db"),
+		NULL,
+		&MainDb,
+		"",
+		PGC_POSTMASTER,
+		GUC_STANDARD,
+		NULL, NULL, NULL);
+
 	DefineCustomIntVariable(
 		"citus.max_adaptive_executor_pool_size",
 		gettext_noop("Sets the maximum number of connections per worker node used by "

src/backend/distributed/utils/maintenanced.c

@@ -99,6 +99,7 @@ int Recover2PCInterval = 60000;
 int DeferShardDeleteInterval = 15000;
 int BackgroundTaskQueueCheckInterval = 5000;
 int MaxBackgroundTaskExecutors = 4;
+char *MainDb = "";
 
 /* config variables for metadata sync timeout */
 int MetadataSyncInterval = 60000;
@@ -112,7 +113,7 @@ static MaintenanceDaemonControlData *MaintenanceDaemonControl = NULL;
  * activated.
  */
 static HTAB *MaintenanceDaemonDBHash;
-
+static ErrorContextCallback errorCallback = { 0 };
 static volatile sig_atomic_t got_SIGHUP = false;
 static volatile sig_atomic_t got_SIGTERM = false;
 
@@ -125,6 +126,8 @@ static void MaintenanceDaemonShmemExit(int code, Datum arg);
 static void MaintenanceDaemonErrorContext(void *arg);
 static bool MetadataSyncTriggeredCheckAndReset(MaintenanceDaemonDBData *dbData);
 static void WarnMaintenanceDaemonNotStarted(void);
+static MaintenanceDaemonDBData * GetMaintenanceDaemonDBHashEntry(Oid databaseId,
+																 bool *found);
 
 /*
  * InitializeMaintenanceDaemon, called at server start, is responsible for
@@ -139,6 +142,82 @@ InitializeMaintenanceDaemon(void)
 }
 
 
+/*
+ * GetMaintenanceDaemonDBHashEntry searches the MaintenanceDaemonDBHash for the
+ * databaseId. It returns the entry if found or creates a new entry and initializes
+ * the value with zeroes.
+ */
+MaintenanceDaemonDBData *
+GetMaintenanceDaemonDBHashEntry(Oid databaseId, bool *found)
+{
+	MaintenanceDaemonDBData *dbData = (MaintenanceDaemonDBData *) hash_search(
+		MaintenanceDaemonDBHash,
+		&MyDatabaseId,
+		HASH_ENTER_NULL,
+		found);
+
+	if (!dbData)
+	{
+		elog(LOG,
+			 "cannot create or find the maintenance deamon hash entry for database %u",
+			 databaseId);
+		return NULL;
+	}
+
+	if (!*found)
+	{
+		/* ensure the values in MaintenanceDaemonDBData are zero */
+		memset(((char *) dbData) + sizeof(Oid), 0,
+			   sizeof(MaintenanceDaemonDBData) - sizeof(Oid));
+	}
+
+	return dbData;
+}
+
+
+/*
+ * InitializeMaintenanceDaemonForMainDb is called in _PG_Init
+ * at which stage we are not in a transaction or have databaseOid
+ */
+void
+InitializeMaintenanceDaemonForMainDb(void)
+{
+	if (strcmp(MainDb, "") == 0)
+	{
+		elog(LOG, "There is no designated Main database.");
+		return;
+	}
+
+	BackgroundWorker worker;
+
+	memset(&worker, 0, sizeof(worker));
+
+
+	strcpy_s(worker.bgw_name, sizeof(worker.bgw_name),
+			 "Citus Maintenance Daemon for Main DB");
+
+	/* request ability to connect to target database */
+	worker.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
+
+	/*
+	 * No point in getting started before able to run query, but we do
+	 * want to get started on Hot-Standby.
+	 */
+	worker.bgw_start_time = BgWorkerStart_ConsistentState;
+
+	/* Restart after a bit after errors, but don't bog the system. */
+	worker.bgw_restart_time = 5;
+	strcpy_s(worker.bgw_library_name,
+			 sizeof(worker.bgw_library_name), "citus");
+	strcpy_s(worker.bgw_function_name, sizeof(worker.bgw_library_name),
+			 "CitusMaintenanceDaemonMain");
+
+	worker.bgw_main_arg = (Datum) 0;
+
+	RegisterBackgroundWorker(&worker);
+}
+
+
 /*
  * InitializeMaintenanceDaemonBackend, called at backend start and
  * configuration changes, is responsible for starting a per-database
@@ -148,31 +227,20 @@ void
 InitializeMaintenanceDaemonBackend(void)
 {
 	Oid extensionOwner = CitusExtensionOwner();
-	bool found;
+	bool found = false;
 
 	LWLockAcquire(&MaintenanceDaemonControl->lock, LW_EXCLUSIVE);
 
-	MaintenanceDaemonDBData *dbData = (MaintenanceDaemonDBData *) hash_search(
+	MaintenanceDaemonDBData *dbData = GetMaintenanceDaemonDBHashEntry(MyDatabaseId,
-		MaintenanceDaemonDBHash,
-		&MyDatabaseId,
-		HASH_ENTER_NULL,
 																	  &found);
 
 	if (dbData == NULL)
 	{
 		WarnMaintenanceDaemonNotStarted();
 		LWLockRelease(&MaintenanceDaemonControl->lock);
-
 		return;
 	}
 
-	if (!found)
-	{
-		/* ensure the values in MaintenanceDaemonDBData are zero */
-		memset(((char *) dbData) + sizeof(Oid), 0,
-			   sizeof(MaintenanceDaemonDBData) - sizeof(Oid));
-	}
-
 	if (IsMaintenanceDaemon)
 	{
 		/*
@@ -271,41 +339,71 @@ WarnMaintenanceDaemonNotStarted(void)
 
 
 /*
- * CitusMaintenanceDaemonMain is the maintenance daemon's main routine, it'll
+ * ConnectToDatabase connects to the database for the given databaseOid.
- * be started by the background worker infrastructure.  If it errors out,
+ * if databaseOid is 0, connects to MainDb and then creates a hash entry.
- * it'll be restarted after a few seconds.
+ * If a hash entry cannot be created for MainDb it exits the process requesting a restart.
+ * However for regular databases, it exits without requesting a restart since another
+ * subsequent backend is expected to start the Maintenance Daemon.
+ * If the found hash entry has a valid workerPid, it exits
+ * without requesting a restart since there is already a daemon running.
  */
-void
+static MaintenanceDaemonDBData *
-CitusMaintenanceDaemonMain(Datum main_arg)
+ConnectToDatabase(Oid databaseOid)
 {
-	Oid databaseOid = DatumGetObjectId(main_arg);
+	MaintenanceDaemonDBData *myDbData = NULL;
-	TimestampTz nextStatsCollectionTime USED_WITH_LIBCURL_ONLY =
-		TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 60 * 1000);
-	bool retryStatsCollection USED_WITH_LIBCURL_ONLY = false;
-	TimestampTz lastRecoveryTime = 0;
-	TimestampTz lastShardCleanTime = 0;
-	TimestampTz lastStatStatementsPurgeTime = 0;
-	TimestampTz nextMetadataSyncTime = 0;
 
-	/* state kept for the background tasks queue monitor */
-	TimestampTz lastBackgroundTaskQueueCheck = GetCurrentTimestamp();
-	BackgroundWorkerHandle *backgroundTasksQueueBgwHandle = NULL;
-	bool backgroundTasksQueueWarnedForLock = false;
 
-	/*
+	bool isMainDb = false;
-	 * We do metadata sync in a separate background worker. We need its
-	 * handle to be able to check its status.
-	 */
-	BackgroundWorkerHandle *metadataSyncBgwHandle = NULL;
 
-	/*
-	 * Look up this worker's configuration.
-	 */
 	LWLockAcquire(&MaintenanceDaemonControl->lock, LW_EXCLUSIVE);
 
-	MaintenanceDaemonDBData *myDbData = (MaintenanceDaemonDBData *)
+
+	if (databaseOid == 0)
+	{
+		char *databaseName = MainDb;
+
+		/*
+		 * Since we cannot query databaseOid without initializing Postgres
+		 * first, connect to the database by name.
+		 */
+		BackgroundWorkerInitializeConnection(databaseName, NULL, 0);
+
+		/*
+		 * Now we have a valid MyDatabaseId.
+		 * Insert the hash entry for the database to the Maintenance Deamon Hash.
+		 */
+		bool found = false;
+
+		myDbData = GetMaintenanceDaemonDBHashEntry(MyDatabaseId, &found);
+
+		if (!myDbData)
+		{
+			/*
+			 * If an entry cannot be created,
+			 * return code of 1 requests worker restart
+			 * Since BackgroundWorker for the MainDb is only registered
+			 * once during server startup, we need to retry.
+			 */
+			proc_exit(1);
+		}
+
+		if (found && myDbData->workerPid != 0)
+		{
+			/* Another maintenance daemon is running.*/
+
+			proc_exit(0);
+		}
+
+		databaseOid = MyDatabaseId;
+		myDbData->userOid = GetSessionUserId();
+		isMainDb = true;
+	}
+	else
+	{
+		myDbData = (MaintenanceDaemonDBData *)
 				   hash_search(MaintenanceDaemonDBHash, &databaseOid,
 							   HASH_FIND, NULL);
+
 		if (!myDbData)
 		{
 			/*
@@ -329,8 +427,9 @@ CitusMaintenanceDaemonMain(Datum main_arg)
 
 			proc_exit(0);
 		}
+	}
 
-	before_shmem_exit(MaintenanceDaemonShmemExit, main_arg);
+	before_shmem_exit(MaintenanceDaemonShmemExit, ObjectIdGetDatum(databaseOid));
 
 	/*
 	 * Signal that I am the maintenance daemon now.
@@ -356,25 +455,55 @@ CitusMaintenanceDaemonMain(Datum main_arg)
 
 	LWLockRelease(&MaintenanceDaemonControl->lock);
 
-	/*
-	 * Setup error context so log messages can be properly attributed. Some of
-	 * them otherwise sound like they might be from a normal user connection.
-	 * Do so before setting up signals etc, so we never exit without the
-	 * context setup.
-	 */
-	ErrorContextCallback errorCallback = { 0 };
 	memset(&errorCallback, 0, sizeof(errorCallback));
 	errorCallback.callback = MaintenanceDaemonErrorContext;
 	errorCallback.arg = (void *) myDbData;
 	errorCallback.previous = error_context_stack;
 	error_context_stack = &errorCallback;
 
-
 	elog(LOG, "starting maintenance daemon on database %u user %u",
 		 databaseOid, myDbData->userOid);
 
+	if (!isMainDb)
+	{
 		/* connect to database, after that we can actually access catalogs */
 		BackgroundWorkerInitializeConnectionByOid(databaseOid, myDbData->userOid, 0);
+	}
+
+	return myDbData;
+}
+
+
+/*
+ * CitusMaintenanceDaemonMain is the maintenance daemon's main routine, it'll
+ * be started by the background worker infrastructure.  If it errors out,
+ * it'll be restarted after a few seconds.
+ */
+void
+CitusMaintenanceDaemonMain(Datum main_arg)
+{
+	Oid databaseOid = DatumGetObjectId(main_arg);
+	TimestampTz nextStatsCollectionTime USED_WITH_LIBCURL_ONLY =
+		TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 60 * 1000);
+	bool retryStatsCollection USED_WITH_LIBCURL_ONLY = false;
+	TimestampTz lastRecoveryTime = 0;
+	TimestampTz lastShardCleanTime = 0;
+	TimestampTz lastStatStatementsPurgeTime = 0;
+	TimestampTz nextMetadataSyncTime = 0;
+
+	/* state kept for the background tasks queue monitor */
+	TimestampTz lastBackgroundTaskQueueCheck = GetCurrentTimestamp();
+	BackgroundWorkerHandle *backgroundTasksQueueBgwHandle = NULL;
+	bool backgroundTasksQueueWarnedForLock = false;
+
+
+	/*
+	 * We do metadata sync in a separate background worker. We need its
+	 * handle to be able to check its status.
+	 */
+	BackgroundWorkerHandle *metadataSyncBgwHandle = NULL;
+
+	MaintenanceDaemonDBData *myDbData = ConnectToDatabase(databaseOid);
 
 	/* make worker recognizable in pg_stat_activity */
 	pgstat_report_appname("Citus Maintenance Daemon");
@@ -383,7 +512,7 @@ CitusMaintenanceDaemonMain(Datum main_arg)
 	 * Terminate orphaned metadata sync daemons spawned from previously terminated
 	 * or crashed maintenanced instances.
 	 */
-	SignalMetadataSyncDaemon(databaseOid, SIGTERM);
+	SignalMetadataSyncDaemon(MyDatabaseId, SIGTERM);
 
 	/* enter main loop */
 	while (!got_SIGTERM)
@@ -945,7 +1074,7 @@ MaintenanceDaemonShmemExit(int code, Datum arg)
 }
 
 
-/* MaintenanceDaemonSigTermHandler calls proc_exit(0) */
+/* MaintenanceDaemonSigTermHandler sets the got_SIGTERM flag.*/
 static void
 MaintenanceDaemonSigTermHandler(SIGNAL_ARGS)
 {

src/include/distributed/maintenanced.h

@@ -20,6 +20,7 @@
 
 /* config variable for */
 extern double DistributedDeadlockDetectionTimeoutFactor;
+extern char *MainDb;
 
 extern void StopMaintenanceDaemon(Oid databaseId);
 extern void TriggerNodeMetadataSync(Oid databaseId);
@@ -27,6 +28,7 @@ extern void InitializeMaintenanceDaemon(void);
 extern size_t MaintenanceDaemonShmemSize(void);
 extern void MaintenanceDaemonShmemInit(void);
 extern void InitializeMaintenanceDaemonBackend(void);
+extern void InitializeMaintenanceDaemonForMainDb(void);
 extern bool LockCitusExtension(void);
 
 extern PGDLLEXPORT void CitusMaintenanceDaemonMain(Datum main_arg);

src/test/regress/citus_tests/common.py

@@ -453,6 +453,9 @@ def cleanup_test_leftovers(nodes):
     for node in nodes:
         node.cleanup_schemas()
 
+    for node in nodes:
+        node.cleanup_databases()
+
     for node in nodes:
         node.cleanup_users()
 
@@ -753,6 +756,7 @@ class Postgres(QueryRunner):
         self.subscriptions = set()
         self.publications = set()
         self.replication_slots = set()
+        self.databases = set()
         self.schemas = set()
         self.users = set()
 
@@ -993,6 +997,10 @@ class Postgres(QueryRunner):
             args = sql.SQL("")
         self.sql(sql.SQL("CREATE USER {} {}").format(sql.Identifier(name), args))
 
+    def create_database(self, name):
+        self.databases.add(name)
+        self.sql(sql.SQL("CREATE DATABASE {}").format(sql.Identifier(name)))
+
     def create_schema(self, name):
         self.schemas.add(name)
         self.sql(sql.SQL("CREATE SCHEMA {}").format(sql.Identifier(name)))
@@ -1020,6 +1028,12 @@ class Postgres(QueryRunner):
         for user in self.users:
             self.sql(sql.SQL("DROP USER IF EXISTS {}").format(sql.Identifier(user)))
 
+    def cleanup_databases(self):
+        for database in self.databases:
+            self.sql(
+                sql.SQL("DROP DATABASE IF EXISTS {}").format(sql.Identifier(database))
+            )
+
     def cleanup_schemas(self):
         for schema in self.schemas:
             self.sql(

src/test/regress/citus_tests/test/test_maintenancedeamon.py

@@ -0,0 +1,74 @@
+# This test checks that once citus.main_db is set and the
+# server is restarted. A Citus Maintenance Daemon for the main_db
+# is launched. This should happen even if there is no query run
+# in main_db yet.
+import time
+
+
+def wait_until_maintenance_deamons_start(deamoncount, cluster):
+    i = 0
+    n = 0
+
+    while i < 10:
+        i += 1
+        n = cluster.coordinator.sql_value(
+            "SELECT count(*) FROM pg_stat_activity WHERE application_name = 'Citus Maintenance Daemon';"
+        )
+
+        if n == deamoncount:
+            break
+
+        time.sleep(0.1)
+
+    assert n == deamoncount
+
+
+def test_set_maindb(cluster_factory):
+    cluster = cluster_factory(0)
+
+    # Test that once citus.main_db is set to a database name
+    # there are two maintenance deamons running upon restart.
+    # One maintenance deamon for the database of the current connection
+    # and one for the citus.main_db.
+    cluster.coordinator.create_database("mymaindb")
+    cluster.coordinator.configure("citus.main_db='mymaindb'")
+    cluster.coordinator.restart()
+
+    assert cluster.coordinator.sql_value("SHOW citus.main_db;") == "mymaindb"
+
+    wait_until_maintenance_deamons_start(2, cluster)
+
+    assert (
+        cluster.coordinator.sql_value(
+            "SELECT count(*) FROM pg_stat_activity WHERE application_name = 'Citus Maintenance Daemon' AND datname='mymaindb';"
+        )
+        == 1
+    )
+
+    # Test that once citus.main_db is set to empty string
+    # there is only one maintenance deamon for the database
+    # of the current connection.
+    cluster.coordinator.configure("citus.main_db=''")
+    cluster.coordinator.restart()
+    assert cluster.coordinator.sql_value("SHOW citus.main_db;") == ""
+
+    wait_until_maintenance_deamons_start(1, cluster)
+
+    # Test that after citus.main_db is dropped. The maintenance
+    # deamon for this database is terminated.
+    cluster.coordinator.configure("citus.main_db='mymaindb'")
+    cluster.coordinator.restart()
+    assert cluster.coordinator.sql_value("SHOW citus.main_db;") == "mymaindb"
+
+    wait_until_maintenance_deamons_start(2, cluster)
+
+    cluster.coordinator.sql("DROP DATABASE mymaindb;")
+
+    wait_until_maintenance_deamons_start(1, cluster)
+
+    assert (
+        cluster.coordinator.sql_value(
+            "SELECT count(*) FROM pg_stat_activity WHERE application_name = 'Citus Maintenance Daemon' AND datname='mymaindb';"
+        )
+        == 0
+    )