Include event and handler name as tags with handle_in metrics

lib/prom_ex/plugins/phoenix.ex

@@ -35,14 +35,29 @@ if Code.ensure_loaded?(Phoenix) do
       will be set to `some-route`. You can pass in either a regular expression or a string to match the incoming
       request.
 
+    - `normalize_event_name`: This option is OPTIONAL and allows you to remap the channel event names to a different
+      name. This is useful if you want to limit the number or size of event names that are emitted.
+
+      For example, if you have a channel and only want to emit unique metrics for the 'join' and 'leave' events, you can
+      map those events to themselves and map all other events to an 'unknown' event.
+
+      ```elixir
+        fn
+          "join" -> "join"
+          "leave" -> "leave"
+          _ -> "unknown"
+        end
+      ```
+
     #### Example plugin configuration
 
     ```elixir
     {
       PromEx.Plugins.Phoenix,
       endpoint: MyApp.Endpoint,
       router: MyAppWeb.Public.Router,
-      event_prefix: [:admin, :endpoint]
+      event_prefix: [:admin, :endpoint],
+      normalize_event_name: fn event -> event end)
     }
     ```
 
@@ -159,13 +174,14 @@ if Code.ensure_loaded?(Phoenix) do
       metric_prefix = Keyword.get(opts, :metric_prefix, PromEx.metric_prefix(otp_app, :phoenix))
       phoenix_event_prefixes = fetch_event_prefixes!(opts)
       duration_unit = Keyword.get(opts, :duration_unit, :millisecond)
+      normalize_event_name = Keyword.get(opts, :normalize_event_name, fn event -> event end)
 
       set_up_telemetry_proxy(phoenix_event_prefixes)
 
       # Event metrics definitions
       [
         http_events(metric_prefix, opts),
-        channel_events(metric_prefix, duration_unit),
+        channel_events(metric_prefix, duration_unit, normalize_event_name),
         socket_events(metric_prefix, duration_unit)
       ]
     end
@@ -300,7 +316,7 @@ if Code.ensure_loaded?(Phoenix) do
       )
     end
 
-    defp channel_events(metric_prefix, duration_unit) do
+    defp channel_events(metric_prefix, duration_unit, normalize_event_name) do
       duration_unit_plural = Utils.make_plural_atom(duration_unit)
 
       Event.build(
@@ -330,12 +346,14 @@ if Code.ensure_loaded?(Phoenix) do
             reporter_options: [
               buckets: [10, 100, 500, 1_000, 5_000, 10_000]
             ],
-            tag_values: fn %{socket: %Socket{endpoint: endpoint}} ->
+            tag_values: fn %{socket: %Socket{endpoint: endpoint, handler: handler}, event: event} ->
               %{
-                endpoint: normalize_module_name(endpoint)
+                endpoint: normalize_module_name(endpoint),
+                event: normalize_event_name.(event),
+                handler: handler
               }
             end,
-            tags: [:endpoint],
+            tags: [:endpoint, :handler, :event],
             unit: {:native, duration_unit}
           )
         ]

test/prom_ex/ets_cron_flusher_test.exs

@@ -43,16 +43,16 @@ defmodule PromEx.ETSCronFlusherTest do
       Events.execute_all(:phoenix)
 
       assert length(get_metrics_table(DefaultPromExSetUp)) == 5
-      assert length(get_dist_table(DefaultPromExSetUp)) == 40
+      assert length(get_dist_table(DefaultPromExSetUp)) == 42
 
       Events.execute_all(:phoenix)
 
       assert length(get_metrics_table(DefaultPromExSetUp)) == 5
-      assert length(get_dist_table(DefaultPromExSetUp)) == 80
+      assert length(get_dist_table(DefaultPromExSetUp)) == 84
 
       Process.sleep(8_000)
 
-      assert length(get_metrics_table(DefaultPromExSetUp)) == 11
+      assert length(get_metrics_table(DefaultPromExSetUp)) == 13
       assert get_dist_table(DefaultPromExSetUp) == []
 
       new_timer_ref = get_timer_ref(DefaultPromExSetUp)
@@ -73,16 +73,16 @@ defmodule PromEx.ETSCronFlusherTest do
       Events.execute_all(:phoenix)
 
       assert length(get_metrics_table(ManualPromExSetUp)) == 5
-      assert length(get_dist_table(ManualPromExSetUp)) == 40
+      assert length(get_dist_table(ManualPromExSetUp)) == 42
 
       Events.execute_all(:phoenix)
 
       assert length(get_metrics_table(ManualPromExSetUp)) == 5
-      assert length(get_dist_table(ManualPromExSetUp)) == 80
+      assert length(get_dist_table(ManualPromExSetUp)) == 84
 
       Process.sleep(3_500)
 
-      assert length(get_metrics_table(ManualPromExSetUp)) == 11
+      assert length(get_metrics_table(ManualPromExSetUp)) == 13
       assert get_dist_table(ManualPromExSetUp) == []
 
       new_timer_ref = get_timer_ref(ManualPromExSetUp)

test/prom_ex/plugins/phoenix_test.exs

@@ -36,6 +36,29 @@ defmodule PromEx.Plugins.PhoenixTest do
     end
   end
 
+  defmodule WebApp.PromExSingleEndpointNormalizedChannelEvents do
+    use PromEx, otp_app: :web_app
+
+    @additional_routes [
+      special_label: "/really-cool-route",
+      another_label: ~r(\/another-cool-route)
+    ]
+
+    @impl true
+    def plugins do
+      [
+        {Phoenix,
+         router: TestApp.Router,
+         additional_routes: @additional_routes,
+         endpoint: TestApp.Endpoint,
+         normalize_event_name: fn
+           "test_event" -> "test_event"
+           _ -> "unknown"
+         end}
+      ]
+    end
+  end
+
   test "telemetry events are accumulated for single endpoint configuration" do
     start_supervised!(WebApp.PromExSingleEndpoint)
     Events.execute_all(:phoenix)
@@ -50,6 +73,15 @@ defmodule PromEx.Plugins.PhoenixTest do
     Metrics.assert_prom_ex_metics(WebApp.PromExMultipleEndpoint, :phoenix)
   end
 
+  test "channel events normalize according to normalize_event_name" do
+    start_supervised!(WebApp.PromExSingleEndpointNormalizedChannelEvents)
+    Events.execute_all(:phoenix)
+
+    collected_metrics = Metrics.read_collected(WebApp.PromExSingleEndpointNormalizedChannelEvents)
+
+    assert collected_metrics |> Enum.any?(&String.contains?(&1, "unknown"))
+  end
+
   describe "event_metrics/1" do
     test "should return the correct number of metrics" do
       assert length(Phoenix.event_metrics(otp_app: :prom_ex, router: Some.Module)) == 3

test/support/events/phoenix.exs

@@ -2157,6 +2157,60 @@
       }
     }
   },
+  %{
+    event: [:phoenix, :channel_handled_in],
+    measurements: %{duration: 51792},
+    metadata: %{
+      event: "test_event",
+      params: %{},
+      ref: "10",
+      socket: %{
+        __struct__: Phoenix.Socket,
+        assigns: %{},
+        channel: Phoenix.LiveReloader.Channel,
+        channel_pid: nil,
+        endpoint: WebAppWeb.Endpoint,
+        handler: Phoenix.LiveReloader.Socket,
+        id: nil,
+        joined: true,
+        join_ref: "6",
+        private: %{log_handle_in: :debug, log_join: :info},
+        pubsub_server: WebApp.PubSub,
+        ref: nil,
+        serializer: Phoenix.Socket.V2.JSONSerializer,
+        topic: "phoenix:live_reload",
+        transport: :websocket,
+        transport_pid: nil
+      }
+    }
+  },
+  %{
+    event: [:phoenix, :channel_handled_in],
+    measurements: %{duration: 51792},
+    metadata: %{
+      event: "another_test_event",
+      params: %{},
+      ref: "10",
+      socket: %{
+        __struct__: Phoenix.Socket,
+        assigns: %{},
+        channel: Phoenix.LiveReloader.Channel,
+        channel_pid: nil,
+        endpoint: WebAppWeb.Endpoint,
+        handler: Phoenix.LiveReloader.Socket,
+        id: nil,
+        joined: true,
+        join_ref: "6",
+        private: %{log_handle_in: :debug, log_join: :info},
+        pubsub_server: WebApp.PubSub,
+        ref: nil,
+        serializer: Phoenix.Socket.V2.JSONSerializer,
+        topic: "phoenix:live_reload",
+        transport: :websocket,
+        transport_pid: nil
+      }
+    }
+  },
   %{
     event: [:phoenix, :endpoint, :stop],
     measurements: %{duration: 501_930_627},

test/support/metrics.ex

@@ -12,6 +12,14 @@ defmodule PromEx.Test.Support.Metrics do
     |> sort()
   end
 
+  @spec read_collected(module()) :: [String.t()]
+  @doc false
+  def read_collected(prom_ex_module) do
+    prom_ex_module
+    |> PromEx.get_metrics()
+    |> sort()
+  end
+
   @doc false
   @spec sort(String.t()) :: [String.t()]
   def sort(metrics_string) do
@@ -23,10 +31,7 @@ defmodule PromEx.Test.Support.Metrics do
   @doc false
   @spec assert_prom_ex_metics(module(), atom()) :: :ok
   def assert_prom_ex_metics(prom_ex_module, expected_metrics_lookup) do
-    collected_metrics =
-      prom_ex_module
-      |> PromEx.get_metrics()
-      |> sort()
+    collected_metrics = read_collected(prom_ex_module)
 
     expected_metrics = read_expected(expected_metrics_lookup)
 

test/support/metrics/phoenix.txt

@@ -4,6 +4,26 @@ web_app_prom_ex_prom_ex_status_info 1
 # HELP web_app_prom_ex_phoenix_channel_joined_total The number of channel joins that have occurred.
 # TYPE web_app_prom_ex_phoenix_channel_joined_total counter
 web_app_prom_ex_phoenix_channel_joined_total{endpoint="WebAppWeb.Endpoint",result="ok",transport="websocket"} 17
+# HELP web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds The time it takes for the application to respond to channel messages.
+# TYPE web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds histogram
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="10"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="100"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="500"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="1000"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="5000"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="10000"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="+Inf"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_sum{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket"} 0.051792
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_count{endpoint="WebAppWeb.Endpoint",event="test_event",handler="Elixir.Phoenix.LiveReloader.Socket"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="10"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="100"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="500"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="1000"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="5000"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="10000"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_bucket{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket",le="+Inf"} 1
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_sum{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket"} 0.051792
+web_app_prom_ex_phoenix_channel_handled_in_duration_milliseconds_count{endpoint="WebAppWeb.Endpoint",event="another_test_event",handler="Elixir.Phoenix.LiveReloader.Socket"} 1
 # HELP web_app_prom_ex_phoenix_http_requests_total The number of requests have been serviced.
 # TYPE web_app_prom_ex_phoenix_http_requests_total counter
 web_app_prom_ex_phoenix_http_requests_total{action="index",controller="Phoenix.LiveView.Plug",method="GET",path="/",status="500"} 2