leaderless gossip-based service discovery w/ consul compatibility
use global logger
+92
-39
+14
-13
agent.go
+14
-13
agent.go
···
2
2
3
3
import (
4
4
"fmt"
5
-
"log"
6
5
"sync"
7
6
8
7
"github.com/hashicorp/serf/serf"
8
+
"golang.org/x/exp/slog"
9
9
)
10
10
11
11
// Agent starts and manages a Serf & adds service discovery (TODO)
12
12
type Agent struct {
13
13
serfConf *serf.Config
14
14
eventCh chan serf.Event
15
-
16
15
// This is the underlying Serf we are wrapping
17
16
serf *serf.Serf
18
-
17
+
// a lil logger used by the agent, you know how it be
18
+
logger *slog.Logger
19
19
// this channel receiving a signal = shutdown
20
20
shutdown bool
21
21
shutdownCh chan struct{}
22
22
shutdownLock sync.Mutex
23
23
}
24
24
25
-
func Create(serfConf *serf.Config) *Agent {
25
+
func Create(config *Config, serfConf *serf.Config, logger *slog.Logger) *Agent {
26
26
// Create a channel to listen for events from Serf
27
27
eventCh := make(chan serf.Event, 64)
28
28
serfConf.EventCh = eventCh
···
31
31
agent := &Agent{
32
32
serfConf: serfConf,
33
33
eventCh: eventCh,
34
+
logger: logger,
34
35
shutdownCh: make(chan struct{}),
35
36
}
36
37
···
42
43
for {
43
44
select {
44
45
case e := <-a.eventCh:
45
-
log.Printf("[INFO] cascade: Received event: %s", e.String())
46
+
a.logger.Info("receive event", "event", e.String())
46
47
47
48
case <-serfShutdownCh:
48
-
log.Printf("[INFO] cascade: serf shutdown detected, quitting eventloop")
49
+
a.logger.Info("detect serf shutdown, turning off event loop")
49
50
a.Shutdown()
50
51
return
51
52
···
56
57
}
57
58
58
59
func (a *Agent) Start() error {
59
-
log.Printf("[INFO] cascade: Serf agent starting")
60
+
a.logger.Info("start serf agent")
60
61
61
62
// Create serf first
62
63
serf, err := serf.Create(a.serfConf)
63
64
if err != nil {
64
-
return fmt.Errorf("Error creating Serf: %s", err)
65
+
return err
65
66
}
66
67
a.serf = serf
67
68
···
84
85
goto EXIT
85
86
}
86
87
87
-
log.Println("[INFO] cascade: requesting serf shutdown")
88
+
a.logger.Info("request serf shutdown")
88
89
if err := a.serf.Shutdown(); err != nil {
89
90
return err
90
91
}
91
92
92
93
EXIT:
93
-
log.Println("[INFO] cascade: shutdown complete")
94
+
a.logger.Info("complete serf shutdown")
94
95
a.shutdown = true
95
96
close(a.shutdownCh)
96
97
return nil
···
98
99
99
100
// Join asks the Serf instance to join. See the Serf.Join function.
100
101
func (a *Agent) Join(addrs []string) (n int, err error) {
101
-
log.Printf("[INFO] cascade: issuing join request to '%v'", addrs)
102
+
a.logger.Info("issue join request", "addrs", addrs)
102
103
// we always ignore old events because cascade don't
103
104
// care about the past
104
105
n, err = a.serf.Join(addrs, true)
105
106
if err != nil {
106
-
return n, fmt.Errorf("[ERR] cascade: error joining: %v\n", err)
107
+
return n, fmt.Errorf("Error joining: %v\n", err)
107
108
}
108
109
// TODO: when joining fails, we don't get an error here - serf & memberlist
109
110
// just print to stdout and serf.Join returns without issue.
···
116
117
return nil
117
118
}
118
119
119
-
log.Println("[INFO] cascade: requesting graceful leave from Serf")
120
+
a.logger.Info("request graceful serf leave")
120
121
return a.serf.Leave()
121
122
}
122
123
+21
config.go
+21
config.go
···
4
4
"fmt"
5
5
"net"
6
6
"os"
7
+
8
+
"golang.org/x/exp/slog"
7
9
)
8
10
9
11
const DefaultBindPort int = 4449
···
16
18
17
19
return &Config{
18
20
BindAddr: "0.0.0.0",
21
+
LogLevel: "INFO",
19
22
NodeName: hostname,
20
23
}
21
24
}
···
30
33
31
34
type Config struct {
32
35
BindAddr string
36
+
LogLevel string
33
37
NodeName string
34
38
StartJoin []string
35
39
Services []Service
···
57
61
58
62
return addr.IP.String(), addr.Port, nil
59
63
}
64
+
65
+
func (c *Config) GetLogLevel() slog.Level {
66
+
var lvl slog.Level
67
+
if c.LogLevel == "DEBUG" {
68
+
lvl = slog.LevelDebug
69
+
}
70
+
if c.LogLevel == "INFO" {
71
+
lvl = slog.LevelInfo
72
+
}
73
+
if c.LogLevel == "WARN" || c.LogLevel == "WARNING" {
74
+
lvl = slog.LevelWarn
75
+
}
76
+
if c.LogLevel == "ERR" || c.LogLevel == "ERROR" {
77
+
lvl = slog.LevelError
78
+
}
79
+
return lvl
80
+
}
+6
-3
go.mod
+6
-3
go.mod
···
2
2
3
3
go 1.20
4
4
5
-
require github.com/hashicorp/serf v0.10.1
5
+
require (
6
+
github.com/hashicorp/memberlist v0.5.0
7
+
github.com/hashicorp/serf v0.10.1
8
+
golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb
9
+
)
6
10
7
11
require (
8
12
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da // indirect
···
13
17
github.com/hashicorp/go-multierror v1.1.0 // indirect
14
18
github.com/hashicorp/go-sockaddr v1.0.0 // indirect
15
19
github.com/hashicorp/golang-lru v0.5.0 // indirect
16
-
github.com/hashicorp/memberlist v0.5.0 // indirect
17
20
github.com/miekg/dns v1.1.41 // indirect
18
21
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
19
22
golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1 // indirect
20
-
golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 // indirect
23
+
golang.org/x/sys v0.1.0 // indirect
21
24
)
+4
go.sum
+4
go.sum
···
56
56
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
57
57
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
58
58
golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392/go.mod h1:/lpIB1dKB+9EgE3H3cr1v9wB50oz8l4C4h62xy7jSTY=
59
+
golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb h1:PaBZQdo+iSDyHT053FjUCgZQ/9uqVwPOcl7KSWhKn6w=
60
+
golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
59
61
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
60
62
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
61
63
golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
···
77
79
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
78
80
golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 h1:WIoqL4EROvwiPdUtaip4VcDdpZ4kha7wBWZrbVKCIZg=
79
81
golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
82
+
golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
83
+
golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
80
84
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
81
85
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
82
86
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+47
-23
main.go
+47
-23
main.go
···
10
10
11
11
import (
12
12
"fmt"
13
-
"log"
14
13
"os"
15
14
"os/signal"
16
15
"strings"
···
19
18
20
19
"github.com/hashicorp/memberlist"
21
20
"github.com/hashicorp/serf/serf"
21
+
"golang.org/x/exp/slog"
22
22
)
23
23
24
24
const (
···
29
29
30
30
func main() {
31
31
config := readConfig()
32
-
agent := setupAgent(config)
32
+
logger := setupLogger(config)
33
+
agent := setupAgent(config, logger)
33
34
if err := agent.Start(); err != nil {
34
-
log.Panic(err)
35
+
panic(err)
35
36
}
36
37
defer agent.Shutdown()
37
38
// join any specified startup nodes
38
39
if err := startupJoin(config, agent); err != nil {
39
-
log.Fatal(err)
40
+
panic(err)
40
41
}
41
-
if os.Getenv("CASCADE_DEBUG") != "" {
42
-
go debugPrints(agent)
42
+
go debugPrints(agent)
43
+
if err := handleSignals(config, agent); err != nil {
44
+
panic(err)
45
+
}
46
+
}
47
+
48
+
func setupLogger(config *Config) *slog.Logger {
49
+
debugOn := false
50
+
if config.LogLevel == "DEBUG" {
51
+
debugOn = true
43
52
}
44
-
if err := handleSignals(config, agent); err != nil {
45
-
log.Panic(err)
53
+
opts := slog.HandlerOptions{
54
+
Level: config.GetLogLevel(),
55
+
AddSource: debugOn,
46
56
}
57
+
handler := opts.NewTextHandler(os.Stderr)
58
+
logger := slog.New(handler)
59
+
slog.SetDefault(logger)
60
+
return logger
47
61
}
48
62
49
-
func debugPrints(agent *Agent) {
63
+
64
+
func debugPrints(a *Agent) {
50
65
for {
51
-
for _, m := range agent.serf.Members() {
52
-
log.Printf("[DEBUG] cascade: name '%s' addr '%s' status '%s' \n",
53
-
m.Name, m.Addr, m.Status)
66
+
for _, m := range a.serf.Members() {
67
+
a.logger.Debug("debug-loop", "name", m.Name, "addr", m.Addr, "status", m.Status)
54
68
}
55
69
time.Sleep(time.Second * 5)
56
70
}
···
70
84
// Agent is already shutdown!
71
85
return nil
72
86
}
73
-
log.Printf("Caught signal: %v", sig)
87
+
agent.logger.Info("caught signal", "signal", sig)
74
88
75
89
// Check if we should do a graceful leave
76
90
graceful := false
···
80
94
81
95
// Bail fast if not doing a graceful leave
82
96
if !graceful {
83
-
log.Fatal("[WARN] cascade: non-graceful leave detected")
97
+
agent.logger.Warn("leave cluster with zero grace")
98
+
return nil
84
99
}
85
100
86
101
// Attempt a graceful leave
87
102
gracefulCh := make(chan struct{})
88
-
log.Println("Gracefully shutting down agent...")
103
+
agent.logger.Info("shut down agent gracefully")
89
104
go func() {
90
105
if err := agent.Leave(); err != nil {
91
-
log.Printf("Error: %s", err)
106
+
agent.logger.Error("Error: %s", err)
92
107
return
93
108
}
94
109
close(gracefulCh)
···
97
112
// Wait for leave or another signal
98
113
select {
99
114
case <-signalCh:
100
-
return fmt.Errorf("[WARN] cascade: idk")
115
+
return fmt.Errorf("idfk")
101
116
case <-time.After(gracefulTimeout):
102
-
return fmt.Errorf("[WARN] cascade: leave timed out")
117
+
return fmt.Errorf("leave timed out")
103
118
case <-gracefulCh:
104
119
return nil
105
120
}
···
107
122
108
123
func readConfig() *Config {
109
124
config := DefaultConfig()
110
-
// CASCADE_BIND=127.0.0.1:1234
125
+
// CASCADE_LOGLEVEL=info
126
+
if os.Getenv("CASCADE_LOGLEVEL") != "" {
127
+
config.LogLevel = strings.ToUpper(os.Getenv("CASCADE_LOGLEVEL"))
128
+
}
129
+
// CASCADE_BIND=192.168.0.15:12345
111
130
if os.Getenv("CASCADE_BIND") != "" {
112
131
config.BindAddr = os.Getenv("CASCADE_BIND")
113
132
}
···
132
151
return err
133
152
}
134
153
if n > 0 {
135
-
log.Printf("[INFO] cascade: join request issued to %d nodes", n)
154
+
agent.logger.Info("issue join request", "nodes", n)
136
155
}
137
156
138
157
return nil
139
158
}
140
159
141
-
func setupAgent(config *Config) *Agent {
160
+
func setupAgent(config *Config, logger *slog.Logger) *Agent {
142
161
bindIP, bindPort, err := config.AddrParts(config.BindAddr)
143
162
if err != nil {
144
-
log.Panic(err)
163
+
panic(err)
145
164
}
146
165
147
166
serfConfig := serf.DefaultConfig()
167
+
// logLogger is just a bridge from the old logger to the
168
+
// new slog handler.
169
+
logLogger := slog.NewLogLogger(logger.Handler(), config.GetLogLevel())
170
+
serfConfig.Logger = logLogger
148
171
serfConfig.NodeName = config.NodeName
149
172
serfConfig.ProtocolVersion = uint8(serf.ProtocolVersionMax)
150
173
// TODO: how should cascade handle name conflicts?
···
160
183
serfConfig.ReconnectTimeout = 3 * 24 * time.Hour
161
184
162
185
serfConfig.MemberlistConfig = memberlist.DefaultWANConfig()
186
+
serfConfig.MemberlistConfig.Logger = logLogger
163
187
serfConfig.MemberlistConfig.DeadNodeReclaimTime = 30 * time.Second
164
188
serfConfig.MemberlistConfig.BindAddr = bindIP
165
189
serfConfig.MemberlistConfig.BindPort = bindPort
166
190
167
-
agent := Create(serfConfig)
191
+
agent := Create(config, serfConfig, logger)
168
192
return agent
169
193
}
170
194