Merge pull request #17 from appunite/use-prometheus-instead-of-statsd

Replace Statix with Prometheus

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"