Fix: Added qemu communication channel options to allow VM to receive commands

Author: Andres D. Molins

src/aleph/vm/controllers/configuration.py

@@ -34,6 +34,7 @@ class QemuVMConfiguration(BaseModel):
     image_path: str
     monitor_socket_path: Path
     qmp_socket_path: Path
+    qga_socket_path: Path
     vcpu_count: int
     mem_size_mb: int
     interface_name: str | None = None
@@ -48,6 +49,7 @@ class QemuConfidentialVMConfiguration(BaseModel):
     image_path: str
     monitor_socket_path: Path
     qmp_socket_path: Path
+    qga_socket_path: Path
     vcpu_count: int
     mem_size_mb: int
     interface_name: str | None = None

src/aleph/vm/controllers/qemu/client.py

@@ -1,7 +1,9 @@
+import base64
 import time
 from enum import Enum
 
 import qmp
+import yaml
 from pydantic import BaseModel
 
 
@@ -74,11 +76,19 @@ def __init__(self, vm):
         if not (vm.qmp_socket_path and vm.qmp_socket_path.exists()):
             msg = "VM is not running"
             raise Exception(msg)
-        client = qmp.QEMUMonitorProtocol(str(vm.qmp_socket_path))
-        client.connect()
 
-        # qmp_client = qmp.QEMUMonitorProtocol(address=("localhost", vm.qmp_port))
-        self.qmp_client = client
+        qmp_client = qmp.QEMUMonitorProtocol(str(vm.qmp_socket_path))
+        qmp_client.connect()
+        self.qmp_client = qmp_client
+
+        # QGA (QEMU Guest Agent) uses a separate communication channel over virtio-serial.
+        # The wire protocol is JSON-based and compatible with the QMP library.
+        if vm.qga_socket_path and vm.qga_socket_path.exists():
+            qga_client = qmp.QEMUMonitorProtocol(str(vm.qga_socket_path))
+            qga_client.connect()
+            self.qga_client = qga_client
+        else:
+            self.qga_client = None
 
     def __enter__(self):
         return self
@@ -88,6 +98,8 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
     def close(self) -> None:
         self.qmp_client.close()
+        if self.qga_client:
+            self.qga_client.close()
 
     def query_sev_info(self) -> VmSevInfo:
         caps = self.qmp_client.command("query-sev")
@@ -176,6 +188,13 @@ def migrate_cancel(self) -> None:
         """Cancel ongoing migration."""
         self.qmp_client.command("migrate_cancel")
 
+    def _get_qga_client(self) -> qmp.QEMUMonitorProtocol:
+        """Get the QGA client, raising an error if not available."""
+        if not self.qga_client:
+            msg = "QEMU Guest Agent socket is not available"
+            raise Exception(msg)
+        return self.qga_client
+
     def guest_exec(self, command: str, args: list[str] | None = None, capture_output: bool = True) -> dict:
         """
         Execute a command in the guest via qemu-guest-agent.
@@ -185,10 +204,11 @@ def guest_exec(self, command: str, args: list[str] | None = None, capture_output
         :param capture_output: Whether to capture stdout/stderr
         :return: Dict with 'pid' key for the started process
         """
+        qga = self._get_qga_client()
         exec_args = {"path": command, "capture-output": capture_output}
         if args:
             exec_args["arg"] = args
-        return self.qmp_client.command("guest-exec", **exec_args)
+        return qga.command("guest-exec", **exec_args)
 
     def guest_exec_status(self, pid: int) -> dict:
         """
@@ -197,14 +217,15 @@ def guest_exec_status(self, pid: int) -> dict:
         :param pid: The PID returned by guest_exec
         :return: Dict with 'exited', 'exitcode', 'out-data', 'err-data' keys
         """
-        return self.qmp_client.command("guest-exec-status", pid=pid)
+        qga = self._get_qga_client()
+        return qga.command("guest-exec-status", pid=pid)
 
     def reconfigure_guest_network(
         self,
         new_ip: str,
         gateway: str,
         nameservers: list[str],
-        interface: str = "ens3",
+        interface: str = "eth0",
     ) -> dict:
         """
         Reconfigure guest network via qemu-guest-agent after migration.
@@ -215,35 +236,31 @@ def reconfigure_guest_network(
         :param new_ip: New IP address with CIDR notation (e.g., "10.0.0.5/24")
         :param gateway: Gateway IP address (e.g., "10.0.0.1")
         :param nameservers: List of DNS server IPs (e.g., ["8.8.8.8", "8.8.4.4"])
-        :param interface: Network interface name (default: "ens3")
+        :param interface: Network interface name (default: "eth0")
         :return: Dict with 'pid' key for the started process
         """
-        # Build nameservers YAML list
-        ns_yaml = "\n".join(f"        - {ns}" for ns in nameservers)
-
-        netplan_config = f"""network:
-  version: 2
-  ethernets:
-    {interface}:
-      addresses: [{new_ip}]
-      routes:
-        - to: default
-          via: {gateway}
-      nameservers:
-        addresses:
-{ns_yaml}
-"""
+        network_config = {
+            "network": {
+                "version": 2,
+                "ethernets": {
+                    interface: {
+                        "addresses": [new_ip],
+                        "routes": [{"to": "default", "via": gateway}],
+                        "nameservers": {"addresses": nameservers},
+                    },
+                },
+            },
+        }
+        netplan_yaml = yaml.safe_dump(network_config, default_flow_style=False, sort_keys=False)
 
         # Create a script that writes the netplan config and applies it
         # Use base64 encoding to avoid escaping issues
-        import base64
-
-        config_b64 = base64.b64encode(netplan_config.encode()).decode()
+        config_b64 = base64.b64encode(netplan_yaml.encode()).decode()
 
         script = f"""
-echo '{config_b64}' | base64 -d > /etc/netplan/50-cloud-init.yaml
-netplan apply
-"""
+        echo '{config_b64}' | base64 -d > /etc/netplan/50-cloud-init.yaml
+        netplan apply
+        """
 
         return self.guest_exec("/bin/bash", ["-c", script])
 
@@ -254,11 +271,12 @@ def wait_for_guest_agent(self, timeout_seconds: int = 60) -> bool:
         :param timeout_seconds: Maximum time to wait
         :return: True if agent is available, False if timeout
         """
+        qga = self._get_qga_client()
         start_time = time.monotonic()
         while time.monotonic() - start_time < timeout_seconds:
             try:
                 # Try to ping the guest agent
-                self.qmp_client.command("guest-ping")
+                qga.command("guest-ping")
                 return True
             except Exception:
                 time.sleep(1)

src/aleph/vm/controllers/qemu/instance.py

@@ -200,6 +200,7 @@ async def configure(self, incoming_migration_port: int | None = None):
             image_path=image_path,
             monitor_socket_path=monitor_socket_path,
             qmp_socket_path=self.qmp_socket_path,
+            qga_socket_path=self.qga_socket_path,
             vcpu_count=vcpu_count,
             mem_size_mb=mem_size_mb,
             interface_name=interface_name,
@@ -236,6 +237,10 @@ def save_controller_configuration(self):
     def qmp_socket_path(self) -> Path:
         return settings.EXECUTION_ROOT / f"{self.vm_hash}-qmp.socket"
 
+    @property
+    def qga_socket_path(self) -> Path:
+        return settings.EXECUTION_ROOT / f"{self.vm_hash}-qga.socket"
+
     async def start(self):
         # Start via systemd not here
         raise NotImplementedError()

src/aleph/vm/hypervisors/qemu/qemuvm.py

@@ -23,6 +23,7 @@ class QemuVM:
     image_path: str
     monitor_socket_path: Path
     qmp_socket_path: Path
+    qga_socket_path: Path
     vcpu_count: int
     mem_size_mb: int
     interface_name: str
@@ -45,6 +46,7 @@ def __init__(self, vm_hash, config: QemuVMConfiguration):
         self.image_path = config.image_path
         self.monitor_socket_path = config.monitor_socket_path
         self.qmp_socket_path = config.qmp_socket_path
+        self.qga_socket_path = config.qga_socket_path
         self.vcpu_count = config.vcpu_count
         self.mem_size_mb = config.mem_size_mb
         self.interface_name = config.interface_name
@@ -91,7 +93,7 @@ async def start(
             "-smp",
             str(self.vcpu_count),
             "-drive",
-            f"file={self.image_path},media=disk,if=virtio",
+            f"file={self.image_path},media=disk,if=virtio,file.locking=off",
             # To debug you can pass gtk or curses instead
             "-display",
             "none",
@@ -100,6 +102,13 @@ async def start(
             # Listen for commands on this socket
             "-monitor",
             f"unix:{self.monitor_socket_path},server,nowait",
+            # Qemu Guest Agent communication channel options
+            "-device",
+            "virtio-serial",
+            "-chardev",
+            f"socket,path={self.qga_socket_path},server=on,wait=off,id=qga0",
+            "-device",
+            "virtserialport,chardev=qga0,name=org.qemu.guest_agent.0",
             # Listen for commands on this socket (QMP protocol in json). Supervisor use it to send shutdown or start
             # command
             "-qmp",