Feature/setup monitoring engine

.gitignore

@@ -34,4 +34,6 @@ uplink-*.tar
 
 mnesia
 
-.mnesia
+.mnesia
+
+.envrc

config/dev.exs

@@ -1,10 +1,13 @@
 import Config
 
-config :uplink, Uplink.Secret, "secretsomethingsixteen"
+config :uplink,
+       Uplink.Secret,
+       System.get_env("UPLINK_SECRET", "secretsomethingsixteen")
 
 config :uplink, Uplink.Data, mode: "lite"
 
-config :uplink, Uplink.Clients.Instellar, endpoint: "http://localhost/uplink"
+config :uplink, Uplink.Clients.Instellar,
+  endpoint: "http://localhost:4000/uplink"
 
 config :uplink, :environment, :dev
 

config/test.exs

@@ -2,6 +2,12 @@ import Config
 
 config :uplink, Uplink.Data, mode: "pro"
 
+config :uplink, Uplink.Monitors, enabled: false
+
+config :uplink, Uplink.Metrics.Pipeline,
+  producer_module: Broadway.DummyProducer,
+  producer_options: []
+
 config :uplink, Uplink.Repo,
   username:
     System.get_env("UPLINK_DB_USERNAME") || System.get_env("POSTGRES_USERNAME"),

lib/uplink/application.ex

@@ -22,6 +22,9 @@ defmodule Uplink.Application do
       {Cluster.Supervisor, [topologies, [name: Uplink.ClusterSupervisor]]},
       {Task.Supervisor, name: Uplink.TaskSupervisor},
       {Plug.Cowboy, plug: Uplink.Internal, scheme: :http, port: internal_port},
+      {Pogo.DynamicSupervisor,
+       [name: Uplink.PipelineSupervisor, scope: :uplink]},
+      {Uplink.Monitors, []},
       {
         Plug.Cowboy,
         plug: Uplink.Router,

lib/uplink/clients/instellar.ex

@@ -9,6 +9,7 @@ defmodule Uplink.Clients.Instellar do
     Instance,
     Register,
     Component,
+    Monitor,
     Variable,
     Proxy,
     Self
@@ -35,6 +36,10 @@ defmodule Uplink.Clients.Instellar do
     to: Proxy,
     as: :list
 
+  defdelegate list_monitors,
+    to: Monitor,
+    as: :list
+
   defdelegate deployment_metadata(install),
     to: Installation,
     as: :metadata

lib/uplink/clients/instellar/monitor.ex

@@ -0,0 +1,21 @@
+defmodule Uplink.Clients.Instellar.Monitor do
+  alias Uplink.Clients.Instellar
+
+  def list do
+    headers = Instellar.Self.headers()
+
+    [Instellar.endpoint(), "self", "monitors"]
+    |> Path.join()
+    |> Req.get(headers: headers, max_retries: 1)
+    |> case do
+      {:ok, %{status: 200, body: %{"data" => monitors}}} ->
+        {:ok, monitors}
+
+      {:ok, %{status: _, body: body}} ->
+        {:error, body}
+
+      {:error, error} ->
+        {:error, error}
+    end
+  end
+end

lib/uplink/clients/lxd.ex

@@ -4,6 +4,10 @@ defmodule Uplink.Clients.LXD do
 
   alias Uplink.Clients.LXD
 
+  defdelegate get_node(name),
+    to: __MODULE__.Node.Manager,
+    as: :show
+
   defdelegate list_cluster_members(),
     to: __MODULE__.Cluster.Manager,
     as: :list_members
@@ -16,11 +20,11 @@ defmodule Uplink.Clients.LXD do
     to: __MODULE__.Profile.Manager,
     as: :get
 
-  defdelegate list_instances(project),
-    to: __MODULE__.Instance.Manager,
+  defdelegate list_metrics(options \\ []),
+    to: __MODULE__.Metric.Manager,
     as: :list
 
-  defdelegate list_instances(),
+  defdelegate list_instances(options \\ []),
     to: __MODULE__.Instance.Manager,
     as: :list
 
@@ -33,37 +37,36 @@ defmodule Uplink.Clients.LXD do
     as: :leases
 
   def uplink_leases do
-    Cache.get({:leases, "uplink"}) ||
-      (
-        config = Application.get_env(:uplink, Uplink.Data) || []
-        uplink_project = Keyword.get(config, :project, "default")
-        client = LXD.client()
-
-        uplink_project =
-          client
-          |> Lexdee.get_project(uplink_project)
-          |> case do
-            {:ok, %{body: %{"name" => name}}} -> name
-            {:error, %{"error_code" => 404}} -> "default"
-          end
-
-        case LXD.network_leases(uplink_project) do
-          leases when is_list(leases) ->
-            uplink_addresses =
-              Enum.map(leases, fn lease ->
-                lease.address
-              end)
-
-            Cache.put({:leases, "uplink"}, uplink_addresses,
-              ttl: :timer.hours(3)
-            )
-
-            uplink_addresses
-
-          {:error, error} ->
-            {:error, error}
-        end
-      )
+    Cache.get({:leases, "uplink"}) || fetch_leases()
+  end
+
+  defp fetch_leases do
+    config = Application.get_env(:uplink, Uplink.Data) || []
+    uplink_project = Keyword.get(config, :project, "default")
+    client = LXD.client()
+
+    uplink_project =
+      client
+      |> Lexdee.get_project(uplink_project)
+      |> case do
+        {:ok, %{body: %{"name" => name}}} -> name
+        {:error, %{"error_code" => 404}} -> "default"
+      end
+
+    case LXD.network_leases(uplink_project) do
+      leases when is_list(leases) ->
+        uplink_addresses =
+          Enum.map(leases, fn lease ->
+            lease.address
+          end)
+
+        Cache.put({:leases, "uplink"}, uplink_addresses, ttl: :timer.hours(3))
+
+        uplink_addresses
+
+      {:error, error} ->
+        {:error, error}
+    end
   end
 
   def client do

lib/uplink/clients/lxd/instance.ex

@@ -9,10 +9,13 @@ defmodule Uplink.Clients.LXD.Instance do
     status
     architecture
     profiles
+    project
     description
     created_at
     last_used_at
     expanded_config
+    expanded_devices
+    state
   )a
 
   @required_attrs ~w(
@@ -34,13 +37,16 @@ defmodule Uplink.Clients.LXD.Instance do
     field :status, :string
     field :architecture, :string
     field :profiles, {:array, :string}
+    field :project, :string
     field :description, :string
 
     field :created_at, :utc_datetime_usec
     field :last_used_at, :utc_datetime_usec
 
     field :expanded_config, :map
     field :expanded_devices, :map
+
+    field :state, :map
   end
 
   def changeset(schema, params) do

lib/uplink/clients/lxd/instance/manager.ex

@@ -6,27 +6,25 @@ defmodule Uplink.Clients.LXD.Instance.Manager do
   alias Clients.LXD
   alias LXD.Instance
 
-  def list do
-    LXD.client()
-    |> Lexdee.list_instances(query: [{:recursion, 1}, {"all-projects", true}])
-    |> case do
-      {:ok, %{body: instances}} ->
-        instances =
-          instances
-          |> Enum.map(fn instance ->
-            Instance.parse(instance)
-          end)
+  def list(options \\ []) do
+    project = Keyword.get(options, :project, nil)
+    recursion = Keyword.get(options, :recursion, 1)
 
-        instances
-
-      error ->
-        error
+    if recursion < 1 do
+      raise "recursion must be greater than 0 and less than 3 but got #{recursion}"
     end
-  end
 
-  def list(project) do
+    project_query =
+      if project do
+        [project: project]
+      else
+        [{"all-projects", true}]
+      end
+
+    query = [{:recursion, recursion} | project_query]
+
     LXD.client()
-    |> Lexdee.list_instances(query: [recursion: 1, project: project])
+    |> Lexdee.list_instances(query: query)
     |> case do
       {:ok, %{body: instances}} ->
         instances =

lib/uplink/clients/lxd/metric.ex

@@ -0,0 +1,23 @@
+defmodule Uplink.Clients.LXD.Metric do
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  @primary_key false
+  embedded_schema do
+    field :instance, :string
+    field :label, :string
+    field :value, :string
+  end
+
+  def changeset(metric, params) do
+    metric
+    |> cast(params, [:instance, :type, :value])
+    |> validate_required([:instance, :type, :value])
+  end
+
+  def parse(params) do
+    %__MODULE__{}
+    |> changeset(params)
+    |> apply_action!(:insert)
+  end
+end

lib/uplink/clients/lxd/metric/manager.ex

@@ -0,0 +1,27 @@
+defmodule Uplink.Clients.LXD.Metric.Manager do
+  alias Uplink.Clients.LXD
+
+  def list(_options \\ []) do
+    LXD.client()
+    |> Lexdee.list_metrics()
+    |> case do
+      {:ok, %{body: raw_metrics}} ->
+        raw_metrics
+        |> String.split("\n")
+        |> Enum.map(fn line ->
+          PrometheusParser.parse(line)
+        end)
+        |> Enum.map(fn
+          {:ok, line} -> line
+          _ -> nil
+        end)
+        |> Enum.reject(&is_nil/1)
+        |> Enum.reject(fn line ->
+          line.line_type != "ENTRY"
+        end)
+
+      error ->
+        error
+    end
+  end
+end

lib/uplink/clients/lxd/node.ex

@@ -0,0 +1,29 @@
+defmodule Uplink.Clients.LXD.Node do
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  @primary_key false
+  embedded_schema do
+    field :name, :string
+    field :cpu_cores_count, :integer
+    field :total_memory, :integer
+    field :total_storage, :integer
+  end
+
+  def changeset(node, params) do
+    node
+    |> cast(params, [:name, :cpu_cores_count, :total_memory, :total_storage])
+    |> validate_required([
+      :name,
+      :cpu_cores_count,
+      :total_memory,
+      :total_storage
+    ])
+  end
+
+  def parse(params) do
+    %__MODULE__{}
+    |> changeset(params)
+    |> apply_action!(:insert)
+  end
+end

lib/uplink/clients/lxd/node/manager.ex

@@ -0,0 +1,38 @@
+defmodule Uplink.Clients.LXD.Node.Manager do
+  alias Uplink.Cache
+  alias Uplink.Clients.LXD
+  alias Uplink.Clients.LXD.Node
+
+  def show(name) do
+    Cache.get({:node, name}) || fetch_node(name)
+  end
+
+  defp fetch_node(name) do
+    LXD.client()
+    |> Lexdee.show_resources(name)
+    |> case do
+      {:ok, %{body: node}} ->
+        %{
+          "cpu" => %{"total" => total_cores_count},
+          "memory" => %{"total" => total_memory},
+          "storage" => %{"disks" => disks}
+        } = node
+
+        disk_sizes =
+          Enum.map(disks, fn disk ->
+            disk["size"]
+          end)
+          |> Enum.sum()
+
+        Node.parse(%{
+          name: name,
+          cpu_cores_count: total_cores_count,
+          total_memory: total_memory,
+          total_storage: disk_sizes
+        })
+
+      error ->
+        error
+    end
+  end
+end