Add custom metrics and metric history fork to LiveDashboard (#613)

* Use live_dashboard historical_data branch

* historical_data integration WIP

* possible metadata storage?

* Drop poller behavior, just re-emit immediately

Also rename metric and simplify history to include metadata

poller was based on previous custom count logic,
doesn’t make sense for shadowing something that
works as is without poller.

* Working Custom viewing metrics/history for simple controller actions

* Fix Credo, add moduledocs to all new modules

config/dev.exs

@@ -56,6 +56,9 @@ config :companies, CompaniesWeb.Endpoint,
     ]
   ]
 
+# Do not run appsignal in local dev
+config :appsignal, :config, active: false
+
 # Do not include metadata nor timestamps in development logs
 config :logger, :console, format: "[$level] $message\n"
 

lib/companies/application.ex

@@ -10,6 +10,9 @@ defmodule Companies.Application do
     children = [
       # Start the Ecto repository
       Companies.Repo,
+      CompaniesWeb.Telemetry,
+      CompaniesWeb.RepoMetricsHistory,
+      CompaniesWeb.ViewingStats,
       # Start the PubSub system
       {Phoenix.PubSub, name: Companies.PubSub},
       # Start the endpoint when the application starts
@@ -24,6 +27,9 @@ defmodule Companies.Application do
       nil
     )
 
+    CompaniesWeb.RepoMetricsHistory.setup_handlers()
+    CompaniesWeb.ViewingStats.setup_handlers()
+
     # See https://hexdocs.pm/elixir/Supervisor.html
     # for other strategies and supported options
     opts = [strategy: :one_for_one, name: Companies.Supervisor]

lib/companies_web/controllers/company_controller.ex

@@ -2,6 +2,7 @@ defmodule CompaniesWeb.CompanyController do
   use CompaniesWeb, :controller
 
   alias Companies.{Companies, Industries, Schema.Company}
+  import CompaniesWeb.ViewingStats, only: [telemetry_event: 0]
 
   def recent(conn, _params) do
     companies_count = Companies.count()
@@ -13,6 +14,7 @@ defmodule CompaniesWeb.CompanyController do
   def index(conn, params) do
     companies = Companies.all(params)
     industries = Industries.for_select()
+    :telemetry.execute(telemetry_event(), %{company_index: 1})
 
     render(conn, "index.html", companies: companies, industries: industries)
   end
@@ -39,6 +41,7 @@ defmodule CompaniesWeb.CompanyController do
 
   def show(conn, %{"id" => id}) do
     company = Companies.get!(id, preloads: [:jobs, :industry])
+    :telemetry.execute(telemetry_event(), %{company_show: 1})
     render(conn, "show.html", company: company)
   end
 

lib/companies_web/controllers/job_controller.ex

@@ -2,10 +2,12 @@ defmodule CompaniesWeb.JobController do
   use CompaniesWeb, :controller
 
   alias Companies.{Companies, Jobs, Schema.Job}
+  import CompaniesWeb.ViewingStats, only: [telemetry_event: 0]
   plug :load_companies when action in [:new, :edit, :create, :update]
 
   def index(conn, params) do
     jobs = Jobs.all(params)
+    :telemetry.execute(telemetry_event(), %{job_index: 1})
     render(conn, "index.html", jobs: jobs)
   end
 
@@ -18,6 +20,8 @@ defmodule CompaniesWeb.JobController do
   def create(conn, %{"job" => params}) do
     case Jobs.create(params, current_user(conn)) do
       {:ok, _job} ->
+        :telemetry.execute(telemetry_event(), %{job_create: 1})
+
         conn
         |> put_flash(:info, "Thank you! Your listing will be review and should appear on the site shortly.")
         |> redirect(to: Routes.company_path(conn, :recent, locale(conn)))

lib/companies_web/historical_data.ex

@@ -0,0 +1,16 @@
+defmodule CompaniesWeb.HistoricalData do
+  @moduledoc """
+  Single source of truth to gather all individual modules that provide historical
+  data to live dashboard and merge their seperate history maps into one. If adding
+  a new source of metric history for LiveDashboard, it only needs to implement
+  it's own signatures/0 method as the other modules here do, and be added the the
+  list of signatures below.
+  """
+  alias CompaniesWeb.{RepoMetricsHistory, ViewingStats}
+
+  def signatures do
+    for module <- [RepoMetricsHistory, ViewingStats], reduce: %{} do
+      acc -> Map.merge(acc, apply(module, :signatures, []))
+    end
+  end
+end

lib/companies_web/repo_metrics_history.ex

@@ -0,0 +1,68 @@
+defmodule CompaniesWeb.RepoMetricsHistory do
+  @moduledoc """
+  GenServer to handle historical data for Ecto queries and rebroadcast to
+  LiveDashboard under seperate history-focused telemetry namespace.
+  """
+
+  use GenServer
+
+  @telemetry_event [:companies, :repo, :query]
+  @historic_metric [:ecto, :dashboard, :query]
+  @history_buffer_size 500
+
+  def signatures do
+    %{
+      @historic_metric => {__MODULE__, :data, []}
+    }
+  end
+
+  def telemetry_event, do: @telemetry_event
+
+  def start_link([]) do
+    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+  end
+
+  def init(_state) do
+    {:ok, %{history: CircularBuffer.new(@history_buffer_size)}}
+  end
+
+  def data(%{event_name: event_name} = metric) do
+    if List.starts_with?(event_name, @historic_metric) do
+      GenServer.call(__MODULE__, {:data, metric})
+    else
+      []
+    end
+  end
+
+  def setup_handlers do
+    :telemetry.attach(
+      "aggregation-handler-#{__MODULE__}",
+      @telemetry_event,
+      &__MODULE__.handle_event/4,
+      nil
+    )
+  end
+
+  def handle_event(@telemetry_event, metric_map, metadata, config) do
+    GenServer.cast(__MODULE__, {:telemetry_metric, metric_map, metadata, config})
+  end
+
+  def handle_call({:data, _metric}, _from, %{history: history}) do
+    {:reply, CircularBuffer.to_list(history), %{history: history}}
+  end
+
+  def handle_cast({:telemetry_metric, metric_map, metadata, _config}, %{history: history}) do
+    time = System.system_time(:second)
+    :telemetry.execute(@historic_metric, metric_map, metadata)
+
+    new_history = CircularBuffer.insert(history, %{data: metric_map, time: time, metadata: pruned_metadata(metadata)})
+
+    {:noreply, %{history: new_history}}
+  end
+
+  defp pruned_metadata(metadata) do
+    # for now keep it all, reminder to either keep or drop selected fields based on dashboard usage to conserve memory,
+    # ideally via some published source of truth hook from dahsboard module to ensure correctness
+    metadata
+  end
+end

lib/companies_web/router.ex

@@ -38,7 +38,10 @@ defmodule CompaniesWeb.Router do
 
   scope "/dashboard", CompaniesWeb do
     pipe_through [:browser, :auth]
-    live_dashboard "/", metrics: CompaniesWeb.Telemetry
+
+    live_dashboard "/",
+      metrics: CompaniesWeb.Telemetry,
+      historical_data: CompaniesWeb.HistoricalData.signatures()
   end
 
   scope "/", CompaniesWeb do

lib/companies_web/telemetry.ex

@@ -36,6 +36,19 @@ defmodule CompaniesWeb.Telemetry do
       summary("companies.repo.query.queue_time", unit: {:native, :millisecond}),
       summary("companies.repo.query.idle_time", unit: {:native, :millisecond}),
 
+      # Database Time Metrics with History
+      summary("ecto.dashboard.query.total_time", unit: {:native, :millisecond}),
+      summary("ecto.dashboard.query.decode_time", unit: {:native, :millisecond}),
+      summary("ecto.dashboard.query.query_time", unit: {:native, :millisecond}),
+      summary("ecto.dashboard.query.queue_time", unit: {:native, :millisecond}),
+      summary("ecto.dashboard.query.idle_time", unit: {:native, :millisecond}),
+
+      # Page view stats
+      summary("page_views.companies_web.company_index"),
+      summary("page_views.companies_web.company_show"),
+      summary("page_views.companies_web.job_index"),
+      summary("page_views.companies_web.job_create"),
+
       # VM Metrics
       summary("vm.memory.total", unit: {:byte, :kilobyte}),
       summary("vm.total_run_queue_lengths.total"),
@@ -45,6 +58,6 @@ defmodule CompaniesWeb.Telemetry do
   end
 
   defp periodic_measurements do
-    []
+    [{CompaniesWeb.ViewingStats, :emit, []}]
   end
 end

lib/companies_web/viewing_stats.ex

@@ -0,0 +1,88 @@
+defmodule CompaniesWeb.ViewingStats do
+  @moduledoc """
+  GenServer to handle aggregating simple app stats and history for telemetry_poller
+  to pass on to LiveDashboard.
+  """
+
+  use GenServer
+
+  @telemetry_event [:page_views, :count_events]
+  @historic_metrics [:page_views, :companies_web]
+  @history_buffer_size 500
+
+  def signatures do
+    %{
+      @historic_metrics => {__MODULE__, :data, []}
+    }
+  end
+
+  def telemetry_event, do: @telemetry_event
+
+  def start_link([]) do
+    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+  end
+
+  def init(_state) do
+    {:ok, %{history: CircularBuffer.new(@history_buffer_size), current: %{}}}
+  end
+
+  def data(%{event_name: event_name} = metric) do
+    if List.starts_with?(event_name, @historic_metrics) do
+      GenServer.call(__MODULE__, {:data, metric})
+    else
+      []
+    end
+  end
+
+  def setup_handlers do
+    :telemetry.attach(
+      "aggregation-handler-#{__MODULE__}",
+      @telemetry_event,
+      &__MODULE__.handle_event/4,
+      nil
+    )
+  end
+
+  def handle_event(@telemetry_event, map, metadata, config) do
+    GenServer.cast(__MODULE__, {:telemetry_metric, map, metadata, config})
+  end
+
+  def emit do
+    GenServer.cast(__MODULE__, :emit_telemetry)
+  end
+
+  def handle_call({:raw_data, _metric}, _from, %{history: history} = state) do
+    {:reply, history, state}
+  end
+
+  def handle_call({:data, metric}, _from, %{history: history} = state) do
+    local_metric = List.last(metric.name)
+
+    reply =
+      for {time, time_metrics} <- history,
+          {^local_metric, data} <- time_metrics do
+        %{data: %{local_metric => data}, time: time}
+      end
+
+    {:reply, reply, state}
+  end
+
+  def handle_cast(:emit_telemetry, %{history: history, current: current}) do
+    time = System.system_time(:second)
+
+    for {key, value} <- current do
+      :telemetry.execute(@historic_metrics, %{key => value})
+    end
+
+    {:noreply, %{history: CircularBuffer.insert(history, {time, current}), current: %{}}}
+  end
+
+  def handle_cast({:telemetry_metric, metric_map, _metadata, _config}, %{current: current} = state) do
+    updated_current =
+      for {key, value} <- metric_map, reduce: current do
+        acc -> Map.put_new(acc, key, 0) |> update_in([key], &(&1 + value))
+      end
+
+    {:noreply, %{state | current: updated_current}}
+  end
+end

mix.exs

@@ -45,7 +45,9 @@ defmodule Companies.MixProject do
       {:phoenix, "~> 1.5.1", override: true},
       {:phoenix_ecto, "~> 4.1"},
       {:phoenix_html, "~> 2.14"},
-      {:phoenix_live_dashboard, "~> 0.2"},
+      {:phoenix_live_dashboard,
+       git: "https://github.com/bglusman/phoenix_live_dashboard.git", branch: "historical_data"},
+      {:circular_buffer, "~> 0.2"},
       {:phoenix_pubsub, "~> 2.0"},
       {:plug_cowboy, "~> 2.2"},
       {:postgrex, ">= 0.0.0"},

mix.lock

@@ -4,6 +4,7 @@
   "bunt": {:hex, :bunt, "0.2.0", "951c6e801e8b1d2cbe58ebbd3e616a869061ddadcc4863d0a2182541acae9a38", [:mix], [], "hexpm", "7af5c7e09fe1d40f76c8e4f9dd2be7cebd83909f31fee7cd0e9eadc567da8353"},
   "bypass": {:hex, :bypass, "1.0.0", "b78b3dcb832a71aca5259c1a704b2e14b55fd4e1327ff942598b4e7d1a7ad83d", [:mix], [{:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 1.0 or ~> 2.0", [hex: :plug_cowboy, repo: "hexpm", optional: false]}], "hexpm", "5a1dc855dfcc86160458c7a70d25f65d498bd8012bd4c06a8d3baa368dda3c45"},
   "certifi": {:hex, :certifi, "2.5.1", "867ce347f7c7d78563450a18a6a28a8090331e77fa02380b4a21962a65d36ee5", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm", "805abd97539caf89ec6d4732c91e62ba9da0cda51ac462380bbd28ee697a8c42"},
+  "circular_buffer": {:hex, :circular_buffer, "0.2.0", "c42be0740855831d04e22e17318a4e186acc3b9ebe9aad8db90b0a08ac0ff589", [:mix], [], "hexpm", "b2a05705485c7576373eeff99b2f76b0048731d9c80c7fa42d6c7b71eb69521e"},
   "combine": {:hex, :combine, "0.10.0", "eff8224eeb56498a2af13011d142c5e7997a80c8f5b97c499f84c841032e429f", [:mix], [], "hexpm", "1b1dbc1790073076580d0d1d64e42eae2366583e7aecd455d1215b0d16f2451b"},
   "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm", "4a0850c9be22a43af9920a71ab17c051f5f7d45c209e40269a1938832510e4d9"},
   "cowboy": {:hex, :cowboy, "2.7.0", "91ed100138a764355f43316b1d23d7ff6bdb0de4ea618cb5d8677c93a7a2f115", [:rebar3], [{:cowlib, "~> 2.8.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "~> 1.7.1", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "04fd8c6a39edc6aaa9c26123009200fc61f92a3a94f3178c527b70b767c6e605"},
@@ -35,11 +36,11 @@
   "phoenix": {:hex, :phoenix, "1.5.1", "95156589879dc69201d5fc0ebdbfdfc7901a09a3616ea611ec297f81340275a2", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_html, "~> 2.13", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:plug, "~> 1.10", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 1.0 or ~> 2.2", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.1.2 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "fc272b38e79d2881790fccae6f67a9fbe9b790103d6878175ea03d23003152eb"},
   "phoenix_ecto": {:hex, :phoenix_ecto, "4.1.0", "a044d0756d0464c5a541b4a0bf4bcaf89bffcaf92468862408290682c73ae50d", [:mix], [{:ecto, "~> 3.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.9", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "c5e666a341ff104d0399d8f0e4ff094559b2fde13a5985d4cb5023b2c2ac558b"},
   "phoenix_html": {:hex, :phoenix_html, "2.14.2", "b8a3899a72050f3f48a36430da507dd99caf0ac2d06c77529b1646964f3d563e", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "58061c8dfd25da5df1ea0ca47c972f161beb6c875cd293917045b92ffe1bf617"},
-  "phoenix_live_dashboard": {:hex, :phoenix_live_dashboard, "0.2.2", "22b551e7ecc8fb77226f20a876c0a241f0f2df39a5309c3d4d35fd64eeb193a3", [:mix], [{:phoenix_html, "~> 2.14.1 or ~> 2.15", [hex: :phoenix_html, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 0.12.0", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.4.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "fe7c810febd63847404eb045ad2689cb4510ae825b73a892fe81b99f5e56558f"},
+  "phoenix_live_dashboard": {:git, "https://github.com/bglusman/phoenix_live_dashboard.git", "9259f4480c8114099bda8ec5d9bda070aad06d02", [branch: "historical_data"]},
   "phoenix_live_reload": {:hex, :phoenix_live_reload, "1.2.1", "274a4b07c4adbdd7785d45a8b0bb57634d0b4f45b18d2c508b26c0344bd59b8f", [:mix], [{:file_system, "~> 0.2.1 or ~> 0.3", [hex: :file_system, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.4", [hex: :phoenix, repo: "hexpm", optional: false]}], "hexpm", "41b4103a2fa282cfd747d377233baf213c648fdcc7928f432937676532490eee"},
   "phoenix_live_view": {:hex, :phoenix_live_view, "0.12.1", "42f591c781edbf9fab921319076b7ac635d43aa23e6748d2644563326236d7e4", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix, "~> 1.4.16 or ~> 1.5.0", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.14", [hex: :phoenix_html, repo: "hexpm", optional: false]}], "hexpm", "585321e98df1cd5943e370b9784e950a37ca073744eb534660c9048967c52ab6"},
   "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.0.0", "a1ae76717bb168cdeb10ec9d92d1480fec99e3080f011402c0a2d68d47395ffb", [:mix], [], "hexpm", "c52d948c4f261577b9c6fa804be91884b381a7f8f18450c5045975435350f771"},
-  "plug": {:hex, :plug, "1.10.0", "6508295cbeb4c654860845fb95260737e4a8838d34d115ad76cd487584e2fc4d", [:mix], [{:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "422a9727e667be1bf5ab1de03be6fa0ad67b775b2d84ed908f3264415ef29d4a"},
+  "plug": {:hex, :plug, "1.10.1", "c56a6d9da7042d581159bcbaef873ba9d87f15dce85420b0d287bca19f40f9bd", [:mix], [{:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "b5cd52259817eb8a31f2454912ba1cff4990bca7811918878091cb2ab9e52cb8"},
   "plug_cowboy": {:hex, :plug_cowboy, "2.2.1", "fcf58aa33227a4322a050e4783ee99c63c031a2e7f9a2eb7340d55505e17f30f", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "3b43de24460d87c0971887286e7a20d40462e48eb7235954681a20cee25ddeb6"},
   "plug_crypto": {:hex, :plug_crypto, "1.1.2", "bdd187572cc26dbd95b87136290425f2b580a116d3fb1f564216918c9730d227", [:mix], [], "hexpm", "6b8b608f895b6ffcfad49c37c7883e8df98ae19c6a28113b02aa1e9c5b22d6b5"},
   "postgrex": {:hex, :postgrex, "0.15.3", "5806baa8a19a68c4d07c7a624ccdb9b57e89cbc573f1b98099e3741214746ae4", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "4737ce62a31747b4c63c12b20c62307e51bb4fcd730ca0c32c280991e0606c90"},