./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 16 Keys = [cluster_name, node_name_to_insert, last_query_info, expire_time, node_ip_binary],
29 16 maps:with(Keys, maps:merge(defaults(), Opts)).
30
31 defaults() ->
32 16 #{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 32 case is_rdbms_running() of
40 true ->
41 32 try try_register(ClusterName, Node, State) of
42 {Num, Nodes, Info, AddrPairs} ->
43 32 mongoose_node_num:set_node_num(Num),
44 32 {{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 32 try mongoose_wpool:get_worker(rdbms, global) of
57 32 {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 32 prepare(),
66 32 Timestamp = timestamp(),
67 32 {selected, Rows} = select(ClusterName),
68 32 Nodes = [element(1, Row) || Row <- Rows],
69 32 Nums = [element(2, Row) || Row <- Rows],
70 32 Addresses = [element(3, Row) || Row <- Rows],
71 32 AddrPairs = maps:from_list(lists:zip(Nodes, Addresses)),
72 32 AlreadyRegistered = lists:member(Node, Nodes),
73 32 NodeNum =
74 case AlreadyRegistered of
75 true ->
76 15 update_existing(Node, Address, Timestamp),
77 15 {value, {_, Num, _Addr, _TS}} = lists:keysearch(Node, 1, Rows),
78 15 Num;
79 false ->
80 17 Num = first_free_num(lists:usort(Nums)),
81 17 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 17 case insert_new(ClusterName, Node, Num, Address, Timestamp) of
85 1 {error, _} -> 0; %% return default node num
86 16 {updated, 1} -> Num
87 end
88 end,
89 32 RunCleaningResult = run_cleaning(Timestamp, Rows, State),
90 %% This could be used for debugging
91 32 Info = #{already_registered => AlreadyRegistered, timestamp => Timestamp,
92 address => Address,
93 node_num => Num, last_rows => Rows, run_cleaning_result => RunCleaningResult},
94 32 {NodeNum, skip_expired_nodes(Nodes, RunCleaningResult), Info, AddrPairs}.
95
96 skip_expired_nodes(Nodes, {removed, ExpiredNodes}) ->
97 32 (Nodes -- ExpiredNodes).
98
99 run_cleaning(Timestamp, Rows, State) ->
100 32 #{expire_time := ExpireTime, node_name_to_insert := CurrentNode} = State,
101 32 ExpiredNodes = [DbNode || {DbNode, _Num, _Addr, DbTS} <- Rows,
102 43 is_expired(DbTS, Timestamp, ExpireTime),
103 2 DbNode =/= CurrentNode],
104 32 [delete_node_from_db(DbNode) || DbNode <- ExpiredNodes],
105 32 case ExpiredNodes of
106 31 [] -> ok;
107 [_ | _] ->
108 1 ?LOG_WARNING(#{what => cets_expired_nodes,
109 text => <<"Expired nodes are detected in discovery_nodes table">>,
110 
:-(
 expired_nodes => ExpiredNodes})
111 end,
112 32 {removed, ExpiredNodes}.
113
114 is_expired(DbTS, Timestamp, ExpireTime) when is_integer(Timestamp),
115 is_integer(ExpireTime),
116 is_integer(DbTS) ->
117 43 (Timestamp - DbTS) > ExpireTime. %% compare seconds
118
119 prepare() ->
120 32 T = discovery_nodes,
121 32 mongoose_rdbms_timestamp:prepare(),
122 32 mongoose_rdbms:prepare(cets_disco_select, T, [cluster_name], select()),
123 32 mongoose_rdbms:prepare(cets_disco_insert_new, T,
124 [cluster_name, node_name, node_num, address, updated_timestamp],
125 insert_new()),
126 32 mongoose_rdbms:prepare(cets_disco_update_existing, T,
127 [updated_timestamp, address, node_name], update_existing()),
128 32 mongoose_rdbms:prepare(cets_delete_node_from_db, T,
129 [node_name], delete_node_from_db()).
130
131 select() ->
132 32 <<"SELECT node_name, node_num, address, updated_timestamp FROM discovery_nodes WHERE cluster_name = ?">>.
133
134 select(ClusterName) ->
135 35 mongoose_rdbms:execute_successfully(global, cets_disco_select, [ClusterName]).
136
137 insert_new() ->
138 32 <<"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 23 mongoose_rdbms:execute(global, cets_disco_insert_new,
143 [ClusterName, NodeName, NodeNum, Address, UpdatedTimestamp]).
144
145 update_existing() ->
146 32 <<"UPDATE discovery_nodes SET updated_timestamp = ?, address = ? WHERE node_name = ?">>.
147
148 update_existing(NodeName, Address, UpdatedTimestamp) ->
149 16 mongoose_rdbms:execute(global, cets_disco_update_existing, [UpdatedTimestamp, Address, NodeName]).
150
151 delete_node_from_db() ->
152 32 <<"DELETE FROM discovery_nodes WHERE node_name = ?">>.
153
154 delete_node_from_db(Node) ->
155 21 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 32 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 17 [FirstFreeNum | _] = lists:seq(1, length(Nums)+1) -- Nums,
167 17 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