fix(monitor): the storage check reads the node name without waiting for the Proxmox API

AppImage/scripts/proxmox_storage_monitor.py

@@ -6,6 +6,8 @@ Monitors configured Proxmox storages and tracks unavailable storages
 """
 
 import json
+import os
+import re
 import subprocess
 import socket
 import time
@@ -44,6 +46,7 @@ class ProxmoxStorageMonitor:
     
     # Cache TTL: 177 seconds (~3 min) - offset to avoid sync with other processes
     _CACHE_TTL = 177
+    _PVE_LOCAL_NODE_LINK = "/etc/pve/local"
     
     def __init__(self):
         self.configured_storages: Dict[str, Dict[str, Any]] = {}
@@ -53,7 +56,9 @@ class ProxmoxStorageMonitor:
         self._load_configured_storages()
     
     def _get_node_name(self) -> str:
-        """Get current Proxmox node name (cached)"""
+        """The name of this node in Proxmox (cached). `/etc/pve/local` is a
+        link to the folder of the local node, so reading it does not wait
+        for the API; the hostname is used while `/etc/pve` is not mounted."""
         current_time = time.time()
         cache = self._node_name_cache
 
@@ -62,28 +67,14 @@ class ProxmoxStorageMonitor:
             return cache['name']
 
         try:
-            result = subprocess.run(
+            name = os.path.basename(os.readlink(self._PVE_LOCAL_NODE_LINK))
-                ['pvesh', 'get', '/nodes', '--output-format', 'json'],
+        except OSError:
-                capture_output=True,
+            name = ''
-                text=True,
+        if re.fullmatch(r'[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?', name):
-                timeout=5
-            )
-            if result.returncode == 0:
-                nodes = json.loads(result.stdout)
-                hostname = socket.gethostname()
-                for node in nodes:
-                    if node.get('node') == hostname:
-                        cache['name'] = hostname
-                        cache['time'] = current_time
-                        return hostname
-                if nodes:
-                    name = nodes[0].get('node', hostname)
             cache['name'] = name
             cache['time'] = current_time
             return name
         return socket.gethostname()
-        except Exception:
-            return socket.gethostname()
 
     def _load_configured_storages(self) -> None:
         """Load configured storages from Proxmox configuration"""
@@ -95,7 +86,7 @@ class ProxmoxStorageMonitor:
                 ['pvesh', 'get', '/storage', '--output-format', 'json'],
                 capture_output=True,
                 text=True,
-                timeout=5
+                timeout=15
             )
             
             if result.returncode != 0:

AppImage/scripts/tests/test_proxmox_node_name.py

@@ -57,5 +57,33 @@ class NodeNameTests(unittest.TestCase):
         self.assertEqual(self.name(), socket.gethostname().split('.', 1)[0].lower())
 
 
+class StorageMonitorNodeNameTests(unittest.TestCase):
+    """The storage monitor finds the node the same way."""
+
+    def setUp(self):
+        import subprocess
+        import sys
+        from unittest.mock import patch
+        sys.path.insert(0, str(SERVER.parent))
+        with patch.object(subprocess, 'run', side_effect=OSError('no Proxmox tools in the test')):
+            import proxmox_storage_monitor
+        self.module = proxmox_storage_monitor
+        folder = tempfile.TemporaryDirectory()
+        self.addCleanup(folder.cleanup)
+        self.link = os.path.join(folder.name, 'local')
+        self.monitor = proxmox_storage_monitor.ProxmoxStorageMonitor.__new__(proxmox_storage_monitor.ProxmoxStorageMonitor)
+        self.monitor._node_name_cache = {'name': None, 'time': 0}
+        self.monitor._PVE_LOCAL_NODE_LINK = self.link
+
+    def test_the_name_comes_from_the_link_without_running_anything(self):
+        from unittest.mock import patch
+        os.symlink('nodes/pve2', self.link)
+        with patch.object(self.module.subprocess, 'run', side_effect=AssertionError('no command')):
+            self.assertEqual(self.monitor._get_node_name(), 'pve2')
+
+    def test_without_the_link_the_hostname_is_used(self):
+        self.assertEqual(self.monitor._get_node_name(), socket.gethostname())
+
+
 if __name__ == '__main__':
     unittest.main()