Optimize docker checker

config/config.exs

@@ -31,7 +31,8 @@ config :bob,
     ],
     [
       module: Bob.Job.DockerChecker,
-      period: {15, :min}
+      period: {15, :min},
+      queue: true
     ]
   ],
   agent_schedule: [

lib/bob/application.ex

@@ -2,8 +2,6 @@ defmodule Bob.Application do
   use Application
 
   def start(_type, _args) do
-    opts = [port: port(), compress: true]
-
     :logger.add_handler(:sentry_handler, Sentry.LoggerHandler, %{})
 
     setup_docker()
@@ -14,16 +12,14 @@ defmodule Bob.Application do
 
     File.mkdir_p!(Bob.tmp_dir())
 
-    # TODO: Do not start webserver if we are an agent
-    Plug.Cowboy.http(Bob.Router, [], opts)
-
     children = [
       {Task.Supervisor, [name: Bob.Tasks]},
       Bob.DockerHub.Auth,
       Bob.DockerHub.Cache,
       Bob.Queue,
       runner_spec(),
-      {Bob.Schedule, [schedule()]}
+      {Bob.Schedule, [schedule()]},
+      {Plug.Cowboy, scheme: :http, plug: Bob.Router, options: [port: port(), compress: true]}
     ]
 
     opts = [strategy: :one_for_one, name: Bob.Supervisor]

lib/bob/docker_hub.ex

@@ -9,23 +9,28 @@ defmodule Bob.DockerHub do
 
     {:ok, 200, _headers, body} =
       Bob.HTTP.retry("DockerHub #{url}", fn ->
-        :hackney.request(:post, url, headers, Jason.encode!(body), opts)
+        :hackney.request(:post, url, headers, JSON.encode!(body), opts)
       end)
 
-    result = Jason.decode!(body)
+    result = JSON.decode!(body)
     Application.put_env(:bob, :dockerhub_token, result["token"])
   end
 
   def fetch_repo_tags(repo) do
-    (@dockerhub_url <> "v2/repositories/#{repo}/tags?page=${page}&page_size=100")
-    |> dockerhub_request()
+    url = @dockerhub_url <> "v2/repositories/#{repo}/tags?page=${page}&page_size=100"
+    {:ok, server} = Bob.DockerHub.Pager.start_link(url)
+    Bob.DockerHub.Pager.wait(server)
   end
 
   def fetch_repo_tags_from_cache(repo) do
-    Bob.DockerHub.Cache.lookup(repo, fn ->
-      (@dockerhub_url <> "v2/repositories/#{repo}/tags?page=${page}&page_size=100")
-      |> dockerhub_request()
-    end)
+    :ok =
+      Bob.DockerHub.Cache.lookup(repo, fn on_result ->
+        url = @dockerhub_url <> "v2/repositories/#{repo}/tags?page=${page}&page_size=100"
+        {:ok, server} = Bob.DockerHub.Pager.start_link(url, on_result)
+        Bob.DockerHub.Pager.wait(server)
+      end)
+
+    Bob.DockerHub.Cache.stream(repo)
   end
 
   def fetch_tag(repo, tag) do
@@ -40,9 +45,7 @@ defmodule Bob.DockerHub do
 
     case result do
       {:ok, 200, _headers, body} ->
-        body
-        |> Jason.decode!()
-        |> parse()
+        parse(JSON.decode!(body))
 
       {:ok, 404, _headers, _body} ->
         nil
@@ -82,12 +85,7 @@ defmodule Bob.DockerHub do
     else
       # DockerHub returns dupes sometimes?
       archs = result["images"] |> Enum.map(&:binary.copy(&1["architecture"])) |> Enum.uniq()
-      {result["name"], archs}
+      {:binary.copy(result["name"]), archs}
     end
   end
-
-  defp dockerhub_request(url) do
-    {:ok, server} = Bob.DockerHub.Pager.start_link(url)
-    Bob.DockerHub.Pager.wait(server)
-  end
 end

lib/bob/docker_hub/cache.ex

@@ -94,16 +94,15 @@ defmodule Bob.DockerHub.Cache do
         try do
           case GenServer.call(__MODULE__, {:lock, repo}, @timeout + @timeout_grace_time) do
             :aquired ->
-              result = fun.()
-
-              :ets.insert(
-                __MODULE__,
-                Enum.map(result, fn {tag, archs} -> {{:data, repo, tag}, archs} end)
-              )
+              on_result = fn results ->
+                Enum.each(results, fn {tag, archs} ->
+                  :ets.insert(__MODULE__, {{:data, repo, tag}, archs})
+                end)
+              end
 
+              :ok = fun.(on_result)
               :ets.insert(__MODULE__, {{:status, repo}, true})
-
-              result
+              :ok
 
             :done ->
               lookup(repo, fun)
@@ -113,8 +112,21 @@ defmodule Bob.DockerHub.Cache do
         end
 
       [{_, true}] ->
-        :ets.select(__MODULE__, [{{{:data, repo, :"$1"}, :"$2"}, [], [:"$$"]}])
-        |> Enum.map(&List.to_tuple/1)
+        :ok
     end
   end
+
+  def stream(repo) do
+    Stream.resource(
+      fn -> :ets.match(__MODULE__, {{:data, repo, :"$1"}, :"$2"}, 1000) end,
+      fn
+        :"$end_of_table" ->
+          {:halt, nil}
+
+        {matches, cont} ->
+          {Enum.map(matches, fn [tag, archs] -> {tag, archs} end), :ets.match(cont)}
+      end,
+      fn _ -> :ok end
+    )
+  end
 end

lib/bob/docker_hub/pager.ex

@@ -5,36 +5,58 @@ defmodule Bob.DockerHub.Pager do
   @timeout 60 * 60 * 1000
 
   def start_link(url) do
-    GenServer.start_link(__MODULE__, url)
+    GenServer.start_link(__MODULE__, {url, nil})
+  end
+
+  def start_link(url, on_result) do
+    GenServer.start_link(__MODULE__, {url, on_result})
   end
 
   def wait(server) do
     GenServer.call(server, :wait, @timeout)
   end
 
-  def init(url) do
-    {:ok, next_request(%{url: url, page: 1, tasks: MapSet.new(), results: [], reply: nil})}
+  def init({url, on_result}) do
+    {:ok,
+     next_request(%{
+       url: url,
+       on_result: on_result,
+       page: 1,
+       tasks: MapSet.new(),
+       results: [],
+       reply: nil
+     })}
   end
 
   def handle_call(:wait, from, state) do
     if MapSet.size(state.tasks) == 0 do
-      {:stop, :normal, Enum.concat(state.results), state}
+      result = if state.on_result, do: :ok, else: Enum.concat(state.results)
+      {:stop, :normal, result, state}
     else
       state = %{state | reply: from}
       {:noreply, state}
     end
   end
 
   def handle_info({ref, {:ok, result}}, state) do
-    state = %{state | tasks: MapSet.delete(state.tasks, ref), results: [result | state.results]}
+    state =
+      if state.on_result do
+        state.on_result.(result)
+        state
+      else
+        %{state | results: [result | state.results]}
+      end
+
+    state = %{state | tasks: MapSet.delete(state.tasks, ref)}
     {:noreply, next_request(state)}
   end
 
   def handle_info({ref, :done}, state) do
     state = %{state | tasks: MapSet.delete(state.tasks, ref)}
 
     if MapSet.size(state.tasks) == 0 do
-      GenServer.reply(state.reply, Enum.concat(state.results))
+      result = if state.on_result, do: :ok, else: Enum.concat(state.results)
+      GenServer.reply(state.reply, result)
       {:stop, :normal, state}
     else
       {:noreply, state}
@@ -60,8 +82,8 @@ defmodule Bob.DockerHub.Pager do
 
           case result do
             {:ok, 200, _headers, body} ->
-              body = Jason.decode!(body)
-              {:ok, Enum.flat_map(body["results"], &List.wrap(Bob.DockerHub.parse(&1)))}
+              decoded = JSON.decode!(body)
+              {:ok, Enum.flat_map(decoded["results"], &List.wrap(Bob.DockerHub.parse(&1)))}
 
             {:ok, 404, _headers, _body} ->
               :done

lib/bob/github.ex

@@ -27,7 +27,9 @@ defmodule Bob.GitHub do
   end
 
   defp response_to_refs(response) do
-    Enum.map(response, &{&1["name"], &1["commit"]["sha"]})
+    Enum.map(response, fn item ->
+      {:binary.copy(item["name"]), :binary.copy(item["commit"]["sha"])}
+    end)
   end
 
   defp github_request(url) do
@@ -39,7 +41,7 @@ defmodule Bob.GitHub do
     {:ok, 200, headers, body} =
       Bob.HTTP.retry("GitHub #{url}", fn -> :hackney.request(:get, url, [], "", opts) end)
 
-    body = Jason.decode!(body)
+    body = JSON.decode!(body)
 
     if url = next_link(headers) do
       body ++ github_request(url)

lib/bob/job/docker_checker.ex

@@ -278,7 +278,7 @@ defmodule Bob.Job.DockerChecker do
     Stream.flat_map(erlang_tags(), fn {erlang, os, os_version, erlang_arch} ->
       if not skip_elixir_for_erlang?(erlang) and os_version in builds[os] do
         Stream.flat_map(refs, fn {"v" <> elixir, otp_major} ->
-          if not skip_elixir?(elixir) and compatible_elixir_and_erlang?(otp_major, erlang) do
+          if compatible_elixir_and_erlang?(otp_major, erlang) do
             [{elixir, erlang, os, os_version, erlang_arch}]
           else
             []
@@ -291,16 +291,14 @@ defmodule Bob.Job.DockerChecker do
   end
 
   def elixir_builds() do
-    all_builds =
-      "builds/elixir"
-      |> Bob.Repo.fetch_built_refs()
-      |> Stream.map(fn {build_name, _ref} -> build_name end)
-      |> Stream.map(&split_elixir_build/1)
-      |> Stream.filter(&build_elixir_ref?/1)
-      |> Enum.sort(&cmp_elixir_tags/2)
-
-    all_builds
-    |> Stream.reject(fn {_elixir, otp} -> otp == nil end)
+    "builds/elixir"
+    |> Bob.Repo.fetch_built_refs()
+    |> Stream.map(fn {build_name, _ref} -> build_name end)
+    |> Stream.map(&split_elixir_build/1)
+    |> Stream.filter(&build_elixir_ref?/1)
+    |> Enum.sort(&cmp_elixir_tags/2)
+    |> Enum.reject(fn {_elixir, otp} -> otp == nil end)
+    |> Enum.reject(fn {"v" <> elixir, _otp} -> skip_elixir?(elixir) end)
   end
 
   def elixir_tags() do
@@ -310,7 +308,7 @@ defmodule Bob.Job.DockerChecker do
   def elixir_tags(arch) do
     "hexpm/elixir-#{arch}"
     |> Bob.DockerHub.fetch_repo_tags_from_cache()
-    |> Enum.map(fn {tag, [^arch]} ->
+    |> Stream.map(fn {tag, [^arch]} ->
       [elixir, erlang, os, os_version] =
         Regex.run(@elixir_tag_regex, tag, capture: :all_but_first)
 

lib/bob/router.ex

@@ -16,7 +16,7 @@ defmodule Bob.Router do
   plug(Plug.Parsers,
     pass: ["application/json", "application/vnd.bob+erlang"],
     parsers: [:json, Bob.Plug.Parser],
-    json_decoder: Jason
+    json_decoder: JSON
   )
 
   plug(Sentry.PlugContext)

mix.exs

@@ -23,7 +23,6 @@ defmodule Bob.Mixfile do
     [
       {:ex_aws_s3, "~> 2.0"},
       {:hackney, "~> 1.11"},
-      {:jason, "~> 1.1"},
       {:plug_cowboy, "~> 2.0"},
       {:porcelain, "~> 2.0"},
       {:sentry, "~> 10.2"},