Auto discovery (#17)

* elasticache cluster auto discovery

Author: Djordje

Makefile

@@ -24,10 +24,11 @@ clean:
 
 test: logs_clean all
 	@$(REBAR)  skip_deps=true ct
+	@$(REBAR)  skip_deps=true verbose=1 eunit
 
 testfast:
 	@$(REBAR)  skip_deps=true ct
+	@$(REBAR)  skip_deps=true eunit
 
 xref:
 	@$(REBAR) skip_deps=true xref
-

README.md

@@ -69,6 +69,22 @@ algorithms available are `shard_phash2` and `shard_crc32`.
 
 ```
 
+Cluster Auto Discovery
+======================
+
+Configuration can also be implemented to support auto discovery of the cluster as opposed to hardcoding nodes. Provide the configuration endpoint ([AWS Reference](http://docs.aws.amazon.com/AmazonElastiCache/latest/UserGuide/AutoDiscovery.html)) and port as in the following example.
+
+```
+    [{cluster_b,
+        [{servers, {elasticache, "ConfigEndpointHostname.com", PortNumber}},
+             {sharding_algorithm, {mero, shard_crc32}},
+             {workers_per_shard, 1},
+             {pool_worker_module, mero_wrk_tcp_binary}]
+    },
+    ...
+]
+
+```
 
 Using Mero:
 ===============

src/mero_sup.erl

@@ -49,11 +49,14 @@
 -spec start_link(ClusterConfig :: list({ClusterName :: atom(),
                                         Config :: list()})) ->
     {ok, Pid :: pid()} | {error, Reason :: term()}.
-start_link(ClusterConfig) ->
+start_link(Config) ->
+    ClusterConfig = process_server_specs(Config),
+
     ok = mero_cluster:load_clusters(ClusterConfig),
     PoolDefs = mero_cluster:child_definitions(),
     supervisor:start_link({local, ?MODULE}, ?MODULE, [PoolDefs]).
 
+
 %%%===================================================================
 %%% Supervisor callbacks
 %%%===================================================================
@@ -68,3 +71,77 @@ init([PoolDefs]) ->
 child(I, Type, {ClusterName, Host, Port, Name, WrkModule}) ->
     {Name, {I, start_link, [ClusterName, Host, Port, Name, WrkModule]}, permanent,
       5000, Type, [I]}.
+
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+%% Given an elasticache config Endpoint and port, return parsed list of {host, port} nodes in cluster
+-spec get_cluster_config(string(), integer()) -> list({string(), integer()}).
+get_cluster_config(ConfigHost, ConfigPort) ->
+    {ok, [{banner, <<"CONFIG cluster", _/binary>>},
+          {version, VersionLine},
+          {hosts, HostLine},
+          {crlf, <<"\r\n">>},
+          {eom, <<"END\r\n">>}]} =
+        request_response(ConfigHost, ConfigPort,
+                         <<"config get cluster\n">>,
+                         [banner, version, hosts, crlf, eom]),
+    {_Version, Hosts} = parse_cluster_config(HostLine, VersionLine),
+    Hosts.
+
+process_server_specs(L) ->
+    lists:foldl(fun ({ClusterName, AttrPlist}, Acc) ->
+                        [{ClusterName, [process_value(Attr)
+                                        || Attr <- AttrPlist]} | Acc]
+                end, [], L).
+
+process_value({servers, {elasticache, ConfigEndpoint, ConfigPort}}) ->
+    HostsPorts = get_cluster_config(ConfigEndpoint, ConfigPort),
+    {servers, HostsPorts};
+process_value(V) ->
+    V.
+
+request_response(Host, Port, Command, Names) ->
+    Opts = [binary, {packet, line}, {active, false}],
+    {ok, Socket} = gen_tcp:connect(Host, Port, Opts),
+    ok = gen_tcp:send(Socket, Command),
+    Lines = [{Name, begin
+                        {ok, Line} = gen_tcp:recv(Socket, 0, 1000),
+                        Line
+                    end}
+             || Name <- Names],
+    ok = gen_tcp:close(Socket),
+    {ok, Lines}.
+
+%% Parse host and version lines to return version and list of {host, port} cluster nodes
+-spec parse_cluster_config(binary(), binary()) -> {integer(), [{string(), integer()}]}.
+parse_cluster_config(HostLine, VersionLine) ->
+    HostSpecs = re:split(butlast(HostLine), <<" ">>),
+    {binary_to_integer(butlast(VersionLine)),
+     [begin
+          [Host, _IP, Port] = re:split(HIP, "\\|"),
+          {binary_to_list(Host), binary_to_integer(Port)}
+      end
+      || HIP <- HostSpecs]}.
+
+butlast(<<>>) -> <<>>;
+butlast(Bin) -> binary:part(Bin, {0, size(Bin) - 1}).
+
+%%%===================================================================
+%%% Unit tests
+%%%===================================================================
+
+-ifdef(TEST).
+
+-include_lib("eunit/include/eunit.hrl").
+
+get_cluster_config_test() ->
+    VersionLine = <<"1\n">>,
+    HostLine = <<"server1.cache.amazonaws.com|10.100.100.100|11211 server2.cache.amazonaws.com|10.101.101.00|11211 server3.cache.amazonaws.com|10.102.00.102|11211\n">>,
+    ExpectedParse = {1, [{"server1.cache.amazonaws.com", 11211}, {"server2.cache.amazonaws.com", 11211}, {"server3.cache.amazonaws.com", 11211}]},
+
+    ?assertEqual(ExpectedParse, parse_cluster_config(HostLine, VersionLine)).
+
+-endif.