leaderless gossip-based service discovery w/ consul compatibility
huge refactor
+394
-313
-128
agent.go
-128
agent.go
···
1
-
package main
2
-
3
-
import (
4
-
"fmt"
5
-
"sync"
6
-
7
-
"github.com/hashicorp/serf/serf"
8
-
"golang.org/x/exp/slog"
9
-
)
10
-
11
-
// Agent starts and manages a Serf & adds service discovery (TODO)
12
-
type Agent struct {
13
-
serfConf *serf.Config
14
-
eventCh chan serf.Event
15
-
// This is the underlying Serf we are wrapping
16
-
serf *serf.Serf
17
-
// a lil logger used by the agent, you know how it be
18
-
logger *slog.Logger
19
-
// this channel receiving a signal = shutdown
20
-
shutdown bool
21
-
shutdownCh chan struct{}
22
-
shutdownLock sync.Mutex
23
-
}
24
-
25
-
func Create(config *Config, serfConf *serf.Config, logger *slog.Logger) *Agent {
26
-
// Create a channel to listen for events from Serf
27
-
eventCh := make(chan serf.Event, 64)
28
-
serfConf.EventCh = eventCh
29
-
30
-
// Setup the agent
31
-
agent := &Agent{
32
-
serfConf: serfConf,
33
-
eventCh: eventCh,
34
-
logger: logger,
35
-
shutdownCh: make(chan struct{}),
36
-
}
37
-
38
-
return agent
39
-
}
40
-
41
-
func (a *Agent) eventLoop() {
42
-
serfShutdownCh := a.serf.ShutdownCh()
43
-
for {
44
-
select {
45
-
case e := <-a.eventCh:
46
-
a.logger.Info("receive event", "event", e.String())
47
-
48
-
case <-serfShutdownCh:
49
-
a.logger.Info("detect serf shutdown, turning off event loop")
50
-
a.Shutdown()
51
-
return
52
-
53
-
case <-a.shutdownCh:
54
-
return
55
-
}
56
-
}
57
-
}
58
-
59
-
func (a *Agent) Start() error {
60
-
a.logger.Info("start serf agent")
61
-
62
-
// Create serf first
63
-
serf, err := serf.Create(a.serfConf)
64
-
if err != nil {
65
-
return err
66
-
}
67
-
a.serf = serf
68
-
69
-
// Start event loop
70
-
go a.eventLoop()
71
-
return nil
72
-
}
73
-
74
-
// Shutdown closes this agent and all of its processes. Should be preceded
75
-
// by a Leave for a graceful shutdown.
76
-
func (a *Agent) Shutdown() error {
77
-
a.shutdownLock.Lock()
78
-
defer a.shutdownLock.Unlock()
79
-
80
-
if a.shutdown {
81
-
return nil
82
-
}
83
-
84
-
if a.serf == nil {
85
-
goto EXIT
86
-
}
87
-
88
-
a.logger.Info("request serf shutdown")
89
-
if err := a.serf.Shutdown(); err != nil {
90
-
return err
91
-
}
92
-
93
-
EXIT:
94
-
a.logger.Info("complete serf shutdown")
95
-
a.shutdown = true
96
-
close(a.shutdownCh)
97
-
return nil
98
-
}
99
-
100
-
// Join asks the Serf instance to join. See the Serf.Join function.
101
-
func (a *Agent) Join(addrs []string) (n int, err error) {
102
-
a.logger.Info("issue join request", "addrs", addrs)
103
-
// we always ignore old events because cascade don't
104
-
// care about the past
105
-
n, err = a.serf.Join(addrs, true)
106
-
if err != nil {
107
-
return n, fmt.Errorf("Error joining: %v\n", err)
108
-
}
109
-
// TODO: when joining fails, we don't get an error here - serf & memberlist
110
-
// just print to stdout and serf.Join returns without issue.
111
-
return n, err
112
-
}
113
-
114
-
// Leave prepares for a graceful shutdown of the agent and its processes
115
-
func (a *Agent) Leave() error {
116
-
if a.serf == nil {
117
-
return nil
118
-
}
119
-
120
-
a.logger.Info("request graceful serf leave")
121
-
return a.serf.Leave()
122
-
}
123
-
124
-
// ShutdownCh returns a channel that can be selected to wait
125
-
// for the agent to perform a shutdown.
126
-
func (a *Agent) ShutdownCh() <-chan struct{} {
127
-
return a.shutdownCh
128
-
}
+351
agent/agent.go
+351
agent/agent.go
···
1
+
// design touchstones
2
+
// simple to configure/implement
3
+
// - minimal configurable options
4
+
// - sparse use of commands/flags/env vars
5
+
// - fast to start/join a cluster
6
+
// - http-only api
7
+
// api auth/encryption is best implemented by a reverse proxy (like caddy or nginx)
8
+
// - no RPC (it's too annoying to deal with)
9
+
// predictable state
10
+
// - no runtime configuration allowed (no reload support)
11
+
// for a gossip cluster to work, nodes in the cluster must
12
+
// be the source of truth. we do not allow the user to change
13
+
// the cluster's view of reality in any way - the user may
14
+
// only inspect it.
15
+
// - HTTP API only supports read-only commands
16
+
// this ensures that cascade's starting state never
17
+
// diverges from its running state.
18
+
// (restarts are very fast + interruptionless)
19
+
// this also helps with security
20
+
// one massive global cluster
21
+
// - for simplicity, operability, and because gossip can handle it
22
+
// compatability
23
+
// - attempt to have a consul-compatible API where it matters?
24
+
// todo
25
+
// dns resolver for services
26
+
// read-write exception for maintenance mode?
27
+
// opentelemetry metrics
28
+
29
+
// key differences from consul
30
+
// - services are gossip'd
31
+
32
+
// in cascade, node == member, and catalog == agent
33
+
// agent api (http-only)
34
+
// GET /v1/agent/metrics -> opentelemetry metrics for this agent
35
+
// GET /v1/agent/members -> show serf cluster membership state
36
+
// GET /v1/agent/services -> show services this agent owns
37
+
// aliases: GET /v1/catalog/nodes
38
+
// catalog api (http & dns)
39
+
// GET /v1/catalog/nodes
40
+
// GET /v1/catalog/services
41
+
// dns: dig @127.0.0.1 -p 8600 nostromo.node.cascade. ANY
42
+
// dns: dig @127.0.0.1 -p 8600 redis.service.cascade. ANY
43
+
44
+
// consul uses /catalog because the masters maintain a service
45
+
// catalog separate than the agent state
46
+
47
+
// in cascade, the agents _are_ expected to maintain service catalogues,
48
+
// so we do away with the catalog concept entirely.
49
+
50
+
package agent
51
+
52
+
import (
53
+
"fmt"
54
+
"os"
55
+
"os/signal"
56
+
"strings"
57
+
"sync"
58
+
"syscall"
59
+
"time"
60
+
61
+
"github.com/hashicorp/memberlist"
62
+
"github.com/hashicorp/serf/serf"
63
+
"golang.org/x/exp/slog"
64
+
)
65
+
66
+
const (
67
+
// gracefulTimeout controls how long we wait before forcefully terminating
68
+
// note that this value interacts with serfConfig.LeavePropagateDelay
69
+
gracefulTimeout = 10 * time.Second
70
+
)
71
+
72
+
func Run() {
73
+
config := readConfig()
74
+
logger := config.SetupLogger()
75
+
agent := setupAgent(config, logger)
76
+
if err := agent.Start(); err != nil {
77
+
fmt.Println(err)
78
+
os.Exit(1)
79
+
}
80
+
defer agent.Shutdown()
81
+
// join any specified startup nodes
82
+
if err := startupJoin(config, agent); err != nil {
83
+
fmt.Println(err)
84
+
os.Exit(1)
85
+
}
86
+
if err := handleSignals(config, agent); err != nil {
87
+
fmt.Println(err)
88
+
os.Exit(1)
89
+
}
90
+
}
91
+
92
+
// handleSignals blocks until we get an exit-causing signal
93
+
func handleSignals(config *Config, agent *Agent) error {
94
+
signalCh := make(chan os.Signal, 4)
95
+
signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM)
96
+
97
+
// Wait for a signal
98
+
var sig os.Signal
99
+
select {
100
+
case s := <-signalCh:
101
+
sig = s
102
+
case <-agent.ShutdownCh():
103
+
// Agent is already shutdown!
104
+
return nil
105
+
}
106
+
agent.logger.Info("caught signal", "signal", sig)
107
+
108
+
// Check if we should do a graceful leave
109
+
graceful := false
110
+
if sig == os.Interrupt || sig == syscall.SIGTERM {
111
+
graceful = true
112
+
}
113
+
114
+
// Bail fast if not doing a graceful leave
115
+
if !graceful {
116
+
agent.logger.Warn("leave cluster with zero grace")
117
+
return nil
118
+
}
119
+
120
+
// Attempt a graceful leave
121
+
gracefulCh := make(chan struct{})
122
+
agent.logger.Info("shut down agent gracefully")
123
+
go func() {
124
+
if err := agent.Leave(); err != nil {
125
+
agent.logger.Error("Error: %s", err)
126
+
return
127
+
}
128
+
close(gracefulCh)
129
+
}()
130
+
131
+
// Wait for leave or another signal
132
+
select {
133
+
case <-signalCh:
134
+
return fmt.Errorf("idfk")
135
+
case <-time.After(gracefulTimeout):
136
+
return fmt.Errorf("leave timed out")
137
+
case <-gracefulCh:
138
+
return nil
139
+
}
140
+
}
141
+
142
+
func readConfig() *Config {
143
+
config := DefaultConfig()
144
+
// CASCADE_LOGLEVEL=info
145
+
l := os.Getenv("CASCADE_LOGLEVEL")
146
+
if l != "" {
147
+
config.LogLevel = l
148
+
if l == "DEBUG" {
149
+
// config.
150
+
}
151
+
}
152
+
// CASCADE_BIND=192.168.0.15:12345
153
+
if os.Getenv("CASCADE_BIND") != "" {
154
+
config.BindAddr = os.Getenv("CASCADE_BIND")
155
+
}
156
+
// CASCADE_JOIN=127.0.0.1,127.0.0.5
157
+
if os.Getenv("CASCADE_JOIN") != "" {
158
+
config.StartJoin = strings.Split(os.Getenv("CASCADE_JOIN"), ",")
159
+
}
160
+
// CASCADE_NAME=nostromo.j3s.sh
161
+
if os.Getenv("CASCADE_NAME") != "" {
162
+
config.NodeName = os.Getenv("CASCADE_NAME")
163
+
}
164
+
return config
165
+
}
166
+
167
+
func startupJoin(config *Config, agent *Agent) error {
168
+
if len(config.StartJoin) == 0 {
169
+
return nil
170
+
}
171
+
172
+
n, err := agent.Join(config.StartJoin)
173
+
if err != nil {
174
+
return err
175
+
}
176
+
if n > 0 {
177
+
agent.logger.Info("issue join request", "nodes", n)
178
+
}
179
+
180
+
return nil
181
+
}
182
+
183
+
func setupAgent(config *Config, logger *slog.Logger) *Agent {
184
+
bindIP, bindPort, err := config.AddrParts(config.BindAddr)
185
+
if err != nil {
186
+
panic(err)
187
+
}
188
+
189
+
serfConfig := serf.DefaultConfig()
190
+
// logLogger is just a bridge from the old logger to the
191
+
// new slog handler.
192
+
logLogger := slog.NewLogLogger(logger.Handler(), config.GetLogLevel())
193
+
serfConfig.Logger = logLogger
194
+
serfConfig.NodeName = config.NodeName
195
+
serfConfig.ProtocolVersion = uint8(serf.ProtocolVersionMax)
196
+
// TODO: how should cascade handle name conflicts?
197
+
// defaulting to just blowing up for now, but
198
+
// we _could_ take the tailscale route & append
199
+
// -1 or whatever to the node. that would be more user friendly.
200
+
// TODO: some of these serf settings were pulled
201
+
// from consul[1]. re-examine them eventually.
202
+
serfConfig.EnableNameConflictResolution = false
203
+
serfConfig.LeavePropagateDelay = 3 * time.Second
204
+
serfConfig.MinQueueDepth = 4096
205
+
serfConfig.QueueDepthWarning = 1000000
206
+
serfConfig.ReconnectTimeout = 3 * 24 * time.Hour
207
+
208
+
serfConfig.MemberlistConfig = memberlist.DefaultWANConfig()
209
+
serfConfig.MemberlistConfig.Logger = logLogger
210
+
serfConfig.MemberlistConfig.DeadNodeReclaimTime = 30 * time.Second
211
+
serfConfig.MemberlistConfig.BindAddr = bindIP
212
+
serfConfig.MemberlistConfig.BindPort = bindPort
213
+
214
+
agent := Create(config, serfConfig, logger)
215
+
agent.logger.Info("setup cascade agent", "nodename", config.NodeName, "bindport", bindPort, "bindip", bindIP, "loglevel", config.GetLogLevel())
216
+
return agent
217
+
}
218
+
219
+
// [1]: sources for consul serf tweaks
220
+
// https://github.com/hashicorp/consul/blob/v1.14.4/agent/consul/config.go
221
+
// https://github.com/hashicorp/consul/blob/v1.14.4/lib/serf/serf.go
222
+
223
+
// Agent starts and manages a Serf & adds service discovery (TODO)
224
+
type Agent struct {
225
+
serfConf *serf.Config
226
+
eventCh chan serf.Event
227
+
// This is the underlying Serf we are wrapping
228
+
serf *serf.Serf
229
+
// a lil logger used by the agent, you know how it be
230
+
logger *slog.Logger
231
+
// this channel receiving a signal = shutdown
232
+
shutdown bool
233
+
shutdownCh chan struct{}
234
+
shutdownLock sync.Mutex
235
+
}
236
+
237
+
func Create(config *Config, serfConf *serf.Config, logger *slog.Logger) *Agent {
238
+
// Create a channel to listen for events from Serf
239
+
eventCh := make(chan serf.Event, 64)
240
+
serfConf.EventCh = eventCh
241
+
242
+
// Setup the agent
243
+
agent := &Agent{
244
+
serfConf: serfConf,
245
+
eventCh: eventCh,
246
+
logger: logger,
247
+
shutdownCh: make(chan struct{}),
248
+
}
249
+
250
+
return agent
251
+
}
252
+
253
+
func (a *Agent) eventLoop() {
254
+
serfShutdownCh := a.serf.ShutdownCh()
255
+
for {
256
+
select {
257
+
case e := <-a.eventCh:
258
+
a.logger.Info("receive event", "event", e.String())
259
+
260
+
case <-serfShutdownCh:
261
+
a.logger.Info("detect serf shutdown, turning off event loop")
262
+
a.Shutdown()
263
+
return
264
+
265
+
case <-a.shutdownCh:
266
+
return
267
+
}
268
+
}
269
+
}
270
+
271
+
func (a *Agent) Start() error {
272
+
// Create serf first
273
+
serf, err := serf.Create(a.serfConf)
274
+
if err != nil {
275
+
return err
276
+
}
277
+
a.serf = serf
278
+
279
+
// Start event loop
280
+
go a.eventLoop()
281
+
return nil
282
+
}
283
+
284
+
// Shutdown closes this agent and all of its processes. Should be preceded
285
+
// by a Leave for a graceful shutdown.
286
+
func (a *Agent) Shutdown() error {
287
+
a.shutdownLock.Lock()
288
+
defer a.shutdownLock.Unlock()
289
+
290
+
if a.shutdown {
291
+
return nil
292
+
}
293
+
294
+
if a.serf == nil {
295
+
goto EXIT
296
+
}
297
+
298
+
a.logger.Info("request serf shutdown")
299
+
if err := a.serf.Shutdown(); err != nil {
300
+
return err
301
+
}
302
+
303
+
EXIT:
304
+
a.logger.Info("complete serf shutdown")
305
+
a.shutdown = true
306
+
close(a.shutdownCh)
307
+
return nil
308
+
}
309
+
310
+
// Join asks the Serf instance to join. See the Serf.Join function.
311
+
func (a *Agent) Join(addrs []string) (n int, err error) {
312
+
a.logger.Info("issue join request", "addrs", addrs)
313
+
// we always ignore old events because cascade don't
314
+
// care about the past
315
+
n, err = a.serf.Join(addrs, true)
316
+
if err != nil {
317
+
return n, fmt.Errorf("Error joining: %v\n", err)
318
+
}
319
+
// TODO: when joining fails, we don't get an error here - serf & memberlist
320
+
// just print to stdout and serf.Join returns without issue.
321
+
return n, err
322
+
}
323
+
324
+
// Leave prepares for a graceful shutdown of the agent and its processes
325
+
func (a *Agent) Leave() error {
326
+
if a.serf == nil {
327
+
return nil
328
+
}
329
+
330
+
a.logger.Info("request graceful serf leave")
331
+
return a.serf.Leave()
332
+
}
333
+
334
+
// ShutdownCh returns a channel that can be selected to wait
335
+
// for the agent to perform a shutdown.
336
+
func (a *Agent) ShutdownCh() <-chan struct{} {
337
+
return a.shutdownCh
338
+
}
339
+
340
+
func (c *Config) SetupLogger() *slog.Logger {
341
+
opts := slog.HandlerOptions{
342
+
Level: c.GetLogLevel(),
343
+
}
344
+
if c.LogLevel == "DEBUG" {
345
+
opts.AddSource = true
346
+
}
347
+
handler := opts.NewTextHandler(os.Stderr)
348
+
logger := slog.New(handler)
349
+
slog.SetDefault(logger)
350
+
return logger
351
+
}
+4
-2
config.go
agent/config.go
+4
-2
config.go
agent/config.go
···
1
-
package main
1
+
package agent
2
2
3
3
import (
4
4
"fmt"
···
8
8
"golang.org/x/exp/slog"
9
9
)
10
10
11
-
const DefaultBindPort int = 4449
11
+
const DefaultBindPort int = 4440
12
12
13
13
func DefaultConfig() *Config {
14
14
hostname, err := os.Hostname()
···
20
20
BindAddr: "0.0.0.0",
21
21
LogLevel: "INFO",
22
22
NodeName: hostname,
23
+
RpcAddr: "127.0.0.1:4441",
23
24
}
24
25
}
25
26
···
35
36
BindAddr string
36
37
LogLevel string
37
38
NodeName string
39
+
RpcAddr string
38
40
StartJoin []string
39
41
Services []Service
40
42
}
+1
go.mod
+1
go.mod
···
17
17
github.com/hashicorp/go-multierror v1.1.0 // indirect
18
18
github.com/hashicorp/go-sockaddr v1.0.0 // indirect
19
19
github.com/hashicorp/golang-lru v0.5.0 // indirect
20
+
github.com/hashicorp/logutils v1.0.0 // indirect
20
21
github.com/miekg/dns v1.1.41 // indirect
21
22
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
22
23
golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1 // indirect
+1
go.sum
+1
go.sum
···
26
26
github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
27
27
github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
28
28
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
29
+
github.com/hashicorp/logutils v1.0.0 h1:dLEQVugN8vlakKOUE3ihGLTZJRB4j+M2cdTm/ORI65Y=
29
30
github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=
30
31
github.com/hashicorp/mdns v1.0.4/go.mod h1:mtBihi+LeNXGtG8L9dX59gAEa12BDtBQSp4v/YAJqrc=
31
32
github.com/hashicorp/memberlist v0.5.0 h1:EtYPN8DpAURiapus508I4n9CzHs2W+8NZGbmmR/prTM=
+16
list/nodes.go
+16
list/nodes.go
+21
-183
main.go
+21
-183
main.go
···
1
-
// design touchstones
2
-
// configured entirely via environment variables
3
-
// minimal configurable options
4
-
// a single global cluster
5
-
// easy cluster formation
6
-
// todo
7
-
// dns resolver for services
8
-
9
1
package main
10
2
11
3
import (
12
4
"fmt"
13
5
"os"
14
-
"os/signal"
15
-
"strings"
16
-
"syscall"
17
-
"time"
18
6
19
-
"github.com/hashicorp/memberlist"
20
-
"github.com/hashicorp/serf/serf"
21
-
"golang.org/x/exp/slog"
7
+
"git.j3s.sh/cascade/agent"
8
+
"git.j3s.sh/cascade/list"
22
9
)
23
10
24
-
const (
25
-
// gracefulTimeout controls how long we wait before forcefully terminating
26
-
// note that this value interacts with serfConfig.LeavePropagateDelay
27
-
gracefulTimeout = 10 * time.Second
28
-
)
11
+
// TODO: rename agent to something cooler
12
+
const help = `cascade agent start a cascade agent
13
+
cascade list|ls list nodes or services
14
+
cascade members show serf cluster members
15
+
cascade rtt estimate latency between nodes
16
+
`
29
17
30
18
func main() {
31
-
config := readConfig()
32
-
logger := setupLogger(config)
33
-
agent := setupAgent(config, logger)
34
-
if err := agent.Start(); err != nil {
35
-
panic(err)
36
-
}
37
-
defer agent.Shutdown()
38
-
// join any specified startup nodes
39
-
if err := startupJoin(config, agent); err != nil {
40
-
panic(err)
41
-
}
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
52
-
}
53
-
opts := slog.HandlerOptions{
54
-
Level: config.GetLogLevel(),
55
-
AddSource: debugOn,
56
-
}
57
-
handler := opts.NewTextHandler(os.Stderr)
58
-
logger := slog.New(handler)
59
-
slog.SetDefault(logger)
60
-
return logger
61
-
}
62
-
63
-
64
-
func debugPrints(a *Agent) {
65
-
for {
66
-
for _, m := range a.serf.Members() {
67
-
a.logger.Debug("debug-loop", "name", m.Name, "addr", m.Addr, "status", m.Status)
68
-
}
69
-
time.Sleep(time.Second * 5)
70
-
}
71
-
}
72
-
73
-
// handleSignals blocks until we get an exit-causing signal
74
-
func handleSignals(config *Config, agent *Agent) error {
75
-
signalCh := make(chan os.Signal, 4)
76
-
signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM)
77
-
78
-
// Wait for a signal
79
-
var sig os.Signal
80
-
select {
81
-
case s := <-signalCh:
82
-
sig = s
83
-
case <-agent.ShutdownCh():
84
-
// Agent is already shutdown!
85
-
return nil
86
-
}
87
-
agent.logger.Info("caught signal", "signal", sig)
88
-
89
-
// Check if we should do a graceful leave
90
-
graceful := false
91
-
if sig == os.Interrupt || sig == syscall.SIGTERM {
92
-
graceful = true
93
-
}
94
-
95
-
// Bail fast if not doing a graceful leave
96
-
if !graceful {
97
-
agent.logger.Warn("leave cluster with zero grace")
98
-
return nil
99
-
}
100
-
101
-
// Attempt a graceful leave
102
-
gracefulCh := make(chan struct{})
103
-
agent.logger.Info("shut down agent gracefully")
104
-
go func() {
105
-
if err := agent.Leave(); err != nil {
106
-
agent.logger.Error("Error: %s", err)
107
-
return
108
-
}
109
-
close(gracefulCh)
110
-
}()
111
-
112
-
// Wait for leave or another signal
113
-
select {
114
-
case <-signalCh:
115
-
return fmt.Errorf("idfk")
116
-
case <-time.After(gracefulTimeout):
117
-
return fmt.Errorf("leave timed out")
118
-
case <-gracefulCh:
119
-
return nil
120
-
}
121
-
}
122
-
123
-
func readConfig() *Config {
124
-
config := DefaultConfig()
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
130
-
if os.Getenv("CASCADE_BIND") != "" {
131
-
config.BindAddr = os.Getenv("CASCADE_BIND")
132
-
}
133
-
// CASCADE_JOIN=127.0.0.1,127.0.0.5
134
-
if os.Getenv("CASCADE_JOIN") != "" {
135
-
config.StartJoin = strings.Split(os.Getenv("CASCADE_JOIN"), ",")
136
-
}
137
-
// CASCADE_NAME=nostromo.j3s.sh
138
-
if os.Getenv("CASCADE_NAME") != "" {
139
-
config.NodeName = os.Getenv("CASCADE_NAME")
140
-
}
141
-
return config
142
-
}
143
-
144
-
func startupJoin(config *Config, agent *Agent) error {
145
-
if len(config.StartJoin) == 0 {
146
-
return nil
147
-
}
148
-
149
-
n, err := agent.Join(config.StartJoin)
150
-
if err != nil {
151
-
return err
152
-
}
153
-
if n > 0 {
154
-
agent.logger.Info("issue join request", "nodes", n)
19
+
if len(os.Args) == 1 {
20
+
fmt.Fprintf(os.Stderr, "%s", help)
21
+
os.Exit(1)
155
22
}
156
23
157
-
return nil
158
-
}
159
-
160
-
func setupAgent(config *Config, logger *slog.Logger) *Agent {
161
-
bindIP, bindPort, err := config.AddrParts(config.BindAddr)
162
-
if err != nil {
163
-
panic(err)
24
+
cmd := os.Args[1]
25
+
args := os.Args[2:]
26
+
switch cmd {
27
+
case "agent":
28
+
agent.Run()
29
+
case "list", "ls":
30
+
list.Run(args)
31
+
default:
32
+
fmt.Fprintf(os.Stderr, "'%s' is not a valid command\n\n%s", cmd, help)
33
+
os.Exit(1)
164
34
}
165
-
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
171
-
serfConfig.NodeName = config.NodeName
172
-
serfConfig.ProtocolVersion = uint8(serf.ProtocolVersionMax)
173
-
// TODO: how should cascade handle name conflicts?
174
-
// defaulting to just blowing up for now, but
175
-
// we _could_ take the tailscale route & append
176
-
// -1 or whatever to the node. that would be more user friendly.
177
-
// TODO: some of these serf settings were pulled
178
-
// from consul[1]. re-examine them eventually.
179
-
serfConfig.EnableNameConflictResolution = false
180
-
serfConfig.LeavePropagateDelay = 3 * time.Second
181
-
serfConfig.MinQueueDepth = 4096
182
-
serfConfig.QueueDepthWarning = 1000000
183
-
serfConfig.ReconnectTimeout = 3 * 24 * time.Hour
184
-
185
-
serfConfig.MemberlistConfig = memberlist.DefaultWANConfig()
186
-
serfConfig.MemberlistConfig.Logger = logLogger
187
-
serfConfig.MemberlistConfig.DeadNodeReclaimTime = 30 * time.Second
188
-
serfConfig.MemberlistConfig.BindAddr = bindIP
189
-
serfConfig.MemberlistConfig.BindPort = bindPort
190
-
191
-
agent := Create(config, serfConfig, logger)
192
-
return agent
193
35
}
194
-
195
-
// [1]: sources for consul serf tweaks
196
-
// https://github.com/hashicorp/consul/blob/v1.14.4/agent/consul/config.go
197
-
// https://github.com/hashicorp/consul/blob/v1.14.4/lib/serf/serf.go