From 117f31125d4d1e39f6bf639360ff37a44664dbb2 Mon Sep 17 00:00:00 2001 From: Evgeniy Khramtsov Date: Tue, 28 Mar 2017 16:31:37 +0300 Subject: [PATCH] Add SQL as router RAM backend --- sql/lite.sql | 11 ++ sql/mssql.sql | 14 +++ sql/mysql.sql | 11 ++ sql/pg.sql | 11 ++ src/ejabberd_local.erl | 7 +- src/ejabberd_router.erl | 10 +- src/ejabberd_router_mnesia.erl | 10 +- src/ejabberd_router_sql.erl | 181 +++++++++++++++++++++++++++++++++ 8 files changed, 247 insertions(+), 8 deletions(-) create mode 100644 src/ejabberd_router_sql.erl diff --git a/sql/lite.sql b/sql/lite.sql index 3e9231768..bc6a6e706 100644 --- a/sql/lite.sql +++ b/sql/lite.sql @@ -319,3 +319,14 @@ CREATE TABLE oauth_token ( scope text NOT NULL, expire bigint NOT NULL ); + +CREATE TABLE route ( + domain text NOT NULL, + server_host text NOT NULL, + node text NOT NULL, + pid text NOT NULL, + local_hint text NOT NULL +); + +CREATE UNIQUE INDEX i_route ON route(domain, server_host, node, pid); +CREATE INDEX i_route_domain ON route(domain); diff --git a/sql/mssql.sql b/sql/mssql.sql index a3b814e02..06f73aea9 100644 --- a/sql/mssql.sql +++ b/sql/mssql.sql @@ -490,3 +490,17 @@ CREATE TABLE [dbo].[oauth_token] ( [token] ASC )WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ) TEXTIMAGE_ON [PRIMARY]; + +CREATE TABLE [dbo].[route] ( + [domain] [varchar] (255) NOT NULL, + [server_host] [varchar] (255) NOT NULL, + [node] [varchar] (255) NOT NULL, + [pid] [varchar](100) NOT NULL, + [local_hint] text NOT NULL +); + +CREATE UNIQUE CLUSTERED INDEX [route_i] ON [route] (domain, server_host, node, pid) +WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON); + +CREATE INDEX [route_domain] ON [route] (domain) +WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON); diff --git a/sql/mysql.sql b/sql/mysql.sql index 9b2114ae5..c4f3d1f02 100644 --- a/sql/mysql.sql +++ b/sql/mysql.sql @@ -335,3 +335,14 @@ CREATE TABLE oauth_token ( scope text NOT NULL, expire bigint NOT NULL ) ENGINE=InnoDB CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; + +CREATE TABLE route ( + domain text NOT NULL, + server_host text NOT NULL, + node text NOT NULL, + pid text NOT NULL, + local_hint text NOT NULL +) ENGINE=InnoDB CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; + +CREATE UNIQUE INDEX i_route ON route(domain(75), server_host(75), node(75), pid(75)); +CREATE INDEX i_route_domain ON route(domain(75)); diff --git a/sql/pg.sql b/sql/pg.sql index 2ce9d1379..fac806e8a 100644 --- a/sql/pg.sql +++ b/sql/pg.sql @@ -339,3 +339,14 @@ CREATE TABLE oauth_token ( ); CREATE UNIQUE INDEX i_oauth_token_token ON oauth_token USING btree (token); + +CREATE TABLE route ( + domain text NOT NULL, + server_host text NOT NULL, + node text NOT NULL, + pid text NOT NULL, + local_hint text NOT NULL +); + +CREATE UNIQUE INDEX i_route ON route USING btree (domain, server_host, node, pid); +CREATE INDEX i_route_domain ON route USING btree (domain); diff --git a/src/ejabberd_local.erl b/src/ejabberd_local.erl index 196e50324..d2f1b20db 100644 --- a/src/ejabberd_local.erl +++ b/src/ejabberd_local.erl @@ -197,6 +197,7 @@ get_features(_, _, XMLNSs) -> %%==================================================================== init([]) -> + process_flag(trap_exit, true), lists:foreach(fun host_up/1, ?MYHOSTS), ejabberd_hooks:add(host_up, ?MODULE, host_up, 10), ejabberd_hooks:add(host_down, ?MODULE, host_down, 100), @@ -288,7 +289,11 @@ host_up(Host) -> ?MODULE, bounce_resource_packet, 100). host_down(Host) -> - ejabberd_router:unregister_route(Host), + Owner = case whereis(?MODULE) of + undefined -> self(); + Pid -> Pid + end, + ejabberd_router:unregister_route(Host, Owner), ejabberd_hooks:delete(local_send_to_resource_hook, Host, ?MODULE, bounce_resource_packet, 100). diff --git a/src/ejabberd_router.erl b/src/ejabberd_router.erl index 596dd52c8..dca3ac25d 100644 --- a/src/ejabberd_router.erl +++ b/src/ejabberd_router.erl @@ -44,6 +44,7 @@ host_of_route/1, process_iq/1, unregister_route/1, + unregister_route/2, unregister_routes/1, get_all_routes/0, is_my_route/1, @@ -67,7 +68,7 @@ -callback init() -> any(). -callback register_route(binary(), binary(), local_hint(), undefined | pos_integer(), pid()) -> ok | {error, term()}. --callback unregister_route(binary(), undefined | pos_integer()) -> ok | {error, term()}. +-callback unregister_route(binary(), undefined | pos_integer(), pid()) -> ok | {error, term()}. -callback find_routes(binary()) -> [#route{}]. -callback host_of_route(binary()) -> {ok, binary()} | error. -callback is_my_route(binary()) -> boolean(). @@ -171,12 +172,17 @@ register_routes(Domains) -> -spec unregister_route(binary()) -> ok. unregister_route(Domain) -> + unregister_route(Domain, self()). + +-spec unregister_route(binary(), pid()) -> ok. +unregister_route(Domain, Pid) -> case jid:nameprep(Domain) of error -> erlang:error({invalid_domain, Domain}); LDomain -> Mod = get_backend(), - case Mod:unregister_route(LDomain, get_component_number(LDomain)) of + case Mod:unregister_route( + LDomain, get_component_number(LDomain), Pid) of ok -> ?DEBUG("Route unregistered: ~s", [LDomain]); {error, Err} -> diff --git a/src/ejabberd_router_mnesia.erl b/src/ejabberd_router_mnesia.erl index 3600c08dc..15cdf64c0 100644 --- a/src/ejabberd_router_mnesia.erl +++ b/src/ejabberd_router_mnesia.erl @@ -24,7 +24,7 @@ -behaviour(gen_server). %% API --export([init/0, register_route/5, unregister_route/2, find_routes/1, +-export([init/0, register_route/5, unregister_route/3, find_routes/1, host_of_route/1, is_my_route/1, is_my_host/1, get_all_routes/0]). %% gen_server callbacks -export([init/1, handle_cast/2, handle_call/3, handle_info/2, @@ -96,19 +96,19 @@ register_route(Domain, ServerHost, _LocalHint, N, Pid) -> end, transaction(F). -unregister_route(Domain, undefined) -> +unregister_route(Domain, undefined, Pid) -> F = fun () -> case mnesia:match_object( - #route{domain = Domain, pid = self(), _ = '_'}) of + #route{domain = Domain, pid = Pid, _ = '_'}) of [R] -> mnesia:delete_object(R); _ -> ok end end, transaction(F); -unregister_route(Domain, _) -> +unregister_route(Domain, _, Pid) -> F = fun () -> case mnesia:match_object( - #route{domain = Domain, pid = self(), _ = '_'}) of + #route{domain = Domain, pid = Pid, _ = '_'}) of [R] -> I = R#route.local_hint, ServerHost = R#route.server_host, diff --git a/src/ejabberd_router_sql.erl b/src/ejabberd_router_sql.erl new file mode 100644 index 000000000..1daa92fb1 --- /dev/null +++ b/src/ejabberd_router_sql.erl @@ -0,0 +1,181 @@ +%%%------------------------------------------------------------------- +%%% @author Evgeny Khramtsov +%%% Created : 28 Mar 2017 by Evgeny Khramtsov +%%% +%%% +%%% ejabberd, Copyright (C) 2002-2017 ProcessOne +%%% +%%% This program is free software; you can redistribute it and/or +%%% modify it under the terms of the GNU General Public License as +%%% published by the Free Software Foundation; either version 2 of the +%%% License, or (at your option) any later version. +%%% +%%% This program is distributed in the hope that it will be useful, +%%% but WITHOUT ANY WARRANTY; without even the implied warranty of +%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +%%% General Public License for more details. +%%% +%%% You should have received a copy of the GNU General Public License along +%%% with this program; if not, write to the Free Software Foundation, Inc., +%%% 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. +%%% +%%%------------------------------------------------------------------- +-module(ejabberd_router_sql). +-behaviour(ejabberd_router). + +-compile([{parse_transform, ejabberd_sql_pt}]). + +%% API +-export([init/0, register_route/5, unregister_route/3, find_routes/1, + host_of_route/1, is_my_route/1, is_my_host/1, get_all_routes/0]). + +-include("ejabberd.hrl"). +-include("logger.hrl"). +-include("ejabberd_sql_pt.hrl"). +-include("ejabberd_router.hrl"). + +%%%=================================================================== +%%% API +%%%=================================================================== +init() -> + Node = erlang:atom_to_binary(node(), latin1), + ?INFO_MSG("Cleaning SQL 'route' table...", []), + case ejabberd_sql:sql_query( + ?MYNAME, ?SQL("delete from route where node=%(Node)s")) of + {updated, _} -> + ok; + Err -> + ?ERROR_MSG("failed to clean 'route' table: ~p", [Err]), + Err + end. + +register_route(Domain, ServerHost, LocalHint, _, Pid) -> + PidS = enc_pid(Pid), + LocalHintS = enc_local_hint(LocalHint), + Node = erlang:atom_to_binary(node(Pid), latin1), + case ?SQL_UPSERT(?MYNAME, "route", + ["!domain=%(Domain)s", + "!server_host=%(ServerHost)s", + "!node=%(Node)s", + "!pid=%(PidS)s", + "local_hint=%(LocalHintS)s"]) of + ok -> + ok; + Err -> + ?ERROR_MSG("failed to update 'route' table: ~p", [Err]), + {error, Err} + end. + +unregister_route(Domain, _, Pid) -> + PidS = enc_pid(Pid), + Node = erlang:atom_to_binary(node(Pid), latin1), + ejabberd_sql:sql_query( + ?MYNAME, + ?SQL("delete from route where domain=%(Domain)s " + "and pid=%(PidS)s and node=%(Node)s")), + %% TODO: return meaningful error + ok. + +find_routes(Domain) -> + case ejabberd_sql:sql_query( + ?MYNAME, + ?SQL("select @(server_host)s, @(node)s, @(pid)s, @(local_hint)s " + "from route where domain=%(Domain)s")) of + {selected, Rows} -> + lists:flatmap( + fun(Row) -> + row_to_route(Domain, Row) + end, Rows); + Err -> + ?ERROR_MSG("failed to select from 'route' table: ~p", [Err]), + {error, Err} + end. + +host_of_route(Domain) -> + case ejabberd_sql:sql_query( + ?MYNAME, + ?SQL("select @(server_host)s from route where domain=%(Domain)s")) of + {selected, [{ServerHost}|_]} -> + {ok, ServerHost}; + {selected, []} -> + error; + Err -> + ?ERROR_MSG("failed to select from 'route' table: ~p", [Err]), + error + end. + +is_my_route(Domain) -> + case host_of_route(Domain) of + {ok, _} -> true; + _ -> false + end. + +is_my_host(Domain) -> + {ok, Domain} == host_of_route(Domain). + +get_all_routes() -> + case ejabberd_sql:sql_query( + ?MYNAME, + ?SQL("select @(domain)s from route where domain <> server_host")) of + {selected, Domains} -> + [Domain || {Domain} <- Domains]; + Err -> + ?ERROR_MSG("failed to select from 'route' table: ~p", [Err]), + [] + end. + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== +enc_local_hint(undefined) -> + <<"">>; +enc_local_hint(LocalHint) -> + jlib:term_to_expr(LocalHint). + +dec_local_hint(<<"">>) -> + undefined; +dec_local_hint(S) -> + ejabberd_sql:decode_term(S). + +-spec enc_pid(pid()) -> binary(). +enc_pid(Pid) -> + list_to_binary(erlang:pid_to_list(Pid)). + +-spec dec_pid(binary(), binary()) -> pid(). +dec_pid(PidBin, NodeBin) -> + PidStr = binary_to_list(PidBin), + Pid = erlang:list_to_pid(PidStr), + case erlang:binary_to_atom(NodeBin, latin1) of + Node when Node == node() -> + Pid; + Node -> + try set_node_id(PidStr, NodeBin) + catch _:badarg -> + erlang:error({node_down, Node}) + end + end. + +-spec set_node_id(string(), binary()) -> pid(). +set_node_id(PidStr, NodeBin) -> + ExtPidStr = erlang:pid_to_list( + binary_to_term( + <<131,103,100,(size(NodeBin)):16,NodeBin/binary,0:72>>)), + [H|_] = string:tokens(ExtPidStr, "."), + [_|T] = string:tokens(PidStr, "."), + erlang:list_to_pid(string:join([H|T], ".")). + +row_to_route(Domain, {ServerHost, NodeS, PidS, LocalHintS} = Row) -> + try [#route{domain = Domain, + server_host = ServerHost, + pid = dec_pid(PidS, NodeS), + local_hint = dec_local_hint(LocalHintS)}] + catch _:{node_down, _} -> + []; + E:R -> + ?ERROR_MSG("failed to decode row from 'route' table:~n" + "Row = ~p~n" + "Domain = ~s~n" + "Reason = ~p", + [Row, Domain, {E, {R, erlang:get_stacktrace()}}]), + [] + end.