✨ SlotProcessorServer with ProcessMetrics

Author: RTLS

lib/sequin/postgres/replication_connection.ex

@@ -140,6 +140,8 @@ defmodule Sequin.Postgres.ReplicationConnection do
 
   @behaviour :gen_statem
 
+  use Sequin.ProcessMetrics.Decorator
+
   import Bitwise
 
   alias Postgrex.Protocol
@@ -492,19 +494,17 @@ defmodule Sequin.Postgres.ReplicationConnection do
   end
 
   def handle_event(:info, msg, @state, %{protocol: protocol, streaming: streaming} = s) do
-    execute_timed(:handle_data, fn ->
-      case Protocol.handle_copy_recv(msg, streaming, protocol) do
-        {:ok, copies, protocol} ->
-          handle_data(copies, %{s | protocol: protocol})
+    case Protocol.handle_copy_recv(msg, streaming, protocol) do
+      {:ok, copies, protocol} ->
+        handle_data(copies, %{s | protocol: protocol})
 
-        :unknown ->
-          %{state: {mod, mod_state}} = s
-          maybe_handle(mod, :handle_info, [msg, mod_state], s)
+      :unknown ->
+        %{state: {mod, mod_state}} = s
+        maybe_handle(mod, :handle_info, [msg, mod_state], s)
 
-        {error, reason, protocol} ->
-          reconnect_or_stop(error, reason, protocol, s)
-      end
-    end)
+      {error, reason, protocol} ->
+        reconnect_or_stop(error, reason, protocol, s)
+    end
   end
 
   def handle_event({:timeout, :reconnect}, nil, @state, s) do
@@ -626,16 +626,4 @@ defmodule Sequin.Postgres.ReplicationConnection do
 
   defp opts, do: Process.get(__MODULE__)
   defp put_opts(opts), do: Process.put(__MODULE__, opts)
-
-  defp incr_counter(name, amount \\ 1) do
-    current = Process.get(name, 0)
-    Process.put(name, current + amount)
-  end
-
-  defp execute_timed(name, fun) do
-    {time, result} = :timer.tc(fun, :millisecond)
-    incr_counter(:"#{name}_total_ms", time)
-    incr_counter(:"#{name}_count")
-    result
-  end
 end

lib/sequin/process_metrics.ex

@@ -131,19 +131,23 @@ defmodule Sequin.ProcessMetrics do
 
       # Add the handle_info callback for process_logging
       def handle_info(:process_logging, state) do
+        handle_process_logging()
+
+        Process.send_after(self(), :process_logging, process_metrics_interval())
+        Sequin.ProcessMetrics.no_reply(state)
+      end
+
+      defp handle_process_logging do
         # Get dynamic tags from process dictionary
         dynamic_tags = Sequin.ProcessMetrics.get_metadata()
         # Merge static and dynamic tags
         tags = Map.merge(process_metrics_tags(), dynamic_tags)
 
         Sequin.ProcessMetrics.handle_process_logging(
-          interval: process_metrics_interval(),
           metric_prefix: process_metrics_metric_prefix(),
           logger_prefix: process_metrics_logger_prefix(),
           tags: tags
         )
-
-        Sequin.ProcessMetrics.no_reply(state)
       end
     end
   end
@@ -277,7 +281,7 @@ defmodule Sequin.ProcessMetrics do
   * `metric_prefix` - The prefix to use for StatsD metrics
   * `tags` - Additional tags to include in StatsD metrics
   """
-  def handle_process_logging(interval: interval, metric_prefix: metric_prefix, logger_prefix: logger_prefix, tags: tags) do
+  def handle_process_logging(metric_prefix: metric_prefix, logger_prefix: logger_prefix, tags: tags) do
     now = System.monotonic_time(:millisecond)
     last_logged_at = Process.get(@metrics_last_logged_at_key)
     interval_ms = if last_logged_at, do: now - last_logged_at
@@ -435,15 +439,8 @@ defmodule Sequin.ProcessMetrics do
     put_metrics(@default_state)
 
     # Schedule next logging and update last logged time
-    schedule_process_logging(interval)
     Process.put(@metrics_last_logged_at_key, now)
   end
-
-  # Private functions
-
-  defp schedule_process_logging(interval) do
-    Process.send_after(self(), :process_logging, interval)
-  end
 end
 
 defmodule Sequin.ProcessMetrics.Decorator do