Add statsd telemetry (#4)

.env.dev

@@ -0,0 +1 @@
+DATABASE_URL=postgresql://postgres:password@localhost:54321/electric

sync_service/config/runtime.exs

@@ -1,4 +1,9 @@
 import Config
+import Dotenvy
+
+if config_env() in [:dev, :test] do
+  source!([".env.#{config_env()}", ".env.#{config_env()}.local", System.get_env()])
+end
 
 config :logger, level: :debug
 
@@ -7,5 +12,13 @@ if Config.config_env() == :test do
     database_config:
       PostgresqlUri.parse("postgresql://postgres:password@localhost:54321/postgres")
 else
-  config :electric, database_config: PostgresqlUri.parse(System.fetch_env!("DATABASE_URL"))
+  config :electric,
+    database_config: PostgresqlUri.parse(env!("DATABASE_URL"))
 end
+
+statsd_host = env!("STATSD_HOST", :string?, nil)
+
+config :electric,
+  # Used in telemetry
+  instance_id: env!("ELECTRIC_INSTANCE_ID", :string, Electric.Utils.uuid4()),
+  telemetry_statsd_host: statsd_host

sync_service/lib/electric.ex

@@ -1,18 +1,11 @@
 defmodule Electric do
-  @moduledoc """
-  Documentation for `Electric`.
-  """
-
   @doc """
-  Hello world.
-
-  ## Examples
-
-      iex> Electric.hello()
-      :world
+  Every electric cluster belongs to a particular console database instance
 
+  This is that database instance id
   """
-  def hello do
-    :world
+  @spec instance_id() :: binary | no_return
+  def instance_id do
+    Application.fetch_env!(:electric, :instance_id)
   end
 end

sync_service/lib/electric/application.ex

@@ -10,6 +10,7 @@ defmodule Electric.Application do
   @impl true
   def start(_type, _args) do
     children = [
+      Electric.Telemetry,
       {Electric.InMemShapeCache, []},
       {Electric.Replication.ShapeLogStorage, []},
       {Registry,

sync_service/lib/electric/telemetry.ex

@@ -0,0 +1,65 @@
+defmodule Electric.Telemetry do
+  use Supervisor
+  import Telemetry.Metrics
+
+  def start_link(init_arg) do
+    Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
+  end
+
+  def init(_) do
+    children = [
+      {:telemetry_poller, measurements: periodic_measurements(), period: 2_000}
+    ]
+
+    children
+    |> add_statsd_reporter(Application.fetch_env!(:electric, :telemetry_statsd_host))
+    |> Supervisor.init(strategy: :one_for_one)
+  end
+
+  defp add_statsd_reporter(children, nil), do: children
+
+  defp add_statsd_reporter(children, host) do
+    children ++
+      [
+        {TelemetryMetricsStatsd,
+         host: host,
+         formatter: :datadog,
+         global_tags: [instance_id: Electric.instance_id()],
+         metrics: statsd_metrics()}
+      ]
+  end
+
+  defp statsd_metrics() do
+    [
+      last_value("vm.memory.total", unit: :byte),
+      last_value("vm.memory.processes_used", unit: :byte),
+      last_value("vm.memory.binary", unit: :byte),
+      last_value("vm.memory.ets", unit: :byte),
+      last_value("vm.total_run_queue_lengths.total"),
+      last_value("vm.total_run_queue_lengths.cpu"),
+      last_value("vm.total_run_queue_lengths.io"),
+      summary("plug.router_dispatch.stop.duration",
+        tags: [:route],
+        unit: {:native, :millisecond}
+      ),
+      summary("plug.router_dispatch.exception.duration",
+        tags: [:route],
+        unit: {:native, :millisecond}
+      )
+    ]
+    |> Enum.map(&%{&1 | tags: [:instance_id | &1.tags]})
+  end
+
+  defp periodic_measurements do
+    [
+      # A module, function and arguments to be invoked periodically.
+      {__MODULE__, :uptime_event, []}
+    ]
+  end
+
+  def uptime_event do
+    :telemetry.execute([:vm, :uptime], %{
+      total: :erlang.monotonic_time() - :erlang.system_info(:start_time)
+    })
+  end
+end

sync_service/lib/electric/utils.ex

@@ -1,4 +1,14 @@
 defmodule Electric.Utils do
+  @doc """
+  Generate a random UUID v4.
+
+  Code taken from Ecto: https://github.com/elixir-ecto/ecto/blob/v3.10.2/lib/ecto/uuid.ex#L174
+  """
+  def uuid4() do
+    <<u0::48, _::4, u1::12, _::2, u2::62>> = :crypto.strong_rand_bytes(16)
+    encode_uuid(<<u0::48, 4::4, u1::12, 2::2, u2::62>>)
+  end
+
   def encode_uuid(
         <<a1::4, a2::4, a3::4, a4::4, a5::4, a6::4, a7::4, a8::4, b1::4, b2::4, b3::4, b4::4,
           c1::4, c2::4, c3::4, c4::4, d1::4, d2::4, d3::4, d4::4, e1::4, e2::4, e3::4, e4::4,

sync_service/mix.exs

@@ -31,7 +31,10 @@ defmodule Electric.MixProject do
       {:postgrex, "~> 0.18"},
       {:postgresql_uri, "~> 0.1"},
       {:jason, "~> 1.4"},
-      {:nimble_options, "~> 1.1"}
+      {:nimble_options, "~> 1.1"},
+      {:telemetry_metrics_statsd, "~> 0.7"},
+      {:dotenvy, "~> 0.8"},
+      {:telemetry_poller, "~> 1.1"}
     ]
   end
 

sync_service/mix.lock

@@ -3,6 +3,7 @@
   "bandit": {:hex, :bandit, "1.5.5", "df28f1c41f745401fe9e85a6882033f5f3442ab6d30c8a2948554062a4ab56e0", [:mix], [{:hpax, "~> 0.2.0", [hex: :hpax, repo: "hexpm", optional: false]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:thousand_island, "~> 1.0", [hex: :thousand_island, repo: "hexpm", optional: false]}, {:websock, "~> 0.5", [hex: :websock, repo: "hexpm", optional: false]}], "hexpm", "f21579a29ea4bc08440343b2b5f16f7cddf2fea5725d31b72cf973ec729079e1"},
   "db_connection": {:hex, :db_connection, "2.6.0", "77d835c472b5b67fc4f29556dee74bf511bbafecdcaf98c27d27fa5918152086", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "c2f992d15725e721ec7fbc1189d4ecdb8afef76648c746a8e1cad35e3b8a35f3"},
   "decimal": {:hex, :decimal, "2.1.1", "5611dca5d4b2c3dd497dec8f68751f1f1a54755e8ed2a966c2633cf885973ad6", [:mix], [], "hexpm", "53cfe5f497ed0e7771ae1a475575603d77425099ba5faef9394932b35020ffcc"},
+  "dotenvy": {:hex, :dotenvy, "0.8.0", "777486ad485668317c56afc53a7cbcd74f43e4e34588ba8e95a73e15a360050e", [:mix], [], "hexpm", "1f535066282388cbd109743d337ac46ff0708195780d4b5778bb83491ab1b654"},
   "elixir_make": {:hex, :elixir_make, "0.8.4", "4960a03ce79081dee8fe119d80ad372c4e7badb84c493cc75983f9d3bc8bde0f", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:certifi, "~> 2.0", [hex: :certifi, repo: "hexpm", optional: true]}], "hexpm", "6e7f1d619b5f61dfabd0a20aa268e575572b542ac31723293a4c1a567d5ef040"},
   "epgsql": {:hex, :epgsql, "4.7.1", "d4e47cae46c18c8afa88e34d59a9b4bae16368d7ce1eb3da24fa755eb28393eb", [:rebar3], [], "hexpm", "b6d86b7dc42c8555b1d4e20880e5099d6d6d053148000e188e548f98e4e01836"},
   "gen_stage": {:hex, :gen_stage, "1.2.1", "19d8b5e9a5996d813b8245338a28246307fd8b9c99d1237de199d21efc4c76a1", [:mix], [], "hexpm", "83e8be657fa05b992ffa6ac1e3af6d57aa50aace8f691fcf696ff02f8335b001"},
@@ -21,6 +22,9 @@
   "postgrex": {:hex, :postgrex, "0.18.0", "f34664101eaca11ff24481ed4c378492fed2ff416cd9b06c399e90f321867d7e", [:mix], [{:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a042989ba1bc1cca7383ebb9e461398e3f89f868c92ce6671feb7ef132a252d1"},
   "protox": {:hex, :protox, "1.7.3", "dff5488a648850c95cbd1cca5430be7ccedc99e4102aa934dbf60abfa30e64c1", [:mix], [{:decimal, "~> 1.9 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "b936c0654b68b306c4be853db23bb5623e2be89e11238908f2ff6da10fc0275f"},
   "telemetry": {:hex, :telemetry, "1.2.1", "68fdfe8d8f05a8428483a97d7aab2f268aaff24b49e0f599faa091f1d4e7f61c", [:rebar3], [], "hexpm", "dad9ce9d8effc621708f99eac538ef1cbe05d6a874dd741de2e689c47feafed5"},
+  "telemetry_metrics": {:hex, :telemetry_metrics, "0.6.2", "2caabe9344ec17eafe5403304771c3539f3b6e2f7fb6a6f602558c825d0d0bfb", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "9b43db0dc33863930b9ef9d27137e78974756f5f198cae18409970ed6fa5b561"},
+  "telemetry_metrics_statsd": {:hex, :telemetry_metrics_statsd, "0.7.0", "92732fae63db31ef2508df6faee7d81401883e33f2976715a82f296a33a45cee", [:mix], [{:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}], "hexpm", "797e34a856376dfd4e96347da0f747fcff4e0cadf6e6f0f989598f563cad05ff"},
+  "telemetry_poller": {:hex, :telemetry_poller, "1.1.0", "58fa7c216257291caaf8d05678c8d01bd45f4bdbc1286838a28c4bb62ef32999", [:rebar3], [{:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "9eb9d9cbfd81cbd7cdd24682f8711b6e2b691289a0de6826e58452f28c103c8f"},
   "thousand_island": {:hex, :thousand_island, "1.3.5", "6022b6338f1635b3d32406ff98d68b843ba73b3aa95cfc27154223244f3a6ca5", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "2be6954916fdfe4756af3239fb6b6d75d0b8063b5df03ba76fd8a4c87849e180"},
   "tz": {:hex, :tz, "0.26.5", "bfe8efa345670f90351c5c31d22455d0307c5d9895fbdede7deeb215a7b60dbe", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:mint, "~> 1.5", [hex: :mint, repo: "hexpm", optional: true]}], "hexpm", "c4f9392d710582c7108b6b8c635f4981120ec4b2072adbd242290fc842338183"},
   "websock": {:hex, :websock, "0.5.3", "2f69a6ebe810328555b6fe5c831a851f485e303a7c8ce6c5f675abeb20ebdadc", [:mix], [], "hexpm", "6105453d7fac22c712ad66fab1d45abdf049868f253cf719b625151460b8b453"},