rabbitmq_peer_discovery_consul: Handle locking inside list_nodes/0

[Why]
The new implementation of `rabbit_peer_discovery` acquires the lock only
when a node needs to join another one. This is meant to disappear in the
medium/long term anyway.

Here, we need to lock the query to Consul to make sure that queries
happen sequentially, not concurrently. This is a work in progress and we
may not keep it either.

Author: dumbbell

deps/rabbitmq_peer_discovery_consul/src/rabbit_peer_discovery_consul.erl

@@ -68,21 +68,30 @@ list_nodes() ->
                    {ok, {[], disc}}
            end,
     Fun2 = fun(Proplist) ->
-                   M = maps:from_list(Proplist),
-                   Path = rabbit_peer_discovery_httpc:build_path([v1, health, service, service_name()]),
-                   HttpOpts = http_options(M),
-                   case rabbit_peer_discovery_httpc:get(get_config_key(consul_scheme, M),
-                                                        get_config_key(consul_host, M),
-                                                        get_integer_config_key(consul_port, M),
-                                                        Path,
-                                                        list_nodes_query_args(),
-                                                        maybe_add_acl([]),
-                                                        HttpOpts) of
-                       {ok, Nodes} ->
-                           IncludeWithWarnings = get_config_key(consul_include_nodes_with_warnings, M),
-                           Result = extract_nodes(
-                                      filter_nodes(Nodes, IncludeWithWarnings)),
-                           {ok, {Result, disc}};
+                   case internal_lock() of
+                       {ok, Priv} ->
+                           try
+                               M = maps:from_list(Proplist),
+                               Path = rabbit_peer_discovery_httpc:build_path([v1, health, service, service_name()]),
+                               HttpOpts = http_options(M),
+                               case rabbit_peer_discovery_httpc:get(get_config_key(consul_scheme, M),
+                                                                    get_config_key(consul_host, M),
+                                                                    get_integer_config_key(consul_port, M),
+                                                                    Path,
+                                                                    list_nodes_query_args(),
+                                                                    maybe_add_acl([]),
+                                                                    HttpOpts) of
+                                   {ok, Nodes} ->
+                                       IncludeWithWarnings = get_config_key(consul_include_nodes_with_warnings, M),
+                                       Result = extract_nodes(
+                                                  filter_nodes(Nodes, IncludeWithWarnings)),
+                                       {ok, {Result, disc}};
+                                   {error, _} = Error ->
+                                       Error
+                               end
+                           after
+                               internal_unlock(Priv)
+                           end;
                        {error, _} = Error ->
                            Error
                    end
@@ -164,9 +173,20 @@ post_registration() ->
     ok.
 
 -spec lock(Nodes :: [node()]) ->
-    {ok, Data :: term()} | {error, Reason :: string()}.
+    not_supported.
 
 lock(_Nodes) ->
+    not_supported.
+
+-spec unlock(Data :: term()) -> ok.
+
+unlock(_Data) ->
+    ok.
+
+-spec internal_lock() ->
+    {ok, Data :: term()} | {error, Reason :: string()}.
+
+internal_lock() ->
     M = ?CONFIG_MODULE:config_map(?BACKEND_CONFIG_KEY),
     ?LOG_DEBUG(
        "Effective Consul peer discovery configuration: ~tp", [M],
@@ -179,13 +199,13 @@ lock(_Nodes) ->
             EndTime = Now + get_config_key(lock_wait_time, M),
             lock(TRef, SessionId, Now, EndTime);
         {error, Reason} ->
-            {error, lists:flatten(io_lib:format("Error while creating a session, reason: ~ts",
+            {error, lists:flatten(io_lib:format("Error while creating a session, reason: ~0p",
                                                 [Reason]))}
     end.
 
--spec unlock({SessionId :: string(), TRef :: timer:tref()}) -> ok.
+-spec internal_unlock({SessionId :: string(), TRef :: timer:tref()}) -> ok.
 
-unlock({SessionId, TRef}) ->
+internal_unlock({SessionId, TRef}) ->
     _ = timer:cancel(TRef),
     ?LOG_DEBUG(
        "Stopped session renewal",
@@ -620,7 +640,7 @@ wait_for_list_nodes(N) ->
 %% Create a session to be acquired for a common key
 %% @end
 %%--------------------------------------------------------------------
--spec create_session(atom(), pos_integer()) -> {ok, string()} | {error, Reason::string()}.
+-spec create_session(atom(), pos_integer()) -> {ok, string()} | {error, Reason::any()}.
 create_session(Name, TTL) ->
     case consul_session_create([], maybe_add_acl([]),
                                [{'Name', Name},
@@ -705,7 +725,7 @@ start_session_ttl_updater(SessionId) ->
 %% Tries to acquire lock. If the lock is held by someone else, waits until it
 %% is released, or too much time has passed
 %% @end
--spec lock(timer:tref(), string(), pos_integer(), pos_integer()) -> {ok, string()} | {error, string()}.
+-spec lock(timer:tref(), string(), pos_integer(), pos_integer()) -> {ok, {SessionId :: string(), TRef :: timer:tref()}} | {error, string()}.
 lock(TRef, _, Now, EndTime) when EndTime < Now ->
     _ = timer:cancel(TRef),
     {error, "Acquiring lock taking too long, bailing out"};

deps/rabbitmq_peer_discovery_consul/src/rabbitmq_peer_discovery_consul.erl

@@ -42,7 +42,7 @@ unregister() ->
 post_registration() ->
     ?DELEGATE:post_registration().
 
--spec lock(Nodes :: [node()]) -> {ok, Data :: term()} | {error, Reason :: string()}.
+-spec lock(Nodes :: [node()]) -> not_supported.
 lock(Node) ->
     ?DELEGATE:lock(Node).
 

deps/rabbitmq_peer_discovery_consul/test/rabbitmq_peer_discovery_consul_SUITE.erl

@@ -345,6 +345,15 @@ list_nodes_return_value_basic_test(_Config) ->
                                                            {consul_port, 8500}
                                                           ]}
                         ]),
+    meck:expect(rabbit_peer_discovery_httpc, put,
+             fun
+                 (_, _, _, "v1/session/create", _, _, _, _) ->
+                     Body = "{\"ID\":\"some-session-id\"}",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body));
+                 (_, _, _, "v1/kv/rabbitmq/default/startup_lock", _, _, _, _) ->
+                     Body = "true",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body))
+             end),
     meck:expect(rabbit_peer_discovery_httpc, get,
              fun(_, _, _, _, _, _, _) ->
                Body = "[{\"Node\": {\"Node\": \"rabbit2.internal.domain\", \"Address\": \"10.20.16.160\"}, \"Checks\": [{\"Node\": \"rabbit2.internal.domain\", \"CheckID\": \"service:rabbitmq\", \"Name\": \"Service \'rabbitmq\' check\", \"ServiceName\": \"rabbitmq\", \"Notes\": \"Connect to the port internally every 30 seconds\", \"Status\": \"passing\", \"ServiceID\": \"rabbitmq\", \"Output\": \"\"}, {\"Node\": \"rabbit2.internal.domain\", \"CheckID\": \"serfHealth\", \"Name\": \"Serf Health Status\", \"ServiceName\": \"\", \"Notes\": \"\", \"Status\": \"passing\", \"ServiceID\": \"\", \"Output\": \"Agent alive and reachable\"}], \"Service\": {\"Address\": \"\", \"Port\": 5672, \"ID\": \"rabbitmq\", \"Service\": \"rabbitmq\", \"Tags\": [\"amqp\"]}}, {\"Node\": {\"Node\": \"rabbit1.internal.domain\", \"Address\": \"10.20.16.159\"}, \"Checks\": [{\"Node\": \"rabbit1.internal.domain\", \"CheckID\": \"service:rabbitmq\", \"Name\": \"Service \'rabbitmq\' check\", \"ServiceName\": \"rabbitmq\", \"Notes\": \"Connect to the port internally every 30 seconds\", \"Status\": \"passing\", \"ServiceID\": \"rabbitmq\", \"Output\": \"\"}, {\"Node\": \"rabbit1.internal.domain\", \"CheckID\": \"serfHealth\", \"Name\": \"Serf Health Status\", \"ServiceName\": \"\", \"Notes\": \"\", \"Status\": \"passing\", \"ServiceID\": \"\", \"Output\": \"Agent alive and reachable\"}], \"Service\": {\"Address\": \"\", \"Port\": 5672, \"ID\": \"rabbitmq\", \"Service\": \"rabbitmq\", \"Tags\": [\"amqp\"]}}]",
@@ -364,6 +373,15 @@ list_nodes_return_value_basic_long_node_name_test(_Config) ->
                                                            {consul_port,         8500}
                                                           ]}
                         ]),
+    meck:expect(rabbit_peer_discovery_httpc, put,
+             fun
+                 (_, _, _, "v1/session/create", _, _, _, _) ->
+                     Body = "{\"ID\":\"some-session-id\"}",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body));
+                 (_, _, _, "v1/kv/rabbitmq/default/startup_lock", _, _, _, _) ->
+                     Body = "true",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body))
+             end),
     meck:expect(rabbit_peer_discovery_httpc, get,
              fun(_, _, _, _, _, _, _) ->
                Body = "[{\"Node\": {\"Node\": \"rabbit2\", \"Address\": \"10.20.16.160\"}, \"Checks\": [{\"Node\": \"rabbit2\", \"CheckID\": \"service:rabbitmq\", \"Name\": \"Service \'rabbitmq\' check\", \"ServiceName\": \"rabbitmq\", \"Notes\": \"Connect to the port internally every 30 seconds\", \"Status\": \"passing\", \"ServiceID\": \"rabbitmq\", \"Output\": \"\"}, {\"Node\": \"rabbit2\", \"CheckID\": \"serfHealth\", \"Name\": \"Serf Health Status\", \"ServiceName\": \"\", \"Notes\": \"\", \"Status\": \"passing\", \"ServiceID\": \"\", \"Output\": \"Agent alive and reachable\"}], \"Service\": {\"Address\": \"\", \"Port\": 5672, \"ID\": \"rabbitmq\", \"Service\": \"rabbitmq\", \"Tags\": [\"amqp\"]}}, {\"Node\": {\"Node\": \"rabbit1\", \"Address\": \"10.20.16.159\"}, \"Checks\": [{\"Node\": \"rabbit1\", \"CheckID\": \"service:rabbitmq\", \"Name\": \"Service \'rabbitmq\' check\", \"ServiceName\": \"rabbitmq\", \"Notes\": \"Connect to the port internally every 30 seconds\", \"Status\": \"passing\", \"ServiceID\": \"rabbitmq\", \"Output\": \"\"}, {\"Node\": \"rabbit1\", \"CheckID\": \"serfHealth\", \"Name\": \"Serf Health Status\", \"ServiceName\": \"\", \"Notes\": \"\", \"Status\": \"passing\", \"ServiceID\": \"\", \"Output\": \"Agent alive and reachable\"}], \"Service\": {\"Address\": \"\", \"Port\": 5672, \"ID\": \"rabbitmq\", \"Service\": \"rabbitmq\", \"Tags\": [\"amqp\"]}}]",
@@ -384,6 +402,15 @@ list_nodes_return_value_long_node_name_and_custom_domain_test(_Config) ->
                                                            {consul_domain,       "internal"}
                                                           ]}
                         ]),
+    meck:expect(rabbit_peer_discovery_httpc, put,
+             fun
+                 (_, _, _, "v1/session/create", _, _, _, _) ->
+                     Body = "{\"ID\":\"some-session-id\"}",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body));
+                 (_, _, _, "v1/kv/rabbitmq/default/startup_lock", _, _, _, _) ->
+                     Body = "true",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body))
+             end),
     meck:expect(rabbit_peer_discovery_httpc, get,
                 fun(_, _, _, _, _, _, _) ->
                         Body = "[{\"Node\": {\"Node\": \"rabbit2\", \"Address\": \"10.20.16.160\"}, \"Checks\": [{\"Node\": \"rabbit2\", \"CheckID\": \"service:rabbitmq\", \"Name\": \"Service \'rabbitmq\' check\", \"ServiceName\": \"rabbitmq\", \"Notes\": \"Connect to the port internally every 30 seconds\", \"Status\": \"passing\", \"ServiceID\": \"rabbitmq\", \"Output\": \"\"}, {\"Node\": \"rabbit2\", \"CheckID\": \"serfHealth\", \"Name\": \"Serf Health Status\", \"ServiceName\": \"\", \"Notes\": \"\", \"Status\": \"passing\", \"ServiceID\": \"\", \"Output\": \"Agent alive and reachable\"}], \"Service\": {\"Address\": \"\", \"Port\": 5672, \"ID\": \"rabbitmq\", \"Service\": \"rabbitmq\", \"Tags\": [\"amqp\"]}}, {\"Node\": {\"Node\": \"rabbit1\", \"Address\": \"10.20.16.159\"}, \"Checks\": [{\"Node\": \"rabbit1\", \"CheckID\": \"service:rabbitmq\", \"Name\": \"Service \'rabbitmq\' check\", \"ServiceName\": \"rabbitmq\", \"Notes\": \"Connect to the port internally every 30 seconds\", \"Status\": \"passing\", \"ServiceID\": \"rabbitmq\", \"Output\": \"\"}, {\"Node\": \"rabbit1\", \"CheckID\": \"serfHealth\", \"Name\": \"Serf Health Status\", \"ServiceName\": \"\", \"Notes\": \"\", \"Status\": \"passing\", \"ServiceID\": \"\", \"Output\": \"Agent alive and reachable\"}], \"Service\": {\"Address\": \"\", \"Port\": 5672, \"ID\": \"rabbitmq\", \"Service\": \"rabbitmq\", \"Tags\": [\"amqp\"]}}]",
@@ -405,6 +432,15 @@ list_nodes_return_value_srv_address_test(_Config) ->
                                                            {consul_port,         8500}
                                                           ]}
                         ]),
+    meck:expect(rabbit_peer_discovery_httpc, put,
+             fun
+                 (_, _, _, "v1/session/create", _, _, _, _) ->
+                     Body = "{\"ID\":\"some-session-id\"}",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body));
+                 (_, _, _, "v1/kv/rabbitmq/default/startup_lock", _, _, _, _) ->
+                     Body = "true",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body))
+             end),
     meck:expect(rabbit_peer_discovery_httpc, get,
              fun(_, _, _, _, _, _, _) ->
                Body = "[{\"Node\": {\"Node\": \"rabbit2.internal.domain\", \"Address\": \"10.20.16.160\"}, \"Checks\": [{\"Node\": \"rabbit2.internal.domain\", \"CheckID\": \"service:rabbitmq\", \"Name\": \"Service \'rabbitmq\' check\", \"ServiceName\": \"rabbitmq\", \"Notes\": \"Connect to the port internally every 30 seconds\", \"Status\": \"passing\", \"ServiceID\": \"rabbitmq:172.172.16.4.50\", \"Output\": \"\"}, {\"Node\": \"rabbit2.internal.domain\", \"CheckID\": \"serfHealth\", \"Name\": \"Serf Health Status\", \"ServiceName\": \"\", \"Notes\": \"\", \"Status\": \"passing\", \"ServiceID\": \"\", \"Output\": \"Agent alive and reachable\"}], \"Service\": {\"Address\": \"172.16.4.51\", \"Port\": 5672, \"ID\": \"rabbitmq:172.16.4.51\", \"Service\": \"rabbitmq\", \"Tags\": [\"amqp\"]}}, {\"Node\": {\"Node\": \"rabbit1.internal.domain\", \"Address\": \"10.20.16.159\"}, \"Checks\": [{\"Node\": \"rabbit1.internal.domain\", \"CheckID\": \"service:rabbitmq\", \"Name\": \"Service \'rabbitmq\' check\", \"ServiceName\": \"rabbitmq\", \"Notes\": \"Connect to the port internally every 30 seconds\", \"Status\": \"passing\", \"ServiceID\": \"rabbitmq\", \"Output\": \"\"}, {\"Node\": \"rabbit1.internal.domain\", \"CheckID\": \"serfHealth\", \"Name\": \"Serf Health Status\", \"ServiceName\": \"\", \"Notes\": \"\", \"Status\": \"passing\", \"ServiceID\": \"\", \"Output\": \"Agent alive and reachable\"}], \"Service\": {\"Address\": \"172.172.16.51\", \"Port\": 5672, \"ID\": \"rabbitmq:172.172.16.51\", \"Service\": \"rabbitmq\", \"Tags\": [\"amqp\"]}}]",
@@ -423,6 +459,15 @@ list_nodes_return_value_nodes_in_warning_state_included_test(_Config) ->
                                                            {consul_port,         8500}
                                                           ]}
                         ]),
+    meck:expect(rabbit_peer_discovery_httpc, put,
+             fun
+                 (_, _, _, "v1/session/create", _, _, _, _) ->
+                     Body = "{\"ID\":\"some-session-id\"}",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body));
+                 (_, _, _, "v1/kv/rabbitmq/default/startup_lock", _, _, _, _) ->
+                     Body = "true",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body))
+             end),
     meck:expect(rabbit_peer_discovery_httpc, get,
                 fun(_, _, _, _, [], _, _) ->
                         rabbit_json:try_decode(list_of_nodes_with_warnings());
@@ -443,6 +488,15 @@ list_nodes_return_value_nodes_in_warning_state_filtered_out_test(_Config) ->
                                                            {consul_port,         8500}
                                                           ]}
                         ]),
+    meck:expect(rabbit_peer_discovery_httpc, put,
+             fun
+                 (_, _, _, "v1/session/create", _, _, _, _) ->
+                     Body = "{\"ID\":\"some-session-id\"}",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body));
+                 (_, _, _, "v1/kv/rabbitmq/default/startup_lock", _, _, _, _) ->
+                     Body = "true",
+                     rabbit_json:try_decode(rabbit_data_coercion:to_binary(Body))
+             end),
     meck:expect(rabbit_peer_discovery_httpc, get,
              fun(_, _, _, _, [], _, _) ->
                      rabbit_json:try_decode(list_of_nodes_with_warnings());