[#55] Receiver for storing ethereum statistics

.circleci/config.yml

@@ -8,6 +8,12 @@ jobs:
       # specify the version here
       - image: circleci/elixir:1.6
 
+      - image: circleci/postgres:10.1-alpine  # database image
+        environment:  # environment variables for database
+          POSTGRES_USER: postgres
+          POSTGRES_DB: poabackend_stats_test
+          POSTGRES_PASSWORD: postgres
+
     working_directory: ~/repo
     steps:
       - checkout

config/config.exs

@@ -8,10 +8,14 @@ config :plug, :statuses, %{
 }
 
 config :poa_backend,
-  ecto_repos: [POABackend.Auth.Repo]
+  ecto_repos: [
+    POABackend.Auth.Repo,
+    POABackend.Receivers.Repo
+  ]
 
 # here we configure the needed data for Ecto and Mnesia (DB)
 config :poa_backend, POABackend.Auth.Repo,
+  priv: "priv/auth",
   adapter: EctoMnesia.Adapter,
   host: Kernel.node(),
   storage_type: :disc_copies # this will store the data on disk and memory

config/prod.exs

@@ -27,7 +27,8 @@ config :poa_backend,
             ws_url: "/ws",
             port: 8181,
             ws_secret: "wssecret"
-            ]}
+            ]},
+          {:store_eth_stats, POABackend.Receivers.EthStats, []}
        ]
 
 # here we define the type of metrics we accept. We will create a GenStage Producer per each type
@@ -43,7 +44,8 @@ config :poa_backend,
 config :poa_backend,
        :subscriptions,
        [
-         {:dashboard_receiver, [:ethereum_metrics, :networking_metrics]}
+         {:dashboard_receiver, [:ethereum_metrics, :networking_metrics]},
+         {:store_eth_stats, [:ethereum_metrics]}
        ]
 
 # here we define the configuration for the Authorisation endpoint
@@ -58,7 +60,7 @@ config :poa_backend,
 
 config :poa_backend, POABackend.Auth.Guardian,
   issuer: "poa_backend",
-  secret_key: "LQYmeqQfrphbxUjJltkwH4xnosLc+2S2e8KuYWctMenNY9bmgwnrH8r3ii9FP/8V"
+  secret_key: "LQYmeqQfrphbxUjJltkwH4xnosLc+2S2e8KuYWctMenNY9bmgwnrH8r3ii9FP/8V" # generate your own secret key here!
 
 # this is a list of admins/passwords for authorisation endpoints
 config :poa_backend,
@@ -77,4 +79,12 @@ config :poa_backend,
        jwt_ttl: {1, :hour}
 
 config :mnesia,
-  dir: 'priv/data/mnesia' # make sure this directory exists!
+  dir: 'priv/data/mnesia' # make sure this directory exists!
+
+config :poa_backend, POABackend.Receivers.Repo,
+  priv: "priv/receivers",
+  adapter: Ecto.Adapters.Postgres,
+  database: "poabackend_stats",
+  username: "postgres",
+  password: "postgres",
+  hostname: "localhost"

config/test.exs

@@ -19,7 +19,8 @@ config :poa_backend,
             ws_url: "/ws",
             port: 8181,
             ws_secret: "mywssecret"
-            ]}
+            ]},
+          {:store_eth_stats, POABackend.Receivers.EthStats, []}
        ]
 
 # here we define the type of metrics we accept. We will create a GenStage Producer per each type
@@ -34,7 +35,8 @@ config :poa_backend,
 config :poa_backend,
        :subscriptions,
        [
-         {:dashboard_receiver, [:ethereum_metrics]}
+         {:dashboard_receiver, [:ethereum_metrics]},
+         {:store_eth_stats, [:ethereum_metrics]}
        ]
 
 # here we define the configuration for the Authorisation endpoint
@@ -63,3 +65,11 @@ config :poa_backend,
 # configuration for mnesia DB
 config :mnesia,
   dir: '_build/test' # make sure this directory exists!
+
+config :poa_backend, POABackend.Receivers.Repo,
+  priv: "priv/receivers",
+  adapter: Ecto.Adapters.Postgres,
+  database: "poabackend_stats_test",
+  username: "postgres",
+  password: "postgres",
+  hostname: "localhost"

lib/poa_backend/receivers/eth_stats.ex

@@ -0,0 +1,76 @@
+defmodule POABackend.Receivers.EthStats do
+  use POABackend.Receiver
+
+  @moduledoc false
+
+  alias POABackend.Protocol.Message
+  alias POABackend.Receivers.Models.EthStats
+  alias POABackend.Receivers.Repo
+
+  def init_receiver(_args) do
+    {:ok, :no_state}
+  end
+
+  def metrics_received([%Message{agent_id: agent_id, data: %{"type" => "statistics", "body" => stats}}], _from, state) do
+    :ok = save_data(stats, agent_id)
+    {:ok, state}
+  end
+  def metrics_received(_metrics, _from, state) do
+    {:ok, state}
+  end
+
+  def handle_message(_message, state) do
+    {:ok, state}
+  end
+
+  def handle_inactive(_agent_id, state) do
+    {:ok, state}
+  end
+
+  def terminate(_state) do
+    :ok
+  end
+
+  defp save_data(%{"active" => active, "gasPrice" => gas_price, "hashrate" => hashrate, "mining" => mining, "peers" => peers, "syncing" => syncing}, agent_id) do
+    current_block = current_block(syncing)
+    highest_block = highest_block(syncing)
+    starting_block = starting_block(syncing)
+    
+    EthStats.new()
+    |> EthStats.date(NaiveDateTime.utc_now())
+    |> EthStats.agent_id(agent_id)
+    |> EthStats.active(active)
+    |> EthStats.mining(mining)
+    |> EthStats.hashrate(hashrate)
+    |> EthStats.peers(peers)
+    |> EthStats.gas_price(gas_price)
+    |> EthStats.current_block(current_block)
+    |> EthStats.highest_block(highest_block)
+    |> EthStats.starting_block(starting_block)
+    |> Repo.insert()
+
+    :ok
+  end
+  defp save_data(_data, _agent_id) do
+    :ok
+  end
+
+  defp current_block(syncing) do
+    get_from_sync(syncing, "currentBlock")
+  end
+
+  defp highest_block(syncing) do
+    get_from_sync(syncing, "highestBlock")
+  end
+
+  defp starting_block(syncing) do
+    get_from_sync(syncing, "startingBlock")
+  end
+
+  defp get_from_sync(syncing, key) when is_map(syncing) do
+    Map.get(syncing, key)
+  end
+  defp get_from_sync(_, _) do
+    nil
+  end
+end

lib/poa_backend/receivers/models/eth_stats.ex

@@ -0,0 +1,89 @@
+defmodule POABackend.Receivers.Models.EthStats do
+  use Ecto.Schema
+
+  @moduledoc false
+
+  @primary_key false
+
+  schema "eth_stats" do
+    field :date, :naive_datetime, primary_key: true
+    field :agent_id, :string, primary_key: true
+    field :active, :boolean
+    field :mining, :boolean
+    field :hashrate, :integer
+    field :peers, :integer
+    field :gas_price, :integer
+    field :current_block, :string
+    field :highest_block, :string
+    field :starting_block, :string
+  end
+
+  @type t :: %__MODULE__{
+    date: NaiveDateTime.t,
+    agent_id: String.t,
+    active: Boolean.t,
+    mining: Boolean.t,
+    hashrate: Integer.t,
+    peers: Integer.t,
+    gas_price: Integer.t,
+    current_block: String.t,
+    highest_block: String.t,
+    starting_block: String.t
+  }
+
+  @spec new() :: __MODULE__.t
+  def new do
+    %__MODULE__{}
+  end
+
+  @spec date(__MODULE__.t, NaiveDateTime.t) :: __MODULE__.t
+  def date(eth_stats, date) do
+    %__MODULE__{eth_stats | date: date}
+  end
+
+  @spec agent_id(__MODULE__.t, String.t) :: __MODULE__.t
+  def agent_id(eth_stats, agent_id) do
+    %__MODULE__{eth_stats | agent_id: agent_id}
+  end
+
+  @spec active(__MODULE__.t, Boolean.t) :: __MODULE__.t
+  def active(eth_stats, active) do
+    %__MODULE__{eth_stats | active: active}
+  end
+
+  @spec mining(__MODULE__.t, Boolean.t) :: __MODULE__.t
+  def mining(eth_stats, mining) do
+    %__MODULE__{eth_stats | mining: mining}
+  end
+
+  @spec hashrate(__MODULE__.t, Integer.t) :: __MODULE__.t
+  def hashrate(eth_stats, hashrate) do
+    %__MODULE__{eth_stats | hashrate: hashrate}
+  end
+
+  @spec peers(__MODULE__.t, Integer.t) :: __MODULE__.t
+  def peers(eth_stats, peers) do
+    %__MODULE__{eth_stats | peers: peers}
+  end
+
+  @spec gas_price(__MODULE__.t, Integer.t) :: __MODULE__.t
+  def gas_price(eth_stats, gas_price) do
+    %__MODULE__{eth_stats | gas_price: gas_price}
+  end
+
+  @spec current_block(__MODULE__.t, String.t) :: __MODULE__.t
+  def current_block(eth_stats, current_block) do
+    %__MODULE__{eth_stats | current_block: current_block}
+  end
+
+  @spec highest_block(__MODULE__.t, String.t) :: __MODULE__.t
+  def highest_block(eth_stats, highest_block) do
+    %__MODULE__{eth_stats | highest_block: highest_block}
+  end
+
+  @spec starting_block(__MODULE__.t, String.t) :: __MODULE__.t
+  def starting_block(eth_stats, starting_block) do
+    %__MODULE__{eth_stats | starting_block: starting_block}
+  end
+
+end

lib/poa_backend/receivers/repo.ex

@@ -0,0 +1,3 @@
+defmodule POABackend.Receivers.Repo do
+  use Ecto.Repo, otp_app: :poa_backend
+end

lib/poa_backend/receivers/supervisor.ex

@@ -11,10 +11,14 @@ defmodule POABackend.Receivers.Supervisor do
     receivers = Application.get_env(:poa_backend, :receivers)
     subscriptions = Application.get_env(:poa_backend, :subscriptions)
 
+    # getting the children from the config file
     children = for {name, module, args} <- receivers do
       worker(module, [%{name: name, subscribe_to: subscriptions[name], args: args}])
     end
 
+    # we have to add the Repo to the children too
+    children = [supervisor(POABackend.Receivers.Repo, []) | children]
+
     opts = [strategy: :one_for_one]
     Supervisor.init(children, opts)
   end

mix.exs

@@ -43,6 +43,7 @@ defmodule POABackend.MixProject do
       {:guardian, "~> 1.1"},
       {:comeonin, "~> 4.0"},
       {:bcrypt_elixir, "~> 0.12"},
+      {:postgrex, "~> 0.13.5"},
 
       # Tests
       {:credo, "~> 0.9", only: [:dev, :test], runtime: false},

mix.lock

@@ -5,9 +5,11 @@
   "certifi": {:hex, :certifi, "2.3.1", "d0f424232390bf47d82da8478022301c561cf6445b5b5fb6a84d49a9e76d2639", [], [{:parse_trans, "3.2.0", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm"},
   "comeonin": {:hex, :comeonin, "4.1.1", "c7304fc29b45b897b34142a91122bc72757bc0c295e9e824999d5179ffc08416", [], [{:argon2_elixir, "~> 1.2", [hex: :argon2_elixir, repo: "hexpm", optional: true]}, {:bcrypt_elixir, "~> 0.12.1 or ~> 1.0", [hex: :bcrypt_elixir, repo: "hexpm", optional: true]}, {:pbkdf2_elixir, "~> 0.12", [hex: :pbkdf2_elixir, repo: "hexpm", optional: true]}], "hexpm"},
   "confex": {:hex, :confex, "3.3.1", "8febaf751bf293a16a1ed2cbd258459cdcc7ca53cfa61d3f83d49dd276a992b4", [], [], "hexpm"},
+  "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [], [], "hexpm"},
   "cowboy": {:hex, :cowboy, "1.0.4", "a324a8df9f2316c833a470d918aaf73ae894278b8aa6226ce7a9bf699388f878", [], [{:cowlib, "~> 1.0.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "~> 1.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm"},
   "cowlib": {:hex, :cowlib, "1.0.2", "9d769a1d062c9c3ac753096f868ca121e2730b9a377de23dec0f7e08b1df84ee", [], [], "hexpm"},
   "credo": {:hex, :credo, "0.10.0", "66234a95effaf9067edb19fc5d0cd5c6b461ad841baac42467afed96c78e5e9e", [], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm"},
+  "db_connection": {:hex, :db_connection, "1.1.3", "89b30ca1ef0a3b469b1c779579590688561d586694a3ce8792985d4d7e575a61", [], [{:connection, "~> 1.0.2", [hex: :connection, repo: "hexpm", optional: false]}, {:poolboy, "~> 1.5", [hex: :poolboy, repo: "hexpm", optional: true]}, {:sbroker, "~> 1.0", [hex: :sbroker, repo: "hexpm", optional: true]}], "hexpm"},
   "decimal": {:hex, :decimal, "1.5.0", "b0433a36d0e2430e3d50291b1c65f53c37d56f83665b43d79963684865beab68", [], [], "hexpm"},
   "dialyxir": {:hex, :dialyxir, "0.5.1", "b331b091720fd93e878137add264bac4f644e1ddae07a70bf7062c7862c4b952", [], [], "hexpm"},
   "distillery": {:hex, :distillery, "1.5.3", "b2f4fc34ec71ab4f1202a796f9290e068883b042319aa8c9aa45377ecac8597a", [], [], "hexpm"},
@@ -36,6 +38,7 @@
   "plug": {:hex, :plug, "1.6.1", "c62fe7623d035020cf989820b38490460e6903ab7eee29e234b7586e9b6c91d6", [], [{:cowboy, "~> 1.0.1 or ~> 1.1 or ~> 2.4", [hex: :cowboy, repo: "hexpm", optional: true]}, {:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}], "hexpm"},
   "poison": {:hex, :poison, "3.1.0", "d9eb636610e096f86f25d9a46f35a9facac35609a7591b3be3326e99a0484665", [], [], "hexpm"},
   "poolboy": {:hex, :poolboy, "1.5.1", "6b46163901cfd0a1b43d692657ed9d7e599853b3b21b95ae5ae0a777cf9b6ca8", [], [], "hexpm"},
+  "postgrex": {:hex, :postgrex, "0.13.5", "3d931aba29363e1443da167a4b12f06dcd171103c424de15e5f3fc2ba3e6d9c5", [], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 1.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.0", [hex: :decimal, repo: "hexpm", optional: false]}], "hexpm"},
   "ranch": {:hex, :ranch, "1.5.0", "f04166f456790fee2ac1aa05a02745cc75783c2bfb26d39faf6aefc9a3d3a58a", [], [], "hexpm"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.1", "28a4d65b7f59893bc2c7de786dec1e1555bd742d336043fe644ae956c3497fbe", [], [], "hexpm"},
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.3.1", "a1f612a7b512638634a603c8f401892afbf99b8ce93a45041f8aaca99cadb85e", [], [], "hexpm"},

priv/receivers/migrations/20180828142653_create_eth_stats.exs

@@ -0,0 +1,18 @@
+defmodule POABackend.Receivers.Repo.Migrations.CreateEthStats do
+  use Ecto.Migration
+
+  def change do
+    create table(:eth_stats, primary_key: false) do
+      add :date, :naive_datetime, primary_key: true
+      add :agent_id, :string, primary_key: true
+      add :active, :boolean
+      add :mining, :boolean
+      add :hashrate, :integer
+      add :peers, :integer
+      add :gas_price, :integer
+      add :current_block, :string
+      add :highest_block, :string
+      add :starting_block, :string
+    end
+  end
+end

test/ancillary/utils.ex

@@ -4,6 +4,14 @@ defmodule POABackend.Ancillary.Utils do
   def clear_db do
     :mnesia.clear_table(:users)
     :mnesia.clear_table(:banned_tokens)
+
+    truncate(POABackend.Receivers.Models.EthStats)
+  end
+
+  defp truncate(schema) do
+    table_name = schema.__schema__(:source)
+    POABackend.Receivers.Repo.query("TRUNCATE #{table_name}", [])
+    :ok
   end
 
 end

test/receivers/dashboard_test.exs

@@ -30,7 +30,7 @@ defmodule Receivers.DashboardTest do
   end
 
   test "connection success" do
-    {result, pid} = Client.start_link("http://localhost:8181/ws", :state, [{:extra_headers, [{"wssecret", "mywssecret"}]}])
+    {result, pid} = Client.start_link("http://localhost:8181/ws", self(), [{:extra_headers, [{"wssecret", "mywssecret"}]}])
 
     assert :ok == result
     assert is_pid(pid)

test/receivers/eth_stats_test.exs

@@ -0,0 +1,62 @@
+defmodule Receivers.EthStatsTest do
+  use ExUnit.Case
+
+  alias POABackend.Protocol.Message
+  alias POABackend.Ancillary.Utils
+  alias POABackend.Receivers.Repo
+  alias POABackend.Receivers.Models.EthStats
+
+  setup do
+    Utils.clear_db()
+
+    on_exit fn ->
+      Utils.clear_db()
+    end
+
+    []
+  end
+
+  test "storing stats" do
+    assert [] == Repo.all(EthStats)
+
+    POABackend.Metric.add(:ethereum_metrics, [raw_stats()])
+
+    # this will create an entry in the DB
+    Process.sleep(5000)
+
+    [stats] = Repo.all(EthStats)
+
+    assert "agent_id1" == stats.agent_id
+    assert false == stats.mining
+    assert 0 == stats.hashrate
+    assert 3 == stats.peers
+    assert 0 == stats.gas_price
+    assert "0x44ee0" == stats.current_block
+    assert "0x44ee2" == stats.highest_block
+    assert "0x40ad3" == stats.starting_block
+  end
+
+  defp raw_stats do
+    stats = %{
+      "active" => true,
+      "gasPrice" => 0,
+      "hashrate" => 0,
+      "mining" => false,
+      "peers" => 3,
+      "syncing" => %{
+        "currentBlock" => "0x44ee0",
+        "highestBlock" => "0x44ee2",
+        "startingBlock" => "0x40ad3"
+      }
+    }
+
+    %Message{
+      agent_id: "agent_id1",
+      data: %{
+        "type" => "statistics",
+        "body" => stats
+        }
+      }
+  end
+
+end