feat(grpc): LoadImage RPC -- load OCI images via gRPC

Author: markovejnovic

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),

proto/hyper/grpc/v0/hyper.proto

@@ -52,6 +52,20 @@ service Hyper {
 
   // List every microVM currently known to the cluster, across all nodes.
   rpc ListVms(google.protobuf.Empty) returns (ListVmsResponse);
+
+  // Load an OCI image into the cluster's shared media store and image database.
+  //
+  // Pulls the referenced image for this node's architecture, flattens it, builds
+  // an ext4 rootfs, content-addresses it, and records a base image. Runs on the
+  // node that receives the call; the result is visible cluster-wide. Blocks until
+  // the load completes -- this can take minutes -- so set a generous deadline.
+  // Returns the content-addressed `img_id` to pass to CreateVm.
+  //
+  // Errors:
+  //   INVALID_ARGUMENT    -- `image_ref` is empty or malformed.
+  //   FAILED_PRECONDITION -- the node lacks the required tools (skopeo/umoci/mke2fs).
+  //   INTERNAL            -- the pull, unpack, build, publish, or DB record failed.
+  rpc LoadImage(LoadImageRequest) returns (LoadImageResponse);
 }
 
 // A fixed (vCPU, memory, disk) size, like a cloud instance class. Each step up
@@ -142,3 +156,18 @@ message Vm {
   // The cluster node (Erlang node name) the VM runs on.
   string node = 2;
 }
+
+// Request to load an OCI image.
+message LoadImageRequest {
+  // Required. An OCI image reference, e.g. "docker.io/library/alpine:3.19".
+  string image_ref = 1;
+
+  // Optional. A human-readable label stored with the image (defaults to image_ref).
+  optional string label = 2;
+}
+
+// Result of a successful LoadImage.
+message LoadImageResponse {
+  // The content-addressed image id; pass it to CreateVm.
+  string img_id = 1;
+}

test/hyper/grpc/codec_test.exs

@@ -0,0 +1,50 @@
+defmodule Hyper.Grpc.CodecTest do
+  use ExUnit.Case, async: true
+
+  alias Hyper.Grpc.Codec
+  alias Hyper.Grpc.V0.{LoadImageRequest, LoadImageResponse}
+
+  describe "from_grpc/1 (LoadImageRequest)" do
+    test "rejects a blank image_ref" do
+      assert Codec.from_grpc(%LoadImageRequest{image_ref: ""}) == {:error, :missing_image_ref}
+      assert Codec.from_grpc(%LoadImageRequest{image_ref: nil}) == {:error, :missing_image_ref}
+    end
+
+    test "passes the ref through with no opts when label is unset" do
+      assert Codec.from_grpc(%LoadImageRequest{image_ref: "alpine:3.19"}) ==
+               {:ok, {"alpine:3.19", []}}
+
+      assert Codec.from_grpc(%LoadImageRequest{image_ref: "alpine:3.19", label: ""}) ==
+               {:ok, {"alpine:3.19", []}}
+    end
+
+    test "carries label as an opt when set" do
+      assert Codec.from_grpc(%LoadImageRequest{image_ref: "alpine:3.19", label: "base"}) ==
+               {:ok, {"alpine:3.19", [label: "base"]}}
+    end
+  end
+
+  describe "to_grpc/1 (loaded)" do
+    test "wraps the img_id in a LoadImageResponse" do
+      assert Codec.to_grpc({:loaded, "oci-abc"}) == %LoadImageResponse{img_id: "oci-abc"}
+    end
+  end
+
+  describe "to_grpc/1 (LoadImage error mapping)" do
+    test "missing/invalid ref -> INVALID_ARGUMENT" do
+      assert %GRPC.RPCError{status: status} = Codec.to_grpc({:error, :missing_image_ref})
+      assert status == GRPC.Status.invalid_argument()
+      assert Codec.to_grpc({:error, :invalid_ref}).status == GRPC.Status.invalid_argument()
+    end
+
+    test "missing tools -> FAILED_PRECONDITION" do
+      err = Codec.to_grpc({:error, {:missing_tools, ["skopeo", "umoci"]}})
+      assert err.status == GRPC.Status.failed_precondition()
+    end
+
+    test "tool/build/db failures -> INTERNAL" do
+      assert Codec.to_grpc({:error, {:skopeo_failed, 1, "boom"}}).status == GRPC.Status.internal()
+      assert Codec.to_grpc({:error, {:record_failed, :blob, :db}}).status == GRPC.Status.internal()
+    end
+  end
+end