Add connection draining for cluster load balancing

Implements automatic connection draining to balance load across cluster nodes:
- ConnectionMonitor tracks CPU usage and connection counts across nodes
- Triggers draining when node is overloaded (>70% CPU or 2x avg connections)
- LiveViews gracefully disconnect and reconnect to less loaded nodes
- Helps prevent single pod from handling all WebSocket connections

Also fixes Mic-E packet parsing to properly extract altitude data and
remove telemetry suffixes from comments.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

CHANGELOG.md

@@ -16,6 +16,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Generated column `has_weather` with trigger for query optimization
 - Region and data_type composite indexes for filtering
 - BRIN indexes consideration for time-series data
+- Connection draining and load balancing for clustered deployments
+  - Monitors CPU usage and connection counts across cluster nodes
+  - Automatically drains connections when node is overloaded (>70% CPU or 2x average connections)
+  - Gracefully reconnects clients to less loaded nodes
+  - Improves overall cluster stability and performance
 
 ### Changed
 - Query performance improved by 50-90% for common operations
@@ -26,6 +31,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - PostgreSQL notify trigger now sends all required fields for info page updates
 - Info page real-time updates now display all packet details correctly
 - Enhanced PostgreSQL notify to include complete packet data (device info, weather data, SSIDs, etc.)
+- Mic-E packet comment parsing now properly extracts altitude data and removes telemetry suffixes
+  - Altitude data (e.g., "6;}" = 218 feet) is now parsed and stored
+  - Telemetry markers (_%...) are removed from comments
+  - Non-human-readable encoded data is no longer displayed as comments
 
 ## [0.2.0] - 2025-07-26
 

assets/js/app.js

@@ -321,6 +321,19 @@ topbar.config({ barColors: { 0: "#29d" }, shadowColor: "rgba(0, 0, 0, .3)" });
 window.addEventListener("phx:page-loading-start", (_info) => topbar.show(100));
 window.addEventListener("phx:page-loading-stop", (_info) => topbar.hide());
 
+// Handle connection draining reconnect events
+window.addEventListener("phx:reconnect", (e) => {
+  const delay = e.detail.delay || 1000;
+  console.log(`[LiveSocket] Reconnecting in ${delay}ms due to connection draining...`);
+  setTimeout(() => {
+    // Disconnect and reconnect to potentially land on a different server
+    liveSocket.disconnect();
+    setTimeout(() => {
+      liveSocket.connect();
+    }, 100);
+  }, delay);
+});
+
 // connect if there are any LiveViews on the page
 liveSocket.connect();
 

lib/aprsme/application.ex

@@ -140,7 +140,8 @@ defmodule Aprsme.Application do
             Aprsme.DynamicSupervisor,
             Aprsme.Cluster.LeaderElection,
             Aprsme.Cluster.ConnectionManager,
-            Aprsme.Cluster.PacketReceiver
+            Aprsme.Cluster.PacketReceiver,
+            Aprsme.ConnectionMonitor
           ]
         end
 

lib/aprsme/connection_monitor.ex

@@ -0,0 +1,239 @@
+defmodule Aprsme.ConnectionMonitor do
+  @moduledoc """
+  Monitors system load and LiveView connections to implement connection draining
+  when load is imbalanced across cluster nodes.
+  """
+  use GenServer
+
+  require Logger
+
+  @check_interval to_timeout(second: 30)
+  # Start draining if CPU usage > 70%
+  @cpu_threshold 0.7
+  # Drain if this node has 2x more connections than average
+  @connection_imbalance_ratio 2.0
+  # Drain 10% of connections when triggered
+  @drain_percentage 0.1
+
+  defstruct [
+    :node_stats,
+    :local_connections,
+    :draining,
+    :last_check
+  ]
+
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl true
+  def init(_opts) do
+    # Only start if clustering is enabled
+    if cluster_enabled?() do
+      schedule_check()
+
+      {:ok,
+       %__MODULE__{
+         node_stats: %{},
+         local_connections: 0,
+         draining: false,
+         last_check: System.monotonic_time(:second)
+       }}
+    else
+      :ignore
+    end
+  end
+
+  @doc """
+  Register a new LiveView connection
+  """
+  def register_connection do
+    if cluster_enabled?() do
+      GenServer.cast(__MODULE__, :register_connection)
+    end
+  end
+
+  @doc """
+  Unregister a LiveView connection
+  """
+  def unregister_connection do
+    if cluster_enabled?() do
+      GenServer.cast(__MODULE__, :unregister_connection)
+    end
+  end
+
+  @doc """
+  Check if this node should accept new connections
+  """
+  def accepting_connections? do
+    if cluster_enabled?() do
+      GenServer.call(__MODULE__, :accepting_connections?)
+    else
+      true
+    end
+  end
+
+  @doc """
+  Get current node statistics
+  """
+  def get_stats do
+    if cluster_enabled?() do
+      GenServer.call(__MODULE__, :get_stats)
+    else
+      %{connections: 0, cpu: 0.0, memory: 0.0}
+    end
+  end
+
+  @impl true
+  def handle_cast(:register_connection, state) do
+    {:noreply, %{state | local_connections: state.local_connections + 1}}
+  end
+
+  @impl true
+  def handle_cast(:unregister_connection, state) do
+    {:noreply, %{state | local_connections: max(0, state.local_connections - 1)}}
+  end
+
+  @impl true
+  def handle_call(:accepting_connections?, _from, state) do
+    {:reply, not state.draining, state}
+  end
+
+  @impl true
+  def handle_call(:get_stats, _from, state) do
+    stats = %{
+      connections: state.local_connections,
+      cpu: get_cpu_usage(),
+      memory: get_memory_usage(),
+      draining: state.draining
+    }
+
+    {:reply, stats, state}
+  end
+
+  @impl true
+  def handle_info(:check_load, state) do
+    new_state =
+      state
+      |> gather_cluster_stats()
+      |> analyze_load()
+      |> maybe_trigger_draining()
+
+    schedule_check()
+    {:noreply, new_state}
+  end
+
+  defp gather_cluster_stats(state) do
+    # Get stats from all nodes
+    nodes = [Node.self() | Node.list()]
+
+    node_stats =
+      Enum.reduce(nodes, %{}, fn node, acc ->
+        stats = :rpc.call(node, __MODULE__, :get_stats, [])
+
+        case stats do
+          {:badrpc, _} -> acc
+          stats -> Map.put(acc, node, stats)
+        end
+      end)
+
+    %{state | node_stats: node_stats}
+  end
+
+  defp analyze_load(state) do
+    local_stats = Map.get(state.node_stats, Node.self(), %{})
+    cpu_usage = Map.get(local_stats, :cpu, 0.0)
+
+    # Calculate average connections across cluster
+    total_connections = Enum.sum(for {_, stats} <- state.node_stats, do: Map.get(stats, :connections, 0))
+    node_count = map_size(state.node_stats)
+    avg_connections = if node_count > 0, do: total_connections / node_count, else: 0
+
+    # Determine if we should be draining
+    should_drain =
+      cond do
+        # High CPU usage
+        cpu_usage > @cpu_threshold ->
+          Logger.info("Node #{Node.self()} CPU usage high: #{Float.round(cpu_usage * 100, 1)}%")
+          true
+
+        # Too many connections compared to average
+        node_count > 1 and state.local_connections > avg_connections * @connection_imbalance_ratio ->
+          Logger.info(
+            "Node #{Node.self()} has #{state.local_connections} connections, avg: #{Float.round(avg_connections, 1)}"
+          )
+
+          true
+
+        # Otherwise, stop draining if we were
+        true ->
+          false
+      end
+
+    %{state | draining: should_drain}
+  end
+
+  defp maybe_trigger_draining(%{draining: true, local_connections: connections} = state) when connections > 0 do
+    # Calculate how many connections to drain
+    to_drain = round(connections * @drain_percentage)
+    # Drain 1-10 connections at a time
+    to_drain = max(1, min(to_drain, 10))
+
+    Logger.info("Draining #{to_drain} connections from node #{Node.self()}")
+
+    # Broadcast drain event to LiveViews
+    Phoenix.PubSub.broadcast(
+      Aprsme.PubSub,
+      "connection:drain:#{Node.self()}",
+      {:drain_connections, to_drain}
+    )
+
+    state
+  end
+
+  defp maybe_trigger_draining(state), do: state
+
+  defp get_cpu_usage do
+    # Get CPU usage from scheduler utilization
+    # Use Erlang's cpu_sup if available, otherwise estimate from scheduler utilization
+    case :cpu_sup.util() do
+      {:ok, busy, _, _} ->
+        busy / 100.0
+
+      _ ->
+        # Fallback to scheduler wall time
+        :scheduler_wall_time_all
+        |> :erlang.statistics()
+        |> Enum.map(fn {_, active, total} ->
+          if total > 0, do: active / total, else: 0.0
+        end)
+        |> Enum.sum()
+        |> Kernel./(System.schedulers_online())
+    end
+  rescue
+    _ -> 0.0
+  end
+
+  defp get_memory_usage do
+    # Get memory usage as a percentage
+    mem_data = :erlang.memory()
+    total = Keyword.get(mem_data, :total, 0)
+    system = Keyword.get(mem_data, :system, 0)
+
+    if system > 0 do
+      total / system
+    else
+      0.0
+    end
+  rescue
+    _ -> 0.0
+  end
+
+  defp schedule_check do
+    Process.send_after(self(), :check_load, @check_interval)
+  end
+
+  defp cluster_enabled? do
+    Application.get_env(:aprsme, :cluster_enabled, false)
+  end
+end

lib/aprsme_web/live/map_live/index.ex

@@ -94,33 +94,51 @@ defmodule AprsmeWeb.MapLive.Index do
   end
 
   defp do_setup_subscriptions(socket, true) do
-    # Generate a unique client ID for this LiveView instance
+    # Check if we should accept new connections
-    client_id = "liveview_#{:erlang.phash2(self())}"
+    if Application.get_env(:aprsme, :cluster_enabled, false) and
+         not Aprsme.ConnectionMonitor.accepting_connections?() do
+      # Redirect to another node or show message
+      socket
+      |> put_flash(:info, "This server is currently at capacity. Please try again in a moment.")
+      |> push_event("redirect_to_least_loaded", %{})
+      |> assign(:connection_draining, true)
+    else
+      # Generate a unique client ID for this LiveView instance
+      client_id = "liveview_#{:erlang.phash2(self())}"
 
-    # Register with spatial PubSub (will get viewport later)
+      # Register with connection monitor
-    # Start with a default viewport that will be updated when we get actual bounds
+      if Application.get_env(:aprsme, :cluster_enabled, false) do
-    default_bounds = %{north: 90.0, south: -90.0, east: 180.0, west: -180.0}
+        Aprsme.ConnectionMonitor.register_connection()
-    {:ok, spatial_topic} = Aprsme.SpatialPubSub.register_viewport(client_id, default_bounds)
+        # Subscribe to drain events for this node
+        Phoenix.PubSub.subscribe(Aprsme.PubSub, "connection:drain:#{Node.self()}")
+      end
 
-    # Subscribe to the spatial topic for this client
+      # Register with spatial PubSub (will get viewport later)
-    Phoenix.PubSub.subscribe(Aprsme.PubSub, spatial_topic)
+      # Start with a default viewport that will be updated when we get actual bounds
+      default_bounds = %{north: 90.0, south: -90.0, east: 180.0, west: -180.0}
+      {:ok, spatial_topic} = Aprsme.SpatialPubSub.register_viewport(client_id, default_bounds)
 
-    # Still subscribe to bad packets (they don't have location)
+      # Subscribe to the spatial topic for this client
-    Phoenix.PubSub.subscribe(Aprsme.PubSub, "bad_packets")
+      Phoenix.PubSub.subscribe(Aprsme.PubSub, spatial_topic)
 
-    # Subscribe to deployment events
+      # Still subscribe to bad packets (they don't have location)
-    Phoenix.PubSub.subscribe(Aprsme.PubSub, "deployment_events")
+      Phoenix.PubSub.subscribe(Aprsme.PubSub, "bad_packets")
 
-    # Subscribe to StreamingPacketsPubSub with initial bounds
+      # Subscribe to deployment events
-    Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), default_bounds)
+      Phoenix.PubSub.subscribe(Aprsme.PubSub, "deployment_events")
 
-    Process.send_after(self(), :cleanup_old_packets, 60_000)
+      # Subscribe to StreamingPacketsPubSub with initial bounds
-    # Schedule UI update timer to refresh "time ago" display every 30 seconds
+      Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), default_bounds)
-    Process.send_after(self(), :update_time_display, 30_000)
 
-    socket
+      Process.send_after(self(), :cleanup_old_packets, 60_000)
-    |> assign(:spatial_client_id, client_id)
+      # Schedule UI update timer to refresh "time ago" display every 30 seconds
-    |> assign(:spatial_topic, spatial_topic)
+      Process.send_after(self(), :update_time_display, 30_000)
+
+      socket
+      |> assign(:spatial_client_id, client_id)
+      |> assign(:spatial_topic, spatial_topic)
+      |> assign(:connection_draining, false)
+    end
   end
 
   defp do_setup_subscriptions(socket, false), do: socket
@@ -682,6 +700,19 @@ defmodule AprsmeWeb.MapLive.Index do
     {:noreply, assign(socket, :deployed_at, deployed_at)}
   end
 
+  def handle_info({:drain_connections, to_drain}, socket) do
+    # Check if this connection should be drained
+    # Use a random selection to determine if this connection should disconnect
+    if :rand.uniform(100) <= to_drain * 10 do
+      # Gracefully disconnect this client
+      socket
+      |> put_flash(:info, "Server load balancing in progress. Reconnecting...")
+      |> push_event("reconnect", %{delay: :rand.uniform(5000)})
+    else
+      {:noreply, socket}
+    end
+  end
+
   def handle_info({:load_rf_path_station_packets, stations}, socket) do
     # Load the most recent packet for each RF path station
     station_packets =
@@ -1543,6 +1574,12 @@ defmodule AprsmeWeb.MapLive.Index do
 
   @impl true
   def terminate(_reason, socket) do
+    # Unregister from connection monitor
+    if Application.get_env(:aprsme, :cluster_enabled, false) and
+         not socket.assigns[:connection_draining] do
+      Aprsme.ConnectionMonitor.unregister_connection()
+    end
+
     # Cleanup spatial registration if we have a client ID
     if socket.assigns[:spatial_client_id] do
       Aprsme.SpatialPubSub.unregister_client(socket.assigns.spatial_client_id)