Adding node name information into the metric table key

Author: thiagoesteves

lib/observer_web/telemetry/storage.ex

@@ -11,7 +11,6 @@ defmodule ObserverWeb.Telemetry.Storage do
   @behaviour ObserverWeb.Telemetry.Adapter
 
   @metric_keys "metric-keys"
-  @metric_table :observer_web_metrics
 
   @one_minute_in_milliseconds 60_000
   @retention_data_delete_interval :timer.minutes(1)
@@ -28,10 +27,11 @@ defmodule ObserverWeb.Telemetry.Storage do
   @impl true
   def init(_args) do
     node = Node.self()
+    metric_table = metric_table(node)
 
     # Create metric tables for the node
-    :ets.new(@metric_table, [:set, :protected, :named_table])
-    :ets.insert(@metric_table, {@metric_keys, []})
+    :ets.new(metric_table, [:set, :protected, :named_table])
+    :ets.insert(metric_table, {@metric_keys, []})
 
     persist_data? =
       if data_retention_period() do
@@ -41,14 +41,14 @@ defmodule ObserverWeb.Telemetry.Storage do
         false
       end
 
-    {:ok, %{node: node, persist_data?: persist_data?}}
+    {:ok, %{node: node, persist_data?: persist_data?, metric_table: metric_table}}
   end
 
   @impl true
   def handle_cast(
         {:observer_web_telemetry,
          %{metrics: metrics, reporter: reporter, measurements: measurements}},
-        %{node: node, persist_data?: persist_data?} = state
+        %{node: node, persist_data?: persist_data?, metric_table: metric_table} = state
       )
       when reporter in [node] do
     now = System.os_time(:millisecond)
@@ -63,12 +63,12 @@ defmodule ObserverWeb.Telemetry.Storage do
         # credo:disable-for-lines:3
         if persist_data? do
           current_data =
-            case :ets.lookup(@metric_table, timed_key) do
+            case :ets.lookup(metric_table, timed_key) do
               [{_, current_list_data}] -> [data | current_list_data]
               _ -> [data]
             end
 
-          :ets.insert(@metric_table, {timed_key, current_data})
+          :ets.insert(metric_table, {timed_key, current_data})
         end
 
         Phoenix.PubSub.broadcast(
@@ -85,7 +85,7 @@ defmodule ObserverWeb.Telemetry.Storage do
       end)
 
     if new_keys != [] do
-      :ets.insert(@metric_table, {@metric_keys, new_keys ++ keys})
+      :ets.insert(metric_table, {@metric_keys, new_keys ++ keys})
 
       Phoenix.PubSub.broadcast(
         ObserverWeb.PubSub,
@@ -106,7 +106,7 @@ defmodule ObserverWeb.Telemetry.Storage do
 
     prune_keys = fn key ->
       Enum.each(deletion_period_from..deletion_period_to, fn timestamp ->
-        :ets.delete(@metric_table, metric_key(key, timestamp))
+        :ets.delete(state.metric_table, metric_key(key, timestamp))
       end)
     end
 
@@ -162,7 +162,7 @@ defmodule ObserverWeb.Telemetry.Storage do
                node,
                :ets,
                :lookup,
-               [@metric_table, metric_key(key, minute)],
+               [metric_table(node), metric_key(key, minute)],
                :infinity
              ) do
           [{_, value}] ->
@@ -180,7 +180,7 @@ defmodule ObserverWeb.Telemetry.Storage do
   def get_keys_by_node(nil), do: []
 
   def get_keys_by_node(node) do
-    case Rpc.call(node, :ets, :lookup, [@metric_table, @metric_keys], :infinity) do
+    case Rpc.call(node, :ets, :lookup, [metric_table(node), @metric_keys], :infinity) do
       [{_, value}] ->
         value
 
@@ -201,6 +201,8 @@ defmodule ObserverWeb.Telemetry.Storage do
     defp data_retention_period, do: :timer.minutes(1)
   end
 
+  defp metric_table(node), do: String.to_atom("#{node}::observer-web-metrics")
+
   defp metric_key(metric, timestamp), do: "#{metric}|#{timestamp}"
 
   defp unix_to_minutes(time \\ System.os_time(:millisecond)),

test/observer_web/telemetry_storage_test.exs

@@ -10,14 +10,15 @@ defmodule ObserverWeb.TelemetryStorageTest do
   setup [
     :set_mox_global,
     :verify_on_exit!,
-    :create_consumer
+    :create_consumer,
+    :metric_table
   ]
 
-  test "[un]subscribe_for_new_keys/0", %{node: node} do
+  test "[un]subscribe_for_new_keys/0", %{node: node, metric_table: metric_table} do
     Storage.subscribe_for_new_keys()
 
     ObserverWeb.RpcMock
-    |> stub(:call, fn ^node, :ets, :lookup, [:observer_web_metrics, "metric-keys"], :infinity ->
+    |> stub(:call, fn ^node, :ets, :lookup, [^metric_table, "metric-keys"], :infinity ->
       [{"metric-keys", []}]
     end)
 
@@ -28,11 +29,11 @@ defmodule ObserverWeb.TelemetryStorageTest do
     assert_receive {:metrics_new_keys, ^node, ["vm.memory.total"]}, 1_000
   end
 
-  test "[un]subscribe_for_new_data/0", %{node: node} do
+  test "[un]subscribe_for_new_data/0", %{node: node, metric_table: metric_table} do
     Storage.subscribe_for_new_data(node, "vm.memory.total")
 
     ObserverWeb.RpcMock
-    |> stub(:call, fn ^node, :ets, :lookup, [:observer_web_metrics, "metric-keys"], :infinity ->
+    |> stub(:call, fn ^node, :ets, :lookup, [^metric_table, "metric-keys"], :infinity ->
       [{"metric-keys", []}]
     end)
 
@@ -48,12 +49,12 @@ defmodule ObserverWeb.TelemetryStorageTest do
     Storage.unsubscribe_for_new_data(node, "vm.memory.total")
   end
 
-  test "get_keys_by_node/1 valid node", %{node: node} do
+  test "get_keys_by_node/1 valid node", %{node: node, metric_table: metric_table} do
     Storage.subscribe_for_new_data(node, "vm.memory.total")
     test_pid = self()
 
     ObserverWeb.RpcMock
-    |> stub(:call, fn ^node, :ets, :lookup, [:observer_web_metrics, "metric-keys"], :infinity ->
+    |> stub(:call, fn ^node, :ets, :lookup, [^metric_table, "metric-keys"], :infinity ->
       if test_pid != self() do
         # GenServer Cast
         [{"metric-keys", []}]
@@ -78,7 +79,7 @@ defmodule ObserverWeb.TelemetryStorageTest do
     assert [] == Storage.get_keys_by_node(nil)
   end
 
-  test "list_data_by_node_key/3", %{node: node} do
+  test "list_data_by_node_key/3", %{node: node, metric_table: metric_table} do
     key_name = "test.phoenix"
 
     Storage.subscribe_for_new_data(node, key_name)
@@ -87,7 +88,7 @@ defmodule ObserverWeb.TelemetryStorageTest do
     |> stub(
       :call,
       fn
-        ^node, :ets, :lookup, [:observer_web_metrics, "metric-keys"], :infinity ->
+        ^node, :ets, :lookup, [^metric_table, "metric-keys"], :infinity ->
           # First time: Empty keys
           # Second time: Added key
           called = Process.get("ets_lookup", 0)
@@ -134,7 +135,7 @@ defmodule ObserverWeb.TelemetryStorageTest do
            ] = Storage.list_data_by_node_key(node |> to_string(), key_name)
   end
 
-  test "Pruning expiring entries", %{node: node, pid: pid} do
+  test "Pruning expiring entries", %{node: node, pid: pid, metric_table: metric_table} do
     key_name = "test.phoenix"
 
     now = System.os_time(:millisecond)
@@ -145,7 +146,7 @@ defmodule ObserverWeb.TelemetryStorageTest do
     |> stub(
       :call,
       fn
-        ^node, :ets, :lookup, [:observer_web_metrics, "metric-keys"], :infinity ->
+        ^node, :ets, :lookup, [^metric_table, "metric-keys"], :infinity ->
           # First time: Empty keys
           # Second time: Added key
           called = Process.get("ets_lookup", 0)
@@ -208,4 +209,9 @@ defmodule ObserverWeb.TelemetryStorageTest do
     |> Map.put(:node, node)
     |> Map.put(:pid, pid)
   end
+
+  defp metric_table(context) do
+    node = Node.self()
+    Map.put(context, :metric_table, String.to_atom("#{node}::observer-web-metrics"))
+  end
 end