%% @private -module(eetcd_conn). -include("eetcd.hrl"). -behaviour(gen_server). %% API -export([ open/1, close/1, round_robin_select/1, pick_member/2, check_health/1, refresh_token/2, set_credentails/3, unset_credentails/1, update_member_list/2, member_id_hex/1 ]). -export([info/1]). -export([ init/1, handle_call/3, handle_cast/2, handle_info/2, handle_continue/2, terminate/2, code_change/3 ]). -define(check_health, check_health_msg). -define(auto_sync, auto_sync). -define(refresh_token, refresh_token_msg). -define(reconnect, reconnect). -type state() :: #{ name := atom(), mode := random | connect_all, members := members(), credentials := undefined | #{name := string(), password := string()}, auth_token := undefined | iodata(), auto_sync_interval_ms := non_neg_integer(), gun_opts := gun:opts(), options := eetcd:opts(), health_ref := undefined | reference(), reconn_ref := undefined | reference(), sync_ref := undefined | reference(), active_conns := [conn()], opening_conns := [conn()] }. -type conn() :: {member_id(), GunPid :: pid(), MonitorRef :: reference()}. -type members() :: #{member_id() => {Host :: string(), inet:port_number(), tcp | tls}}. -define(DEFAULT_REPLY_TIMEOUT, 10000). -define(DEFAULT_REQ_OPTS, [{reply_timeout, ?DEFAULT_REPLY_TIMEOUT}]). %%%=================================================================== %%% API %%%=================================================================== open({EtcdName, _, _} = Args) -> case gen_server:start_link({local, EtcdName}, ?MODULE, Args, []) of {ok, Pid} -> {ok, Pid}; {error, _} = Err -> Err end. close(EtcdName) -> eetcd:close(EtcdName). round_robin_select(EtcdName) -> case gen_server:call(EtcdName, round_robin_select) of {ok, GunPid, undefined} -> {ok, GunPid, ?HEADERS}; {ok, GunPid, Token} -> {ok, GunPid, [{<<"token">>, Token} | ?HEADERS]}; {error, _} = E -> E end. pick_member(EtcdName, MemberId) -> case gen_server:call(EtcdName, {pick_member, MemberId}) of {ok, GunPid, undefined} -> {ok, GunPid, ?HEADERS}; {ok, GunPid, Token} -> {ok, GunPid, [{<<"token">>, Token} | ?HEADERS]}; {error, _} = E -> E end. set_credentails(EtcdName, UserName, Password) -> gen_server:call(EtcdName, {set_credentials, {UserName, Password}}). unset_credentails(EtcdName) -> gen_server:call(EtcdName, {set_credentials, undefined}). update_member_list(_Name, Members) when map_size(Members) =:= 0 -> {error, empty_member_list}; update_member_list(EtcdName, Members) -> gen_server:cast(EtcdName, {update_member_list, Members}). check_health(EtcdName) -> gen_server:cast(EtcdName, ?check_health). refresh_token(EtcdName, Headers) -> case gen_server:call(EtcdName, ?refresh_token) of {ok, undefined} -> lists:keydelete(<<"token">>, 1, Headers); {ok, NewToken} -> lists:keyreplace(<<"token">>, 1, Headers, {<<"token">>, NewToken}); {error, Reason} -> ?LOG_WARNING("Refresh token failed: ~p", [Reason]), Headers end. %% @private info(ServerRef) -> gen_server:call(ServerRef, info). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== init({EtcdName, Hosts, Options}) -> erlang:process_flag(trap_exit, true), GunOpts = #{ protocols => [http2], connect_timeout => proplists:get_value(connect_timeout, Options, 1000), domain_lookup_timeout => proplists:get_value(domain_lookup_timeout, Options, 1000), tls_handshake_timeout => proplists:get_value(tls_handshake_timeout, Options, 3000), http2_opts => #{keepalive => infinity}, retry_timeout => proplists:get_value(retry_timeout, Options, 5000), transport => proplists:get_value(transport, Options, tcp), tcp_opts => proplists:get_value(tcp_opts, Options, []), tls_opts => proplists:get_value(tls_opts, Options, []) }, AutoSyncInterval = proplists:get_value(auto_sync_interval_ms, Options, 0), try Members = case ets:lookup(?ETCD_CLIENT_CACHE, EtcdName) of [#eetcd_client{members = CachedMembers}] -> ?LOG_NOTICE("eetcd client ~p found cached members: ~p", [EtcdName, CachedMembers]), CachedMembers; [] -> NewMembers = get_member_list(EtcdName, shuffle(Hosts), GunOpts), ?LOG_NOTICE("eetcd client ~p found members: ~p", [EtcdName, NewMembers]), ets:insert(?ETCD_CLIENT_CACHE, #eetcd_client{name = EtcdName, members = NewMembers}), NewMembers end, State0 = #{ name => EtcdName, mode => proplists:get_value(mode, Options, connect_all), members => Members, credentials => undefined, auth_token => undefined, auto_sync_interval_ms => AutoSyncInterval, gun_opts => GunOpts, options => Options, health_ref => undefined, reconn_ref => undefined, sync_ref => undefined, active_conns => [], opening_conns => [] }, State1 = put_in_credentials(State0, Options), #{active_conns := [{_Id, GunPid, _MRef} | _]} = State2 = init_connect(Members, EtcdName, State1), case token_remote(GunPid, State2) of {ok, State3} -> {_, State4} = handle_continue(next_health_check, State3), {_, State5} = handle_continue(reconnect, State4), {_, State6} = handle_continue(auto_sync, State5), {ok, State6}; {error, R} -> error(R) end catch error:Reason0 -> ?LOG_ERROR("~p failed to connect etcd endpoints: ~p ~p", [EtcdName, Hosts, Reason0]), {stop, Reason0} end. handle_call(info, _From, State) -> {reply, maps:with([name, members, active_conns, opening_conns], State), State}; handle_call(round_robin_select, _From, #{active_conns := []} = State) -> {reply, {error, eetcd_conn_unavailable}, State}; handle_call(round_robin_select, _From, #{active_conns := [{_Host, GunPid, _MRef} = Conn | Rest], auth_token := Token} = State) -> {reply, {ok, GunPid, Token}, State#{active_conns => Rest ++ [Conn]}}; handle_call({set_credentails, undefined}, _From, State) -> {reply, ok, State#{credentials => undefined, auth_token => undefined}}; handle_call({set_credentails, {UserName, Password}}, _From, State) -> {reply, ok, State#{credentials => #{name => UserName, password => Password}}}; handle_call({pick_member, _MemberId}, _From, #{mode := random} = State) -> {reply, {error, random_mode}, State}; handle_call({pick_member, MemberId}, _From, #{mode := connect_all, members := Members, auth_token := Token, active_conns := Actives} = State) when is_map_key(MemberId, Members) -> case lists:keyfind(MemberId, 1, Actives) of {_MemberId, GunPid, _MRef} -> {reply, {ok, GunPid, Token}, State}; false -> {reply, {error, eetcd_conn_unavailable}, State} end; handle_call({pick_member, _MemberId}, _From, State) -> {reply, {error, eetcd_member_not_found}, State}; handle_call(?refresh_token, _From, #{active_conns := [{_Host, GunPid, _MRef} | _]} = State) -> case token_remote(GunPid, State) of {ok, #{auth_token := NewToken} = NewState} -> {reply, {ok, NewToken}, NewState}; {error, Reason} -> {reply, {error, Reason}, State} end. handle_cast(?check_health, State) -> {noreply, do_check_health(State)}; handle_cast({update_member_list, NewMembers}, State) when map_size(NewMembers) =:= 0 -> {noreply, State}; handle_cast({update_member_list, NewMembers}, #{name := EtcdName, active_conns := Actives, members := OldMembers, opening_conns := Openings} = State) -> %% Close all connections to members that are not in the new list Removes = [R || {Id, _GunPid, _MRef} = R <- Actives ++ Openings, not is_map_key(Id, NewMembers)], lists:foreach(fun({Id, GunPid, MRef}) when is_reference(MRef) -> {Host, Port, _Transport} = maps:get(Id, OldMembers), ?LOG_NOTICE("~p removing member (~s) ~s:~p, conn_pid: ~p", [EtcdName, member_id_hex(Id), Host, Port, GunPid]), erlang:demonitor(MRef, [flush]), ensure_close(GunPid) end, Removes), Added = maps:without(maps:keys(OldMembers), NewMembers), %% elp:ignore W0034 [ ?LOG_NOTICE("~p found new member (~s) ~s:~p", [EtcdName, member_id_hex(Id), Host, Port]) || {Id, {Host, Port, _}} <- maps:to_list(Added) ], ets:insert(?ETCD_CLIENT_CACHE, #eetcd_client{name = EtcdName, members = NewMembers}), {noreply, State#{members => NewMembers}, {continue, reconnect_now}}; handle_cast(_Req, State) -> {noreply, State}. handle_info({timeout, TRef, ?auto_sync}, #{name := EtcdName, sync_ref := TRef, active_conns := []} = State) -> erlang:cancel_timer(TRef), ?LOG_WARNING("~p failed to sync member list: no active connections", [EtcdName]), {noreply, State#{sync_ref => undefined}, {continue, auto_sync}}; handle_info({timeout, TRef, ?auto_sync}, #{name := EtcdName, sync_ref := TRef, active_conns := Actives} = State) -> erlang:cancel_timer(TRef), Self = self(), spawn(fun() -> check_member_list(Self, EtcdName, shuffle(Actives)) end), {noreply, State#{sync_ref => undefined}, {continue, auto_sync}}; handle_info({timeout, TRef, ?reconnect}, #{reconn_ref := TRef} = State) -> erlang:cancel_timer(TRef), {noreply, do_reconnect(State#{reconn_ref => undefined}), {continue, reconnect}}; handle_info({timeout, _TRef, ?reconnect}, State) -> {noreply, State}; handle_info({timeout, TRef, ?check_health}, #{health_ref := TRef} = State) -> erlang:cancel_timer(TRef), {noreply, do_check_health(State#{health_ref => undefined}), {continue, next_health_check}}; handle_info({gun_up, GunPid, http2}, State) -> handle_gun_up(GunPid, State), {noreply, State}; handle_info({await_check_ok, GunPid}, #{opening_conns := Openings, active_conns := Actives, members := Members} = State) -> case lists:keytake(GunPid, 2, Openings) of {value, {Id, GunPid, MRef}, Rest} -> #{Id := {Host, Port, _Transport}} = Members, ?LOG_NOTICE("Connection established to etcd member (~s) ~s:~p", [member_id_hex(Id), Host, Port]), {noreply, State#{active_conns => Actives ++ [{Id, GunPid, MRef}], opening_conns => Rest}}; false -> ?LOG_WARNING("Received unknown connection pid up: ~p", [GunPid]), {noreply, State} end; handle_info({await_check_error, GunPid, Reason}, State) -> {noreply, handle_gun_down(GunPid, Reason, State)}; handle_info({gun_down, GunPid, http2, Reason, _Streams}, State) -> {noreply, handle_gun_down(GunPid, Reason, State)}; handle_info({'DOWN', MRef, process, GunPid, Reason}, State) -> erlang:demonitor(MRef, [flush]), {noreply, handle_gun_down(GunPid, Reason, State)}; handle_info(_Msg, State) -> {noreply, State}. handle_continue(next_health_check, #{options := Opts, health_ref := undefined} = State) -> After = case proplists:get_value(health_check_ms, Opts, 15000) of N when is_integer(N), N > 0 -> N; _ -> 15000 end, TRef = erlang:start_timer(After, self(), ?check_health), {noreply, State#{health_ref => TRef}}; handle_continue(next_health_check, #{} = State) -> {noreply, State}; handle_continue(auto_sync, #{auto_sync_interval_ms := Interval, sync_ref := undefined} = State) when Interval > 0 -> TRef = erlang:start_timer(Interval, self(), ?auto_sync), {noreply, State#{sync_ref => TRef}}; handle_continue(auto_sync, #{} = State) -> {noreply, State}; handle_continue(reconnect_now, #{reconn_ref := undefined} = State) -> {noreply, do_reconnect(State), {continue, reconnect}}; handle_continue(reconnect_now, #{reconn_ref := TRef} = State) -> erlang:cancel_timer(TRef), {noreply, do_reconnect(State#{reconn_ref => undefined}), {continue, reconnect}}; handle_continue(reconnect, #{reconn_ref := undefined} = State) -> TRef = erlang:start_timer(2000, self(), ?reconnect), {noreply, State#{reconn_ref => TRef}}; handle_continue(reconnect, State) -> {noreply, State}. terminate(Reason, #{name := EtcdName, active_conns := Actives, opening_conns := Openings} = _State) -> ?LOG_NOTICE("eetcd client ~p terminating with reason: ~p", [EtcdName, Reason]), is_normal(Reason) andalso ets:delete(?ETCD_CLIENT_CACHE, EtcdName), [ensure_close(GunPid) || {_Host, GunPid, _Token} <- Actives ++ Openings], ok. code_change(_OldVsntateName, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== is_normal(normal) -> true; is_normal(shutdown) -> true; is_normal({shutdown, _}) -> true; is_normal(_) -> false. ensure_close(undefined) -> ok; ensure_close({GunPid, MRef}) when is_pid(GunPid), is_reference(MRef) -> erlang:demonitor(MRef, [flush]), ensure_close(GunPid); ensure_close(GunPid) when is_pid(GunPid) -> try gun:close(GunPid) catch _:_ -> ok end; ensure_close(_) -> ok. -spec init_connect(members(), atom(), state()) -> state() | no_return(). init_connect(Members, EtcdName, #{mode := connect_all} = State) -> connect_all(Members, EtcdName, State); init_connect(Members, EtcdName, #{mode := random} = State) -> connect_random(shuffle(maps:to_list(Members)), EtcdName, State); init_connect(_, _, _) -> error(unsupported_mode). connect_all(Members, EtcdName, #{gun_opts := GunOpts} = State) -> case fold_connect(maps:to_list(Members), EtcdName, GunOpts, [], []) of {Ok, []} -> State#{ mode => connect_all, active_conns => Ok}; {Ok, Failed} when length(Ok) > length(Failed) -> State#{ mode => connect_all, active_conns => Ok}; {Ok, Failed} -> [ensure_close({GunPid, MRef}) || {_Id, GunPid, MRef} <- Ok], error({shutdown, Failed}) end. connect_random([], _Name, _State) -> error(eetcd_conn_unavailable); connect_random([{Id, {_Host, _Port, _Transport}} = Member | Rest], EtcdName, #{gun_opts := GunOpts} = State) -> case connect_and_await_up(EtcdName, Member, GunOpts) of {ok, {GunPid, MRef}} -> State#{active_conns => [{Id, GunPid, MRef}]}; {error, _Reason} -> connect_random(Rest, EtcdName, State) end. fold_connect([], _Name, _GunOpts, Ok, Fail) -> {Ok, Fail}; fold_connect([{Id, {_Host, _Port, _Transport}} = Member | Rest], EtcdName, GunOpts, Ok, Fail) -> case connect_and_await_up(EtcdName, Member, GunOpts) of {ok, {GunPid, MRef}} -> NewOk = [{Id, GunPid, MRef} | Ok], fold_connect(Rest, EtcdName, GunOpts, NewOk, Fail); {error, Reason} -> NewFail = [{Id, Reason} | Fail], fold_connect(Rest, EtcdName, GunOpts, Ok, NewFail) end. -spec connect_and_await_up(atom(), {member_id(), {Host :: string(), inet:port_number(), tcp | tls}}, gun:opts()) -> {ok, {GunPid :: pid(), MRef :: reference()}} | {error, any()}. connect_and_await_up(EtcdName, {Id, {Host, Port, Transport}}, GunOpts0) -> try GunOpts1 = GunOpts0#{retry => 0}, GunOpts = case Transport of tcp -> GunOpts1; tls -> GunOpts1#{transport => tls} end, GunPid = case gun:open(Host, Port, GunOpts) of {ok, Gun0} -> Gun0; {error, GunReason0} -> error({undefined, GunReason0}) end, Retries = maps:get(retry, GunOpts, 0), AwaitTime = case {maps:get(connect_timeout, GunOpts), maps:get(retry_timeout, GunOpts)} of {infinity, _} -> infinity; {ConnectTimeout, RetryTimeout} when ConnectTimeout > 0, RetryTimeout > 0 -> (Retries + 1) * ConnectTimeout + Retries * RetryTimeout end, case gun:await_up(GunPid, AwaitTime) of {ok, http2} -> ok; {error, GunReason} -> error({GunPid, GunReason}) end, try ok = await_check(EtcdName, GunPid), MRef = erlang:monitor(process, GunPid), ?LOG_NOTICE("Connection established to etcd member (~s) ~s:~p", [member_id_hex(Id), Host, Port]), {ok, {GunPid, MRef}} catch _:Reason1 -> error({GunPid, Reason1}) end catch error:{GunPid0, Reason} -> ensure_close(GunPid0), ?LOG_WARNING("~p failed to connect etcd member (~s) ~s:~p by ~p", [EtcdName, member_id_hex(Id), Host, Port, Reason]), {error, Reason} end. -spec await_check(atom(), pid()) -> ok | no_return(). await_check(EtcdName, GunPid) -> case check_health_remote(EtcdName, GunPid) of ok -> ok; {error, HealthReason} -> error(HealthReason) end. do_reconnect(#{mode := random, members := Members, active_conns := [], opening_conns := [], gun_opts := GunOpts} = State) -> NewOpenings = open_members(shuffle(maps:to_list(Members)), GunOpts, [], one), State#{opening_conns => NewOpenings}; do_reconnect(#{mode := random} = State) -> State; do_reconnect(#{mode := connect_all, members := Members, active_conns := Actives, opening_conns := Openings, gun_opts := GunOpts} = State) -> Retries0 = maps:keys(Members) -- [Id || {Id, _GunPid, _MRef} <- Actives], Retries = Retries0 -- [Id || {Id, _} <- Openings], RetriesMembers = maps:to_list(maps:with(Retries, Members)), NewOpenings = open_members(RetriesMembers, GunOpts, [], all), State#{opening_conns => Openings ++ NewOpenings}. open_members([], _GunOpts, Result, _) -> Result; open_members([{Id, {Host, Port, Transport}} | Rest], GunOpts0, Result, AllorOne) -> GunOpts1 = GunOpts0#{retry => 0}, GunOpts = case Transport of tcp -> GunOpts1; tls -> GunOpts1#{transport => tls} end, case gun:open(Host, Port, GunOpts) of {ok, GunPid} -> MRef = erlang:monitor(process, GunPid), ?LOG_NOTICE("Opening connection to etcd member ~s:~p", [Host, Port]), case AllorOne of all -> open_members(Rest, GunOpts, [{Id, GunPid, MRef} | Result], AllorOne); one -> [{Id, GunPid, MRef}] end; {error, Reason} -> ?LOG_WARNING("Failed to open connection to etcd ~s:~p, reason: ~p", [Host, Port, Reason]), open_members(Rest, GunOpts, Result, AllorOne) end. handle_gun_up(GunPid, #{name := EtcdName, opening_conns := Openings} = _State) -> case lists:keytake(GunPid, 2, Openings) of {value, {_Id, GunPid, _MRef}, _Rest} -> Self = self(), spawn(fun() -> try ok = await_check(EtcdName, GunPid), Self ! {await_check_ok, GunPid} catch _:Reason -> Self ! {await_check_error, GunPid, Reason} end end); false -> ?LOG_WARNING("Received unknown connection pid up: ~p", [GunPid]) end, ok. handle_gun_down(GunPid, Reason, #{members := Members, active_conns := Actives, opening_conns := Openings} = State) -> case {lists:keyfind(GunPid, 2, Actives), lists:keyfind(GunPid, 2, Openings)} of {{Id, GunPid, MRef}, false} -> erlang:demonitor(MRef, [flush]), {Host, Port, _Transport} = maps:get(Id, Members), ?LOG_WARNING("etcd connection ~p to member ~s:~p down: ~p", [GunPid, Host, Port, Reason]), State#{active_conns => lists:keydelete(GunPid, 2, Actives)}; {false, {Id, GunPid, MRef}} -> erlang:demonitor(MRef, [flush]), {Host, Port, _Transport} = maps:get(Id, Members), ?LOG_WARNING("etcd connecting ~p to member ~s:~p failed: ~p", [GunPid, Host, Port, Reason]), State#{opening_conns => lists:keydelete(GunPid, 2, Openings)}; {false, false} -> ?LOG_WARNING("unknown etcd connection ~p down: ~p", [GunPid, Reason]), State end. do_check_health(#{active_conns := Actives} = State) -> do_check_health(Actives, State). do_check_health([], #{} = State) -> State; do_check_health([{Id, GunPid, MRef} | Rest], #{name := EtcdName, active_conns := Actives} = State) -> try case check_health_remote(EtcdName, GunPid) of ok -> do_check_health(Rest, State); {error, Reason1} -> error({health, Reason1}) end catch error:{Type, Reason} -> ConnInfo = gun:info(GunPid), erlang:demonitor(MRef, [flush]), ensure_close(GunPid), ?LOG_ERROR("~p check member (~s) ~s:~p (~p) failed: ~p", [EtcdName, member_id_hex(Id), inet:ntoa(maps:get(sock_ip, ConnInfo)), maps:get(sock_port, ConnInfo), Type, Reason]), State1 = State#{active_conns => lists:keydelete(Id, 1, Actives)}, do_check_health(Rest, State1) end. %% UNKNOWN = 0; %% SERVING = 1; %% NOT_SERVING = 2; %% SERVICE_UNKNOWN = 3; // Used only by the Watch method. -spec check_health_remote(atom(), pid()) -> ok | {error, any()}. check_health_remote(EtcdName, GunPid) -> case eetcd_health_gen:check({EtcdName, {GunPid, ?HEADERS}}, #{}, ?DEFAULT_REQ_OPTS) of {ok, #{status := 'SERVING'}} -> ok; {ok, #{status := 1}} -> ok; {ok, #{status := 'UNKNOWN'}} -> ok; {ok, #{status := 0}} -> ok; %% etcd does not support health checks in early versions of v3 API {error, {grpc_error, #{'grpc-message' := <<"unknown service grpc.health.v1.Health">>, 'grpc-status' := ?GRPC_STATUS_UNIMPLEMENTED}}} -> ok; {ok, #{status := Status}} -> {error, {unhealthy, Status}}; {error, _Reason} = Err -> Err end. -spec token_remote(pid(), state()) -> {ok, state()} | {error, any()}. token_remote(GunPid, #{name := EtcdName, credentials := #{name := UserName, password := Password}} = State) -> Request = #{name => UserName, password => Password}, Opts = ?DEFAULT_REQ_OPTS, case eetcd_auth_gen:authenticate({EtcdName, {GunPid, ?HEADERS}}, Request, Opts) of {ok, #{token := Token}} -> {ok, State#{auth_token => Token}}; {error, _Reason} = Err -> Err end; token_remote(_GunPid, State) -> {ok, State#{auth_token => undefined}}. -spec with_gun(atom(), Host :: string(), inet:port_number(), gun:opts(), Fun) -> Result when Fun :: fun((atom(), pid()) -> T), Result :: T | {error, any()}. with_gun(EtcdName, Host, Port, GunOpts, Fun) -> case gun:open(Host, Port, GunOpts) of {ok, GunPid} -> try case gun:await_up(GunPid, 5000) of {ok, http2} -> Fun(EtcdName, GunPid); {error, _Reason} = E -> E end after ensure_close(GunPid) end; Other -> Other end. get_member_list(_Name, [], _GunOpts) -> error(no_available_members); get_member_list(EtcdName, [{Host, Port}| Rest], GunOpts) -> case with_gun(EtcdName, Host, Port, GunOpts, fun member_list/2) of {ok, Members} -> Members; {error, Reason} -> ?LOG_WARNING("Failed to get member list from ~s:~p, reason: ~p", [Host, Port, Reason]), get_member_list(EtcdName, Rest, GunOpts) end. check_member_list(_SelfPid, EtcdName, []) -> ?LOG_WARNING("~p failed to sync member list: no active connections", [EtcdName]); check_member_list(SelfPid, EtcdName, [{_Id, GunPid, _MRef} | _]) -> try member_list(EtcdName, GunPid) of {ok, Members} -> case update_member_list(SelfPid, Members) of ok -> ok; {error, Reason} -> ?LOG_WARNING("~p failed to update member list: ~p", [EtcdName, Reason]), ok end; {error, Reason} -> ?LOG_WARNING("~p failed to get member list: ~p", [EtcdName, Reason]), ok catch error:_Reason -> ok end. -spec member_list(atom(), pid()) -> {ok, members()} | {error, any()}. member_list(EtcdName, GunPid) -> Conn = {EtcdName, {GunPid, ?HEADERS}}, Opts = [{reply_timeout, 10000}], case eetcd_cluster_gen:member_list(Conn, #{}, Opts) of {ok, #{members := Members0}} when is_list(Members0) -> {Members, _} = lists:partition(fun(#{isLearner := 1}) -> false; (#{isLearner := true}) -> false; (_) -> true end, Members0), {ok, parse_members(Members)}; {error, _Reason} = Err -> Err end. -spec parse_members([router_pb:'Etcd.Member'()]) -> members() | no_return(). parse_members(Members) -> Result = [{Id, parse_client_url(Url)} || #{'ID' := Id, clientURLs := [Url | _]} <- Members], maps:from_list(Result). parse_client_url(Url) when is_binary(Url) -> parse_client_url(binary_to_list(Url)); parse_client_url(Url) when is_list(Url) -> case uri_string:parse(Url) of #{host := Host, port := Port, scheme := "http"} -> {Host, Port, tcp}; #{host := Host, port := Port, scheme := "https"} -> {Host, Port, tls}; _ -> ?LOG_WARNING("Invalid client URL: ~s", [Url]), error(invalid_url) end. put_in_credentials(Data, Options) -> case {proplists:get_value(name, Options, undefined), proplists:get_value(password, Options, undefined)} of {undefined, _} -> Data; {_, undefined} -> Data; {UserName, Password} -> Data#{credentials => #{name => UserName, password => Password}} end. shuffle(List) -> [X || {_, X} <- lists:sort([{rand:uniform(), N} || N <- List])]. member_id_hex(Id) when is_integer(Id) -> string:lowercase(binary:encode_hex(<>)).