Add command to get installed packages on managers (#3773)

# Description

Added a `MANAGERS_PACKAGES` HTEX interchange command and associated HTEX
method to retrieve a dict mapping each manager ID to a dict of installed
packages and their versions. This is useful for debugging and monitoring
worker environments.

We gather the package information from the manager upon registration,
but do not include it in the standard `MANAGERS` command, which runs
every 5 seconds. This avoids unnecessary overhead.

## Type of change

- New feature

Author: rjmello

parsl/executors/high_throughput/executor.py

@@ -617,6 +617,12 @@ def connected_managers(self) -> List[Dict[str, typing.Any]]:
         """
         return self.command_client.run("MANAGERS")
 
+    def connected_managers_packages(self) -> Dict[str, Dict[str, str]]:
+        """Returns a dict mapping each manager ID to a dict of installed
+        packages and their versions
+        """
+        return self.command_client.run("MANAGERS_PACKAGES")
+
     def connected_blocks(self) -> List[str]:
         """List of connected block ids"""
         return self.command_client.run("CONNECTED_BLOCKS")

parsl/executors/high_throughput/interchange.py

@@ -257,6 +257,13 @@ def process_command(self, monitoring_radio: Optional[MonitoringRadioSender]) ->
                             'draining': m['draining']}
                     reply.append(resp)
 
+            elif command_req == "MANAGERS_PACKAGES":
+                reply = {}
+                for manager_id in self._ready_managers:
+                    m = self._ready_managers[manager_id]
+                    manager_id_str = manager_id.decode('utf-8')
+                    reply[manager_id_str] = m["packages"]
+
             elif command_req.startswith("HOLD_WORKER"):
                 cmd, s_manager = command_req.split(';')
                 manager_id = s_manager.encode('utf-8')

parsl/executors/high_throughput/manager_record.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, List, Optional
+from typing import Any, Dict, List, Optional
 
 from typing_extensions import TypedDict
 
@@ -18,3 +18,4 @@ class ManagerRecord(TypedDict, total=False):
     timestamp: datetime
     parsl_version: str
     python_version: str
+    packages: Dict[str, str]

parsl/executors/high_throughput/process_worker_pool.py

@@ -14,6 +14,7 @@
 import threading
 import time
 import uuid
+from importlib.metadata import distributions
 from multiprocessing.managers import DictProxy
 from multiprocessing.sharedctypes import Synchronized
 from typing import Dict, List, Optional, Sequence
@@ -265,6 +266,7 @@ def create_reg_message(self):
                'python_v': "{}.{}.{}".format(sys.version_info.major,
                                              sys.version_info.minor,
                                              sys.version_info.micro),
+               'packages': {dist.metadata['Name']: dist.version for dist in distributions()},
                'worker_count': self.worker_count,
                'uid': self.uid,
                'block_id': self.block_id,

parsl/tests/test_htex/test_managers_command.py

@@ -1,6 +1,6 @@
-import logging
 import sys
 
+import packaging.version
 import pytest
 
 import parsl
@@ -34,3 +34,20 @@ def test_connected_managers():
     assert 'parsl_version' in manager_info
     assert manager_info['parsl_version'] == parsl.__version__
     assert manager_info['python_version'] == f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
+
+
+@pytest.mark.local
+def test_connected_managers_packages():
+    # Run dummy function to ensure a manager is online
+    f = dummy()
+    assert f.result() is None
+
+    htex: parsl.HighThroughputExecutor = parsl.dfk().executors['htex_local']
+    managers_info_list = htex.connected_managers()
+    managers_packages = htex.connected_managers_packages()
+
+    assert len(managers_packages) == len(managers_info_list) == 1
+    manager_id, packages = list(managers_packages.items())[0]
+    assert manager_id == managers_info_list[0]['manager']
+    normalized_version = str(packaging.version.parse(parsl.__version__))
+    assert packages['parsl'] == normalized_version