Add Qemu support for instances

examples/qemu_message_from_aleph.json

@@ -0,0 +1,66 @@
+{
+    "chain": "ETH",
+    "item_hash": "fake-hash-fake-hash-fake-hash-fake-hash-fake-hash-fake-hash-hash",
+    "sender": "0x9319Ad3B7A8E0eE24f2E639c40D8eD124C5520Ba",
+    "type": "INSTANCE",
+    "channel": "Fun-dApps",
+    "confirmed": true,
+    "content": {
+        "address": "0x9319Ad3B7A8E0eE24f2E639c40D8eD124C5520Ba",
+        "allow_amend": false,
+        "variables": {
+            "VM_CUSTOM_NUMBER": "32"
+        },
+        "environment": {
+            "reproducible": true,
+            "internet": true,
+            "aleph_api": true,
+            "shared_cache": true,
+            "hypervisor": "qemu"
+        },
+        "resources": {
+            "vcpus": 1,
+            "memory": 512,
+            "seconds": 30
+        },
+        "rootfs": {
+            "parent": {
+                "ref": "549ec451d9b099cad112d4aaa2c00ac40fb6729a92ff252ff22eef0b5c3cb613",
+                "use_latest": false
+            },
+            "persistence": "host",
+            "size_mib": 5000
+        },
+        "authorized_keys": [
+            "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDj95BHGUx0/z2G/tTrEi8o49i70xvjcEUdSs3j4A33jE7pAphrfRVbuFMgFubcm8n9r5ftd/H8SjjTL4hY9YvWV5ZuMf92GUga3n4wgevvPlBszYZCy/idxFl0vtHYC1CcK9v4tVb9onhDt8FOJkf2m6PmDyvC+6tl6LwoerXTeeiKr5VnTB4KOBkammtFmix3d1X1SZd/cxdwZIHcQ7BNsqBm2w/YzVba6Z4ZnFUelBkQtMQqNs2aV51O1pFFqtZp2mM71D5d8vn9pOtqJ5QmY5IW6NypcyqKJZg5o6QguK5rdXLkc7AWro27BiaHIENl3w0wazp9EDO9zPAGJ6lz olivier@lanius"
+        ],
+        "volumes": [
+            {
+                "mount": "/opt/venv",
+                "ref": "5f31b0706f59404fad3d0bff97ef89ddf24da4761608ea0646329362c662ba51",
+                "use_latest": false
+            },
+            {
+                "comment": "Working data persisted on the VM supervisor, not available on other nodes",
+                "mount": "/var/lib/example",
+                "name": "data",
+                "persistence": "host",
+                "size_mib": 5
+            }
+        ],
+        "replaces": "0x9319Ad3B7A8E0eE24f2E639c40D8eD124C5520Ba",
+        "time": 1619017773.8950517
+    },
+    "item_content": "{\"address\":\"0x9319Ad3B7A8E0eE24f2E639c40D8eD124C5520Ba\",\"allow_amend\":false,\"variables\":{\"VM_CUSTOM_NUMBER\":\"32\"},\"environment\":{\"reproducible\":true,\"internet\":true,\"aleph_api\":true,\"shared_cache\":true},\"resources\":{\"vcpus\":1,\"memory\":128,\"seconds\":30},\"rootfs\":{\"parent\":{\"ref\":\"549ec451d9b099cad112d4aaa2c00ac40fb6729a92ff252ff22eef0b5c3cb613\",\"use_latest\":true},\"persistence\":\"host\",\"size_mib\":20000},\"cloud_config\":{\"password\":\"password\",\"chpasswd\":{\"expire\":\"False\"}},\"volumes\":[{\"mount\":\"/opt/venv\",\"ref\":\"5f31b0706f59404fad3d0bff97ef89ddf24da4761608ea0646329362c662ba51\",\"use_latest\":false},{\"comment\":\"Working data persisted on the VM supervisor, not available on other nodes\",\"mount\":\"/var/lib/example\",\"name\":\"data\",\"persistence\":\"host\",\"size_mib\":5}],\"replaces\":\"0x9319Ad3B7A8E0eE24f2E639c40D8eD124C5520Ba\",\"time\":1619017773.8950517}",
+    "item_type": "inline",
+    "signature": "0x372da8230552b8c3e65c05b31a0ff3a24666d66c575f8e11019f62579bf48c2b7fe2f0bbe907a2a5bf8050989cdaf8a59ff8a1cbcafcdef0656c54279b4aa0c71b",
+    "size": 749,
+    "time": 1619017773.8950577,
+    "confirmations": [
+        {
+            "chain": "ETH",
+            "height": 12284734,
+            "hash": "0x67f2f3cde5e94e70615c92629c70d22dc959a118f46e9411b29659c2fce87cdc"
+        }
+    ]
+}

pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
   "msgpack~=1.0.7",
   "packaging~=23.2",
   "jsonschema==4.19.1",
+  "qmp==0.0.1"
 ]
 
 [project.urls]

src/aleph/vm/conf.py

@@ -212,6 +212,8 @@ class Settings(BaseSettings):
     # hashlib.sha256(b"secret-token").hexdigest()
     ALLOCATION_TOKEN_HASH = "151ba92f2eb90bce67e912af2f7a5c17d8654b3d29895b042107ea312a7eebda"
 
+    ENABLE_QEMU_SUPPORT: bool = Field(default=False)
+
     # Tests on programs
 
     FAKE_DATA_PROGRAM: Optional[Path] = None
@@ -291,6 +293,18 @@ def check(self):
         if self.USE_NDP_PROXY:
             assert is_command_available("ndppd"), "Command `ndppd` not found, run `apt install ndppd`"
 
+        # Necessary for cloud-init customisation of instance
+        assert is_command_available(
+            "cloud-localds"
+        ), "Command `cloud-localds` not found, run `apt install cloud-image-utils`"
+
+        if settings.ENABLE_QEMU_SUPPORT:
+            # Qemu support
+            assert is_command_available("qemu-img"), "Command `qemu-img` not found, run `apt install qemu-utils`"
+            assert is_command_available(
+                "qemu-system-x86_64"
+            ), "Command `qemu-system-x86_64` not found, run `apt install qemu-system-x86`"
+
     def setup(self):
         os.makedirs(self.MESSAGE_CACHE, exist_ok=True)
         os.makedirs(self.CODE_CACHE, exist_ok=True)

src/aleph/vm/controllers/firecracker/executable.py

@@ -4,7 +4,6 @@
 
 import asyncio
 import logging
-import subprocess
 from dataclasses import dataclass, field
 from multiprocessing import Process, set_start_method
 from os.path import exists, isfile
@@ -17,6 +16,7 @@
 
 from aleph.vm.conf import settings
 from aleph.vm.controllers.firecracker.snapshots import CompressedDiskVolumeSnapshot
+from aleph.vm.controllers.interface import AlephControllerInterface
 from aleph.vm.guest_api.__main__ import run_guest_api
 from aleph.vm.hypervisors.firecracker.microvm import FirecrackerConfig, MicroVM
 from aleph.vm.network.firewall import teardown_nftables_for_vm
@@ -137,7 +137,7 @@ class VmInitNotConnectedError(Exception):
 ConfigurationType = TypeVar("ConfigurationType")
 
 
-class AlephFirecrackerExecutable(Generic[ConfigurationType]):
+class AlephFirecrackerExecutable(Generic[ConfigurationType], AlephControllerInterface):
     vm_id: int
     vm_hash: ItemHash
     resources: AlephFirecrackerResources
@@ -150,6 +150,7 @@ class AlephFirecrackerExecutable(Generic[ConfigurationType]):
     guest_api_process: Optional[Process] = None
     is_instance: bool
     _firecracker_config: Optional[FirecrackerConfig] = None
+    support_snapshot: bool
 
     def __init__(
         self,
@@ -185,26 +186,6 @@ def __init__(
         self.guest_api_process = None
         self._firecracker_config = None
 
-    def get_vm_ip(self) -> Optional[str]:
-        if self.tap_interface:
-            return self.tap_interface.guest_ip.with_prefixlen
-        return None
-
-    def get_vm_route(self) -> Optional[str]:
-        if self.tap_interface:
-            return str(self.tap_interface.host_ip).split("/", 1)[0]
-        return None
-
-    def get_vm_ipv6(self) -> Optional[str]:
-        if self.tap_interface:
-            return self.tap_interface.guest_ipv6.with_prefixlen
-        return None
-
-    def get_vm_ipv6_gateway(self) -> Optional[str]:
-        if self.tap_interface:
-            return str(self.tap_interface.host_ipv6.ip)
-        return None
-
     def to_dict(self):
         """Dict representation of the virtual machine. Used to record resource usage and for JSON serialization."""
         if self.fvm.proc and psutil:
@@ -300,3 +281,18 @@ async def teardown(self):
 
     async def create_snapshot(self) -> CompressedDiskVolumeSnapshot:
         raise NotImplementedError()
+
+    async def get_log_queue(self) -> asyncio.Queue:
+        queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
+        # Limit the number of queues per VM
+
+        if len(self.fvm.log_queues) > 20:
+            logger.warning("Too many log queues, dropping the oldest one")
+            self.fvm.log_queues.pop(0)
+        self.fvm.log_queues.append(queue)
+        return queue
+
+    async def unregister_queue(self, queue: asyncio.Queue):
+        if queue in self.fvm.log_queues:
+            self.fvm.log_queues.remove(queue)
+        queue.empty()

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

@@ -58,6 +58,7 @@ class AlephFirecrackerInstance(AlephFirecrackerExecutable):
     resources: AlephInstanceResources
     latest_snapshot: Optional[DiskVolumeSnapshot]
     is_instance = True
+    support_snapshot = False
 
     def __init__(
         self,

src/aleph/vm/controllers/firecracker/program.py

@@ -258,6 +258,7 @@ class AlephFirecrackerProgram(AlephFirecrackerExecutable[ProgramVmConfiguration]
     vm_configuration: ProgramVmConfiguration | None
     resources: AlephProgramResources
     is_instance = False
+    support_snapshot = False
 
     def __init__(
         self,

src/aleph/vm/controllers/firecracker/snapshot_manager.py

@@ -96,7 +96,7 @@ def run_snapshots(self) -> None:
         job_thread.start()
 
     async def start_for(self, vm: AlephFirecrackerExecutable, frequency: Optional[int] = None) -> None:
-        if not vm.is_instance:
+        if not vm.support_snapshot:
             msg = "Snapshots are not implemented for programs."
             raise NotImplementedError(msg)
 

src/aleph/vm/controllers/interface.py

@@ -0,0 +1,91 @@
+import asyncio
+import logging
+from abc import ABC
+from asyncio.subprocess import Process
+from typing import Any, Coroutine, Optional
+
+from aleph_message.models import ItemHash
+from aleph_message.models.execution.environment import MachineResources
+
+from aleph.vm.controllers.firecracker.snapshots import CompressedDiskVolumeSnapshot
+from aleph.vm.network.interfaces import TapInterface
+
+logger = logging.getLogger(__name__)
+
+
+class AlephControllerInterface(ABC):
+    vm_id: int
+    "id in the VMPool, attributed at execution"
+    vm_hash: ItemHash
+    "identifier for the VM definition, linked to an Aleph Message"
+    resources: Any
+    "local resource for the machine"
+    enable_console: bool
+    enable_networking: bool
+    "enable networking for this VM"
+    hardware_resources: MachineResources
+    support_snapshot: bool
+    "Does this controller support snapshotting"
+    guest_api_process: Optional[Process] = None
+    tap_interface: Optional[TapInterface] = None
+    "Network interface used for this VM"
+
+    def get_vm_ip(self) -> Optional[str]:
+        if self.tap_interface:
+            return self.tap_interface.guest_ip.with_prefixlen
+        return None
+
+    def get_vm_route(self) -> Optional[str]:
+        if self.tap_interface:
+            return str(self.tap_interface.host_ip).split("/", 1)[0]
+        return None
+
+    def get_vm_ipv6(self) -> Optional[str]:
+        if self.tap_interface:
+            return self.tap_interface.guest_ipv6.with_prefixlen
+        return None
+
+    def get_vm_ipv6_gateway(self) -> Optional[str]:
+        if self.tap_interface:
+            return str(self.tap_interface.host_ipv6.ip)
+        return None
+
+    def to_dict(self):
+        """Dict representation of the virtual machine. Used to record resource usage and for JSON serialization."""
+        raise NotImplementedError()
+
+    async def setup(self):
+        """Configuration done before the VM process is started"""
+        raise NotImplementedError()
+
+    async def start(self):
+        """Start the VM process"""
+        raise NotImplementedError()
+
+    async def wait_for_init(self) -> None:
+        """Wait for the init process of the virtual machine to be ready.
+        May be empty."""
+        pass
+
+    async def configure(self) -> None:
+        """Configuration done after the VM process is started"""
+        raise NotImplementedError()
+
+    async def start_guest_api(self):
+        raise NotImplementedError()
+
+    async def stop_guest_api(self):
+        raise NotImplementedError()
+
+    async def teardown(self) -> Coroutine:
+        raise NotImplementedError()
+
+    async def create_snapshot(self) -> CompressedDiskVolumeSnapshot:
+        "Must be implement if self.support_snapshot is True"
+        raise NotImplementedError()
+
+    async def get_log_queue(self) -> asyncio.Queue:
+        raise NotImplementedError()
+
+    async def unregister_queue(self, queue: asyncio.Queue):
+        raise NotImplementedError()