forked from canonical/go-dqlite
-
Notifications
You must be signed in to change notification settings - Fork 0
/
node.go
166 lines (142 loc) · 4.11 KB
/
node.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
package dqlite
import (
"time"
"github.com/canonical/go-dqlite/client"
"github.com/canonical/go-dqlite/internal/bindings"
"github.com/pkg/errors"
)
// Node runs a dqlite node.
type Node struct {
log client.LogFunc // Logger
server *bindings.Node // Low-level C implementation
acceptCh chan error // Receives connection handling errors
id uint64
address string
bindAddress string
}
// NodeInfo is a convenience alias for client.NodeInfo.
type NodeInfo = client.NodeInfo
// Option can be used to tweak node parameters.
type Option func(*options)
// WithDialFunc sets a custom dial function for the server.
func WithDialFunc(dial client.DialFunc) Option {
return func(options *options) {
options.DialFunc = dial
}
}
// WithBindAddress sets a custom bind address for the server.
func WithBindAddress(address string) Option {
return func(options *options) {
options.BindAddress = address
}
}
// WithNetworkLatency sets the average one-way network latency.
func WithNetworkLatency(latency time.Duration) Option {
return func(options *options) {
options.NetworkLatency = uint64(latency.Nanoseconds())
}
}
// WithFailureDomain sets the code of the failure domain the node belongs to.
func WithFailureDomain(code uint64) Option {
return func(options *options) {
options.FailureDomain = code
}
}
// New creates a new Node instance.
func New(id uint64, address string, dir string, options ...Option) (*Node, error) {
o := defaultOptions()
for _, option := range options {
option(o)
}
server, err := bindings.NewNode(id, address, dir)
if err != nil {
return nil, err
}
if o.DialFunc != nil {
if err := server.SetDialFunc(o.DialFunc); err != nil {
return nil, err
}
}
if o.BindAddress != "" {
if err := server.SetBindAddress(o.BindAddress); err != nil {
return nil, err
}
}
if o.NetworkLatency != 0 {
if err := server.SetNetworkLatency(o.NetworkLatency); err != nil {
return nil, err
}
}
if o.FailureDomain != 0 {
if err := server.SetFailureDomain(o.FailureDomain); err != nil {
return nil, err
}
}
s := &Node{
server: server,
acceptCh: make(chan error, 1),
id: id,
address: address,
bindAddress: o.BindAddress,
}
return s, nil
}
// BindAddress returns the network address the node is listening to.
func (s *Node) BindAddress() string {
return s.server.GetBindAddress()
}
// Start serving requests.
func (s *Node) Start() error {
return s.server.Start()
}
// Recover a node by forcing a new cluster configuration.
//
// DEPRECATED: Use ReconfigureMembership instead, which does not require
// instantiating a new Node object.
func (s *Node) Recover(cluster []NodeInfo) error {
return s.server.Recover(cluster)
}
// Hold configuration options for a dqlite server.
type options struct {
Log client.LogFunc
DialFunc client.DialFunc
BindAddress string
NetworkLatency uint64
FailureDomain uint64
}
// Close the server, releasing all resources it created.
func (s *Node) Close() error {
// Send a stop signal to the dqlite event loop.
if err := s.server.Stop(); err != nil {
return errors.Wrap(err, "server failed to stop")
}
s.server.Close()
return nil
}
// BootstrapID is a magic ID that should be used for the fist node in a
// cluster. Alternatively ID 1 can be used as well.
const BootstrapID = 0x2dc171858c3155be
// GenerateID generates a unique ID for a new node, based on a hash of its
// address and the current time.
func GenerateID(address string) uint64 {
return bindings.GenerateID(address)
}
// ReconfigureMembership can be used to recover a cluster whose majority of
// nodes have died, and therefore has become unavailable.
//
// It forces appending a new configuration to the raft log stored in the given
// directory, effectively replacing the current configuration.
func ReconfigureMembership(dir string, cluster []NodeInfo) error {
server, err := bindings.NewNode(1, "1", dir)
if err != nil {
return err
}
defer server.Close()
return server.Recover(cluster)
}
// Create a options object with sane defaults.
func defaultOptions() *options {
return &options{
DialFunc: client.DefaultDialFunc,
}
}