CHANGES:

- More robust handling of clusterdb.pkl on database password changes.

Author: karel26

core/data/impl/nodeimpl.py

@@ -302,23 +302,7 @@ def read_locals(self) -> dict:
             return node
         raise FatalException(f"No {config_file} found. Exiting.")
 
-    async def init_db(self):
-        async def check_db(url: str) -> str | None:
-            max_attempts = self.locals.get("database", self.config.get('database')).get('max_retries', 10)
-            for attempt in range(max_attempts + 1):
-                try:
-                    aconn = await psycopg.AsyncConnection.connect(url, connect_timeout=5)
-                    async with aconn:
-                        cursor = await aconn.execute("SHOW server_version")
-                        return (await cursor.fetchone())[0]
-                except ConnectionTimeout:
-                    if attempt < max_attempts:
-                        self.log.warning("- Database not available (yet), trying again ...")
-                        continue
-                    raise
-            # we will never be here
-            return None
-
+    def get_database_urls(self):
         cpool_url = self.config.get("database", self.locals.get('database'))['url']
         lpool_url = self.locals.get("database", self.config.get('database'))['url']
 
@@ -343,7 +327,26 @@ async def check_db(url: str) -> str | None:
 
         cpool_url = cpool_url.replace('SECRET', quote(cpool_pwd) or '')
         lpool_url = lpool_url.replace('SECRET', quote(lpool_pwd) or '')
+        return cpool_url, lpool_url
+
+    async def init_db(self):
+        async def check_db(url: str) -> str | None:
+            max_attempts = self.locals.get("database", self.config.get('database')).get('max_retries', 10)
+            for attempt in range(max_attempts + 1):
+                try:
+                    aconn = await psycopg.AsyncConnection.connect(url, connect_timeout=5)
+                    async with aconn:
+                        cursor = await aconn.execute("SHOW server_version")
+                        return (await cursor.fetchone())[0]
+                except ConnectionTimeout:
+                    if attempt < max_attempts:
+                        self.log.warning("- Database not available (yet), trying again ...")
+                        continue
+                    raise
+            # we will never be here
+            return None
 
+        cpool_url, lpool_url = self.get_database_urls()
         version = await check_db(lpool_url)
         if lpool_url != cpool_url:
             await check_db(cpool_url)

services/servicebus/service.py

@@ -48,22 +48,15 @@ def __init__(self, node):
         self.init_servers()
         self.udp_server = None
         self.executor = None
+
         if 'DCS' in self.locals and self.node.locals['DCS'].get('desanitize', True):
             if not self.node.locals['DCS'].get('cloud', False) or self.master:
                 utils.desanitize(self)
         self.loop = asyncio.get_event_loop()
-        # main.yaml database connection has priority for intercom
-        try:
-            password = utils.get_password('clusterdb', self.node.config_dir)
-        except ValueError:
-            password = utils.get_password('database', self.node.config_dir)
-        url = self.node.config.get("database", self.node.locals.get('database'))['url'].replace(
-            'SECRET', password)
-        self.intercom_channel = PubSub(self.node, 'intercom', url, self.handle_rpc)
-        # nodes.yaml database connection has priority for broadcasts
-        url = self.node.locals.get("database", self.node.config.get('database'))['url'].replace(
-            'SECRET', utils.get_password('database', self.node.config_dir))
-        self.broadcasts_channel = PubSub(self.node, 'broadcasts', url, self.handle_broadcast_event)
+
+        cpool_url, lpool_url = self.node.get_database_urls()
+        self.intercom_channel = PubSub(self.node, 'intercom', cpool_url, self.handle_rpc)
+        self.broadcasts_channel = PubSub(self.node, 'broadcasts', lpool_url, self.handle_broadcast_event)
         self._lock = asyncio.Lock()
 
     async def start(self):