diff --git a/CLAUDE.md b/CLAUDE.md index 7f60934..f27666e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -127,6 +127,7 @@ The application supports Kubernetes deployment with manifests in `k8s/` director The app is deployed in a k3s cluster with the following structure: - **App name**: `aprs` - **Namespace**: `aprs` +- **K8s manifests location**: `~/dev/infra/clusters/aprs/` Common kubectl commands for debugging: ```bash @@ -150,4 +151,37 @@ kubectl rollout status deployment/aprs -n aprs # Execute commands in the pod kubectl exec -it deployment/aprs -n aprs -- /app/bin/aprsme remote -``` \ No newline at end of file + +# Check cluster membership +kubectl exec -it -n aprs -- /app/bin/aprsme eval "Node.list()" + +# Check leader status +kubectl exec -it -n aprs -- /app/bin/aprsme eval "Aprsme.Cluster.LeaderElection.is_leader?()" +``` + +### Clustering Architecture + +The application uses distributed Erlang clustering to ensure only one APRS-IS connection across multiple replicas: + +1. **Leader Election**: Uses `:global` registry for distributed leader election + - Only the elected leader maintains the APRS-IS connection + - Automatic failover when leader goes down + - Leader election managed by `Aprsme.Cluster.LeaderElection` + +2. **Connection Management**: + - `Aprsme.Cluster.ConnectionManager` starts/stops APRS-IS based on leadership + - Uses `DynamicSupervisor` to manage connection lifecycle + - Prevents duplicate connections and packet processing + +3. **Kubernetes Configuration**: + - Headless service (`aprs-headless`) for node discovery + - DNS-based clustering via libcluster + - Environment variables: + - `CLUSTER_ENABLED=true` - Enables clustering + - `RELEASE_NODE` - Erlang node name + - `RELEASE_COOKIE` - Erlang distribution cookie + +4. **Deployment**: + - Default replicas: 3 (configurable in `aprs-deployment.yaml`) + - Only leader processes APRS packets + - All nodes serve web traffic \ No newline at end of file diff --git a/config/runtime.exs b/config/runtime.exs index 0f18793..ef10c93 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -46,6 +46,9 @@ if config_env() == :prod do host = System.get_env("PHX_HOST") || "example.com" port = String.to_integer(System.get_env("PORT") || "4000") + # Configure clustering based on environment + cluster_enabled = System.get_env("CLUSTER_ENABLED", "false") == "true" + # ## Configuring the mailer # # Configure Resend for email delivery @@ -116,6 +119,8 @@ if config_env() == :prod do "https://tile.openstreetmap.org" ] + config :aprsme, :cluster_enabled, cluster_enabled + # Optional: Set the default "from" email address config :aprsme, default_from_email: "w5isp@aprs.me" @@ -134,19 +139,35 @@ if config_env() == :prod do config :hammer, backend: {Hammer.Backend.ETS, [expiry_ms: 60_000 * 60 * 4, cleanup_interval_ms: 60_000 * 10]} - # Configure libcluster for Dokku clustering - config :libcluster, - topologies: [ - dokku: [ - strategy: Cluster.Strategy.Gossip, - config: [ - port: 45_892, - if_addr: "0.0.0.0", - multicast_addr: "230.1.1.1", - multicast_ttl: 1 + # Configure libcluster topology based on environment + if cluster_enabled do + config :libcluster, + topologies: [ + kubernetes: [ + strategy: Cluster.Strategy.Kubernetes.DNS, + config: [ + service: "aprs-headless", + application_name: "aprs", + namespace: System.get_env("POD_NAMESPACE", "aprs"), + polling_interval: 10_000 + ] ] ] - ] + else + # Keep existing Dokku configuration for non-Kubernetes deployments + config :libcluster, + topologies: [ + dokku: [ + strategy: Cluster.Strategy.Gossip, + config: [ + port: 45_892, + if_addr: "0.0.0.0", + multicast_addr: "230.1.1.1", + multicast_ttl: 1 + ] + ] + ] + end # # For this example you need include a HTTP client required by Swoosh API client. diff --git a/lib/aprsme/application.ex b/lib/aprsme/application.ex index 61fd775..331da6f 100644 --- a/lib/aprsme/application.ex +++ b/lib/aprsme/application.ex @@ -16,8 +16,6 @@ defmodule Aprsme.Application do config: %{metadata: [:file, :line]} }) - topologies = Application.get_env(:libcluster, :topologies) || [] - children = [ # Start the Telemetry supervisor AprsmeWeb.Telemetry, @@ -49,7 +47,8 @@ defmodule Aprsme.Application do # Start a worker by calling: Aprsme.Worker.start_link(arg) # {Aprsme.Worker, arg} {Registry, keys: :duplicate, name: Registry.PubSub, partitions: System.schedulers_online()}, - {Cluster.Supervisor, [topologies, [name: Aprsme.ClusterSupervisor]]}, + # Start clustering + Aprsme.Cluster.Topology, # Start Oban for background jobs {Oban, :aprsme |> Application.get_env(Oban, []) |> Keyword.put(:queues, default: 10, maintenance: 2)}, Aprsme.Presence, @@ -58,6 +57,7 @@ defmodule Aprsme.Application do Aprsme.PacketPipelineSupervisor ] + children = maybe_add_cluster_components(children) children = maybe_add_is_supervisor(children, Application.get_env(:aprsme, :env)) children = maybe_add_aprs_connection(children, Application.get_env(:aprsme, :env)) @@ -99,10 +99,30 @@ defmodule Aprsme.Application do :ok end + defp maybe_add_cluster_components(children) do + if Application.get_env(:aprsme, :cluster_enabled, false) do + children ++ + [ + # Dynamic supervisor for processes managed by cluster leader + Aprsme.DynamicSupervisor, + # Leader election process + Aprsme.Cluster.LeaderElection, + # Connection manager that starts/stops APRS-IS based on leadership + Aprsme.Cluster.ConnectionManager, + # Packet receiver for distributed packets on non-leader nodes + Aprsme.Cluster.PacketReceiver + ] + else + children + end + end + defp maybe_add_is_supervisor(children, env) do disable_connection = Application.get_env(:aprsme, :disable_aprs_connection, false) + cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false) - if env in [:prod, :dev] and not disable_connection do + # Only add Is.IsSupervisor directly if clustering is disabled + if env in [:prod, :dev] and not disable_connection and not cluster_enabled do children ++ [Aprsme.Is.IsSupervisor] else children diff --git a/lib/aprsme/cluster/connection_manager.ex b/lib/aprsme/cluster/connection_manager.ex new file mode 100644 index 0000000..8e3743c --- /dev/null +++ b/lib/aprsme/cluster/connection_manager.ex @@ -0,0 +1,87 @@ +defmodule Aprsme.Cluster.ConnectionManager do + @moduledoc """ + Manages the APRS-IS connection based on cluster leadership. + Only starts the connection when this node is elected as leader. + """ + use GenServer + + alias Aprsme.Is.IsSupervisor + + require Logger + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + @impl true + def init(_opts) do + # Subscribe to leadership changes + Phoenix.PubSub.subscribe(Aprsme.PubSub, "cluster:leadership") + + # Check initial leadership state + Process.send_after(self(), :check_initial_state, 1000) + + {:ok, %{connection_started: false}} + end + + @impl true + def handle_info(:check_initial_state, state) do + if Aprsme.Cluster.LeaderElection.is_leader?() do + Logger.info("This node is the leader, starting APRS-IS connection") + start_aprs_connection() + {:noreply, %{state | connection_started: true}} + else + Logger.info("This node is not the leader, APRS-IS connection will not be started") + {:noreply, state} + end + end + + @impl true + def handle_info({:leadership_change, node, became_leader}, state) do + cond do + became_leader and node == node() and not state.connection_started -> + Logger.info("This node became the leader, starting APRS-IS connection") + start_aprs_connection() + {:noreply, %{state | connection_started: true}} + + not became_leader and node == node() and state.connection_started -> + Logger.info("This node lost leadership, stopping APRS-IS connection") + stop_aprs_connection() + {:noreply, %{state | connection_started: false}} + + true -> + {:noreply, state} + end + end + + defp start_aprs_connection do + # Start the APRS-IS connection supervisor if not already started + case DynamicSupervisor.start_child( + Aprsme.DynamicSupervisor, + {IsSupervisor, []} + ) do + {:ok, _pid} -> + Logger.info("APRS-IS connection started successfully") + + {:error, {:already_started, _pid}} -> + Logger.info("APRS-IS connection was already running") + + {:error, reason} -> + Logger.error("Failed to start APRS-IS connection: #{inspect(reason)}") + end + end + + defp stop_aprs_connection do + # Find and terminate the APRS-IS connection supervisor + children = DynamicSupervisor.which_children(Aprsme.DynamicSupervisor) + + Enum.each(children, fn + {_, pid, _, [IsSupervisor]} when is_pid(pid) -> + Logger.info("Stopping APRS-IS connection supervisor") + DynamicSupervisor.terminate_child(Aprsme.DynamicSupervisor, pid) + + _ -> + :ok + end) + end +end diff --git a/lib/aprsme/cluster/leader_election.ex b/lib/aprsme/cluster/leader_election.ex new file mode 100644 index 0000000..ab9c42a --- /dev/null +++ b/lib/aprsme/cluster/leader_election.ex @@ -0,0 +1,106 @@ +defmodule Aprsme.Cluster.LeaderElection do + @moduledoc """ + Manages leader election for APRS-IS connection using distributed Erlang. + Only the elected leader will maintain the APRS-IS connection. + """ + use GenServer + + require Logger + + @election_key {:aprs_is_leader, __MODULE__} + @check_interval 5_000 + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + def is_leader? do + GenServer.call(__MODULE__, :is_leader?) + end + + def current_leader do + GenServer.call(__MODULE__, :current_leader) + end + + @impl true + def init(_opts) do + Logger.info("Starting leader election process") + + # Schedule initial election + Process.send_after(self(), :attempt_election, 100) + + # Schedule periodic checks + Process.send_after(self(), :check_leadership, @check_interval) + + {:ok, %{is_leader: false, leader_node: nil}} + end + + @impl true + def handle_info(:attempt_election, state) do + case :global.register_name(@election_key, self(), &resolve_conflict/3) do + :yes -> + Logger.info("Elected as APRS-IS connection leader on node #{node()}") + notify_leadership_change(true) + {:noreply, %{state | is_leader: true, leader_node: node()}} + + :no -> + leader_pid = :global.whereis_name(@election_key) + leader_node = if leader_pid, do: node(leader_pid) + Logger.info("Not elected as leader. Current leader is on node #{inspect(leader_node)}") + {:noreply, %{state | is_leader: false, leader_node: leader_node}} + end + end + + @impl true + def handle_info(:check_leadership, state) do + # Re-attempt election if we're not leader + if not state.is_leader do + Process.send_after(self(), :attempt_election, 100) + end + + # Schedule next check + Process.send_after(self(), :check_leadership, @check_interval) + + {:noreply, state} + end + + @impl true + def handle_call(:is_leader?, _from, state) do + {:reply, state.is_leader, state} + end + + @impl true + def handle_call(:current_leader, _from, state) do + {:reply, state.leader_node, state} + end + + @impl true + def terminate(reason, state) do + if state.is_leader do + Logger.info("Leader stepping down due to: #{inspect(reason)}") + :global.unregister_name(@election_key) + notify_leadership_change(false) + end + + :ok + end + + # Conflict resolution - prefer the process that's been running longer + defp resolve_conflict(_name, pid1, pid2) do + info1 = Process.info(pid1, [:registered_name, :current_function]) + info2 = Process.info(pid2, [:registered_name, :current_function]) + + Logger.debug("Resolving leader conflict between #{inspect(info1)} and #{inspect(info2)}") + + # Keep the first registered process + pid1 + end + + defp notify_leadership_change(became_leader) do + Phoenix.PubSub.broadcast( + Aprsme.PubSub, + "cluster:leadership", + {:leadership_change, node(), became_leader} + ) + end +end diff --git a/lib/aprsme/cluster/packet_distributor.ex b/lib/aprsme/cluster/packet_distributor.ex new file mode 100644 index 0000000..2e97eda --- /dev/null +++ b/lib/aprsme/cluster/packet_distributor.ex @@ -0,0 +1,38 @@ +defmodule Aprsme.Cluster.PacketDistributor do + @moduledoc """ + Distributes APRS packets from the leader node to all cluster members. + This ensures all nodes can serve real-time updates via LiveView while + only the leader maintains the APRS-IS connection. + """ + require Logger + + @pubsub_topic "cluster:packets" + + def distribute_packet(packet) do + # Only distribute if we're the leader + if Aprsme.Cluster.LeaderElection.is_leader?() do + # Broadcast to all nodes including self + Phoenix.PubSub.broadcast( + Aprsme.PubSub, + @pubsub_topic, + {:distributed_packet, packet} + ) + + Logger.debug("Distributed packet #{packet.raw} to cluster") + end + end + + def subscribe do + Phoenix.PubSub.subscribe(Aprsme.PubSub, @pubsub_topic) + end + + def handle_distributed_packet({:distributed_packet, packet}) do + # Broadcast to local LiveView clients + Aprsme.StreamingPacketsPubSub.broadcast_packet(packet) + + # Update packet store for LiveView + AprsmeWeb.MapLive.PacketStore.store_packet(packet) + + Logger.debug("Received distributed packet on node #{node()}") + end +end diff --git a/lib/aprsme/cluster/packet_receiver.ex b/lib/aprsme/cluster/packet_receiver.ex new file mode 100644 index 0000000..b762aac --- /dev/null +++ b/lib/aprsme/cluster/packet_receiver.ex @@ -0,0 +1,41 @@ +defmodule Aprsme.Cluster.PacketReceiver do + @moduledoc """ + Receives distributed packets from the cluster leader on non-leader nodes. + Ensures all nodes can serve real-time updates even though only the leader + processes APRS packets. + """ + use GenServer + + alias Aprsme.Cluster.PacketDistributor + + require Logger + + def start_link(opts) do + GenServer.start_link(__MODULE__, opts, name: __MODULE__) + end + + @impl true + def init(_opts) do + # Subscribe to distributed packets + PacketDistributor.subscribe() + + Logger.info("Started packet receiver on node #{node()}") + + {:ok, %{}} + end + + @impl true + def handle_info({:distributed_packet, packet}, state) do + # Only process if we're not the leader (leader already processed locally) + if !Aprsme.Cluster.LeaderElection.is_leader?() do + PacketDistributor.handle_distributed_packet({:distributed_packet, packet}) + end + + {:noreply, state} + end + + @impl true + def handle_info(_msg, state) do + {:noreply, state} + end +end diff --git a/lib/aprsme/cluster/topology.ex b/lib/aprsme/cluster/topology.ex new file mode 100644 index 0000000..70b10de --- /dev/null +++ b/lib/aprsme/cluster/topology.ex @@ -0,0 +1,43 @@ +defmodule Aprsme.Cluster.Topology do + @moduledoc """ + Cluster topology configuration for different environments. + """ + + def child_spec(opts) do + cluster_config = config() + + if cluster_config do + {Cluster.Supervisor, [cluster_config, opts]} + else + # Return a no-op spec when clustering is disabled + %{ + id: __MODULE__, + start: {__MODULE__, :start_link, [opts]}, + type: :worker + } + end + end + + def start_link(_opts), do: :ignore + + defp config do + if Application.get_env(:aprsme, :cluster_enabled, false) do + kubernetes_config() + end + end + + defp kubernetes_config do + [ + kubernetes: [ + strategy: Cluster.Strategy.Kubernetes, + config: [ + mode: :dns, + kubernetes_node_basename: "aprs", + kubernetes_selector: "app=aprs", + kubernetes_namespace: "aprs", + polling_interval: 10_000 + ] + ] + ] + end +end diff --git a/lib/aprsme/dynamic_supervisor.ex b/lib/aprsme/dynamic_supervisor.ex new file mode 100644 index 0000000..4cde145 --- /dev/null +++ b/lib/aprsme/dynamic_supervisor.ex @@ -0,0 +1,16 @@ +defmodule Aprsme.DynamicSupervisor do + @moduledoc """ + Dynamic supervisor for managing processes that may be started/stopped at runtime. + Used primarily for the APRS-IS connection which is managed by cluster leadership. + """ + use DynamicSupervisor + + def start_link(init_arg) do + DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__) + end + + @impl true + def init(_init_arg) do + DynamicSupervisor.init(strategy: :one_for_one) + end +end diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 780b92a..ffb7c42 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -251,6 +251,8 @@ defmodule Aprsme.PacketConsumer do # Broadcast packets asynchronously to avoid blocking defp broadcast_packets_async(packets) do Task.start(fn -> + cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false) + Enum.each(packets, fn packet_attrs -> # Convert back to a format suitable for broadcasting packet = %{ @@ -265,7 +267,13 @@ defmodule Aprsme.PacketConsumer do comment: packet_attrs[:comment] } - Aprsme.StreamingPacketsPubSub.broadcast_packet(packet) + if cluster_enabled do + # Use cluster distributor to broadcast to all nodes + Aprsme.Cluster.PacketDistributor.distribute_packet(packet) + else + # Normal single-node broadcasting + Aprsme.StreamingPacketsPubSub.broadcast_packet(packet) + end end) end) end