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