Setup supervision tree for pipelines

lib/uplink/metrics/pipeline.ex

@@ -7,12 +7,12 @@ defmodule Uplink.Metrics.Pipeline do
   alias Uplink.Metrics.Document
 
   def start_link(opts) do
-    monitor = Keyword.fetch!(opts, :monitor)
+    monitors = Keyword.fetch!(opts, :monitors)
 
     Broadway.start_link(__MODULE__,
-      name: __MODULE__,
+      name: {:global, __MODULE__},
       context: %{
-        monitor: monitor
+        monitors: monitors
       },
       producer: [
         module: {Uplink.Metrics.Producer, [poll_interval: :timer.seconds(15)]},
@@ -31,6 +31,14 @@ defmodule Uplink.Metrics.Pipeline do
         ]
       ]
     )
+    |> case do
+      {:ok, pid} ->
+        {:ok, pid}
+
+      {:error, {:already_started, pid}} ->
+        Process.link(pid)
+        {:ok, pid}
+    end
   end
 
   def handle_message(_, %Message{data: data} = message, _) do
@@ -56,7 +64,10 @@ defmodule Uplink.Metrics.Pipeline do
   def handle_batch(_, messages, _batch_info, context) do
     documents = to_ndjson(messages)
 
-    Metrics.push!(context.monitor, documents) |> IO.inspect()
+    context.monitors
+    |> Enum.map(fn monitor ->
+      Metrics.push!(context.monitor, documents)
+    end)
 
     messages
   end

lib/uplink/monitors.ex

@@ -0,0 +1,33 @@
+defmodule Uplink.Monitors do
+  use Task
+
+  alias Uplink.Cache
+  alias Uplink.Pipelines
+  alias Uplink.Clients.Instellar
+
+  def start_link(options) do
+    Task.start_link(__MODULE__, :run, options)
+  end
+
+  def run(options) do
+    Instellar.list_monitors()
+    |> case do
+      {:ok, %{body: monitors}} ->
+        state = maybe_start_pipeline(monitors)
+
+      error ->
+        {:error, error}
+    end
+  end
+
+  defp maybe_start_pipeline(monitors) do
+    Cache.transaction([keys: [:monitors]], fn ->
+      started_monitors = Cache.get(:monitors)
+
+      not_started_monitors =
+        Enum.filter(monitors, fn monitor ->
+          monitor["attributes"]["id"] not in started_monitors
+        end)
+    end)
+  end
+end

lib/uplink/pipelines.ex

@@ -0,0 +1,17 @@
+defmodule Uplink.Pipelines do
+  use DynamicSupervisor
+
+  def start_link(options) do
+    DynamicSupervisor.start_link(__MODULE__, options, name: __MODULE__)
+  end
+
+  def start_metrics(monitors) do
+    spec = {Uplink.Metrics.Pipeline, monitors: monitors}
+
+    DynamicSupervisor.start_child(__MODULE__, spec)
+  end
+
+  def init(_options) do
+    DynamicSupervisor.init(strategy: :one_for_one)
+  end
+end