this repo has no description
Write initial config from ClusterNode
+87
-29
+55
-15
lib/cluster_node.ex
+55
-15
lib/cluster_node.ex
···
1
1
defmodule Hobbes.ClusterNode do
2
2
use GenServer
3
-
alias Trinity.SimServer
3
+
alias Trinity.{SimServer, SimProcess}
4
4
5
+
alias Hobbes.ClusterConfig
5
6
alias Hobbes.Servers.{Coordinator, ServerSupervisor}
6
7
7
8
import ExUnit.Assertions, only: [assert: 1]
···
9
10
@spec start_link(keyword) :: term
10
11
def start_link(config) when is_list(config), do: SimServer.start_link(__MODULE__, config)
11
12
13
+
defmodule NodeConfig do
14
+
@enforce_keys [
15
+
:cluster_name,
16
+
17
+
:coordinator_names,
18
+
:coordinator_id,
19
+
:initial_cluster_config,
20
+
21
+
:slots,
22
+
]
23
+
defstruct @enforce_keys
24
+
25
+
def from_opts(opts) do
26
+
# TODO: validate config with helpful error messages
27
+
%NodeConfig{
28
+
cluster_name: Keyword.fetch!(opts, :cluster),
29
+
coordinator_names: Keyword.fetch!(opts, :coordinators),
30
+
coordinator_id: Keyword.get(opts, :coordinator_id),
31
+
initial_cluster_config: Keyword.fetch!(opts, :initial_cluster_config),
32
+
slots: Keyword.fetch!(opts, :slots),
33
+
}
34
+
end
35
+
end
36
+
12
37
defmodule State do
13
38
@enforce_keys [
14
-
:config,
39
+
:node_config,
15
40
:coordinator_pid,
16
41
:server_supervisor_pid,
17
42
]
18
43
defstruct @enforce_keys
19
44
end
20
45
21
-
def init(config_opts) do
46
+
def init(opts) do
22
47
#SimProcess.flag(:trap_exit, true)
23
-
24
-
# TODO: validate config with helpful error messages
25
-
config = parse_config(config_opts)
48
+
node_config = NodeConfig.from_opts(opts)
26
49
27
-
coordinator_pid = start_coordinator(config)
28
-
server_supervisor_pid = start_server_supervisor(config)
50
+
coordinator_pid = start_coordinator(node_config)
51
+
server_supervisor_pid = start_server_supervisor(node_config)
29
52
30
53
state = %State{
31
-
config: config,
54
+
node_config: node_config,
32
55
coordinator_pid: coordinator_pid,
33
56
server_supervisor_pid: server_supervisor_pid,
34
57
}
35
58
59
+
:ok = ensure_cluster_config(state)
36
60
{:ok, state}
37
61
end
38
62
···
53
77
pid
54
78
end
55
79
56
-
defp parse_config(config) do
80
+
defp ensure_cluster_config(%State{} = state) do
57
81
%{
58
-
cluster_name: Keyword.fetch!(config, :cluster),
59
-
coordinator_names: Keyword.fetch!(config, :coordinators),
60
-
coordinator_id: Keyword.get(config, :coordinator_id),
61
-
slots: Keyword.fetch!(config, :slots),
62
-
}
82
+
coordinator_pid: coordinator_pid,
83
+
node_config: %NodeConfig{
84
+
initial_cluster_config: %ClusterConfig{} = cluster_config,
85
+
},
86
+
} = state
87
+
88
+
SimProcess.spawn_link(fn ->
89
+
do_ensure_cluster_config(coordinator_pid, cluster_config)
90
+
end)
91
+
:ok
92
+
end
93
+
94
+
defp do_ensure_cluster_config(coordinator_pid, %ClusterConfig{} = config) do
95
+
case Coordinator.write_initial_config(coordinator_pid, config) do
96
+
:ok -> :noop
97
+
{:error, {:not_primary, _}} -> :noop
98
+
{:error, :config_already_written} -> :noop
99
+
{:error, :timeout} ->
100
+
SimProcess.sleep(100)
101
+
do_ensure_cluster_config(coordinator_pid, config)
102
+
end
63
103
end
64
104
end
+12
-13
lib/hobbes.ex
+12
-13
lib/hobbes.ex
···
3
3
alias Hobbes.Servers.{Coordinator, Manager}
4
4
alias Hobbes.Structs.Cluster
5
5
6
-
alias Trinity.{Sim, SimProcess}
6
+
alias Trinity.Sim
7
7
8
8
defmodule AppShim do
9
9
use Application
···
19
19
end
20
20
21
21
# Sim-only, relies on simulated nodes
22
-
defp init_distributed_cluster(num_coordinators) do
22
+
defp init_distributed_cluster(num_coordinators, %ClusterConfig{} = cluster_config) do
23
23
coordinators =
24
24
Enum.map(0..(num_coordinators - 1), fn i ->
25
25
{
···
31
31
coordinators
32
32
|> Enum.with_index()
33
33
|> Enum.each(fn {{_name, node}, i} ->
34
-
config = [
34
+
node_config = [
35
35
cluster: "cluster",
36
36
coordinators: coordinators,
37
37
coordinator_id: i,
38
+
initial_cluster_config: cluster_config,
39
+
38
40
slots: [
39
41
stateless: 6,
40
42
tlog: [
···
47
49
mfa = {
48
50
Hobbes.AppShim,
49
51
:start,
50
-
[nil, [config]],
52
+
[nil, [node_config]],
51
53
}
52
54
Sim.create_node(node, mfa)
53
55
Sim.start_node(node)
···
56
58
coordinators
57
59
end
58
60
59
-
defp init_local_cluster(num_coordinators) do
61
+
defp init_local_cluster(num_coordinators, %ClusterConfig{} = cluster_config) do
60
62
coordinators =
61
63
Enum.map(0..(num_coordinators - 1), fn i ->
62
64
String.to_atom("coordinator-#{i}")
···
69
71
cluster: "cluster",
70
72
coordinators: coordinators,
71
73
coordinator_id: i,
74
+
initial_cluster_config: cluster_config,
75
+
72
76
slots: [
73
77
stateless: 6,
74
78
tlog: [
···
87
91
@spec start_cluster(Keyword.t) :: {:ok, [pid]}
88
92
def start_cluster(opts) do
89
93
num_coordinators = Keyword.get(opts, :num_coordinators, 3)
94
+
config = ClusterConfig.from_opts(opts)
90
95
91
96
coordinators =
92
97
case Sim.simulated?() do
93
-
true -> init_distributed_cluster(num_coordinators)
94
-
false -> init_local_cluster(num_coordinators)
98
+
true -> init_distributed_cluster(num_coordinators, config)
99
+
false -> init_local_cluster(num_coordinators, config)
95
100
end
96
-
97
-
SimProcess.sleep(1000)
98
-
99
-
config = ClusterConfig.from_opts(opts)
100
-
config_pairs = ClusterConfig.to_pairs(config)
101
-
:ok = Coordinator.write(hd(coordinators), config_pairs)
102
101
103
102
{
104
103
:ok,
+6
lib/servers/coordinator.ex
+6
lib/servers/coordinator.ex
···
5
5
6
6
import ExUnit.Assertions, only: [assert: 1]
7
7
8
+
alias Hobbes.ClusterConfig
8
9
alias Hobbes.Servers.{ServerSupervisor, Manager}
9
10
alias Hobbes.Servers.Coordinator.Commands
10
11
alias Hobbes.Structs.TLogGeneration
···
170
171
@spec dump(server) :: {:ok, [{binary, binary}]} | request_error
171
172
def dump(server) do
172
173
request(server, {Commands, :dump, []})
174
+
end
175
+
176
+
@spec write_initial_config(server, ClusterConfig.t) :: :ok | {:error, :config_already_written} | request_error
177
+
def write_initial_config(server, %ClusterConfig{} = config) do
178
+
request(server, {Commands, :write_initial_config, [config]})
173
179
end
174
180
175
181
@spec inc_generation(server) :: {:ok, map} | {:error, :no_config} | request_error
+14
-1
lib/servers/coordinator/commands.ex
+14
-1
lib/servers/coordinator/commands.ex
···
1
1
defmodule Hobbes.Servers.Coordinator.Commands do
2
-
import Hobbes.Utils
2
+
alias Hobbes.ClusterConfig
3
3
alias Hobbes.Structs.TLogGeneration
4
+
5
+
import Hobbes.Utils
4
6
5
7
def write(_get, _scan, [pairs]) when is_list(pairs) do
6
8
{pairs, :ok}
···
14
16
def dump(_get, scan, []) do
15
17
pairs = scan.("", special_prefix())
16
18
{[], {:ok, pairs}}
19
+
end
20
+
21
+
def write_initial_config(_get, scan, [%ClusterConfig{} = config]) do
22
+
case scan.(coordinator_config_prefix(), coordinator_config_end()) do
23
+
[] ->
24
+
config_pairs = ClusterConfig.to_pairs(config)
25
+
{config_pairs, :ok}
26
+
27
+
[_ | _] ->
28
+
{[], {:error, :config_already_written}}
29
+
end
17
30
end
18
31
19
32
def inc_generation(get, scan, []) do