%%%------------------------------------------------------------------- %%% @doc Macula UDP Multicast Gossip Cluster Strategy. %%% %%% Zero-config cluster formation using UDP multicast for LAN discovery. %%% Equivalent to libcluster's Cluster.Strategy.Gossip. %%% %%% == How It Works == %%% %%% Each node: %%% 1. Joins a multicast group (default 230.1.1.251) %%% 2. Periodically broadcasts its node name %%% 3. Listens for broadcasts from other nodes %%% 4. Connects to discovered nodes via Erlang distribution %%% %%% == Configuration == %%% %%% Start with options: %%% %%% ``` %%% {ok, _Pid} = macula_cluster_gossip:start_link(#{ %%% multicast_addr => {230, 1, 1, 251}, %% Multicast group %%% port => 45892, %% UDP port %%% broadcast_interval => 1500, %% ms between broadcasts %%% multicast_ttl => 1, %% TTL (1 = same subnet) %%% secret => <<"my-cluster-secret">> %% Optional: filter by secret %%% }). %%% ''' %%% %%% Or from environment variables: %%% %%% ``` %%% %% MACULA_GOSSIP_ADDR=230.1.1.251 %%% %% MACULA_GOSSIP_PORT=45892 %%% %% MACULA_GOSSIP_SECRET=my-cluster-secret %%% {ok, _Pid} = macula_cluster_gossip:start_link(#{}). %%% ''' %%% %%% == Network Requirements == %%% %%% - Nodes must be on the same multicast-enabled network %%% - UDP port must be open in firewalls %%% - Multicast must be enabled on network interfaces %%% - For Docker, use `network_mode: host` or macvlan %%% %%% == Callbacks == %%% %%% Register a callback to receive cluster events: %%% %%% ``` %%% {ok, _Pid} = macula_cluster_gossip:start_link(#{ %%% callback => self() %% PID or {Module, Function} %%% }). %%% %% Receives: {macula_cluster, nodeup, Node} %%% %% Receives: {macula_cluster, nodedown, Node} %%% ''' %%% %%% @copyright 2026 Macula.io Apache-2.0 %%% @end %%%------------------------------------------------------------------- -module(macula_cluster_gossip). -behaviour(gen_server). -include_lib("kernel/include/logger.hrl"). %% API -export([ start_link/0, start_link/1, stop/0, stop/1, get_discovered/0, get_discovered/1, get_connected/0, get_connected/1, broadcast_now/0, broadcast_now/1 ]). %% gen_server callbacks -export([ init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2 ]). -define(SERVER, ?MODULE). -define(DEFAULT_MULTICAST_ADDR, {230, 1, 1, 251}). -define(DEFAULT_PORT, 45892). -define(DEFAULT_BROADCAST_INTERVAL, 1500). -define(DEFAULT_MULTICAST_TTL, 1). -define(HEARTBEAT_PREFIX, <<"MACULA_GOSSIP:">>). -define(HEARTBEAT_PREFIX_LEN, 14). %% Length of "MACULA_GOSSIP:" -record(state, { %% UDP socket for multicast socket :: inet:socket() | undefined, %% Multicast address multicast_addr :: inet:ip4_address(), %% UDP port port :: inet:port_number(), %% Broadcast interval in ms broadcast_interval :: pos_integer(), %% Multicast TTL multicast_ttl :: non_neg_integer(), %% Optional secret for filtering secret :: binary() | undefined, %% Discovered nodes (may not be connected yet) discovered :: sets:set(atom()), %% Currently connected nodes connected :: sets:set(atom()), %% Broadcast timer reference broadcast_timer :: reference() | undefined, %% Callback for cluster events callback :: pid() | {module(), atom()} | undefined }). %%%=================================================================== %%% API %%%=================================================================== %% @doc Start the gossip cluster strategy with default options. -spec start_link() -> {ok, pid()} | {error, term()}. start_link() -> start_link(#{}). %% @doc Start the gossip cluster strategy with options. %% %% Options: %% - multicast_addr: IPv4 multicast address (default {230, 1, 1, 251}) %% - port: UDP port (default 45892) %% - broadcast_interval: Milliseconds between broadcasts (default 1500) %% - multicast_ttl: Multicast TTL, 1 = same subnet (default 1) %% - secret: Optional binary secret for filtering broadcasts %% - callback: PID or {Module, Function} to receive cluster events -spec start_link(map()) -> {ok, pid()} | {error, term()}. start_link(Opts) -> gen_server:start_link({local, ?SERVER}, ?MODULE, Opts, []). %% @doc Stop the gossip cluster strategy. -spec stop() -> ok. stop() -> stop(?SERVER). %% @doc Stop a named gossip cluster strategy. -spec stop(atom() | pid()) -> ok. stop(NameOrPid) -> gen_server:stop(NameOrPid). %% @doc Get the list of discovered nodes. -spec get_discovered() -> [atom()]. get_discovered() -> get_discovered(?SERVER). %% @doc Get discovered nodes from a named instance. -spec get_discovered(atom() | pid()) -> [atom()]. get_discovered(NameOrPid) -> gen_server:call(NameOrPid, get_discovered). %% @doc Get the list of currently connected nodes. -spec get_connected() -> [atom()]. get_connected() -> get_connected(?SERVER). %% @doc Get connected nodes from a named instance. -spec get_connected(atom() | pid()) -> [atom()]. get_connected(NameOrPid) -> gen_server:call(NameOrPid, get_connected). %% @doc Force an immediate broadcast. -spec broadcast_now() -> ok. broadcast_now() -> broadcast_now(?SERVER). %% @doc Force an immediate broadcast on a named instance. -spec broadcast_now(atom() | pid()) -> ok. broadcast_now(NameOrPid) -> gen_server:cast(NameOrPid, broadcast_now). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %% @private init(Opts) -> process_flag(trap_exit, true), %% Resolve configuration MulticastAddr = resolve_multicast_addr(Opts), Port = resolve_port(Opts), BroadcastInterval = maps:get(broadcast_interval, Opts, ?DEFAULT_BROADCAST_INTERVAL), MulticastTTL = maps:get(multicast_ttl, Opts, ?DEFAULT_MULTICAST_TTL), Secret = resolve_secret(Opts), Callback = maps:get(callback, Opts, undefined), %% Open UDP socket with multicast case open_multicast_socket(MulticastAddr, Port, MulticastTTL) of {ok, Socket} -> %% Subscribe to node events ok = net_kernel:monitor_nodes(true), State = #state{ socket = Socket, multicast_addr = MulticastAddr, port = Port, broadcast_interval = BroadcastInterval, multicast_ttl = MulticastTTL, secret = Secret, discovered = sets:new([{version, 2}]), connected = sets:new([{version, 2}]), callback = Callback }, ?LOG_INFO("[macula_cluster_gossip] Started on ~p:~p (TTL=~p)", [MulticastAddr, Port, MulticastTTL]), %% Start broadcasting immediately self() ! broadcast, {ok, State}; {error, Reason} -> ?LOG_ERROR("[macula_cluster_gossip] Failed to open socket: ~p", [Reason]), {stop, {socket_open_failed, Reason}} end. %% @private handle_call(get_discovered, _From, State) -> {reply, sets:to_list(State#state.discovered), State}; handle_call(get_connected, _From, State) -> {reply, sets:to_list(State#state.connected), State}; handle_call(_Request, _From, State) -> {reply, {error, unknown_request}, State}. %% @private handle_cast(broadcast_now, State) -> do_broadcast(State), {noreply, State}; handle_cast(_Msg, State) -> {noreply, State}. %% @private %% Periodic broadcast handle_info(broadcast, State) -> do_broadcast(State), Timer = schedule_broadcast(State#state.broadcast_interval), {noreply, State#state{broadcast_timer = Timer}}; %% Incoming UDP packet handle_info({udp, Socket, _SrcIP, _SrcPort, Packet}, #state{socket = Socket} = State) -> NewState = handle_gossip_packet(Packet, State), {noreply, NewState}; %% Node joined the cluster handle_info({nodeup, Node}, State) -> case sets:is_element(Node, State#state.discovered) of true -> ?LOG_INFO("[macula_cluster_gossip] Node ~p connected", [Node]), NewConnected = sets:add_element(Node, State#state.connected), notify_callback(State#state.callback, nodeup, Node), {noreply, State#state{connected = NewConnected}}; false -> %% Not a node we discovered {noreply, State} end; %% Node left the cluster handle_info({nodedown, Node}, State) -> case sets:is_element(Node, State#state.connected) of true -> ?LOG_WARNING("[macula_cluster_gossip] Node ~p disconnected", [Node]), NewConnected = sets:del_element(Node, State#state.connected), notify_callback(State#state.callback, nodedown, Node), {noreply, State#state{connected = NewConnected}}; false -> {noreply, State} end; handle_info(_Info, State) -> {noreply, State}. %% @private terminate(_Reason, State) -> cancel_timer(State#state.broadcast_timer), catch net_kernel:monitor_nodes(false), close_socket(State#state.socket), ok. %%%=================================================================== %%% Internal Functions - Socket Management %%%=================================================================== %% @private Open multicast UDP socket -spec open_multicast_socket(inet:ip4_address(), inet:port_number(), non_neg_integer()) -> {ok, inet:socket()} | {error, term()}. open_multicast_socket(MulticastAddr, Port, TTL) -> SocketOpts = [ binary, {active, true}, {reuseaddr, true}, {multicast_ttl, TTL}, {multicast_loop, true}, %% Receive own broadcasts (useful for testing) {add_membership, {MulticastAddr, {0, 0, 0, 0}}} ], gen_udp:open(Port, SocketOpts). %% @private Close socket -spec close_socket(inet:socket() | undefined) -> ok. close_socket(undefined) -> ok; close_socket(Socket) -> gen_udp:close(Socket), ok. %%%=================================================================== %%% Internal Functions - Gossip Protocol %%%=================================================================== %% @private Send a broadcast with our node name -spec do_broadcast(#state{}) -> ok. do_broadcast(#state{socket = undefined}) -> ok; do_broadcast(#state{socket = Socket, multicast_addr = Addr, port = Port, secret = Secret}) -> NodeName = atom_to_binary(node(), utf8), Payload = build_gossip_payload(NodeName, Secret), case gen_udp:send(Socket, Addr, Port, Payload) of ok -> ?LOG_DEBUG("[macula_cluster_gossip] Broadcast sent: ~p", [node()]), ok; {error, Reason} -> ?LOG_WARNING("[macula_cluster_gossip] Broadcast failed: ~p", [Reason]), ok end. %% @private Build gossip payload -spec build_gossip_payload(binary(), binary() | undefined) -> binary(). build_gossip_payload(NodeName, undefined) -> <>; build_gossip_payload(NodeName, Secret) -> %% Include HMAC for authentication when secret is set %% Use "|" as delimiter since ":" appears in node names Data = <>, HMAC = crypto:mac(hmac, sha256, Secret, Data), <>. %% @private Handle incoming gossip packet -spec handle_gossip_packet(binary(), #state{}) -> #state{}. handle_gossip_packet(Packet, State) -> case parse_gossip_packet(Packet, State#state.secret) of {ok, NodeName} -> handle_discovered_node(NodeName, State); {error, _Reason} -> State end. %% @private Parse gossip packet -spec parse_gossip_packet(binary(), binary() | undefined) -> {ok, atom()} | {error, term()}. parse_gossip_packet(Packet, undefined) -> %% No secret, just extract node name Prefix = ?HEARTBEAT_PREFIX, PrefixLen = ?HEARTBEAT_PREFIX_LEN, case Packet of <> -> %% May have HMAC suffix, strip it NodeNameBin = strip_hmac(Rest), {ok, binary_to_atom(NodeNameBin, utf8)}; _ -> {error, invalid_prefix} end; parse_gossip_packet(Packet, Secret) -> %% Verify HMAC - uses "|" as delimiter since ":" appears in node names verify_gossip_hmac(binary:split(Packet, <<"|">>), Secret). verify_gossip_hmac([Data, HexHMAC], Secret) -> ExpectedHMAC = crypto:mac(hmac, sha256, Secret, Data), ReceivedHMAC = binary:decode_hex(HexHMAC), gossip_hmac_result(crypto:hash_equals(ExpectedHMAC, ReceivedHMAC), Data); verify_gossip_hmac(_, _Secret) -> %% No HMAC but we expect one - could be from a node without secret %% Depending on policy, either accept or reject {error, no_hmac}. gossip_hmac_result(true, Data) -> Prefix = ?HEARTBEAT_PREFIX, PrefixLen = ?HEARTBEAT_PREFIX_LEN, <> = Data, {ok, binary_to_atom(NodeNameBin, utf8)}; gossip_hmac_result(false, _Data) -> {error, hmac_mismatch}. %% @private Strip HMAC suffix if present (uses "|" as delimiter) -spec strip_hmac(binary()) -> binary(). strip_hmac(Bin) -> case binary:split(Bin, <<"|">>) of [NodeName, _HMAC] -> NodeName; [NodeName] -> NodeName end. %% @private Handle a discovered node -spec handle_discovered_node(atom(), #state{}) -> #state{}. handle_discovered_node(Node, State) when Node =:= node() -> %% Ignore our own broadcast State; handle_discovered_node(Node, State) -> case sets:is_element(Node, State#state.discovered) of true -> %% Already known, try to connect if not connected try_connect(Node, State); false -> ?LOG_INFO("[macula_cluster_gossip] Discovered new node: ~p", [Node]), NewDiscovered = sets:add_element(Node, State#state.discovered), try_connect(Node, State#state{discovered = NewDiscovered}) end. %% @private Try to connect to a node -spec try_connect(atom(), #state{}) -> #state{}. try_connect(Node, State) -> try_connect_unconnected(sets:is_element(Node, State#state.connected), Node, State). try_connect_unconnected(true, _Node, State) -> %% Already connected State; try_connect_unconnected(false, Node, State) -> ?LOG_DEBUG("[macula_cluster_gossip] Attempting connection to ~p", [Node]), connect_result(net_kernel:connect_node(Node), Node, State). connect_result(true, Node, State) -> ?LOG_INFO("[macula_cluster_gossip] Connected to ~p", [Node]), NewConnected = sets:add_element(Node, State#state.connected), notify_callback(State#state.callback, nodeup, Node), State#state{connected = NewConnected}; connect_result(false, Node, State) -> ?LOG_DEBUG("[macula_cluster_gossip] Failed to connect to ~p", [Node]), State; connect_result(ignored, _Node, State) -> ?LOG_WARNING("[macula_cluster_gossip] net_kernel not running"), State. %%%=================================================================== %%% Internal Functions - Configuration %%%=================================================================== %% @private Resolve multicast address from options or env -spec resolve_multicast_addr(map()) -> inet:ip4_address(). resolve_multicast_addr(Opts) -> case maps:get(multicast_addr, Opts, undefined) of undefined -> parse_env_addr(); Addr when is_tuple(Addr) -> Addr end. %% @private Parse multicast address from env var -spec parse_env_addr() -> inet:ip4_address(). parse_env_addr() -> parse_env_addr(os:getenv("MACULA_GOSSIP_ADDR")). parse_env_addr(false) -> ?DEFAULT_MULTICAST_ADDR; parse_env_addr("") -> ?DEFAULT_MULTICAST_ADDR; parse_env_addr(AddrStr) -> parsed_addr(inet:parse_ipv4_address(AddrStr)). parsed_addr({ok, Addr}) -> Addr; parsed_addr({error, _}) -> ?DEFAULT_MULTICAST_ADDR. %% @private Resolve port from options or env -spec resolve_port(map()) -> inet:port_number(). resolve_port(Opts) -> case maps:get(port, Opts, undefined) of undefined -> parse_env_port(); Port when is_integer(Port) -> Port end. %% @private Parse port from env var -spec parse_env_port() -> inet:port_number(). parse_env_port() -> parse_port_from_env(os:getenv("MACULA_GOSSIP_PORT")). %% @private Parse port from env value parse_port_from_env(false) -> ?DEFAULT_PORT; parse_port_from_env("") -> ?DEFAULT_PORT; parse_port_from_env(PortStr) -> handle_port_int_parse(catch list_to_integer(PortStr)). %% @private Handle port parse result handle_port_int_parse({'EXIT', _}) -> ?DEFAULT_PORT; handle_port_int_parse(Port) when is_integer(Port) -> Port; handle_port_int_parse(_) -> ?DEFAULT_PORT. %% @private Resolve secret from options or env -spec resolve_secret(map()) -> binary() | undefined. resolve_secret(Opts) -> case maps:get(secret, Opts, undefined) of undefined -> parse_env_secret(); Secret when is_binary(Secret) -> Secret; Secret when is_list(Secret) -> list_to_binary(Secret) end. %% @private Parse secret from env var -spec parse_env_secret() -> binary() | undefined. parse_env_secret() -> case os:getenv("MACULA_GOSSIP_SECRET") of false -> undefined; "" -> undefined; Secret -> list_to_binary(Secret) end. %%%=================================================================== %%% Internal Functions - Utilities %%%=================================================================== %% @private Schedule next broadcast -spec schedule_broadcast(pos_integer()) -> reference(). schedule_broadcast(Interval) -> erlang:send_after(Interval, self(), broadcast). %% @private Cancel timer if defined -spec cancel_timer(reference() | undefined) -> ok. cancel_timer(undefined) -> ok; cancel_timer(Timer) -> erlang:cancel_timer(Timer), ok. %% @private Notify callback of cluster event -spec notify_callback(pid() | {module(), atom()} | undefined, nodeup | nodedown, atom()) -> ok. notify_callback(undefined, _Event, _Node) -> ok; notify_callback(Pid, Event, Node) when is_pid(Pid) -> Pid ! {macula_cluster, Event, Node}, ok; notify_callback({Module, Function}, Event, Node) -> _ = catch Module:Function(Event, Node), ok.