./ct_report/coverage/mongoose_cets_discovery_rdbms.COVER.html

1 %% @doc MongooseIM RDBMS backend for cets_discovery.
2 -module(mongoose_cets_discovery_rdbms).
3 -behaviour(cets_discovery).
4 -export([init/1, get_nodes/1]).
5
6 %% these functions are exported for testing purposes only.
7 -export([select/1, insert_new/5, update_existing/3, delete_node_from_db/1]).
8 -ignore_xref([select/1, insert_new/5, update_existing/3, delete_node_from_db/1]).
9
10 -include("mongoose_logger.hrl").
11
12 -ifdef(TEST).
13 -include_lib("eunit/include/eunit.hrl").
14 -endif.
15
16 -type opts() :: #{cluster_name := binary(), node_name_to_insert := binary(),
17 last_query_info => map(), expire_time => non_neg_integer(),
18 node_ip_binary => binary(),
19 any() => any()}.
20
21 -type state() :: #{cluster_name := binary(), node_name_to_insert := binary(),
22 last_query_info := map(), expire_time := non_neg_integer(),
23 node_ip_binary := binary(), address_pairs := #{binary() => binary()}}.
24
25 -spec init(opts()) -> state().
26 init(Opts = #{cluster_name := ClusterName, node_name_to_insert := Node})
27 when is_binary(ClusterName), is_binary(Node) ->
28 17 Keys = [node_name_to_insert, expire_time, last_query_info, node_ip_binary],
29 17 StateOpts = maps:merge(defaults(), maps:with(Keys, Opts)),
30 17 StateOpts#{cluster_name => cluster_name_with_vsn(ClusterName)}.
31
32 cluster_name_with_vsn(ClusterName) ->
33 17 {ok, CetsVsn} = application:get_key(cets, vsn),
34 17 [MajorVsn, MinorVsn | _] = string:tokens(CetsVsn, "."),
35 17 iolist_to_binary([ClusterName, $-, MajorVsn, $., MinorVsn]).
36
37 defaults() ->
38 17 #{expire_time => 60 * 60 * 1, %% 1 hour in seconds
39 last_query_info => #{},
40 node_ip_binary => <<>>,
41 address_pairs => #{}}.
42
43 -spec get_nodes(state()) -> {cets_discovery:get_nodes_result(), state()}.
44 get_nodes(State = #{cluster_name := ClusterName, node_name_to_insert := Node}) ->
45 33 case is_rdbms_running() of
46 true ->
47 33 try try_register(ClusterName, Node, State) of
48 {Num, Nodes, Info, AddrPairs} ->
49 33 mongoose_node_num:set_node_num(Num),
50 33 {{ok, [binary_to_atom(N) || N <- Nodes]},
51 State#{last_query_info => Info, address_pairs => AddrPairs}}
52 catch Class:Reason:Stacktrace ->
53 
:-(
 ?LOG_ERROR(#{what => discovery_failed_select, class => Class,
54 
:-(
 reason => Reason, stacktrace => Stacktrace}),
55 
:-(
 {{error, Reason}, State}
56 end;
57 false ->
58 
:-(
 {{error, rdbms_not_running}, State}
59 end.
60
61 is_rdbms_running() ->
62 33 try mongoose_wpool:get_worker(rdbms, global) of
63 33 {ok, _} -> true;
64 
:-(
 _ -> false
65 catch _:_ ->
66 
:-(
 false
67 end.
68
69 try_register(ClusterName, Node, State = #{node_ip_binary := Address})
70 when is_binary(Node), is_binary(ClusterName) ->
71 33 prepare(),
72 33 Timestamp = timestamp(),
73 33 {selected, Rows} = select(ClusterName),
74 33 Nodes = [element(1, Row) || Row <- Rows],
75 33 Nums = [element(2, Row) || Row <- Rows],
76 33 Addresses = [element(3, Row) || Row <- Rows],
77 33 AddrPairs = maps:from_list(lists:zip(Nodes, Addresses)),
78 33 AlreadyRegistered = lists:member(Node, Nodes),
79 33 NodeNum =
80 case AlreadyRegistered of
81 true ->
82 14 update_existing(Node, Address, Timestamp),
83 14 {value, {_, Num, _Addr, _TS}} = lists:keysearch(Node, 1, Rows),
84 14 Num;
85 false ->
86 19 Num = first_free_num(lists:usort(Nums)),
87 19 delete_node_from_db(Node), % Delete node if it was a member of another cluster
88 %% Could fail with duplicate node_num reason.
89 %% In this case just wait for the next get_nodes call.
90 19 case insert_new(ClusterName, Node, Num, Address, Timestamp) of
91 2 {error, _} -> 0; %% return default node num
92 17 {updated, 1} -> Num
93 end
94 end,
95 33 RunCleaningResult = run_cleaning(Timestamp, Rows, State),
96 %% This could be used for debugging
97 33 Info = #{already_registered => AlreadyRegistered, timestamp => Timestamp,
98 address => Address,
99 node_num => Num, last_rows => Rows, run_cleaning_result => RunCleaningResult},
100 33 {NodeNum, skip_expired_nodes(Nodes, RunCleaningResult), Info, AddrPairs}.
101
102 skip_expired_nodes(Nodes, {removed, ExpiredNodes}) ->
103 33 (Nodes -- ExpiredNodes).
104
105 run_cleaning(Timestamp, Rows, State) ->
106 33 #{expire_time := ExpireTime, node_name_to_insert := CurrentNode} = State,
107 33 ExpiredNodes = [DbNode || {DbNode, _Num, _Addr, DbTS} <- Rows,
108 41 is_expired(DbTS, Timestamp, ExpireTime),
109 2 DbNode =/= CurrentNode],
110 33 [delete_node_from_db(DbNode) || DbNode <- ExpiredNodes],
111 33 case ExpiredNodes of
112 32 [] -> ok;
113 [_ | _] ->
114 1 ?LOG_WARNING(#{what => cets_expired_nodes,
115 text => <<"Expired nodes are detected in discovery_nodes table">>,
116 
:-(
 expired_nodes => ExpiredNodes})
117 end,
118 33 {removed, ExpiredNodes}.
119
120 is_expired(DbTS, Timestamp, ExpireTime) when is_integer(Timestamp),
121 is_integer(ExpireTime),
122 is_integer(DbTS) ->
123 41 (Timestamp - DbTS) > ExpireTime. %% compare seconds
124
125 prepare() ->
126 33 T = discovery_nodes,
127 33 mongoose_rdbms_timestamp:prepare(),
128 33 mongoose_rdbms:prepare(cets_disco_select, T, [cluster_name], select()),
129 33 mongoose_rdbms:prepare(cets_disco_insert_new, T,
130 [cluster_name, node_name, node_num, address, updated_timestamp], insert_new()),
131 33 mongoose_rdbms:prepare(cets_disco_update_existing, T,
132 [updated_timestamp, address, node_name], update_existing()),
133 33 mongoose_rdbms:prepare(cets_delete_node_from_db, T,
134 [node_name], delete_node_from_db()).
135
136 select() ->
137 33 <<"SELECT node_name, node_num, address, updated_timestamp FROM discovery_nodes WHERE cluster_name = ?">>.
138
139 select(ClusterName) ->
140 36 mongoose_rdbms:execute_successfully(global, cets_disco_select, [ClusterName]).
141
142 insert_new() ->
143 33 <<"INSERT INTO discovery_nodes (cluster_name, node_name, node_num, address, updated_timestamp)"
144 " VALUES (?, ?, ?, ?, ?)">>.
145
146 insert_new(ClusterName, NodeName, NodeNum, Address, UpdatedTimestamp) ->
147 25 mongoose_rdbms:execute(global, cets_disco_insert_new,
148 [ClusterName, NodeName, NodeNum, Address, UpdatedTimestamp]).
149
150 update_existing() ->
151 33 <<"UPDATE discovery_nodes SET updated_timestamp = ?, address = ? WHERE node_name = ?">>.
152
153 update_existing(NodeName, Address, UpdatedTimestamp) ->
154 15 mongoose_rdbms:execute(global, cets_disco_update_existing, [UpdatedTimestamp, Address, NodeName]).
155
156 delete_node_from_db() ->
157 33 <<"DELETE FROM discovery_nodes WHERE node_name = ?">>.
158
159 delete_node_from_db(Node) ->
160 21 mongoose_rdbms:execute_successfully(global, cets_delete_node_from_db, [Node]).
161
162 %% in seconds
163 timestamp() ->
164 % We could use Erlang timestamp os:system_time(second).
165 % But we use the database server time as a central source of truth.
166 33 mongoose_rdbms_timestamp:select().
167
168 %% Returns a next free node id based on the currently registered ids
169 first_free_num(Nums) ->
170 %% 0 is default node_num, so lets start from 1
171 19 [FirstFreeNum | _] = lists:seq(1, length(Nums)+1) -- Nums,
172 19 FirstFreeNum.
173
174 -ifdef(TEST).
175
176 jid_to_opt_binary_test_() ->
177 [?_assertEqual(1, first_free_num([])),
178 ?_assertEqual(3, first_free_num([1, 2, 5])),
179 ?_assertEqual(1, first_free_num([2, 5])),
180 ?_assertEqual(3, first_free_num([1, 2]))].
181
182 -endif.
Line Hits Source