* src/ejabberd_node_groups.erl: Support for node tagging

* src/ejabberd_sup.erl: Likewise
* src/ejabberd_frontend_socket.erl: Use node tags to determine
backend nodes
* src/ejabberd_config.erl: Added node_type and cluster_nodes
options
* src/ejabberd_app.erl: Establish connections to the nodes from
the cluster_nodes option

* src/ejabberd_router.erl: Added balancing method option
* src/ejabberd_config.erl: Likewise

SVN Revision: 676

ChangeLog

@@ -1,3 +1,17 @@
+2006-11-07  Alexey Shchepin  <alexey@sevcom.net>
+
+	* src/ejabberd_node_groups.erl: Support for node tagging
+	* src/ejabberd_sup.erl: Likewise
+	* src/ejabberd_frontend_socket.erl: Use node tags to determine
+	backend nodes
+	* src/ejabberd_config.erl: Added node_type and cluster_nodes
+	options
+	* src/ejabberd_app.erl: Establish connections to the nodes from
+	the cluster_nodes option
+
+	* src/ejabberd_router.erl: Added balancing method option
+	* src/ejabberd_config.erl: Likewise
+
 2006-11-05  Mickael Remond  <mickael.remond@process-one.net>
 
 	* src/mod_private_odbc.erl: Support for MySQL and MSSQL.
@@ -13,8 +27,8 @@
 	* src/odbc/mysql.sql: Likewise.
 	* src/odbc/mssql.sql: Likewise.
 
-	* src/odbc/mysql: Database creation script should now be compliant with
+	* src/odbc/mysql: Database creation script should now be compliant
-	MySQL 4.0.x.
+	with MySQL 4.0.x.
 
 2006-10-29  Mickael Remond  <mickael.remond@process-one.net>
 
@@ -23,8 +37,8 @@
 
 2006-10-28  Mickael Remond  <mickael.remond@process-one.net>
 
-	* src/ejabberd.cfg.example: Changed the anonymous example a bit to work
+	* src/ejabberd.cfg.example: Changed the anonymous example a bit to
-	in most cases.
+	work in most cases.
 	* doc/guide.tex: Likewise.
 
 2006-10-28  Alexey Shchepin  <alexey@sevcom.net>

src/ejabberd_app.erl

@@ -27,6 +27,8 @@ start(normal, _Args) ->
     ejabberd_ctl:init(),
     gen_mod:start(),
     ejabberd_config:start(),
+    start(),
+    connect_nodes(),
     Sup = ejabberd_sup:start_link(),
     ejabberd_rdbms:start(),
     ejabberd_auth:start(),
@@ -35,7 +37,6 @@ start(normal, _Args) ->
     %eprof:start(),
     %eprof:profile([self()]),
     %fprof:trace(start, "/tmp/fprof"),
-    start(),
     load_modules(),
     Sup;
 start(_, _) ->
@@ -103,3 +104,15 @@ load_modules() ->
 	      end
       end, ?MYHOSTS).
 
+connect_nodes() ->
+    case ejabberd_config:get_local_option(cluster_nodes) of
+	undefined ->
+	    ok;
+	Nodes when is_list(Nodes) ->
+	    lists:foreach(fun(Node) ->
+				  net_kernel:connect_node(Node)
+			  end, Nodes)
+    end.
+
+
+

src/ejabberd_config.erl

@@ -115,6 +115,12 @@ process_term(Term, State) ->
 	    add_option(s2s_certfile, CertFile, State);
 	{domain_certfile, Domain, CertFile} ->
 	    add_option({domain_certfile, Domain}, CertFile, State);
+	{node_type, NodeType} ->
+	    add_option(node_type, NodeType, State);
+	{cluster_nodes, Nodes} ->
+	    add_option(cluster_nodes, Nodes, State);
+	{domain_balancing, Domain, Balancing} ->
+	    add_option({domain_balancing, Domain}, Balancing, State);
 	{Opt, Val} ->
 	    lists:foldl(fun(Host, S) -> process_host_term(Term, Host, S) end,
 			State, State#state.hosts)

src/ejabberd_frontend_socket.erl

@@ -13,7 +13,7 @@
 
 %% API
 -export([start/4,
-	 start_link/4,
+	 start_link/5,
 	 %connect/3,
 	 starttls/2,
 	 starttls/3,
@@ -41,9 +41,9 @@
 %% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
 %% Description: Starts the server
 %%--------------------------------------------------------------------
-start_link(Module, SockMod, Socket, Opts) ->
+start_link(Module, SockMod, Socket, Opts, Receiver) ->
     gen_server:start_link(?MODULE,
-			  [Module, SockMod, Socket, Opts], []).
+			  [Module, SockMod, Socket, Opts, Receiver], []).
 
 start(Module, SockMod, Socket, Opts) ->
     case Module:socket_type() of
@@ -60,8 +60,8 @@ start(Module, SockMod, Socket, Opts) ->
 		{error, _Reason} ->
 		    SockMod:close(Socket)
 	    end,
-	    gen_server:start(?MODULE,
+	    supervisor:start_child(ejabberd_frontend_socket_sup,
-			     [Module, SockMod, Socket, Opts, Receiver], []);
+				   [Module, SockMod, Socket, Opts, Receiver]);
 	raw ->
 	    %{ok, Pid} = Module:start({SockMod, Socket}, Opts),
 	    %case SockMod:controlling_process(Socket, Pid) of
@@ -129,8 +129,7 @@ peername(FsmRef) ->
 %% Description: Initiates the server
 %%--------------------------------------------------------------------
 init([Module, SockMod, Socket, Opts, Receiver]) ->
-    Nodes = mnesia:table_info(schema, disc_copies),
+    Node = ejabberd_node_groups:get_closest_node(backend),
-    Node = lists:nth(erlang:phash(now(), length(Nodes)), Nodes),
     {ok, Pid} =
 	rpc:call(Node, Module, start, [{?MODULE, self()}, Opts]),
     ejabberd_receiver:become_controller(Receiver, Pid),

src/ejabberd_node_groups.erl

@@ -0,0 +1,142 @@
+%%%----------------------------------------------------------------------
+%%% File    : ejabberd_node_groups.erl
+%%% Author  : Alexey Shchepin <alexey@process-one.net>
+%%% Purpose : Distributed named node groups based on pg2 module
+%%% Created :  1 Nov 2006 by Alexey Shchepin <alexey@process-one.net>
+%%% Id      : $Id$
+%%%----------------------------------------------------------------------
+
+-module(ejabberd_node_groups).
+-author('alexey@process-one.net').
+-vsn('$Revision$ ').
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0,
+	 join/1,
+	 leave/1,
+	 get_members/1,
+	 get_closest_node/1]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+	 terminate/2, code_change/3]).
+
+-record(state, {}).
+
+%%====================================================================
+%% API
+%%====================================================================
+%%--------------------------------------------------------------------
+%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
+%% Description: Starts the server
+%%--------------------------------------------------------------------
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+join(Name) ->
+    PG = {?MODULE, Name},
+    pg2:create(PG),
+    pg2:join(PG, whereis(?MODULE)).
+
+leave(Name) ->
+    PG = {?MODULE, Name},
+    pg2:leave(PG, whereis(?MODULE)).
+
+get_members(Name) ->
+    PG = {?MODULE, Name},
+    [node(P) || P <- pg2:get_members(PG)].
+
+get_closest_node(Name) ->
+    PG = {?MODULE, Name},
+    node(pg2:get_closest_pid(PG)).
+
+%%====================================================================
+%% gen_server callbacks
+%%====================================================================
+
+%%--------------------------------------------------------------------
+%% Function: init(Args) -> {ok, State} |
+%%                         {ok, State, Timeout} |
+%%                         ignore               |
+%%                         {stop, Reason}
+%% Description: Initiates the server
+%%--------------------------------------------------------------------
+init([]) ->
+    {FE, BE} =
+	case ejabberd_config:get_local_option(node_type) of
+	    frontend ->
+		{true, false};
+	    backend ->
+		{false, true};
+	    generic ->
+		{true, true};
+	    undefined ->
+		{true, true}
+	end,
+    if
+	FE ->
+	    join(frontend);
+	true ->
+	    ok
+    end,
+    if
+	BE ->
+	    join(backend);
+	true ->
+	    ok
+    end,
+    {ok, #state{}}.
+
+%%--------------------------------------------------------------------
+%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
+%%                                      {reply, Reply, State, Timeout} |
+%%                                      {noreply, State} |
+%%                                      {noreply, State, Timeout} |
+%%                                      {stop, Reason, Reply, State} |
+%%                                      {stop, Reason, State}
+%% Description: Handling call messages
+%%--------------------------------------------------------------------
+handle_call(_Request, _From, State) ->
+    Reply = ok,
+    {reply, Reply, State}.
+
+%%--------------------------------------------------------------------
+%% Function: handle_cast(Msg, State) -> {noreply, State} |
+%%                                      {noreply, State, Timeout} |
+%%                                      {stop, Reason, State}
+%% Description: Handling cast messages
+%%--------------------------------------------------------------------
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+%%--------------------------------------------------------------------
+%% Function: handle_info(Info, State) -> {noreply, State} |
+%%                                       {noreply, State, Timeout} |
+%%                                       {stop, Reason, State}
+%% Description: Handling all non call/cast messages
+%%--------------------------------------------------------------------
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+%%--------------------------------------------------------------------
+%% Function: terminate(Reason, State) -> void()
+%% Description: This function is called by a gen_server when it is about to
+%% terminate. It should be the opposite of Module:init/1 and do any necessary
+%% cleaning up. When it returns, the gen_server terminates with Reason.
+%% The return value is ignored.
+%%--------------------------------------------------------------------
+terminate(_Reason, _State) ->
+    ok.
+
+%%--------------------------------------------------------------------
+%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
+%% Description: Convert process state when code is changed
+%%--------------------------------------------------------------------
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%--------------------------------------------------------------------
+%%% Internal functions
+%%--------------------------------------------------------------------

src/ejabberd_router.erl

@@ -220,7 +220,6 @@ code_change(_OldVsn, State, _Extra) ->
 do_route(OrigFrom, OrigTo, OrigPacket) ->
     ?DEBUG("route~n\tfrom ~p~n\tto ~p~n\tpacket ~p~n",
 	   [OrigFrom, OrigTo, OrigPacket]),
-    LOrigDstDomain = OrigTo#jid.lserver,
     case ejabberd_hooks:run_fold(filter_packet,
 				 {OrigFrom, OrigTo, OrigPacket}, []) of
 	{From, To, Packet} ->
@@ -244,12 +243,25 @@ do_route(OrigFrom, OrigTo, OrigPacket) ->
 		Rs ->
 		    case [R || R <- Rs, node(R#route.pid) == node()] of
 			[] ->
-			    R = lists:nth(erlang:phash(now(), length(Rs)), Rs),
+			    Value = case ejabberd_config:get_local_option(
+					   {domain_balancing, LDstDomain}) of
+					source -> jlib:jid_tolower(From);
+					destination -> jlib:jid_tolower(To);
+					random -> now();
+					undefined -> now()
+				    end,
+			    R = lists:nth(erlang:phash(Value, length(Rs)), Rs),
 			    Pid = R#route.pid,
 			    Pid ! {route, From, To, Packet};
 			LRs ->
-			    LRs,
+			    Value = case ejabberd_config:get_local_option(
-			    R = lists:nth(erlang:phash(now(), length(LRs)), LRs),
+					   {domain_balancing, LDstDomain}) of
+					source -> jlib:jid_tolower(From);
+					destination -> jlib:jid_tolower(To);
+					random -> now();
+					undefined -> now()
+				    end,
+			    R = lists:nth(erlang:phash(Value, length(LRs)), LRs),
 			    Pid = R#route.pid,
 			    case R#route.local_hint of
 				{apply, Module, Function} ->

src/ejabberd_sup.erl

@@ -33,6 +33,13 @@ init([]) ->
 	 brutal_kill,
 	 worker,
 	 [stringprep]},
+    NodeGroups =
+	{ejabberd_node_groups,
+	 {ejabberd_node_groups, start_link, []},
+	 permanent,
+	 brutal_kill,
+	 worker,
+	 [ejabberd_node_groups]},
     Router =
 	{ejabberd_router,
 	 {ejabberd_router, start_link, []},
@@ -123,6 +130,14 @@ init([]) ->
 	 infinity,
 	 supervisor,
 	 [ejabberd_tmp_sup]},
+    FrontendSocketSupervisor =
+	{ejabberd_frontend_socket_sup,
+	 {ejabberd_tmp_sup, start_link,
+	  [ejabberd_frontend_socket_sup, ejabberd_frontend_socket]},
+	 permanent,
+	 infinity,
+	 supervisor,
+	 [ejabberd_tmp_sup]},
     IQSupervisor =
 	{ejabberd_iq_sup,
 	 {ejabberd_tmp_sup, start_link,
@@ -134,6 +149,7 @@ init([]) ->
     {ok, {{one_for_one, 10, 1},
 	  [Hooks,
 	   StringPrep,
+	   NodeGroups,
 	   Router,
 	   SM,
 	   S2S,
@@ -146,6 +162,7 @@ init([]) ->
 	   HTTPSupervisor,
 	   HTTPPollSupervisor,
 	   IQSupervisor,
+	   FrontendSocketSupervisor,
 	   Listener]}}.