virtme-ng: use pre-compiled mainline kernels

Allow to download precompiled mainline kernels from the Ubuntu mainline
kernel repository (https://kernel.ubuntu.com/mainline).

Mainline builds are provided by Ubuntu for debugging purposes, they are
basically vanilla kernels (no additional patches applied), built with
the generic Ubuntu .config and packaged as deb.

These precompiled kernels can be used by virtme-ng to test specific
mainline tags, without having to rebuild the kernel from source.

To do so the option `--run` can now accept a Linux tag (i.e., v6.6-rc2).
When a tag is specified, virtme-ng will search in the Ubuntu mainline
repository and fetch the corresponding packages, if available.

Packages are then cached and extracted inside $HOME/.cache/virtme-ng, so
they just need to be downloaded the first time that a mainline tag is
requested.

Example usage:

 $ vng -r v6.6-rc2 -- uname -r
 6.6.0-060600rc2-generic

This allows to save even more time (and energy) when testing mainline
kernel versions, completely cutting out the kernel rebuild time.

Signed-off-by: Andrea Righi <[email protected]>

requirements.txt

@@ -1 +1,2 @@
 argcomplete
+requests

virtme_ng/mainline.py

@@ -0,0 +1,70 @@
+# -*- mode: python -*-
+# Copyright 2023 Andrea Righi
+
+"""virtme-ng: mainline kernel downloader."""
+
+import os
+import re
+import sys
+import subprocess
+from glob import glob
+import requests
+from virtme_ng.utils import CACHE_DIR, spinner_decorator
+
+BASE_URL = "https://kernel.ubuntu.com/mainline"
+
+HTTP_CHUNK = 4096
+HTTP_TIMEOUT = 30
+
+
+class KernelDownloader:
+    def __init__(self, version, arch="amd64", verbose=False):
+        # Fetch and extract precompiled mainline kernel
+        self.kernel_dir = f"{CACHE_DIR}/{version}/{arch}"
+        self.version = version
+        self.arch = arch
+        self.verbose = verbose
+        self.target = f"{self.kernel_dir}/boot/vmlinuz*generic*"
+
+        if not glob(self.target):
+            self._fetch_kernel()
+
+    def _download_file(self, url, destination):
+        response = requests.get(url, stream=True, timeout=HTTP_TIMEOUT)
+        if response.status_code == 200:
+            os.makedirs(self.kernel_dir, exist_ok=True)
+            with open(destination, 'wb') as file:
+                for chunk in response.iter_content(chunk_size=HTTP_CHUNK):
+                    file.write(chunk)
+        else:
+            raise FileNotFoundError(f"failed to download {url}, error: {response.status_code}")
+
+    @spinner_decorator(message="📥 downloading kernel")
+    def _fetch_kernel(self):
+        url = BASE_URL + "/" + self.version + "/" + self.arch
+        response = requests.get(url, timeout=HTTP_TIMEOUT)
+        if response.status_code != 200:
+            url = BASE_URL + "/" + self.version
+            response = requests.get(url, timeout=HTTP_TIMEOUT)
+        if self.verbose:
+            sys.stderr.write(f"use {self.version}/{self.arch} pre-compiled kernel from {url}\n")
+        if response.status_code == 200:
+            href_pattern = re.compile(r'href=["\']([^\s"\']+.deb)["\']')
+            matches = href_pattern.findall(response.text)
+            for match in matches:
+                # Skip headers packages
+                if 'headers' in match:
+                    continue
+                # Skip packages for different architectures
+                if f'{self.arch}.deb' not in match:
+                    continue
+                # Skip if package is already downloaded
+                deb_file = f"{self.kernel_dir}/{match}"
+                if os.path.exists(deb_file):
+                    continue
+                self._download_file(url + "/" + match, deb_file)
+                subprocess.check_call(['dpkg', '-x', deb_file, self.kernel_dir])
+            if not glob(f"{self.kernel_dir}/*.deb"):
+                raise FileNotFoundError(f"could not find kernel packages at {url}")
+        else:
+            raise FileNotFoundError(f"failed to retrieve content, error: {response.status_code}")

virtme_ng/run.py

@@ -28,23 +28,11 @@ def autocomplete(*args, **kwargs):
         pass
 
 from virtme.util import SilentError, uname, get_username
-from virtme_ng.utils import CONF_FILE
-from virtme_ng.spinner import Spinner
+from virtme_ng.utils import CONF_FILE, spinner_decorator
+from virtme_ng.mainline import KernelDownloader
 from virtme_ng.version import VERSION
 
 
-def spinner_decorator(message):
-    def decorator(func):
-        def wrapper(*args, **kwargs):
-            with Spinner(message=message):
-                result = func(*args, **kwargs)
-                return result
-
-        return wrapper
-
-    return decorator
-
-
 def check_call_cmd(command, quiet=False, dry_run=False):
     if dry_run:
         print(" ".join(command))
@@ -776,7 +764,22 @@ def _get_virtme_overlay_rwdir(self, args):
 
     def _get_virtme_run(self, args):
         if args.run is not None:
-            self.virtme_param["kdir"] = "--kimg " + args.run
+            # If an upstream version is specified (using an upstream tag) fetch
+            # and run the corresponding kernel from the Ubuntu mainline
+            # repository.
+            if args.run.startswith('v'):
+                if args.arch is None:
+                    arch = 'amd64'
+                else:
+                    arch = args.arch
+                try:
+                    mainline = KernelDownloader(args.run, arch=arch, verbose=args.verbose)
+                    self.virtme_param["kdir"] = "--kimg " + mainline.target
+                except FileNotFoundError as exc:
+                    sys.stderr.write(str(exc) + "\n")
+                    sys.exit(1)
+            else:
+                self.virtme_param["kdir"] = "--kimg " + args.run
         else:
             self.virtme_param["kdir"] = "--kdir ./"
 

virtme_ng/utils.py

@@ -4,6 +4,20 @@
 """virtme-ng: configuration path."""
 
 from pathlib import Path
+from virtme_ng.spinner import Spinner
 
+CACHE_DIR = Path(Path.home(), ".cache", "virtme-ng")
 CONF_PATH = Path(Path.home(), ".config", "virtme-ng")
 CONF_FILE = Path(CONF_PATH, "virtme-ng.conf")
+
+
+def spinner_decorator(message):
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            with Spinner(message=message):
+                result = func(*args, **kwargs)
+                return result
+
+        return wrapper
+
+    return decorator