Usage system: remove reserved usage for volumes

of vm when calculate available disk space

Jira: ALEPH-420

This will help the scheduler scheduler more properly the available resource on CRN

Alternative approach to #780 as Sparse file were not properly created.

Author: olethanh

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

@@ -13,7 +13,7 @@
 from aiohttp import ClientResponseError
 from aleph_message.models import ExecutableContent, ItemHash
 from aleph_message.models.execution.environment import MachineResources
-from aleph_message.models.execution.volume import PersistentVolume
+from aleph_message.models.execution.volume import MachineVolume, PersistentVolume
 
 from aleph.vm.conf import settings
 from aleph.vm.controllers.configuration import (
@@ -71,6 +71,7 @@ class HostVolume:
     mount: str
     path_on_host: Path
     read_only: bool
+    size_mib: int | None
 
 
 @dataclass
@@ -100,6 +101,35 @@ class AlephFirecrackerResources:
     volumes: list[HostVolume]
     namespace: str
 
+    def get_disk_usage_delta(self) -> int:
+        """Difference between the size requested and what is currently used on disk.
+
+        Count rootfs and volumes.
+        Used to calculate an estimate of space resource available for use.
+        Value in bytes and is negative"""
+
+        total_delta = 0
+        # Root fs
+        if hasattr(self.message_content, "rootfs"):
+            volume = self.message_content.rootfs
+            used_size = self.rootfs_path.stat().st_size if self.rootfs_path.exists() else 0
+            requested_size = int(volume.size_mib * 1024 * 1024)
+            size_delta = used_size - requested_size
+            total_delta += size_delta
+
+        # Count each extra volume
+        for volume in self.volumes:
+            if not volume.size_mib:
+                # planned size not set on immutable volume
+                size_delta = 0
+            else:
+                used_size = volume.path_on_host.stat().st_size if volume.path_on_host.exists() else 0
+                requested_size = int(volume.size_mib * 1024 * 1024)
+
+                size_delta = used_size - requested_size
+            total_delta += size_delta
+        return total_delta
+
     def __init__(self, message_content: ExecutableContent, namespace: str):
         self.message_content = message_content
         self.namespace = namespace
@@ -115,8 +145,9 @@ async def download_kernel(self):
     async def download_volumes(self):
         volumes = []
         # TODO: Download in parallel and prevent duplicated volume names
+        volume: MachineVolume
         for i, volume in enumerate(self.message_content.volumes):
-            # only persistant volume has name and mount
+            # only persistent volume has name and mount
             if isinstance(volume, PersistentVolume):
                 if not volume.name:
                     volume.name = f"unamed_volume_{i}"
@@ -127,6 +158,7 @@ async def download_volumes(self):
                     mount=volume.mount,
                     path_on_host=(await get_volume_path(volume=volume, namespace=self.namespace)),
                     read_only=volume.is_read_only(),
+                    size_mib=getattr(volume, "size_mib", None),
                 )
             )
         self.volumes = volumes

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

@@ -72,7 +72,7 @@ async def make_writable_volume(self, parent_image_path, volume: PersistentVolume
             raise VmSetupError(msg)
 
         dest_path = settings.PERSISTENT_VOLUMES_DIR / self.namespace / f"{volume_name}.qcow2"
-        # Do not override if user asked for host persistance.
+        # Do not override if user asked for host persistence.
         if dest_path.exists() and volume.persistence == VolumePersistence.host:
             return dest_path
 

src/aleph/vm/orchestrator/resources.py

@@ -133,6 +133,7 @@ async def about_system_usage(request: web.Request):
     """Public endpoint to expose information about the system usage."""
     period_start = datetime.now(timezone.utc).replace(second=0, microsecond=0)
     machine_properties = get_machine_properties()
+    pool = request.app["vm_pool"]
 
     usage: MachineUsage = MachineUsage(
         cpu=CpuUsage(
@@ -146,7 +147,8 @@ async def about_system_usage(request: web.Request):
         ),
         disk=DiskUsage(
             total_kB=psutil.disk_usage(str(settings.PERSISTENT_VOLUMES_DIR)).total // 1000,
-            available_kB=psutil.disk_usage(str(settings.PERSISTENT_VOLUMES_DIR)).free // 1000,
+            available_kB=pool.calculate_available_disk() // 1000,
+            # available_kB=psutil.disk_usage(str(settings.PERSISTENT_VOLUMES_DIR)).free // 1000,
         ),
         period=UsagePeriod(
             start_timestamp=period_start,

src/aleph/vm/pool.py

@@ -3,6 +3,7 @@
 import asyncio
 import json
 import logging
+import shutil
 from collections.abc import Iterable
 from datetime import datetime, timedelta, timezone
 from typing import Any, cast
@@ -101,6 +102,29 @@ def teardown(self) -> None:
             #  upon supervisor restart or upgrade.
             pass
 
+    def calculate_available_disk(self):
+        """Disk available for the creation of new VM.
+
+        This take into account the disk request (but not used) for Volume of executions in the pool"""
+        free_space = shutil.disk_usage(str(settings.PERSISTENT_VOLUMES_DIR)).free // 1000
+        # Free disk space reported by system
+
+        # Calculate the reservation
+        total_delta = 0
+        for execution in self.executions.values():
+            if not execution.resources:
+                continue
+            delta = execution.resources.get_disk_usage_delta()
+            logger.warning("Disk usage delta: %d for %s", delta, execution.vm_hash)
+            total_delta += delta
+        available_space = free_space - total_delta
+        logger.info(
+            "Disk: freespace : %.f Mb,   available space (non reserved) %.f Mb",
+            free_space / 1024**2,
+            available_space / 1024**2,
+        )
+        return available_space
+
     async def create_a_vm(
         self, vm_hash: ItemHash, message: ExecutableContent, original: ExecutableContent, persistent: bool
     ) -> VmExecution:

tests/supervisor/test_views.py

@@ -39,18 +39,10 @@ async def test_allocation_fails_on_invalid_item_hash(aiohttp_client):
 
 
 @pytest.mark.asyncio
-async def test_system_usage(aiohttp_client):
+async def test_system_usage(aiohttp_client, mocker, mock_app_with_pool):
     """Test that the usage system endpoints responds. No auth needed"""
 
-    class FakeVmPool:
-        gpus = []
-
-        def get_available_gpus(self):
-            return []
-
-    app = setup_webapp()
-    app["vm_pool"] = FakeVmPool()
-    client = await aiohttp_client(app)
+    client = await aiohttp_client(await mock_app_with_pool)
     response: web.Response = await client.get("/about/usage/system")
     assert response.status == 200
     # check if it is valid json
@@ -60,15 +52,9 @@ def get_available_gpus(self):
 
 
 @pytest.mark.asyncio
-async def test_system_usage_mock(aiohttp_client, mocker):
+async def test_system_usage_mock(aiohttp_client, mocker, mock_app_with_pool):
     """Test that the usage system endpoints response value. No auth needed"""
 
-    class FakeVmPool:
-        gpus = []
-
-        def get_available_gpus(self):
-            return []
-
     mocker.patch(
         "cpuinfo.cpuinfo.get_cpu_info",
         {
@@ -85,9 +71,7 @@ def get_available_gpus(self):
         lambda: 200,
     )
 
-    app = setup_webapp()
-    app["vm_pool"] = FakeVmPool()
-    client = await aiohttp_client(app)
+    client = await aiohttp_client(await mock_app_with_pool)
     response: web.Response = await client.get("/about/usage/system")
     assert response.status == 200
     # check if it is valid json