diff --git a/lib/bob/artifacts.ex b/lib/bob/artifacts.ex index e38f68b..1ae71b6 100644 --- a/lib/bob/artifacts.ex +++ b/lib/bob/artifacts.ex @@ -509,8 +509,8 @@ defmodule Bob.Artifacts do Enum.map(rows, fn [tag, archs] -> {tag, archs} end) end - # build_requests is small, so the reserved tag names are cheaper to collect - # and match against than to reconstruct from the component columns in SQL. + # build_requests is small, so collecting the reserved tag names and matching + # against the list is cheap. defp reserved_targets() do from(br in BuildRequest, where: br.state in ["pending", "completed"]) |> Repo.all() @@ -521,15 +521,19 @@ defmodule Bob.Artifacts do defp unreserved(query), do: where(query, [d], d.tag not in ^reserved_targets()) - defp stale_per_arch(cutoff) do - from(d in DockerTag, - where: d.repo in @docker_cleanup_per_arch_repos and d.built_at < ^cutoff - ) + def docker_cleanup_per_arch_repos(), do: @docker_cleanup_per_arch_repos + + # Only the known per-arch repos, so a caller cannot point the cleanup at a + # manifest repo. + defp stale_per_arch(cutoff, repos) do + repos = Enum.filter(@docker_cleanup_per_arch_repos, &(&1 in repos)) + + from(d in DockerTag, where: d.repo in ^repos and d.built_at < ^cutoff) end @doc "Per-repo count of the tags a run would delete." - def count_stale_per_arch_tags(cutoff) do - stale_per_arch(cutoff) + def count_stale_per_arch_tags(cutoff, repos \\ @docker_cleanup_per_arch_repos) do + stale_per_arch(cutoff, repos) |> unreserved() |> group_by([d], d.repo) |> select([d], {d.repo, count(d.id)}) @@ -538,8 +542,8 @@ defmodule Bob.Artifacts do end @doc "Up to `limit` deletable `{repo, tag}` pairs." - def stale_per_arch_tags(cutoff, limit) do - stale_per_arch(cutoff) + def stale_per_arch_tags(cutoff, limit, repos \\ @docker_cleanup_per_arch_repos) do + stale_per_arch(cutoff, repos) |> unreserved() |> limit(^limit) |> select([d], {d.repo, d.tag}) diff --git a/lib/bob/docker_cleanup.ex b/lib/bob/docker_cleanup.ex index 87903bf..ce3cc92 100644 --- a/lib/bob/docker_cleanup.ex +++ b/lib/bob/docker_cleanup.ex @@ -3,9 +3,13 @@ defmodule Bob.DockerCleanup do Deletes per-arch Docker Hub tags older than 30 days. Tags reserved by a build request are kept. The manifest repos are not pruned. - `run/1` deletes one batch and is what the nightly job calls; `drain/1` loops - until the backlog is empty. `:docker_cleanup_mode` gates whether `run/1` - deletes or only reports. + A live run keeps taking batches until one deletes nothing, so it clears the + whole backlog rather than a single batch. What bounds it is the job's timeout. + `:docker_cleanup_mode` gates whether it deletes or only reports. + + `:repos` narrows a run to some of the per-arch repos. Docker Hub rate limits + deletes per source IP, so pointing one node at each repo doubles throughput, + where two nodes on the same repo just race for the same tags. """ require Logger @@ -16,7 +20,7 @@ defmodule Bob.DockerCleanup do # tags the checker still expects. Asserted in docker_cleanup_test. @per_arch_max_age_days 30 - # Sized to finish inside Bob.Runner's three-hour job timeout. + # Candidate query page size. @default_batch 10_000 # Rows are committed per chunk so a killed run keeps its progress. @@ -29,42 +33,21 @@ defmodule Bob.DockerCleanup do @default_concurrency 25 def run(opts \\ []) do + repos = Keyword.get(opts, :repos, Artifacts.docker_cleanup_per_arch_repos()) + case Keyword.get(opts, :mode, configured_mode()) do - :dry_run -> dry_run() - :live -> live(opts) + :dry_run -> dry_run(repos) + :live -> live(opts, repos) end end - @doc """ - Runs batches until one deletes nothing. Always deletes, whatever - `:docker_cleanup_mode` says, and is not the scheduled path — the job runner - kills anything past three hours, so start this from a supervised task. - """ - def drain(opts \\ []) do - opts = Keyword.put_new(opts, :mode, :live) - batch = Keyword.get(opts, :limit, @default_batch) - - Stream.repeatedly(fn -> run(Keyword.put(opts, :limit, batch)) end) - |> Enum.reduce_while(0, fn {:live, deleted}, total -> - total = total + deleted - - if deleted == 0 do - Logger.info("DOCKER CLEANUP drain finished, deleted #{total} tag(s)") - {:halt, total} - else - Logger.info("DOCKER CLEANUP drain progress, deleted #{total} tag(s) so far") - {:cont, total} - end - end) - end - defp configured_mode(), do: Application.get_env(:bob, :docker_cleanup_mode, :dry_run) - defp dry_run() do - per_arch = Artifacts.count_stale_per_arch_tags(per_arch_cutoff()) + defp dry_run(repos) do + per_arch = Artifacts.count_stale_per_arch_tags(per_arch_cutoff(), repos) backlog = per_arch |> Map.values() |> Enum.sum() - # Report both: the backlog is every candidate, a run stops at the batch. + # The backlog is every candidate; a scheduled run stops at the batch. Logger.info( "DOCKER CLEANUP dry-run: per-arch built_at < #{@per_arch_max_age_days}d -> " <> "#{format_counts(per_arch)}; one scheduled run would delete " <> @@ -74,15 +57,22 @@ defmodule Bob.DockerCleanup do {:dry_run, %{per_arch: per_arch}} end - defp live(opts) do + defp live(opts, repos) do deleter = Keyword.get(opts, :deleter, &Bob.DockerHub.delete_tag/2) limit = Keyword.get(opts, :limit, @default_batch) cutoff = per_arch_cutoff() - candidates = Artifacts.stale_per_arch_tags(cutoff, limit) - - deleted = delete(candidates, deleter, cutoff) - Logger.info("DOCKER CLEANUP deleted #{deleted}/#{length(candidates)} tag(s)") + deleted = + Stream.repeatedly(fn -> + cutoff + |> Artifacts.stale_per_arch_tags(limit, repos) + |> delete(deleter, cutoff) + end) + |> Enum.reduce_while(0, fn batch, total -> + if batch == 0, do: {:halt, total}, else: {:cont, total + batch} + end) + + Logger.info("DOCKER CLEANUP deleted #{deleted} tag(s) from #{Enum.join(repos, ", ")}") {:live, deleted} end @@ -97,8 +87,7 @@ defmodule Bob.DockerCleanup do Artifacts.delete_docker_tags(confirmed) deleted = deleted + length(confirmed) - # Whole failed chunks, not individual errors: concurrent deletes have no - # "consecutive", and a flaky tag shouldn't trip the abort. + # Counted per chunk so a flaky tag does not trip the abort. dead_chunks = if failed > 0 and confirmed == [], do: dead_chunks + 1, else: 0 if dead_chunks >= @dead_chunk_ceiling do diff --git a/lib/bob/job.ex b/lib/bob/job.ex index a462b5e..6d7f1eb 100644 --- a/lib/bob/job.ex +++ b/lib/bob/job.ex @@ -1,5 +1,28 @@ defmodule Bob.Job do @type args :: [term()] + # A job still running after this is wedged, so the runner reaps it and the + # maintenance sweep requeues it. Jobs that legitimately run longer override + # timeout/0. + @default_timeout 3 * 60 * 60 * 1000 + @callback run(args()) :: term() + @callback timeout() :: pos_integer() + + @optional_callbacks timeout: 0 + + def default_timeout(), do: @default_timeout + + @doc "How long `key` may run before it is treated as wedged, in milliseconds." + def timeout({module, _key}), do: timeout(module) + + # ensure_loaded? first: function_exported?/3 answers false for a module that + # has not been loaded yet, which silently hands every job the default. + def timeout(module) do + if Code.ensure_loaded?(module) and function_exported?(module, :timeout, 0) do + module.timeout() + else + @default_timeout + end + end end diff --git a/lib/bob/job/docker_cleanup.ex b/lib/bob/job/docker_cleanup.ex index dfd128c..5155245 100644 --- a/lib/bob/job/docker_cleanup.ex +++ b/lib/bob/job/docker_cleanup.ex @@ -3,6 +3,10 @@ defmodule Bob.Job.DockerCleanup do Bob.DockerCleanup.run() end + # A live run clears the whole backlog, which takes days at Docker Hub's delete + # rate. The next night's run is dedup'd away while this one is still going. + def timeout(), do: 23 * 60 * 60 * 1000 + def priority(), do: 1 def weight(), do: 1 def concurrency(), do: :shared diff --git a/lib/bob/queue/maintenance.ex b/lib/bob/queue/maintenance.ex index 32522b9..0e455fb 100644 --- a/lib/bob/queue/maintenance.ex +++ b/lib/bob/queue/maintenance.ex @@ -15,7 +15,6 @@ defmodule Bob.Queue.Maintenance do alias Bob.Queue.{Job, Failure} @interval_seconds 60 - @job_timeout_seconds 3 * 60 * 60 # A job stuck in running this long means its node died hard (OOM, node loss), # so the work itself is not suspect — requeue it. The cap stops a job that # reliably kills its node or genuinely exceeds the timeout from looping. @@ -67,12 +66,23 @@ defmodule Bob.Queue.Maintenance do defp sweep_stale_running() do now = DateTime.utc_now() - cutoff = DateTime.add(now, -@job_timeout_seconds, :second) - stale = + # Pre-filter on the shortest timeout any job can have, then keep only those + # past their own. + cutoff = DateTime.add(now, -div(Bob.Job.default_timeout(), 1000), :second) + + stale_ids = from(j in Job, - where: j.state == "running" and not is_nil(j.started_at) and j.started_at < ^cutoff + where: j.state == "running" and not is_nil(j.started_at) and j.started_at < ^cutoff, + select: {j.id, j.module_key, j.started_at} ) + |> Repo.all() + |> Enum.filter(fn {_id, module_key, started_at} -> + DateTime.diff(now, started_at, :millisecond) > Bob.Job.timeout(module_key) + end) + |> Enum.map(fn {id, _module_key, _started_at} -> id end) + + stale = from(j in Job, where: j.id in ^stale_ids) {requeued, _} = Repo.update_all( diff --git a/lib/bob/runner.ex b/lib/bob/runner.ex index 038e006..dd5d3b3 100644 --- a/lib/bob/runner.ex +++ b/lib/bob/runner.ex @@ -9,10 +9,8 @@ defmodule Bob.Runner do # A build that hangs (e.g. a wedged `docker build`) blocks its task forever and # keeps counting its weight against the shared budget, so the agent stops - # pulling new builds. Reap a task that outlives the master's stale-job timeout - # (mirrors Bob.Queue.Maintenance @job_timeout_seconds) so the slot is freed and - # the job is reported failed. - @job_timeout 3 * 60 * 60 * 1000 + # pulling new builds. Reap a task that outlives its job's timeout so the slot + # is freed and the job is reported failed. def start_link([]) do GenServer.start_link(__MODULE__, new_state(), name: __MODULE__) @@ -92,7 +90,7 @@ defmodule Bob.Runner do end end - # A task still running past @job_timeout is wedged (e.g. a hung `docker build`). + # A task still running past its timeout is wedged (e.g. a hung `docker build`). # Kill it so it stops leaking its weight, fail the job, and pull fresh work. # The kill triggers an abnormal :DOWN, but the row is already gone so that # clause no-ops. @@ -202,7 +200,7 @@ defmodule Bob.Runner do defp start_job(id, key, args, state) do Logger.info("STARTING #{inspect(key)} #{inspect(args)}") task = Task.Supervisor.async(Bob.Tasks, fn -> run_task(key, args) end) - timer = Process.send_after(self(), {:job_timeout, task.ref}, @job_timeout) + timer = Process.send_after(self(), {:job_timeout, task.ref}, Bob.Job.timeout(key)) put_in(state.tasks[task.ref], {key, args, id, task.pid, timer}) end diff --git a/test/bob/docker_cleanup_test.exs b/test/bob/docker_cleanup_test.exs index 1c0f290..b038ba1 100644 --- a/test/bob/docker_cleanup_test.exs +++ b/test/bob/docker_cleanup_test.exs @@ -131,20 +131,20 @@ defmodule Bob.DockerCleanupTest do [{"1.26.0-erlang-27.0-ubuntu-noble-20250101", ["amd64"]}] end - test "respects the batch limit so a run deletes a bounded slice" do + test "the batch limit pages the candidate query without capping the run" do deleter = fn _repo, _tag -> :ok end for n <- 1..3 do Artifacts.add_docker_tag( "hexpm/elixir-amd64", - "2#{n}.0-ubuntu-noble-20250101", + "1.2#{n}.0-erlang-27.0-ubuntu-noble-20250101", ["amd64"], old() ) end - assert {:live, 2} = DockerCleanup.run(mode: :live, deleter: deleter, limit: 2) - assert length(Artifacts.docker_tags("hexpm/elixir-amd64")) == 1 + assert {:live, 3} = DockerCleanup.run(mode: :live, deleter: deleter, limit: 2) + assert Artifacts.docker_tags("hexpm/elixir-amd64") == [] end test "an ancient manifest tag is never a candidate, whatever the limit" do @@ -216,8 +216,63 @@ defmodule Bob.DockerCleanupTest do end end - describe "drain/1" do - test "keeps going past the batch limit until nothing is left" do + describe "run/1 scoped to some repos" do + setup do + for repo <- ~w(hexpm/elixir-amd64 hexpm/elixir-arm64) do + Artifacts.add_docker_tag( + repo, + "1.18.0-erlang-27.0-ubuntu-noble-20250101", + ["amd64"], + old() + ) + end + + :ok + end + + test "deletes only from the named repo" do + test = self() + deleter = fn repo, _tag -> send(test, {:deleted, repo}) && :ok end + + assert {:live, 1} = + DockerCleanup.run( + mode: :live, + deleter: deleter, + repos: ["hexpm/elixir-amd64"] + ) + + assert_received {:deleted, "hexpm/elixir-amd64"} + refute_received {:deleted, "hexpm/elixir-arm64"} + assert Artifacts.docker_tags("hexpm/elixir-arm64") != [] + end + + test "the dry run counts only the named repo" do + assert {:dry_run, %{per_arch: counts}} = + DockerCleanup.run(mode: :dry_run, repos: ["hexpm/elixir-arm64"]) + + assert counts == %{"hexpm/elixir-arm64" => 1} + end + + # A repo outside the list would be the images users pull. + test "a repo outside the per-arch list is ignored, not deleted from" do + deleter = fn _repo, _tag -> :ok end + + Artifacts.add_docker_tag( + "hexpm/elixir", + "1.17.0-erlang-27.0-ubuntu-noble-20250101", + ["amd64", "arm64"], + ancient() + ) + + assert {:live, 0} = + DockerCleanup.run(mode: :live, deleter: deleter, repos: ["hexpm/elixir"]) + + assert Artifacts.docker_tags("hexpm/elixir") != [] + end + end + + describe "run/1 clears the whole backlog" do + test "keeps taking batches until nothing is left" do deleter = fn _repo, _tag -> :ok end for n <- 1..7 do @@ -229,8 +284,7 @@ defmodule Bob.DockerCleanupTest do ) end - # A single run of this batch size would stop at 2; the drain loops. - assert DockerCleanup.drain(limit: 2, deleter: deleter) == 7 + assert {:live, 7} = DockerCleanup.run(mode: :live, limit: 2, deleter: deleter) assert Artifacts.docker_tags("hexpm/elixir-amd64") == [] end @@ -244,7 +298,7 @@ defmodule Bob.DockerCleanupTest do old() ) - assert DockerCleanup.drain(deleter: deleter) == 0 + assert {:live, 0} = DockerCleanup.run(mode: :live, deleter: deleter) assert length(Artifacts.docker_tags("hexpm/elixir-amd64")) == 1 end @@ -275,7 +329,7 @@ defmodule Bob.DockerCleanupTest do builds_count: 0 }) - assert DockerCleanup.drain(limit: 2, deleter: deleter) == 0 + assert {:live, 0} = DockerCleanup.run(mode: :live, limit: 2, deleter: deleter) assert length(Artifacts.docker_tags("hexpm/elixir-amd64")) == 2 end end diff --git a/test/bob/queue/maintenance_test.exs b/test/bob/queue/maintenance_test.exs index 5815546..7ed89f3 100644 --- a/test/bob/queue/maintenance_test.exs +++ b/test/bob/queue/maintenance_test.exs @@ -74,6 +74,37 @@ defmodule Bob.Queue.MaintenanceTest do assert [%Job{state: "running"}] = Repo.all(Job) end + test "leaves a job running past the default timeout but inside its own" do + old = ago(4 * @hour) + + insert_job( + module_key: Bob.Job.DockerCleanup, + state: "running", + inserted_at: old, + started_at: old + ) + + Maintenance.run() + + # Requeuing a cleanup at three hours starts a second one alongside it. + assert [%Job{state: "running"}] = Repo.all(Job) + end + + test "requeues a long-timeout job once past its own timeout" do + old = ago(24 * @hour) + + insert_job( + module_key: Bob.Job.DockerCleanup, + state: "running", + inserted_at: old, + started_at: old + ) + + Maintenance.run() + + assert [%Job{state: "queued", requeues: 1}] = Repo.all(Job) + end + test "prunes failures older than the expiry window" do insert_failure(last_failed_at: ago(8 * @day))