Better monitoring

libmuscle/python/libmuscle/mcp/protocol.py

@@ -34,6 +34,7 @@ class RequestType(Enum):
     REPORT_RESOURCES = 41
     GET_COMMAND = 42
     REPORT_RESULT = 43
+    MONITOR_USAGE = 44
 
 
 class ResponseType(Enum):
@@ -57,3 +58,4 @@ class AgentCommandType(Enum):
     START = 1
     CANCEL_ALL = 2
     SHUTDOWN = 3
+    ADD_MONITOR = 4

libmuscle/python/libmuscle/mmp_client.py

@@ -59,7 +59,8 @@ def encode_profile_event(event: ProfileEvent) -> Any:
             event.event_type.value,
             event.start_time.nanoseconds, event.stop_time.nanoseconds,
             encoded_port, event.port_length, event.slot,
-            event.message_number, event.message_size, event.message_timestamp]
+            event.message_number, event.message_size, event.message_timestamp,
+            event.cpu_percent, event.memory_usage]
 
 
 def decode_checkpoint_rule(rule: Dict[str, Any]) -> CheckpointRule:

libmuscle/python/libmuscle/native_instantiator/agent/__main__.py

@@ -9,10 +9,9 @@
 from libmuscle.native_instantiator.process_manager import ProcessManager
 from libmuscle.native_instantiator.agent.map_client import MAPClient
 from libmuscle.native_instantiator.agent.agent_commands import (
-        CancelAllCommand, ShutdownCommand, StartCommand)
+        AddMonitorCommand, CancelAllCommand, ShutdownCommand, StartCommand)
 from libmuscle.planner.resources import Core, CoreSet, OnNodeResources
 
-
 _logger = logging.getLogger(__name__)
 
 
@@ -30,6 +29,7 @@ def __init__(self, node_name: str, server_location: str) -> None:
         self._process_manager = ProcessManager()
 
         self._node_name = node_name
+        self._monitor_pids : List[Tuple[str, int]] = []
 
         _logger.info(f'Connecting to manager at {server_location}')
         self._server = MAPClient(self._node_name, server_location)
@@ -51,6 +51,8 @@ def run(self) -> None:
                 self._process_manager.start(
                         command.name, command.work_dir, command.args, command.env,
                         command.stdout, command.stderr)
+            elif isinstance(command, AddMonitorCommand):
+                self._monitor_pids.append((command.instance, command.pid))
             elif isinstance(command, CancelAllCommand):
                 _logger.info('Cancelling all instances')
                 self._process_manager.cancel_all()
@@ -60,12 +62,16 @@ def run(self) -> None:
                 shutting_down = True
                 _logger.info('Agent shutting down')
 
+
             finished = self._process_manager.get_finished()
             if finished:
                 for name, exit_code in finished:
                     _logger.info(f'Process {name} finished with exit code {exit_code}')
                 self._server.report_result(finished)
 
+            if not shutting_down and not finished:
+                self._server.monitor_usage(self._monitor_pids, _logger)
+
             sleep(0.1)
 
         self._server.close()

libmuscle/python/libmuscle/native_instantiator/agent/map_client.py

@@ -2,14 +2,15 @@
 from typing import Any, List, Optional, Tuple
 
 import msgpack
+import psutil
 
 from libmuscle.mcp.protocol import AgentCommandType, RequestType, ResponseType
 from libmuscle.mcp.tcp_transport_client import TcpTransportClient
 from libmuscle.native_instantiator.agent.agent_commands import (
-        AgentCommand, StartCommand, CancelAllCommand, ShutdownCommand)
+        AgentCommand, AddMonitorCommand, StartCommand, CancelAllCommand, 
+        ShutdownCommand)
 from libmuscle.planner.resources import OnNodeResources
 
-
 class MAPClient:
     """The client for the MUSCLE Agent Protocol.
 
@@ -44,6 +45,32 @@ def report_resources(self, resources: OnNodeResources) -> None:
                 resources.node_name, {'cpu': enc_cpu_resources}]
         self._call_agent_manager(request)
 
+    def monitor_usage(self, pids: List[Tuple[str, int]], logger) -> None:
+        """Monitor usage of resources of processes with given (instance_id, pid) on this node."""
+        if len(pids) == 0:
+            """ Nothing to monitor, return """
+            return
+
+        usage : Dict[str, Tuple[float, int]] = {}
+        for instance_id, pid in pids:
+            try:
+                process = psutil.Process(pid)
+                cpu = process.cpu_percent()
+                mem = process.memory_info().vms
+                logger.debug(f'PID {pid}: CPU {cpu}%, Memory {mem}')
+                usage[instance_id] = (cpu, mem)
+            except psutil.NoSuchProcess:
+                logger.debug(f'PID {pid}: Process not found')
+
+        if len(usage) < 1:
+            """ Nothing to monitor """
+            return
+
+        request = [
+                RequestType.MONITOR_USAGE.value,
+                self._node_name, usage]
+        self._call_agent_manager(request)
+
     def get_command(self) -> Optional[AgentCommand]:
         """Get a command from the agent manager.
 
@@ -67,6 +94,9 @@ def get_command(self) -> Optional[AgentCommand]:
             stderr = Path(command[6])
 
             return StartCommand(name, workdir, args, env, stdout, stderr)
+
+        elif command[0] == AgentCommandType.ADD_MONITOR.value:
+            return AddMonitorCommand(command[1], command[2], command[3])
 
         elif command[0] == AgentCommandType.CANCEL_ALL.value:
             return CancelAllCommand()

libmuscle/python/libmuscle/native_instantiator/agent_manager.py

@@ -83,7 +83,7 @@ def start(
 
     def monitor_usage(self, instance: str, hostname: str, pid: int) -> None:
         """Monitor usage of resources."""
-        #self._server.deposit_command(hostname, AddMonitorCommand(instance, hostname, pid))
+        self._server.deposit_command(hostname, AddMonitorCommand(instance, hostname, pid))
 
     def cancel_all(self) -> None:
         """Cancel all processes.

libmuscle/python/libmuscle/native_instantiator/map_server.py

@@ -1,14 +1,14 @@
 import errno
 import logging
-from typing import Any, Dict, cast, List, Optional
+from typing import Any, Dict, cast, List, Optional, Tuple
 
 import msgpack
 
 from libmuscle.mcp.protocol import AgentCommandType, RequestType, ResponseType
 from libmuscle.mcp.tcp_transport_server import TcpTransportServer
 from libmuscle.mcp.transport_server import RequestHandler
 from libmuscle.native_instantiator.agent.agent_commands import (
-        AgentCommand, CancelAllCommand, ShutdownCommand, StartCommand)
+        AgentCommand, AddMonitorCommand, CancelAllCommand, ShutdownCommand, StartCommand)
 from libmuscle.native_instantiator.iagent_manager import IAgentManager
 from libmuscle.planner.resources import Core, CoreSet, OnNodeResources
 from libmuscle.post_office import PostOffice
@@ -49,9 +49,15 @@ def handle_request(self, request: bytes) -> bytes:
             response = self._get_command(*req_args)
         elif req_type == RequestType.REPORT_RESULT.value:
             response = self._report_result(*req_args)
-
+        elif req_type == RequestType.MONITOR_USAGE.value:
+            response = self._monitor_usage(*req_args)
         return cast(bytes, msgpack.packb(response, use_bin_type=True))
 
+    def _monitor_usage(self, node_name: str, usage: Dict[int, Tuple[float, int]]) -> Any:
+        """Handle a monitor usage request."""
+        print(f"Received usage report {usage} for node {node_name}")
+        return [ResponseType.SUCCESS.value]
+
     def _report_resources(
             self, node_name: str, data: Dict[str, Any]) -> Any:
         """Handle a report resources request.
@@ -166,6 +172,10 @@ def deposit_command(self, node_name: str, command: AgentCommand) -> None:
                     AgentCommandType.START.value, command.name, str(command.work_dir),
                     command.args, command.env, str(command.stdout), str(command.stderr)
                     ]
+        elif isinstance(command, AddMonitorCommand):
+            command_obj = [
+                    AgentCommandType.ADD_MONITOR.value, command.instance, command.hostname, command.pid
+                    ]
         elif isinstance(command, CancelAllCommand):
             command_obj = [AgentCommandType.CANCEL_ALL.value]
         elif isinstance(command, ShutdownCommand):