Replace Statix with Prometheus

Prometheus pull method of metrics gathering suits better such
application as it doesn't require gatherer to be configure always. It
also provides metrics in much more "standardized" way as
prometheus_phoenix and prometheus_plugs are quite norm when coming to
the respective library metrics gathering.

There are also predefined Grafana's dashboards that can be found there:

https://github.com/deadtrickster/beam-dashboards

config/config.exs

@@ -9,7 +9,7 @@ use Mix.Config
 config :imager, ImagerWeb.Endpoint,
   url: [host: "localhost"],
   render_errors: [view: ImagerWeb.Views.Error, accepts: ~w(json)],
-  instrumenters: [ImagerWeb.Instrumenter]
+  instrumenters: [Prometheus.PhoenixInstrumenter]
 
 # Configures Elixir's Logger
 config :logger, :console,
@@ -21,10 +21,6 @@ config :phoenix, :format_encoders, json: Jason
 config :ex_aws,
   json_codec: Jason
 
-config :vmstats,
-  sink: Imager.Stats,
-  base_key: "erlang"
-
 config :sentry,
   included_environments: [:prod],
   release: Mix.Project.config()[:version],

lib/imager.ex

@@ -11,7 +11,7 @@ defmodule Imager do
 
   require Logger
 
-  alias Imager.Stats
+  alias Imager.Instrumenter
   alias Imager.Store
   alias Imager.Tool
 
@@ -26,34 +26,31 @@ defmodule Imager do
   def process(store, file_name, commands, opts \\ [])
 
   def process(%{store: store}, file_name, [], opts) do
-    Stats.increment("imager.passthrough", 1, tags: Stats.tags())
+    Instrumenter.Cache.passthrough(store)
 
     Store.retrieve(store, file_name, opts)
   end
 
   def process(%{store: store, cache: cache}, file_name, commands, opts) do
     {mime, result_name} = Tool.result(file_name, commands)
     args = Tool.build_command(commands)
-    tags = Stats.tags(~w(
-      store:#{elem(store, 0)}
-      cache:#{elem(cache, 0)}
-    ))
+
+    Instrumenter.Processing.command(commands)
 
     Logger.metadata(input: file_name, commands: inspect(commands))
     Logger.debug(inspect(args))
 
     with :error <- Store.retrieve(cache, result_name, opts),
          {:ok, {_, _, in_stream}} <- Store.retrieve(store, file_name, opts) do
       Logger.debug("Start processing")
-      Stats.increment("imager.process.started", 1, tags: tags)
 
       {pid, out_stream} = runner().stream(executable(), args)
 
       runner().feed_stream(pid, in_stream)
 
       case runner().wait(pid) do
         :ok ->
-          Stats.increment("imager.process.succeeded", 1, tags: tags)
+          Instrumenter.Processing.succeeded(store)
 
           stream =
             out_stream
@@ -62,14 +59,14 @@ defmodule Imager do
           {:ok, {:unknown, mime, stream}}
 
         _ ->
-          Stats.increment("imager.process.failed", 1, tags: tags)
+          Instrumenter.Processing.failed(store)
 
           :failed
       end
     else
       {:ok, {_, _, _}} = result ->
         Logger.debug("Cache hit")
-        Stats.increment("imager.process.cache_hit", 1, tags: tags)
+        Instrumenter.Cache.cache_hit(cache)
 
         result
 

lib/imager/application.ex

@@ -16,9 +16,10 @@ defmodule Imager.Application do
       Application.get_env(:imager, :sentry_dsn)
     )
 
-    Imager.Stats.start()
     {:ok, _} = Logger.add_backend(Sentry.LoggerBackend)
 
+    prometheus()
+
     opts = [strategy: :one_for_one, name: Imager.Supervisor]
     Supervisor.start_link(children, opts)
   end
@@ -40,4 +41,12 @@ defmodule Imager.Application do
 
     %{id: :exec_app, start: {:exec, :start_link, [opts]}}
   end
+
+  defp prometheus do
+    :prometheus_registry.register_collector(:prometheus_process_collector)
+
+    Imager.Instrumenter.Cache.setup()
+    Imager.Instrumenter.Processing.setup()
+    Imager.Instrumenter.Storage.setup()
+  end
 end

lib/imager/instrumenter/cache.ex

@@ -0,0 +1,15 @@
+defmodule Imager.Instrumenter.Cache do
+  def setup do
+    :prometheus_counter.declare(
+      name: :cache_hits_total,
+      help: "Count of cached images hits",
+      labels: [:type, :store]
+    )
+  end
+
+  def passthrough({store, _}),
+    do: :prometheus_counter.inc(:cache_hits_total, [:passthrough, store])
+
+  def cache_hit({store, _}),
+    do: :prometheus_counter.inc(:cache_hits_total, [:cache, store])
+end

lib/imager/instrumenter/processing.ex

@@ -0,0 +1,26 @@
+defmodule Imager.Instrumenter.Processing do
+  def setup do
+    :prometheus_counter.declare(
+      name: :processed_total,
+      help: "Count of processed images",
+      labels: [:status, :store]
+    )
+
+    :prometheus_counter.declare(
+      name: :process_commands_total,
+      help: "Processing commands defined",
+      labels: [:command, :argument]
+    )
+  end
+
+  def succeeded({store, _}),
+    do: :prometheus_counter.inc(:processed_total, ["ok", store])
+
+  def failed({store, _}),
+    do: :prometheus_counter.inc(:processed_total, ["failed", store])
+
+  def command({command, argument}),
+    do: :prometheus_counter.inc(:process_commands_total, [command, argument])
+
+  def command(list) when is_list(list), do: Enum.each(list, &command/1)
+end

lib/imager/instrumenter/storage.ex

@@ -0,0 +1,49 @@
+defmodule Imager.Instrumenter.Storage do
+  def setup do
+    :prometheus_counter.declare(
+      name: :store_retrieved_total,
+      help: "Count of retrieves from the given store",
+      labels: [:store]
+    )
+
+    :prometheus_counter.declare(
+      name: :store_saved_total,
+      help: "Count of saves into the given store",
+      labels: [:store]
+    )
+
+    :prometheus_summary.declare(
+      name: :store_retrieved_bytes,
+      help: "Bytes retrieved from the store",
+      labels: [:store]
+    )
+
+    :prometheus_summary.declare(
+      name: :store_saved_bytes,
+      help: "Saved bytes to given store",
+      labels: [:store]
+    )
+  end
+
+  def retrieved(stream, store) do
+    :prometheus_counter.inc(:store_retrieved_total, [store])
+
+    Stream.each(
+      stream,
+      &:prometheus_summary.observe(
+        :store_retrieved_bytes,
+        [store],
+        byte_size(&1)
+      )
+    )
+  end
+
+  def saved(stream, store) do
+    :prometheus_counter.inc(:store_saved_total, [store])
+
+    Stream.each(
+      stream,
+      &:prometheus_summary.observe(:store_saved_bytes, [store], byte_size(&1))
+    )
+  end
+end

lib/imager/store.ex

@@ -4,7 +4,7 @@ defmodule Imager.Store do
   in application.
   """
 
-  alias Imager.Stats
+  alias Imager.Instrumenter
 
   @type size :: non_neg_integer()
   @type stream :: Enumerable.t()
@@ -31,21 +31,18 @@ defmodule Imager.Store do
   @spec retrieve(store, binary, keyword) ::
           {:ok, {size, mime, stream}} | :error
   def retrieve({store, glob_opts}, path, options) do
-    Stats.increment(
-      "imager.store.retrieve",
-      1,
-      Stats.tags(~w(module:#{store}))
-    )
-
-    store.retrieve(path, Keyword.merge(glob_opts, options))
+    with {:ok, {size, mime, stream}} <-
+           store.retrieve(path, Keyword.merge(glob_opts, options)),
+         do: {:ok, {size, mime, Instrumenter.Storage.retrieved(stream, store)}}
   end
 
   @doc """
   Save file in store.
   """
   @spec store(stream, store, mime, binary, keyword) :: stream
   def store(stream, {store, glob_opts}, mime, path, options) do
-    Stats.increment("imager.store.store", 1, Stats.tags(~w(module:#{store})))
-    store.store(path, mime, stream, Keyword.merge(glob_opts, options))
+    path
+    |> store.store(mime, stream, Keyword.merge(glob_opts, options))
+    |> Instrumenter.Storage.saved(store)
   end
 end

lib/imager_web/controllers/health.ex

@@ -6,8 +6,10 @@ defmodule ImagerWeb.Controllers.Health do
   def get(conn, _params) do
     json(conn, %{
       status: "pass",
-      version: Imager.Stats.version(),
+      version: version(),
       description: @description
     })
   end
+
+  defp version, do: Application.spec(:imager, :vsn) |> List.to_string()
 end

lib/imager_web/endpoint.ex

@@ -2,6 +2,9 @@ defmodule ImagerWeb.Endpoint do
   use Phoenix.Endpoint, otp_app: :imager
   use Sentry.Phoenix.Endpoint
 
+  plug(ImagerWeb.Plug.PipelineInstrumenter)
+  plug(ImagerWeb.Plug.MetricsExporter)
+
   # Code reloading can be explicitly enabled under the
   # :code_reloader configuration of your endpoint.
   if code_reloading? do
@@ -27,6 +30,9 @@ defmodule ImagerWeb.Endpoint do
   configuration should be loaded from the system environment.
   """
   def init(_key, config) do
+    ImagerWeb.Plug.PipelineInstrumenter.setup()
+    ImagerWeb.Plug.MetricsExporter.setup()
+
     port =
       System.get_env("PORT") || Application.get_env(:imager, :port) ||
         raise "expected the PORT environment variable to be set"