feat: OCI image loader (Hyper.Img.OciLoader)

AGENTS.md

@@ -132,6 +132,16 @@ Coverage is a side effect of good tests, never the target.
 
 ## Other conventions
 
+- **Comments earn their place.** No section-divider banners (`# --- foo ---`),
+  no comment that just restates what the next line does. A comment explains a
+  non-obvious *why* a reader cannot recover from the code — a workaround, an
+  invariant, a deliberate trade-off. Prefer self-documenting code: a named
+  function, a `Unit.*` quantity instead of a bare `1024 * 1024`, a descriptive
+  variable — over a comment narrating the mechanics. If you reach for a comment
+  to explain *what*, rename the thing instead.
+- Don't hand-roll magic numbers for sizes/durations/bandwidth: use the
+  `Unit.*` types (`Unit.Information.mib(8)`, not `8 * 1024 * 1024`) and
+  `use Unit.Operators` for unit-aware arithmetic.
 - Add `@spec` to public functions. Dialyzer runs with `:unmatched_returns`,
   `:extra_return`, `:missing_return` and will fail the gate on a missing/wrong
   spec.

docs/cookbook/intro.md

@@ -17,6 +17,20 @@ with it.
 
 The absolute best way to get started with `Hyper` is to play with it.
 
+### Requirements
+
+Hyper requires the following software be installed on each node running it:
+
+  - [`skopeo`](https://github.com/containers/skopeo)
+  - [`e2fsprogs`](https://github.com/tytso/e2fsprogs)
+
+Hyper has more runtime dependencies, but they are automatically redistributed
+by Hyper.
+
+### Installation
+
+<!-- TODO(markovejnovic): Write this out. -->
+
 ### Configuration
 
 Running `Hyper` is involved and requires a large number of pre-requisites. The
@@ -41,3 +55,27 @@ config :hyper,
   uid_gid_range: {900_000, 999_999},
   layer_dir: "/srv/hyper/layers"
 ```
+
+<!-- TODO(markovejnovic): Update the config section. -->
+
+### Usage
+
+<!-- TODO(markovejnovic): Write out how to boot hyper etc -->
+
+#### Loading Images
+
+Before an image can be booted, it needs to be loaded into Hyper. Currently, the
+only way to load images is through an OCI image, either natively or through the
+native interface, or through [gRPC](../grpc.md):
+
+```elixir
+{:ok, img_id} = Hyper.Img.OciLoader.load("docker.io/library/alpine:3.19")
+```
+
+#### Booting a VM
+
+With the image loaded, and an `img_id` in hand, you can boot it:
+
+```elixir
+{:ok, vm} = Hyper.create_vm(%Hyper.Vm.Spec{ img_id: img_id })
+```

docs/grpc.md

@@ -51,14 +51,31 @@ from hyper.grpc.v0 import hyper_pb2, hyper_pb2_grpc
 client = hyper_pb2_grpc.HyperStub(grpc.aio.insecure_channel("localhost:50051"))
 ```
 
+### Loading Images
+
+Before you can create a VM you need an image in the cluster. `LoadImage` pulls an
+OCI image, builds its rootfs, and records it -- returning the `img_id` you pass to
+`CreateVm`. It blocks until the load finishes (this can take minutes), so set a
+generous deadline.
+
+```python
+loaded = await client.LoadImage(
+    hyper_pb2.LoadImageRequest(
+        image_ref="docker.io/library/alpine:3.19",
+        # label is optional; defaults to image_ref.
+    )
+)
+print(loaded.img_id)  # pass this to CreateVm
+```
+
 ### Creating VMs
 
 You can create new VMs with the `CreateVm` RPC.
 
 ```python
 created = await client.CreateVm(
     hyper_pb2.CreateVmRequest(
-        img_id="img-abc",
+        img_id=loaded.img_id,
         instance_type=hyper_pb2.INSTANCE_TYPE_DECI,
         arch=hyper_pb2.ARCHITECTURE_X86_64,
         # boot_args is optional; omit it for the default kernel cmdline.

lib/hyper.ex

@@ -8,11 +8,6 @@ defmodule Hyper do
     @type id :: String.t()
   end
 
-  defmodule Img do
-    @moduledoc "A content-addressed image: an ordered stack of layers."
-    @type id :: String.t()
-  end
-
   @doc """
   Create a new virtual machine from an image.
 

lib/hyper/config.ex

@@ -19,6 +19,9 @@ defmodule Hyper.Config do
   @losetup_path Application.compile_env(:hyper, :losetup_path, "losetup")
   @dmsetup_path Application.compile_env(:hyper, :dmsetup_path, "dmsetup")
   @blockdev_path Application.compile_env(:hyper, :blockdev_path, "blockdev")
+  @skopeo_path Application.compile_env(:hyper, :skopeo_path, "skopeo")
+  @umoci_path Application.compile_env(:hyper, :umoci_path, nil)
+  @mke2fs_path Application.compile_env(:hyper, :mke2fs_path, "mke2fs")
   @vmlinux Application.compile_env(:hyper, :vmlinux, %{})
 
   @doc """
@@ -62,6 +65,10 @@ defmodule Hyper.Config do
   @spec vmlinux_install_dir :: Path.t()
   def vmlinux_install_dir, do: Path.join(redist_dir(), "vmlinux")
 
+  @doc "Directory where `Hyper.Img.OciLoader.Umoci` installs the default umoci binary."
+  @spec umoci_install_dir :: Path.t()
+  def umoci_install_dir, do: Path.join(redist_dir(), "umoci")
+
   @doc """
   Path to the directory where all VM chroot's are created (`<work_dir>/jails`).
 
@@ -114,6 +121,18 @@ defmodule Hyper.Config do
   @doc "Path to the blockdev binary."
   def blockdev_path, do: @blockdev_path
 
+  @doc "Path to the skopeo binary (used by `Hyper.Img.OciLoader` to pull OCI images)."
+  def skopeo_path, do: @skopeo_path
+
+  @doc """
+  Operator-configured path to the umoci binary, or `nil` (the default) to let
+  `Hyper.Img.OciLoader.Umoci` download and manage a pinned default.
+  """
+  def umoci_path, do: @umoci_path
+
+  @doc "Path to the mke2fs binary (used by `Hyper.Img.OciLoader` to build the ext4 rootfs)."
+  def mke2fs_path, do: @mke2fs_path
+
   @doc """
   Path to the setuid-root device helper (`hyper-suidhelper`). Required: the node
   runs unprivileged and routes every `losetup`/`dmsetup`/`blockdev` operation

lib/hyper/grpc/codec.ex

@@ -15,6 +15,8 @@ defmodule Hyper.Grpc.Codec do
     CreateVmResponse,
     GetVmResponse,
     ListVmsResponse,
+    LoadImageRequest,
+    LoadImageResponse,
     Vm
   }
 
@@ -72,6 +74,16 @@ defmodule Hyper.Grpc.Codec do
     end
   end
 
+  @spec from_grpc(LoadImageRequest.t()) ::
+          {:ok, {String.t(), keyword()}} | {:error, :missing_image_ref}
+  def from_grpc(%LoadImageRequest{image_ref: ref}) when ref in [nil, ""],
+    do: {:error, :missing_image_ref}
+
+  def from_grpc(%LoadImageRequest{image_ref: ref, label: label}) do
+    opts = if label in [nil, ""], do: [], else: [label: label]
+    {:ok, {ref, opts}}
+  end
+
   @doc "Convert a domain result to an outbound response message, or an error to `GRPC.RPCError`."
   @spec to_grpc({:created, Hyper.Vm.id(), node()}) :: CreateVmResponse.t()
   def to_grpc({:created, vm_id, node}) when is_binary(vm_id),
@@ -85,6 +97,10 @@ defmodule Hyper.Grpc.Codec do
   def to_grpc({:vms, vms}),
     do: %ListVmsResponse{vms: Enum.map(vms, &vm/1)}
 
+  @spec to_grpc({:loaded, Hyper.Img.id()}) :: LoadImageResponse.t()
+  def to_grpc({:loaded, img_id}) when is_binary(img_id),
+    do: %LoadImageResponse{img_id: img_id}
+
   @spec to_grpc(:stopped) :: Empty.t()
   def to_grpc(:stopped), do: %Empty{}
 
@@ -124,6 +140,19 @@ defmodule Hyper.Grpc.Codec do
   defp rpc_error(reason) when reason in [:no_capacity, :exhausted],
     do: GRPC.RPCError.exception(:resource_exhausted, "no capacity")
 
+  defp rpc_error(:missing_image_ref),
+    do: GRPC.RPCError.exception(:invalid_argument, "image_ref is required")
+
+  defp rpc_error(:invalid_ref),
+    do: GRPC.RPCError.exception(:invalid_argument, "image_ref is malformed")
+
+  defp rpc_error({:missing_tools, tools}),
+    do:
+      GRPC.RPCError.exception(
+        :failed_precondition,
+        "node is missing required image tools: #{Enum.join(tools, ", ")}"
+      )
+
   defp rpc_error(reason),
     do: GRPC.RPCError.exception(:internal, "internal error: #{inspect(reason)}")
 end

lib/hyper/grpc/server.ex

@@ -17,9 +17,21 @@ defmodule Hyper.Grpc.Server do
     GetVmRequest,
     GetVmResponse,
     ListVmsResponse,
+    LoadImageRequest,
+    LoadImageResponse,
     StopVmRequest
   }
 
+  @spec load_image(LoadImageRequest.t(), GRPC.Server.Stream.t()) :: LoadImageResponse.t()
+  def load_image(%LoadImageRequest{} = req, _stream) do
+    with {:ok, {ref, opts}} <- Codec.from_grpc(req),
+         {:ok, img_id} <- Hyper.Img.OciLoader.load(ref, opts) do
+      Codec.to_grpc({:loaded, img_id})
+    else
+      {:error, reason} -> raise Codec.to_grpc({:error, reason})
+    end
+  end
+
   @spec create_vm(CreateVmRequest.t(), GRPC.Server.Stream.t()) :: CreateVmResponse.t()
   def create_vm(%CreateVmRequest{} = req, _stream) do
     with {:ok, spec} <- Codec.from_grpc(req),

lib/hyper/img.ex

@@ -0,0 +1,161 @@
+defmodule Hyper.Img do
+  @moduledoc """
+  A content-addressed image: an ordered stack of layers, and the entry point for
+  putting one into the cluster.
+
+  `create/2` ingests a prepared image file -- e.g. the ext4 rootfs produced by
+  `Hyper.Img.OciLoader` -- into the shared media store and the image database. It
+  content-addresses the file (sha256 of its bytes = the image id), publishes it
+  into `Hyper.Config.layer_dir/0` at `layer_<id>.img`, then records it as a
+  one-layer base image (`blobs` + `images` + `image_layers`). Producers of image
+  files stay decoupled from the store and DB: they hand a path to `create/2`.
+  """
+
+  use OpenTelemetryDecorator
+
+  require Logger
+
+  alias Hyper.Config
+  alias Hyper.Img.Db.{Blob, Image, ImageLayer, Repo}
+
+  @type id :: String.t()
+
+  # `Ecto.Multi` is an opaque struct; building it through the pipe trips
+  # dialyzer's opacity check (a known Ecto false positive), so silence it for the
+  # one function that assembles a Multi.
+  @dialyzer {:no_opaque, record: 3}
+
+  @doc """
+  Ingest the image file at `path` into the cluster and return its
+  content-addressed id.
+
+  Content-addresses `path` (sha256 of its bytes = the id), publishes it into the
+  media store at `layer_<id>.img`, and records it as a one-layer base image. The
+  file at `path` is consumed -- moved into the store on success, removed on
+  failure -- so the caller hands off ownership.
+
+  `opts[:label]` sets the human-readable `images.label` (defaults to the basename
+  of `path`).
+
+  Idempotent: creating identical bytes again is a no-op that returns the same id.
+  """
+  @spec create(Path.t(), keyword()) :: {:ok, id()} | {:error, term()}
+  @decorate with_span("Hyper.Img.create", include: [:path, :label])
+  def create(path, opts \\ []) do
+    label = Keyword.get(opts, :label, Path.basename(path))
+
+    with {:ok, %File.Stat{size: size}} <- File.stat(path),
+         {:ok, id} <- content_id(path),
+         {:ok, final, origin} <- publish(path, id),
+         :ok <- record_or_rollback(id, label, size, final, origin) do
+      {:ok, id}
+    else
+      {:error, _} = err ->
+        _ = File.rm(path)
+        err
+    end
+  end
+
+  # Record the image; if the DB write fails, roll back a file we just created (a
+  # reused file pre-existed and may back another image, so leave it).
+  @spec record_or_rollback(id(), String.t(), non_neg_integer(), Path.t(), :created | :reused) ::
+          :ok | {:error, term()}
+  defp record_or_rollback(id, label, size, final, origin) do
+    case record(id, label, size) do
+      :ok ->
+        :ok
+
+      {:error, _} = err ->
+        _ = if origin == :created, do: File.rm(final), else: :ok
+        err
+    end
+  end
+
+  # Streaming sha256 of `path`, lowercase hex -- the content address.
+  @spec content_id(Path.t()) :: {:ok, id()} | {:error, term()}
+  @decorate with_span("Hyper.Img.content_id", include: [:path])
+  defp content_id(path) do
+    {:ok, Hyper.Redist.Sha256.file(path)}
+  rescue
+    e -> {:error, {:hash_failed, Exception.message(e)}}
+  end
+
+  # Move `src` into the store at its content-addressed path. If the destination
+  # already exists (identical bytes already published), drop `src` and reuse it.
+  @spec publish(Path.t(), id()) :: {:ok, Path.t(), :created | :reused} | {:error, term()}
+  @decorate with_span("Hyper.Img.publish", include: [:id])
+  defp publish(src, id) do
+    File.mkdir_p!(Config.layer_dir())
+    final = final_path(id)
+
+    if File.exists?(final) do
+      Logger.info("image #{id} already present in store; reusing")
+      _ = File.rm(src)
+      {:ok, final, :reused}
+    else
+      case place(src, final) do
+        {:ok, ^final} -> {:ok, final, :created}
+        {:error, _} = err -> err
+      end
+    end
+  end
+
+  # An atomic rename when `src` is on the store's filesystem; a copy-then-drop
+  # across filesystems (rename can't cross a mount).
+  @spec place(Path.t(), Path.t()) :: {:ok, Path.t()} | {:error, term()}
+  defp place(src, final) do
+    case File.rename(src, final) do
+      :ok ->
+        {:ok, final}
+
+      {:error, :exdev} ->
+        case File.cp(src, final) do
+          :ok ->
+            _ = File.rm(src)
+            {:ok, final}
+
+          {:error, reason} ->
+            _ = File.rm(final)
+            {:error, {:publish_failed, reason}}
+        end
+
+      {:error, reason} ->
+        {:error, {:publish_failed, reason}}
+    end
+  end
+
+  @spec final_path(id()) :: Path.t()
+  defp final_path(id), do: Path.join(Config.layer_dir(), "layer_#{id}.img")
+
+  # Record the base image: one blob, one image (id == blob id), one layer at
+  # position 0. All upserts are idempotent so a re-publish of the same bytes is a
+  # no-op. The blob is inserted before the layer so the FK is satisfied.
+  @spec record(id(), String.t(), non_neg_integer()) :: :ok | {:error, term()}
+  defp record(id, label, size) do
+    multi =
+      Ecto.Multi.new()
+      |> Ecto.Multi.insert(
+        :blob,
+        Blob.changeset(%Blob{}, %{id: id, kind: :base, size: size}),
+        on_conflict: :nothing,
+        conflict_target: :id
+      )
+      |> Ecto.Multi.insert(
+        :image,
+        Image.changeset(%Image{}, %{id: id, label: label}),
+        on_conflict: :nothing,
+        conflict_target: :id
+      )
+      |> Ecto.Multi.insert(
+        :layer,
+        ImageLayer.changeset(%ImageLayer{}, %{image_id: id, position: 0, blob_id: id}),
+        on_conflict: :nothing,
+        conflict_target: [:image_id, :position]
+      )
+
+    case Repo.transaction(multi) do
+      {:ok, _} -> :ok
+      {:error, step, reason, _changes} -> {:error, {:record_failed, step, reason}}
+    end
+  end
+end