etcd peer discovery fixes

Instead of relying on the complex and non-determinstic default node
selection mechanism inside peer discovery this change makes the
etcd backend implemention make the leader selection itself based on
the etcd create_revision of each entry. Although not spelled out anywhere
explicitly is likely that a property called "Create Revision" is going
to remain consistent throughout the lifetime of the etcd key.

Either way this is likely to be an improvement on the current approach.

Author: kjnilsson

deps/rabbit/test/unit_quorum_queue_SUITE.erl

@@ -1,5 +1,6 @@
 -module(unit_quorum_queue_SUITE).
 
+-compile(nowarn_export_all).
 -compile(export_all).
 
 all() ->

deps/rabbitmq_peer_discovery_etcd/src/rabbit_peer_discovery_etcd.erl

@@ -59,9 +59,13 @@ list_nodes() ->
                    {ok, {[], disc}}
            end,
     Fun2 = fun(_Proplist) ->
-                   %% error logging will be done by the client
-                   Nodes = rabbitmq_peer_discovery_etcd_v3_client:list_nodes(),
-                   {ok, {Nodes, disc}}
+                   %% nodes are returned sorted with the create_revision as
+                   %% the first element in the tuple.
+                   %% The node with the lowest create_revision is thus selected
+                   %% based on the assumption that the create_revision remains
+                   %% consistent throughout the lifetime of the etcd key.
+                   [{_, Node} | _] = rabbitmq_peer_discovery_etcd_v3_client:list_nodes(),
+                   {ok, {Node, disc}}
            end,
     rabbit_peer_discovery_util:maybe_backend_configured(?BACKEND_CONFIG_KEY, Fun0, Fun1, Fun2).
 

deps/rabbitmq_peer_discovery_etcd/src/rabbitmq_peer_discovery_etcd_v3_client.erl

@@ -230,16 +230,13 @@ connected({call, From}, list_keys, Data = #statem_data{connection_name = Conn})
     rabbit_log:debug("etcd peer discovery: will use prefix ~ts to query for node keys", [Prefix]),
     {ok, #{kvs := Result}} = eetcd_kv:get(C2),
     rabbit_log:debug("etcd peer discovery returned keys: ~tp", [Result]),
-    Values = [maps:get(value, M) || M <- Result],
-    rabbit_log:debug("etcd peer discovery: listing node keys returned ~b results", [length(Values)]),
-    ParsedNodes = lists:map(fun extract_node/1, Values),
-    {Successes, Failures} = lists:partition(fun filter_node/1, ParsedNodes),
-    JoinedString = lists:join(",", [rabbit_data_coercion:to_list(Node) || Node <- lists:usort(Successes)]),
-    rabbit_log:error("etcd peer discovery: successfully extracted nodes: ~ts", [JoinedString]),
-    lists:foreach(fun(Val) ->
-        rabbit_log:error("etcd peer discovery: failed to extract node name from etcd value ~tp", [Val])
-    end, Failures),
-    gen_statem:reply(From, lists:usort(Successes)),
+    Values = [{maps:get(create_revision, M), maps:get(value, M)} || M <- Result],
+    rabbit_log:debug("etcd peer discovery: listing node keys returned ~b results",
+                     [length(Values)]),
+    ParsedNodes = lists:filtermap(fun extract_node/1, Values),
+    rabbit_log:info("etcd peer discovery: successfully extracted nodes: ~0tp",
+                    [ParsedNodes]),
+    gen_statem:reply(From, lists:usort(ParsedNodes)),
     keep_state_and_data.
 
 
@@ -298,15 +295,18 @@ registration_value(#statem_data{node_key_lease_id = LeaseID, node_key_ttl_in_sec
         <<"ttl">>      => TTL
     })).
 
--spec extract_node(binary()) -> atom() | {error, any()}.
-
-extract_node(Payload) ->
+extract_node({CreatedRev, Payload}) ->
     case rabbit_json:try_decode(Payload) of
-        {error, Error} -> {error, Error};
+        {error, _Error} ->
+            rabbit_log:error("etcd peer discovery: failed to extract node name from etcd value ~tp",
+                             [Payload]),
+            false;
         {ok, Map} ->
             case maps:get(<<"node">>, Map, undefined) of
-                undefined -> undefined;
-                Node      -> rabbit_data_coercion:to_atom(Node)
+                undefined ->
+                    false;
+                Node ->
+                    {true, {CreatedRev, rabbit_data_coercion:to_atom(Node)}}
             end
     end.
 

deps/rabbitmq_peer_discovery_etcd/test/system_SUITE.erl

@@ -265,7 +265,12 @@ registration_with_locking_test(Config) ->
     ?assertEqual(ok, rabbitmq_peer_discovery_etcd_v3_client:unlock(Pid, LockOwnerKey)),
 
     Condition2 = fun() ->
-                    [node()] =:= rabbitmq_peer_discovery_etcd_v3_client:list_nodes(Pid)
+                    case rabbitmq_peer_discovery_etcd_v3_client:list_nodes(Pid) of
+                        [{_, N}] when N =:= node() ->
+                            true;
+                        _ ->
+                            false
+                    end
                  end,
     try
         rabbit_ct_helpers:await_condition(Condition2, 45000)

deps/rabbitmq_peer_discovery_etcd/test/unit_SUITE.erl

@@ -55,10 +55,14 @@ registration_value_test(_Config) ->
 extract_nodes_case1_test(_Config) ->
     Input    = registration_value_of(8488283859587364900, 61),
     Expected = node(),
-
-    ?assertEqual(Expected, rabbitmq_peer_discovery_etcd_v3_client:extract_node(Input)),
-
-    ?assertEqual(undefined, rabbitmq_peer_discovery_etcd_v3_client:extract_node(<<"{}">>)).
+    CreatedRev = ?LINE,
+    ?assertEqual({true, {CreatedRev, Expected}},
+                 rabbitmq_peer_discovery_etcd_v3_client:extract_node(
+                   {CreatedRev, Input})),
+
+    ?assertEqual(false,
+                 rabbitmq_peer_discovery_etcd_v3_client:extract_node(
+                   {CreatedRev, <<"{}">>})).
 
 filter_nodes_test(_Config) ->
     Input    = [node(), undefined, undefined, {error, reason1}, {error, {another, reason}}],