|
| 1 | +defmodule Hyper.Img do |
| 2 | + @moduledoc """ |
| 3 | + A content-addressed image: an ordered stack of layers, and the entry point for |
| 4 | + putting one into the cluster. |
| 5 | +
|
| 6 | + `create/2` ingests a prepared image file -- e.g. the ext4 rootfs produced by |
| 7 | + `Hyper.Img.OciLoader` -- into the shared media store and the image database. It |
| 8 | + content-addresses the file (sha256 of its bytes = the image id), publishes it |
| 9 | + into `Hyper.Config.layer_dir/0` at `layer_<id>.img`, then records it as a |
| 10 | + one-layer base image (`blobs` + `images` + `image_layers`). Producers of image |
| 11 | + files stay decoupled from the store and DB: they hand a path to `create/2`. |
| 12 | + """ |
| 13 | + |
| 14 | + use OpenTelemetryDecorator |
| 15 | + |
| 16 | + require Logger |
| 17 | + |
| 18 | + alias Hyper.Config |
| 19 | + alias Hyper.Img.Db.{Blob, Image, ImageLayer, Repo} |
| 20 | + |
| 21 | + @type id :: String.t() |
| 22 | + |
| 23 | + # `Ecto.Multi` is an opaque struct; building it through the pipe trips |
| 24 | + # dialyzer's opacity check (a known Ecto false positive), so silence it for the |
| 25 | + # one function that assembles a Multi. |
| 26 | + @dialyzer {:no_opaque, record: 3} |
| 27 | + |
| 28 | + @doc """ |
| 29 | + Ingest the image file at `path` into the cluster and return its |
| 30 | + content-addressed id. |
| 31 | +
|
| 32 | + Content-addresses `path` (sha256 of its bytes = the id), publishes it into the |
| 33 | + media store at `layer_<id>.img`, and records it as a one-layer base image. The |
| 34 | + file at `path` is consumed -- moved into the store on success, removed on |
| 35 | + failure -- so the caller hands off ownership. |
| 36 | +
|
| 37 | + `opts[:label]` sets the human-readable `images.label` (defaults to the basename |
| 38 | + of `path`). |
| 39 | +
|
| 40 | + Idempotent: creating identical bytes again is a no-op that returns the same id. |
| 41 | + """ |
| 42 | + @spec create(Path.t(), keyword()) :: {:ok, id()} | {:error, term()} |
| 43 | + @decorate with_span("Hyper.Img.create", include: [:path, :label]) |
| 44 | + def create(path, opts \\ []) do |
| 45 | + label = Keyword.get(opts, :label, Path.basename(path)) |
| 46 | + |
| 47 | + with {:ok, %File.Stat{size: size}} <- File.stat(path), |
| 48 | + {:ok, id} <- content_id(path), |
| 49 | + {:ok, final, origin} <- publish(path, id), |
| 50 | + :ok <- record_or_rollback(id, label, size, final, origin) do |
| 51 | + {:ok, id} |
| 52 | + else |
| 53 | + {:error, _} = err -> |
| 54 | + _ = File.rm(path) |
| 55 | + err |
| 56 | + end |
| 57 | + end |
| 58 | + |
| 59 | + # Record the image; if the DB write fails, roll back a file we just created (a |
| 60 | + # reused file pre-existed and may back another image, so leave it). |
| 61 | + @spec record_or_rollback(id(), String.t(), non_neg_integer(), Path.t(), :created | :reused) :: |
| 62 | + :ok | {:error, term()} |
| 63 | + defp record_or_rollback(id, label, size, final, origin) do |
| 64 | + case record(id, label, size) do |
| 65 | + :ok -> |
| 66 | + :ok |
| 67 | + |
| 68 | + {:error, _} = err -> |
| 69 | + _ = if origin == :created, do: File.rm(final), else: :ok |
| 70 | + err |
| 71 | + end |
| 72 | + end |
| 73 | + |
| 74 | + # Streaming sha256 of `path`, lowercase hex -- the content address. |
| 75 | + @spec content_id(Path.t()) :: {:ok, id()} | {:error, term()} |
| 76 | + @decorate with_span("Hyper.Img.content_id", include: [:path]) |
| 77 | + defp content_id(path) do |
| 78 | + {:ok, Hyper.Redist.Sha256.file(path)} |
| 79 | + rescue |
| 80 | + e -> {:error, {:hash_failed, Exception.message(e)}} |
| 81 | + end |
| 82 | + |
| 83 | + # Move `src` into the store at its content-addressed path. If the destination |
| 84 | + # already exists (identical bytes already published), drop `src` and reuse it. |
| 85 | + @spec publish(Path.t(), id()) :: {:ok, Path.t(), :created | :reused} | {:error, term()} |
| 86 | + @decorate with_span("Hyper.Img.publish", include: [:id]) |
| 87 | + defp publish(src, id) do |
| 88 | + File.mkdir_p!(Config.layer_dir()) |
| 89 | + final = final_path(id) |
| 90 | + |
| 91 | + if File.exists?(final) do |
| 92 | + Logger.info("image #{id} already present in store; reusing") |
| 93 | + _ = File.rm(src) |
| 94 | + {:ok, final, :reused} |
| 95 | + else |
| 96 | + case place(src, final) do |
| 97 | + {:ok, ^final} -> {:ok, final, :created} |
| 98 | + {:error, _} = err -> err |
| 99 | + end |
| 100 | + end |
| 101 | + end |
| 102 | + |
| 103 | + # An atomic rename when `src` is on the store's filesystem; a copy-then-drop |
| 104 | + # across filesystems (rename can't cross a mount). |
| 105 | + @spec place(Path.t(), Path.t()) :: {:ok, Path.t()} | {:error, term()} |
| 106 | + defp place(src, final) do |
| 107 | + case File.rename(src, final) do |
| 108 | + :ok -> |
| 109 | + {:ok, final} |
| 110 | + |
| 111 | + {:error, :exdev} -> |
| 112 | + case File.cp(src, final) do |
| 113 | + :ok -> |
| 114 | + _ = File.rm(src) |
| 115 | + {:ok, final} |
| 116 | + |
| 117 | + {:error, reason} -> |
| 118 | + _ = File.rm(final) |
| 119 | + {:error, {:publish_failed, reason}} |
| 120 | + end |
| 121 | + |
| 122 | + {:error, reason} -> |
| 123 | + {:error, {:publish_failed, reason}} |
| 124 | + end |
| 125 | + end |
| 126 | + |
| 127 | + @spec final_path(id()) :: Path.t() |
| 128 | + defp final_path(id), do: Path.join(Config.layer_dir(), "layer_#{id}.img") |
| 129 | + |
| 130 | + # Record the base image: one blob, one image (id == blob id), one layer at |
| 131 | + # position 0. All upserts are idempotent so a re-publish of the same bytes is a |
| 132 | + # no-op. The blob is inserted before the layer so the FK is satisfied. |
| 133 | + @spec record(id(), String.t(), non_neg_integer()) :: :ok | {:error, term()} |
| 134 | + defp record(id, label, size) do |
| 135 | + multi = |
| 136 | + Ecto.Multi.new() |
| 137 | + |> Ecto.Multi.insert( |
| 138 | + :blob, |
| 139 | + Blob.changeset(%Blob{}, %{id: id, kind: :base, size: size}), |
| 140 | + on_conflict: :nothing, |
| 141 | + conflict_target: :id |
| 142 | + ) |
| 143 | + |> Ecto.Multi.insert( |
| 144 | + :image, |
| 145 | + Image.changeset(%Image{}, %{id: id, label: label}), |
| 146 | + on_conflict: :nothing, |
| 147 | + conflict_target: :id |
| 148 | + ) |
| 149 | + |> Ecto.Multi.insert( |
| 150 | + :layer, |
| 151 | + ImageLayer.changeset(%ImageLayer{}, %{image_id: id, position: 0, blob_id: id}), |
| 152 | + on_conflict: :nothing, |
| 153 | + conflict_target: [:image_id, :position] |
| 154 | + ) |
| 155 | + |
| 156 | + case Repo.transaction(multi) do |
| 157 | + {:ok, _} -> :ok |
| 158 | + {:error, step, reason, _changes} -> {:error, {:record_failed, step, reason}} |
| 159 | + end |
| 160 | + end |
| 161 | +end |
0 commit comments