leaderless gossip-based service discovery w/ consul compatibility
remove command gunk from main()
+5
-182
-25
agent/config.go
-25
agent/config.go
···
1
1
package agent
2
2
3
3
import (
4
-
"fmt"
5
4
"net"
6
5
"os"
7
6
)
···
30
29
NodeName string
31
30
StartJoin []string
32
31
}
33
-
34
-
// BindAddrParts returns the parts of the BindAddr that should be
35
-
// used to configure Serf.
36
-
func (c *Config) AddrParts(address string) (string, int, error) {
37
-
checkAddr := address
38
-
39
-
START:
40
-
_, _, err := net.SplitHostPort(checkAddr)
41
-
if ae, ok := err.(*net.AddrError); ok && ae.Err == "missing port in address" {
42
-
checkAddr = fmt.Sprintf("%s:%d", checkAddr, DefaultSerfPort)
43
-
goto START
44
-
}
45
-
if err != nil {
46
-
return "", 0, err
47
-
}
48
-
49
-
// Get the address
50
-
addr, err := net.ResolveTCPAddr("tcp", checkAddr)
51
-
if err != nil {
52
-
return "", 0, err
53
-
}
54
-
55
-
return addr.IP.String(), addr.Port, nil
56
-
}
-34
ipaddr/ipaddr.go
-34
ipaddr/ipaddr.go
···
1
-
package ipaddr
2
-
3
-
import (
4
-
"log"
5
-
"net"
6
-
"strconv"
7
-
"strings"
8
-
)
9
-
10
-
// resolveHost will take a single host string and convert it to a list of TCPAddrs
11
-
// This will process any port in the input as well as looking up the hostname using
12
-
// normal DNS resolution.
13
-
func ParseIPPort(IPPort string) *net.TCPAddr {
14
-
var addr net.TCPAddr
15
-
ip, portStr, err := net.SplitHostPort(IPPort)
16
-
if err != nil {
17
-
if strings.Contains(err.Error(), "missing port in address") {
18
-
ip = IPPort
19
-
} else {
20
-
log.Printf("error splitting ip+port '%s' into IP and port: %s\n", IPPort, err)
21
-
return &addr
22
-
}
23
-
} else {
24
-
addr.Port, err = strconv.Atoi(portStr)
25
-
if err != nil {
26
-
log.Printf("Parsed port '%s' is not an integer: %s\n", portStr, err)
27
-
return &addr
28
-
}
29
-
}
30
-
31
-
addr.IP = net.ParseIP(ip)
32
-
33
-
return &addr
34
-
}
+5
-123
main.go
+5
-123
main.go
···
2
2
3
3
import (
4
4
"fmt"
5
-
"log"
6
5
"os"
7
-
"os/signal"
8
-
"strings"
9
-
"syscall"
10
-
"time"
11
6
12
-
"git.j3s.sh/cascade/agent"
13
-
"git.j3s.sh/cascade/ipaddr"
7
+
"git.j3s.sh/cascade/command/agent"
14
8
"git.j3s.sh/cascade/list"
15
9
)
16
10
···
20
14
cascade rtt estimate latency between nodes
21
15
`
22
16
23
-
// gracefulTimeout controls how long we wait before forcefully terminating
24
-
// note that this value interacts with serf's LeavePropagateDelay config
25
-
const gracefulTimeout = 10 * time.Second
26
-
27
17
// TODO: rename agent to something cooler
28
18
func main() {
29
19
if len(os.Args) == 1 {
···
31
21
os.Exit(1)
32
22
}
33
23
34
-
run(os.Args[1], os.Args[2:])
35
-
}
36
-
37
-
func run(command string, args []string) {
38
-
switch command {
24
+
command := os.Args[1]
25
+
args := os.Args[2:]
26
+
switch os.Args[1] {
39
27
case "agent":
40
-
handleAgent()
28
+
agent.Run(args)
41
29
case "list", "ls":
42
30
list.Run(args)
43
31
default:
···
45
33
os.Exit(1)
46
34
}
47
35
}
48
-
49
-
func handleAgent() {
50
-
config := getAgentConfig()
51
-
agent := agent.New(config)
52
-
if err := agent.Start(); err != nil {
53
-
fmt.Println(err)
54
-
os.Exit(1)
55
-
}
56
-
defer agent.Shutdown()
57
-
// join any specified startup nodes
58
-
if err := startupJoin(agent); err != nil {
59
-
fmt.Println(err)
60
-
os.Exit(1)
61
-
}
62
-
if err := handleSignals(agent); err != nil {
63
-
fmt.Println(err)
64
-
os.Exit(1)
65
-
}
66
-
}
67
-
68
-
// handleSignals blocks until we get an exit-causing signal
69
-
func handleSignals(agent *agent.Agent) error {
70
-
signalCh := make(chan os.Signal, 4)
71
-
signal.Notify(signalCh, os.Interrupt, syscall.SIGTERM)
72
-
73
-
// Wait for a signal
74
-
var sig os.Signal
75
-
select {
76
-
case s := <-signalCh:
77
-
sig = s
78
-
case <-agent.ShutdownCh():
79
-
// Agent is already shutdown!
80
-
return nil
81
-
}
82
-
fmt.Fprintf(os.Stderr, "caught signal %s", sig)
83
-
84
-
// Check if we should do a graceful leave
85
-
graceful := false
86
-
if sig == os.Interrupt || sig == syscall.SIGTERM {
87
-
graceful = true
88
-
}
89
-
90
-
// Bail fast if not doing a graceful leave
91
-
if !graceful {
92
-
fmt.Fprintf(os.Stderr, "leave cluster with zero grace")
93
-
return nil
94
-
}
95
-
96
-
// Attempt a graceful leave
97
-
gracefulCh := make(chan struct{})
98
-
fmt.Printf("shutting down gracefully")
99
-
go func() {
100
-
if err := agent.Leave(); err != nil {
101
-
fmt.Println("Error: ", err)
102
-
return
103
-
}
104
-
close(gracefulCh)
105
-
}()
106
-
107
-
// Wait for leave or another signal
108
-
select {
109
-
case <-signalCh:
110
-
return fmt.Errorf("idfk")
111
-
case <-time.After(gracefulTimeout):
112
-
return fmt.Errorf("leave timed out")
113
-
case <-gracefulCh:
114
-
return nil
115
-
}
116
-
}
117
-
118
-
func startupJoin(a *agent.Agent) error {
119
-
if len(a.Config.StartJoin) == 0 {
120
-
return nil
121
-
}
122
-
123
-
n, err := a.Join(a.Config.StartJoin)
124
-
if err != nil {
125
-
return err
126
-
}
127
-
if n > 0 {
128
-
log.Printf("issue join request nodes=%d\n", n)
129
-
}
130
-
131
-
return nil
132
-
}
133
-
134
-
func getAgentConfig() *agent.Config {
135
-
config := agent.DefaultConfig()
136
-
// CASCADE_BIND=192.168.0.15:12345
137
-
if os.Getenv("CASCADE_BIND") != "" {
138
-
addr := ipaddr.ParseIPPort(os.Getenv("CASCADE_BIND"))
139
-
config.BindAddr.IP = addr.IP
140
-
if addr.Port != 0 {
141
-
config.BindAddr.Port = addr.Port
142
-
}
143
-
}
144
-
// CASCADE_JOIN=127.0.0.1,127.0.0.5
145
-
if os.Getenv("CASCADE_JOIN") != "" {
146
-
config.StartJoin = strings.Split(os.Getenv("CASCADE_JOIN"), ",")
147
-
}
148
-
// CASCADE_NAME=nostromo.j3s.sh
149
-
if os.Getenv("CASCADE_NAME") != "" {
150
-
config.NodeName = os.Getenv("CASCADE_NAME")
151
-
}
152
-
return config
153
-
}