add clock drift metrics

Author: acco

lib/sequin/runtime/slot_processor_server.ex

@@ -39,6 +39,7 @@ defmodule Sequin.Runtime.SlotProcessorServer do
   alias Sequin.Runtime.PostgresRelationHashCache
   alias Sequin.Runtime.SlotMessageStore
   alias Sequin.Runtime.SlotProcessor.Message
+  alias Sequin.Time
   alias Sequin.Workers.CreateReplicationSlotWorker
 
   require Logger
@@ -456,7 +457,10 @@ defmodule Sequin.Runtime.SlotProcessorServer do
   # Int64           - Server's system clock (microseconds since 2000-01-01 midnight)
   # Byte1           - 1 if reply requested immediately to avoid timeout, 0 otherwise
   # The server is not asking for a reply
-  def handle_data(<<?k, wal_end::64, _clock::64, 0>>, %State{} = state) do
+  def handle_data(<<?k, wal_end::64, clock::64, 0>>, %State{} = state) do
+    diff_ms = Time.microseconds_since_2000_to_ms_since_now(clock)
+    Logger.info("Received keepalive message for slot", clock: clock, wal_end: wal_end, diff_ms: diff_ms)
+
     execute_timed(:handle_data_keepalive, fn ->
       # Because these are <14 Postgres databases, they will not receive heartbeat messages
       # temporarily mark them as healthy if we receive a keepalive message
@@ -469,9 +473,11 @@ defmodule Sequin.Runtime.SlotProcessorServer do
 
       # Check if we should send an ack even though not requested
       if should_send_ack?(state) do
+        Logger.info("Sending ack")
         commit_lsn = get_commit_lsn(state, wal_end)
         reply = ack_message(commit_lsn)
         state = %{state | last_lsn_acked_at: Sequin.utc_now()}
+        log_keepalive_ack(commit_lsn, clock)
         {:keep_state_and_ack, reply, state}
       else
         {:keep_state, state}
@@ -480,11 +486,20 @@ defmodule Sequin.Runtime.SlotProcessorServer do
   end
 
   # The server is asking for a reply
-  def handle_data(<<?k, wal_end::64, _clock::64, 1>>, %State{} = state) do
+  def handle_data(<<?k, wal_end::64, clock::64, 1>>, %State{} = state) do
+    diff_ms = Time.microseconds_since_2000_to_ms_since_now(clock)
+
+    Logger.info("Received keepalive message for slot (expecting reply)",
+      clock: clock,
+      wal_end: wal_end,
+      diff_ms: diff_ms
+    )
+
     execute_timed(:handle_data_keepalive, fn ->
       commit_lsn = get_commit_lsn(state, wal_end)
       reply = ack_message(commit_lsn)
       state = %{state | last_lsn_acked_at: Sequin.utc_now()}
+      log_keepalive_ack(commit_lsn, clock)
       {:keep_state_and_ack, reply, state}
     end)
   end
@@ -1783,4 +1798,16 @@ defmodule Sequin.Runtime.SlotProcessorServer do
     latency_us = DateTime.diff(Sequin.utc_now(), ts, :microsecond)
     Prometheus.observe_ingestion_latency(state.replication_slot.id, state.replication_slot.slot_name, latency_us)
   end
+
+  defp log_keepalive_ack(commit_lsn, clock) do
+    diff_ms = Time.microseconds_since_2000_to_ms_since_now(clock)
+    message = "Responded to keepalive ack in #{diff_ms}ms"
+    tags = [commit_lsn: commit_lsn, diff_ms: diff_ms]
+
+    case diff_ms do
+      diff_ms when diff_ms < 100 -> Logger.info(message, tags)
+      diff_ms when diff_ms < 1000 -> Logger.warning(message, tags)
+      _ -> Logger.error(message, tags)
+    end
+  end
 end

lib/sequin/time.ex

@@ -154,4 +154,63 @@ defmodule Sequin.Time do
   defp convert_to_ms(num, "ms"), do: num
   defp convert_to_ms(num, "s"), do: num * 1000
   defp convert_to_ms(num, "m"), do: num * 60 * 1000
+
+  @doc """
+  Calculate the difference in milliseconds between two timestamps.
+
+  ## Parameters
+    - start: The starting timestamp
+    - finish: The ending timestamp (defaults to now)
+
+  ## Examples
+      iex> start = ~N[2023-01-01 00:00:00]
+      iex> finish = ~N[2023-01-01 00:00:01]
+      iex> Sequin.Time.ms_since(start, finish)
+      1000
+  """
+  @spec ms_since(DateTime.t() | NaiveDateTime.t(), DateTime.t() | NaiveDateTime.t() | nil) :: integer()
+  def ms_since(start, finish \\ nil)
+
+  def ms_since(%DateTime{} = start, %DateTime{} = finish) do
+    DateTime.diff(finish, start, :millisecond)
+  end
+
+  def ms_since(%DateTime{} = start, nil) do
+    ms_since(start, DateTime.utc_now())
+  end
+
+  def ms_since(%NaiveDateTime{} = start, %NaiveDateTime{} = finish) do
+    NaiveDateTime.diff(finish, start, :millisecond)
+  end
+
+  def ms_since(%NaiveDateTime{} = start, nil) do
+    ms_since(start, NaiveDateTime.utc_now())
+  end
+
+  @doc """
+  Convert microseconds since 2000-01-01 to milliseconds difference from now.
+
+  This is primarily used for Postgres replication protocol timestamps.
+
+  ## Parameters
+    - microseconds_since_2000: Microseconds since 2000-01-01 00:00:00 UTC
+
+  ## Returns
+    Millisecond difference between the timestamp and now (positive if timestamp is in future)
+  """
+  @spec microseconds_since_2000_to_ms_since_now(integer()) :: integer()
+  def microseconds_since_2000_to_ms_since_now(microseconds_since_2000) do
+    # Define the epoch (2000-01-01 midnight in UTC)
+    {:ok, epoch_2000} = NaiveDateTime.new(2000, 1, 1, 0, 0, 0)
+    epoch_2000_in_microseconds = NaiveDateTime.diff(epoch_2000, ~N[1970-01-01 00:00:00], :microsecond)
+
+    # Convert microseconds since 2000 to microseconds since Unix epoch
+    unix_microseconds = epoch_2000_in_microseconds + microseconds_since_2000
+
+    # Convert to DateTime
+    timestamp = DateTime.from_unix!(unix_microseconds, :microsecond)
+
+    # Calculate difference in milliseconds
+    ms_since(timestamp, DateTime.utc_now())
+  end
 end