Merge branch 'release/0.19.1'

lib/uplink/monitors.ex

@@ -22,7 +22,7 @@ defmodule Uplink.Monitors do
   end
 
   def run(_options \\ []) do
-    Cache.put_new({:monitors, :metrics}, [])
+    Pipelines.reset_monitors(:metrics)
 
     Instellar.list_monitors()
     |> case do
@@ -39,27 +39,14 @@ defmodule Uplink.Monitors do
   defp start_pipeline(monitors, context) do
     Logger.info("[Uplink.Monitors] Starting pipeline...")
 
-    started_metrics_monitor_ids =
-      Pipelines.get_monitors(context)
-      |> Enum.map(fn monitor ->
-        monitor["attributes"]["id"]
-      end)
-
-    not_started_monitors =
-      Enum.filter(monitors, fn monitor ->
-        monitor["attributes"]["id"] not in started_metrics_monitor_ids
-      end)
-
     grouped_monitors =
-      Enum.group_by(not_started_monitors, fn monitor ->
+      Enum.group_by(monitors, fn monitor ->
         monitor["attributes"]["type"]
       end)
 
     context_monitors = Map.get(grouped_monitors, "#{context}") || []
 
-    if Enum.count(context_monitors) > 0 do
-      Pipelines.append_monitors(context, context_monitors)
-    end
+    Pipelines.update_monitors(context, context_monitors)
 
     module = Map.fetch!(@pipeline_modules, context)
 

lib/uplink/pipelines.ex

@@ -1,14 +1,18 @@
 defmodule Uplink.Pipelines do
   alias Uplink.Cache
 
-  def get_monitors(context) do
+  @valid_contexts [:metrics]
+
+  def get_monitors(context) when context in @valid_contexts do
     Cache.get({:monitors, context}) || []
   end
 
-  def append_monitors(context, monitors) do
-    Cache.get_and_update({:monitors, context}, fn existing_monitors ->
-      {existing_monitors, existing_monitors ++ monitors}
-    end)
+  def reset_monitors(context) when context in @valid_contexts do
+    Cache.put({:monitors, context}, [])
+  end
+
+  def update_monitors(context, monitors) when context in @valid_contexts do
+    Cache.put({:monitors, context}, monitors)
   end
 
   def start(module) do

mix.exs

@@ -4,7 +4,7 @@ defmodule Uplink.MixProject do
   def project do
     [
       app: :uplink,
-      version: "0.19.0",
+      version: "0.19.1",
       elixir: "~> 1.13",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/uplink/monitors/router_test.exs

@@ -45,6 +45,24 @@ defmodule Uplink.Monitors.RouterTest do
       endpoint: "http://localhost:#{bypass.port}/uplink"
     )
 
+    Uplink.Cache.put({:monitors, :metrics}, [
+      %{
+        "attributes" => %{
+          "current_state" => "active",
+          "endpoint" => "https://elastic:9200",
+          "expires_at" => "2024-11-21T03:14:17Z",
+          "id" => 1,
+          "token" => "some-token",
+          "type" => "metrics",
+          "uid" => "some-other-uid"
+        },
+        "id" => "1",
+        "links" => %{"self" => "http://localhost:4000/uplink/self/monitors/1"},
+        "relationships" => %{},
+        "type" => "monitors"
+      }
+    ])
+
     {:ok, bypass: bypass}
   end
 
@@ -59,6 +77,10 @@ defmodule Uplink.Monitors.RouterTest do
     end
 
     test "can refresh monitors list", %{signature: signature, bypass: bypass} do
+      [monitor] = Uplink.Pipelines.get_monitors(:metrics)
+
+      assert monitor["attributes"]["uid"] == "some-other-uid"
+
       Bypass.expect_once(bypass, "GET", "/uplink/self/monitors", fn conn ->
         conn
         |> Plug.Conn.put_resp_header("content-type", "application/json")
@@ -71,6 +93,10 @@ defmodule Uplink.Monitors.RouterTest do
         |> put_req_header("content-type", "application/json")
         |> Router.call(@opts)
 
+      [monitor] = Uplink.Pipelines.get_monitors(:metrics)
+
+      assert monitor["attributes"]["uid"] == "some-uid"
+
       assert conn.status == 200
     end
   end