refactor(sync-service): Convert Logger calls to use static strings with metadata

.changeset/static-logger-strings.md

@@ -0,0 +1,6 @@
+---
+'@core/sync-service': patch
+'@core/electric-telemetry': patch
+---
+
+Convert Logger calls to use static strings with structured metadata

integration-tests/tests/db-connection-scaledown.lux

@@ -72,9 +72,9 @@
   [sleep 2]
 
   !Electric.Postgres.ReplicationClient.name("single_stack") |> GenServer.whereis() |> send(:check_if_idle)
-  ??[notice] Closing all database connections after the replication stream has been idle for
+  ?idle_time_s=.*\[notice\] Closing all database connections after the replication stream has been idle
 
-  ??[debug] Terminating connection manager with reason :shutdown.
+  ?reason=shutdown.*\[debug\] Terminating connection manager
 
 # Confirm that Postgres no longer sees any open connections for the electric user
 [shell psql]
@@ -119,7 +119,7 @@
 
 [shell electric]
   ??[debug] GET /v1/shape
-  ??[debug] Query String: table=items&handle=$handle&offset=0_0&live
+  ?query_string=table=items&handle=$handle&offset=0_0&live.*\[debug\] Query String
 
 
   ??[notice] Acquiring lock from postgres

integration-tests/tests/insufficient-connection-resources.lux

@@ -20,8 +20,8 @@
 
 ## Observe the connection error.
 [shell electric]
-  ??[warning] Database connection in connection_pool mode failed: sorry, too many clients already
-  ??[warning] Reconnecting in 
+  ?connection_mode=connection_pool.*err_msg=.*too many clients.*\[warning\] Database connection failed
+  ?delay_ms=.*\[warning\] Reconnecting
 
 
 ## Restarting Postgres with a higher number of max_connections should fix the issue

integration-tests/tests/ipv6-to-ipv4-fallback.lux

@@ -23,8 +23,8 @@
   ??[notice] Lock acquired
   ??tcp connect (local-ipv4-only.electric-sql.dev:$pg_pooler_host_port): non-existing domain - :nxdomain
   ??[warning] Database connection failed to find valid IPv6 address for local-ipv4-only.electric-sql.dev - falling back to IPv4
-  ?+\[notice\] Connection pool \[admin: $pg_host_name:$pg_host_port\] is ready
-  ?\[notice\] Connection pool \[snapshot: $pg_host_name:$pg_pooler_host_port\] is ready
+  ?+role=admin.*host=$pg_host_name:$pg_host_port.*\[notice\] Connection pool is ready
+  ?role=snapshot.*host=$pg_host_name:$pg_pooler_host_port.*\[notice\] Connection pool is ready
   ??[notice] Starting replication from postgres
 
 

integration-tests/tests/manual-table-publishing.lux

@@ -42,7 +42,7 @@
 
 [shell electric]
   -
-  ?\[warning\] Failed to create snapshot for \d+-\d+: Database table \"public.items\" is missing from the publication \"electric_publication_integration\" and the ELECTRIC_MANUAL_TABLE_PUBLISHING setting prevents Electric from adding it
+  ?shape_handle=\d+-\d+.*err_msg=Database table.*public\.items.*ELECTRIC_MANUAL_TABLE_PUBLISHING.*\[warning\] Failed to create snapshot
   -$fail_pattern
 
 ## Add the table to the publication and try again
@@ -63,7 +63,7 @@
 
 [shell electric]
   -
-  ?\[warning\] Failed to create snapshot for \d+-\d+: Database table \"public.items\" does not have its replica identity set to FULL
+  ?shape_handle=\d+-\d+.*err_msg=Database table.*public\.items.*replica identity.*\[warning\] Failed to create snapshot
   -$fail_pattern
 
 ## Set table's replica identity to FULL and try again

integration-tests/tests/postgres-disconnection.lux

@@ -22,8 +22,8 @@
 
 ## Observe the connection error.
 [shell electric]
-  ??[warning] Database connection in replication mode failed
-  ??[warning] Reconnecting in 
+  ?connection_mode=replication.*\[warning\] Database connection failed
+  ?delay_ms=.*\[warning\] Reconnecting
 
 ## Start the Postgres container back up.
 [invoke resume_pg]

integration-tests/tests/recreation-of-replication-slot.lux

@@ -45,14 +45,14 @@
   -
 
   ??[warning] Publication fatal misconfiguration: %Electric.DbConfigurationError{type: :publication_missing
-  ??[debug] Starting replication client for stack single_stack
+  ??[debug] Starting replication client
 
   -$fail_pattern
 
   ??[debug] ReplicationClient step: create_publication_query
   ??[debug] ReplicationClient step: drop_slot
   ??[debug] ReplicationClient step: create_slot
-  ??[debug] Created new slot at lsn=
+  ?lsn=.*\[debug\] Created new slot
   ??[notice] Purging all shapes.
   ??[notice] Starting replication from postgres
 
@@ -73,7 +73,7 @@
   ??"value":{"val":"3"}
 
 [shell electric]
-  ??[debug] Snapshot started shape_handle: $new_shape_handle
+  ?shape_handle=$new_shape_handle.*\[debug\] Snapshot started
   ??[debug] Chunked 200
 
 [cleanup]

integration-tests/tests/replication-slot-self-conflict.lux

@@ -35,9 +35,9 @@
   # it's blocked by the pending transaction.
   -Starting replication from postgres|$fail_pattern
 
-  ??[debug] Starting replication client for stack single_stack
-  ??[notice] Acquiring lock from postgres with name electric_slot_integration
-  ??[notice] Lock acquired from postgres with name electric_slot_integration
+  ??[debug] Starting replication client
+  ?slot_name=electric_slot_integration.*\[notice\] Acquiring lock from postgres
+  ?slot_name=electric_slot_integration.*\[notice\] Lock acquired from postgres
   ??[debug] ReplicationClient step: create_slot
   ??[warning] Waiting for the replication connection setup to complete...
 
@@ -52,7 +52,7 @@
 
   -Lock acquired from postgres|$fail_pattern
 
-  ??[notice] Acquiring lock from postgres with name electric_slot_integration
+  ?slot_name=electric_slot_integration.*\[notice\] Acquiring lock from postgres
 
 # Rollback the pending transaction to unblock Electric and verify that it completes its initialization.
 [shell psql]
@@ -61,9 +61,9 @@
 [shell electric]
   -$fail_pattern
 
-  ??[notice] Lock acquired from postgres with name electric_slot_integration
+  ?slot_name=electric_slot_integration.*\[notice\] Lock acquired from postgres
   ??[debug] ReplicationClient step: create_slot
-  ??[debug] Created new slot at lsn=
+  ?lsn=.*\[debug\] Created new slot
 
   ??[notice] Starting replication from postgres
   ??[debug] ReplicationClient step: start_streaming

integration-tests/tests/rolling-deploy.lux

@@ -36,8 +36,8 @@
 [invoke setup_electric_shell_with_tenant "electric_1" "3000"]
 
 [shell electric_1]
-  ??[notice] Acquiring lock from postgres with name electric_slot_integration
-  ??[notice] Lock acquired from postgres with name electric_slot_integration
+  ?slot_name=electric_slot_integration.*\[notice\] Acquiring lock from postgres
+  ?slot_name=electric_slot_integration.*\[notice\] Lock acquired from postgres
   ??[notice] Starting replication from postgres
 
 # First service should be health and active
@@ -61,7 +61,7 @@
 ## in the second electric
 [shell electric_2]
   -Lock acquired from postgres|Starting replication from postgres|$fail_pattern
-  ??[notice] Acquiring lock from postgres with name electric_slot_integration
+  ?slot_name=electric_slot_integration.*\[notice\] Acquiring lock from postgres
   [sleep 2]
 
 
@@ -104,7 +104,7 @@
 # Lock should now be acquired and replication starting
 [shell electric_2]
   -$fail_pattern
-  ??[notice] Lock acquired from postgres with name electric_slot_integration
+  ?slot_name=electric_slot_integration.*\[notice\] Lock acquired from postgres
 
 # Add more data after handover
 [shell psql]

integration-tests/tests/shape-suspension-resumption.lux

@@ -46,7 +46,7 @@
 [shell electric]
   [sleep 1]
 
-  ??[debug] Suspending consumer $handle
+  ?shape_handle=$handle.*\[debug\] Suspending consumer
   !{:active_consumer_count, Electric.Shapes.ConsumerRegistry.active_consumer_count("single_stack")}
   ?{:active_consumer_count, 0}
 
@@ -57,7 +57,7 @@
 
 
 [shell electric]
-  ??[debug] Started consumer for existing handle $handle
+  ?shape_handle=$handle.*\[debug\] Started consumer for existing handle
   !{:active_consumer_count, Electric.Shapes.ConsumerRegistry.active_consumer_count("single_stack")}
   ?{:active_consumer_count, 1}
 

integration-tests/tests/stack-telemetry.lux

@@ -12,6 +12,10 @@
 
 [invoke setup_electric_with_env "ELECTRIC_OTLP_ENDPOINT=http://localhost:4318 ELECTRIC_SYSTEM_METRICS_POLL_INTERVAL=1s ELECTRIC_OTEL_EXPORT_PERIOD=2s DO_NOT_START_CONN_MAN_PING=1 ELECTRIC_LOG_LEVEL=info OTEL_RESOURCE_ATTRIBUTES=custom.attr=electric.val"]
 
+[shell electric]
+    # Override the fail pattern to exclude benign OTEL log handler removal messages
+    -(?i)(?!.*removed_failing_handler.*otel_log_handler)(error|fatal|no such)
+
 [shell otel_collector]
   # Verify that the Collector receives metrics from Electric with expected resource attributes
   ??info	ResourceMetrics

integration-tests/tests/startup-delayed-by-pending-transaction.lux

@@ -27,9 +27,9 @@
   # it's blocked by the pending transaction.
   -Starting replication from postgres|$fail_pattern
 
-  ??[debug] Starting replication client for stack single_stack
-  ??[notice] Acquiring lock from postgres with name electric_slot_integration
-  ??[notice] Lock acquired from postgres with name electric_slot_integration
+  ??[debug] Starting replication client
+  ?slot_name=electric_slot_integration.*\[notice\] Acquiring lock from postgres
+  ?slot_name=electric_slot_integration.*\[notice\] Lock acquired from postgres
   ??[debug] ReplicationClient step: create_slot
 
 ## Verify that Electric's health check can report that the service is starting
@@ -54,7 +54,7 @@
 ## in the second Electric
 [shell electric_2]
   -Lock acquired from postgres|Starting replication from postgres|$fail_pattern
-  ??[notice] Acquiring lock from postgres with name electric_slot_integration
+  ?slot_name=electric_slot_integration.*\[notice\] Acquiring lock from postgres
 
 # The 2nd service should be in waiting state, ready to take over
 [shell client]
@@ -102,7 +102,7 @@
 [shell electric_2]
   -$fail_pattern
 
-  ??[notice] Lock acquired from postgres with name electric_slot_integration
+  ?slot_name=electric_slot_integration.*\[notice\] Lock acquired from postgres
   ??[notice] Starting replication from postgres
 
 # The 2nd Electric is now healthy and active

packages/electric-telemetry/lib/electric/telemetry/application_telemetry.ex

@@ -179,7 +179,7 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
   def cpu_utilization(_) do
     case :cpu_sup.util([:per_cpu]) do
       {:error, reason} ->
-        Logger.debug("Failed to collect CPU utilization: #{inspect(reason)}")
+        Logger.debug("Failed to collect CPU utilization", reason: reason)
 
       data ->
         {per_core_utilization, bare_values} =
@@ -304,7 +304,7 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
         |> Enum.reduce(%{}, fn probe, acc ->
           case apply(:cpu_sup, probe, []) do
             {:error, reason} ->
-              Logger.debug("Failed to collect system load #{probe}: #{inspect(reason)}")
+              Logger.debug("Failed to collect system load", probe: probe, reason: reason)
               acc
 
             value ->
@@ -333,9 +333,8 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
     mem_stats =
       cond do
         missing_system_memory_keys != [] ->
-          Logger.warning(
-            "Error gathering system memory stats: " <>
-              "missing data points #{Enum.join(missing_system_memory_keys, ", ")}"
+          Logger.warning("Error gathering system memory stats: missing data points",
+            data_points: missing_system_memory_keys
           )
 
           %{}
@@ -387,9 +386,8 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
       @resident_memory_keys
       |> Enum.reject(&Map.has_key?(system_memory, &1))
 
-    Logger.warning(
-      "Error gathering resident memory stats: " <>
-        "missing data points #{Enum.join(missing_keys, ", ")}"
+    Logger.warning("Error gathering resident memory stats: missing data points",
+      data_points: missing_keys
     )
 
     %{}
@@ -422,9 +420,8 @@ defmodule ElectricTelemetry.ApplicationTelemetry do
   defp swap_stats(_os_type, system_memory) do
     missing_swap_keys = Enum.reject(@required_swap_keys, &Map.has_key?(system_memory, &1))
 
-    Logger.warning(
-      "Error gathering system swap stats: " <>
-        "missing data points #{Enum.join(missing_swap_keys, ", ")}"
+    Logger.warning("Error gathering system swap stats: missing data points",
+      data_points: missing_swap_keys
     )
 
     %{}

packages/electric-telemetry/lib/electric/telemetry/call_home_reporter.ex

@@ -157,7 +157,8 @@ defmodule ElectricTelemetry.CallHomeReporter do
       rescue
         e ->
           Logger.warning(
-            "Reporter function failed while trying to send telemetry data.\nError: #{Exception.format(:error, e, __STACKTRACE__)}"
+            "Reporter function failed while trying to send telemetry data",
+            error: Exception.format(:error, e, __STACKTRACE__)
           )
 
           state

packages/electric-telemetry/lib/electric/telemetry/system_monitor.ex

@@ -38,8 +38,11 @@ defmodule ElectricTelemetry.SystemMonitor do
   def handle_info({:monitor, gc_pid, :long_gc, info}, state) do
     type = proc_type(gc_pid)
 
-    Logger.debug(
-      "Long GC detected for pid #{inspect(gc_pid)} (#{inspect(type)}): took #{Keyword.fetch!(info, :timeout)}ms. #{inspect(info, limit: :infinity)}"
+    Logger.debug("Long GC detected",
+      monitored_pid: gc_pid,
+      process_type: type,
+      timeout_ms: Keyword.fetch!(info, :timeout),
+      gc_info: info
     )
 
     :telemetry.execute(@vm_monitor_long_gc, Map.new(info), %{process_type: type})
@@ -48,8 +51,9 @@ defmodule ElectricTelemetry.SystemMonitor do
   end
 
   def handle_info({:monitor, port, :long_schedule, info}, state) when is_port(port) do
-    Logger.debug(
-      "Long schedule detected for port #{inspect(port)}, took #{Keyword.fetch!(info, :timeout)}ms"
+    Logger.debug("Long schedule detected for port",
+      port: port,
+      timeout_ms: Keyword.fetch!(info, :timeout)
     )
 
     :telemetry.execute(@vm_monitor_long_schedule, Map.new(info), %{process_type: :port})
@@ -60,25 +64,17 @@ defmodule ElectricTelemetry.SystemMonitor do
   def handle_info({:monitor, pid, :long_schedule, info}, state) when is_pid(pid) do
     type = proc_type(pid)
 
-    Logger.debug(fn ->
-      locations =
+    Logger.debug("Long schedule detected for pid",
+      monitored_pid: pid,
+      process_type: type,
+      timeout_ms: Keyword.fetch!(info, :timeout),
+      locations:
         info
         |> Keyword.delete(:timeout)
         |> Map.new(fn {loc, {m, f, a}} when loc in [:in, :out] ->
           {loc, Exception.format_mfa(m, f, a)}
         end)
-
-      locs_str =
-        if map_size(locations) > 0 do
-          "; " <>
-            (locations |> Enum.map(fn {key, val} -> "#{key}: #{val}" end) |> Enum.join(", "))
-        else
-          ""
-        end
-
-      "Long schedule detected for pid #{inspect(pid)} (#{inspect(type)}), took #{Keyword.fetch!(info, :timeout)}ms" <>
-        locs_str
-    end)
+    )
 
     :telemetry.execute(@vm_monitor_long_schedule, %{timeout: Keyword.fetch!(info, :timeout)}, %{
       process_type: type
@@ -90,7 +86,7 @@ defmodule ElectricTelemetry.SystemMonitor do
   def handle_info({:monitor, pid, :long_message_queue, true}, state) do
     type = proc_type(pid)
 
-    Logger.debug("Long message queue detected for pid #{inspect(pid)} (#{inspect(type)})")
+    Logger.debug("Long message queue detected", monitored_pid: pid, process_type: type)
 
     log_long_message_queue_event(pid, type)
 
@@ -102,7 +98,7 @@ defmodule ElectricTelemetry.SystemMonitor do
   end
 
   def handle_info({:monitor, pid, :long_message_queue, false}, state) do
-    Logger.debug("Long message queue no longer detected for pid #{inspect(pid)}")
+    Logger.debug("Long message queue no longer detected", monitored_pid: pid)
 
     {:noreply, %{state | long_message_queue_pids: Map.delete(state.long_message_queue_pids, pid)}}
   end