leaderless gossip-based service discovery w/ consul compatibility
cli ux rework, members -> nodes, register/deregister
+724
-61
+101
api/catalog.go
+101
api/catalog.go
···
1
+
package api
2
+
3
+
import "fmt"
4
+
5
+
// Node is a Consul-shaped representation of a cluster member.
6
+
type Node struct {
7
+
Node string
8
+
Address string
9
+
Meta map[string]string
10
+
}
11
+
12
+
// CatalogService describes a single service instance as exposed by the
13
+
// /v1/catalog/service/<name> endpoint.
14
+
type CatalogService struct {
15
+
Node string
16
+
Address string
17
+
ServiceID string
18
+
ServiceName string
19
+
ServiceAddress string
20
+
ServicePort int
21
+
ServiceTags []string
22
+
ServiceMeta map[string]string
23
+
}
24
+
25
+
// CatalogNode bundles a node and the services it owns, returned by
26
+
// /v1/catalog/node/<name>.
27
+
type CatalogNode struct {
28
+
Node *Node
29
+
Services map[string]*AgentService
30
+
}
31
+
32
+
// Catalog provides access to the cluster-wide read views over gossipped
33
+
// service state.
34
+
type Catalog struct {
35
+
c *Client
36
+
}
37
+
38
+
// Catalog returns a handle to the catalog endpoints.
39
+
func (c *Client) Catalog() *Catalog {
40
+
return &Catalog{c: c}
41
+
}
42
+
43
+
// Services returns the cluster-wide service-name -> tag-union view.
44
+
func (cat *Catalog) Services() (map[string][]string, error) {
45
+
r := cat.c.newRequest("GET", "/v1/catalog/services")
46
+
_, resp, err := requireOK(cat.c.doRequest(r))
47
+
if err != nil {
48
+
return nil, err
49
+
}
50
+
defer closeResponseBody(resp)
51
+
var out map[string][]string
52
+
if err := decodeJSONBody(resp.Body, &out); err != nil {
53
+
return nil, err
54
+
}
55
+
return out, nil
56
+
}
57
+
58
+
// Service returns every known instance of the named service.
59
+
func (cat *Catalog) Service(name string) ([]*CatalogService, error) {
60
+
r := cat.c.newRequest("GET", fmt.Sprintf("/v1/catalog/service/%s", name))
61
+
_, resp, err := requireOK(cat.c.doRequest(r))
62
+
if err != nil {
63
+
return nil, err
64
+
}
65
+
defer closeResponseBody(resp)
66
+
var out []*CatalogService
67
+
if err := decodeJSONBody(resp.Body, &out); err != nil {
68
+
return nil, err
69
+
}
70
+
return out, nil
71
+
}
72
+
73
+
// Nodes returns every node serf knows about.
74
+
func (cat *Catalog) Nodes() ([]*Node, error) {
75
+
r := cat.c.newRequest("GET", "/v1/catalog/nodes")
76
+
_, resp, err := requireOK(cat.c.doRequest(r))
77
+
if err != nil {
78
+
return nil, err
79
+
}
80
+
defer closeResponseBody(resp)
81
+
var out []*Node
82
+
if err := decodeJSONBody(resp.Body, &out); err != nil {
83
+
return nil, err
84
+
}
85
+
return out, nil
86
+
}
87
+
88
+
// Node returns the node and the services it owns.
89
+
func (cat *Catalog) Node(name string) (*CatalogNode, error) {
90
+
r := cat.c.newRequest("GET", fmt.Sprintf("/v1/catalog/node/%s", name))
91
+
_, resp, err := requireOK(cat.c.doRequest(r))
92
+
if err != nil {
93
+
return nil, err
94
+
}
95
+
defer closeResponseBody(resp)
96
+
var out CatalogNode
97
+
if err := decodeJSONBody(resp.Body, &out); err != nil {
98
+
return nil, err
99
+
}
100
+
return &out, nil
101
+
}
+45
internal/agent/agent.go
+45
internal/agent/agent.go
···
231
231
return a.services.Owned()
232
232
}
233
233
234
+
// CatalogServices returns the cluster-wide service-name -> tag-union view.
235
+
func (a *Agent) CatalogServices() map[string][]string {
236
+
return a.services.AllServices()
237
+
}
238
+
239
+
// CatalogServiceInstances returns every known instance of the named service.
240
+
func (a *Agent) CatalogServiceInstances(name string) []NodeService {
241
+
return a.services.ServiceInstances(name)
242
+
}
243
+
244
+
// CatalogNodeServices returns the services owned by the named node.
245
+
func (a *Agent) CatalogNodeServices(node string) map[string]*api.AgentService {
246
+
return a.services.NodeServices(node)
247
+
}
248
+
249
+
// CatalogNodes returns every node serf currently knows about, regardless of
250
+
// whether it has registered services. Nodes that have gossipped services
251
+
// but are not yet visible to serf (rare race during join) are also included.
252
+
func (a *Agent) CatalogNodes() []serf.Member {
253
+
members := a.serf.Members()
254
+
seen := map[string]struct{}{}
255
+
for _, m := range members {
256
+
seen[m.Name] = struct{}{}
257
+
}
258
+
for _, n := range a.services.Nodes() {
259
+
if _, ok := seen[n]; ok {
260
+
continue
261
+
}
262
+
members = append(members, serf.Member{Name: n, Status: serf.StatusNone})
263
+
seen[n] = struct{}{}
264
+
}
265
+
return members
266
+
}
267
+
268
+
// MemberByName returns the serf member with the given name, or zero value
269
+
// if no such member is known.
270
+
func (a *Agent) MemberByName(name string) (serf.Member, bool) {
271
+
for _, m := range a.serf.Members() {
272
+
if m.Name == name {
273
+
return m, true
274
+
}
275
+
}
276
+
return serf.Member{}, false
277
+
}
278
+
234
279
// antiEntropyLoop periodically re-broadcasts this agent's owned services so
235
280
// peers that missed the original user event eventually catch up.
236
281
func (a *Agent) antiEntropyLoop() {
+106
internal/agent/catalog_endpoint.go
+106
internal/agent/catalog_endpoint.go
···
1
+
package agent
2
+
3
+
import (
4
+
"fmt"
5
+
"net/http"
6
+
"strings"
7
+
8
+
"git.j3s.sh/cascade/api"
9
+
)
10
+
11
+
// catalogServices serves GET /v1/catalog/services. Returns the cluster-wide
12
+
// map of service name -> union of tags seen on any instance.
13
+
func (s *HTTPHandlers) catalogServices(w http.ResponseWriter, r *http.Request) {
14
+
out := s.agent.CatalogServices()
15
+
if out == nil {
16
+
out = map[string][]string{}
17
+
}
18
+
s.writeJSON(w, r, out)
19
+
}
20
+
21
+
// catalogService serves GET /v1/catalog/service/<name>. Returns one entry
22
+
// per known instance, joined with the owning node's serf membership info.
23
+
func (s *HTTPHandlers) catalogService(w http.ResponseWriter, r *http.Request) {
24
+
name := strings.TrimPrefix(r.URL.Path, "/v1/catalog/service/")
25
+
if name == "" || strings.Contains(name, "/") {
26
+
http.Error(w, "service name required in path", http.StatusBadRequest)
27
+
return
28
+
}
29
+
30
+
instances := s.agent.CatalogServiceInstances(name)
31
+
out := make([]api.CatalogService, 0, len(instances))
32
+
for _, inst := range instances {
33
+
entry := api.CatalogService{
34
+
Node: inst.Node,
35
+
ServiceID: inst.Service.ID,
36
+
ServiceName: inst.Service.Service,
37
+
ServiceAddress: inst.Service.Address,
38
+
ServicePort: inst.Service.Port,
39
+
ServiceTags: inst.Service.Tags,
40
+
ServiceMeta: inst.Service.Meta,
41
+
}
42
+
if m, ok := s.agent.MemberByName(inst.Node); ok {
43
+
entry.Address = m.Addr.String()
44
+
}
45
+
out = append(out, entry)
46
+
}
47
+
s.writeJSON(w, r, out)
48
+
}
49
+
50
+
// catalogNodes serves GET /v1/catalog/nodes. Returns the serf membership
51
+
// list shaped as Consul-style Node entries.
52
+
func (s *HTTPHandlers) catalogNodes(w http.ResponseWriter, r *http.Request) {
53
+
members := s.agent.CatalogNodes()
54
+
out := make([]api.Node, 0, len(members))
55
+
for _, m := range members {
56
+
out = append(out, api.Node{
57
+
Node: m.Name,
58
+
Address: m.Addr.String(),
59
+
Meta: m.Tags,
60
+
})
61
+
}
62
+
s.writeJSON(w, r, out)
63
+
}
64
+
65
+
// catalogNode serves GET /v1/catalog/node/<name>. Returns the node and the
66
+
// services it owns.
67
+
func (s *HTTPHandlers) catalogNode(w http.ResponseWriter, r *http.Request) {
68
+
name := strings.TrimPrefix(r.URL.Path, "/v1/catalog/node/")
69
+
if name == "" || strings.Contains(name, "/") {
70
+
http.Error(w, "node name required in path", http.StatusBadRequest)
71
+
return
72
+
}
73
+
74
+
out := api.CatalogNode{
75
+
Services: s.agent.CatalogNodeServices(name),
76
+
}
77
+
if out.Services == nil {
78
+
out.Services = map[string]*api.AgentService{}
79
+
}
80
+
if m, ok := s.agent.MemberByName(name); ok {
81
+
out.Node = &api.Node{
82
+
Node: m.Name,
83
+
Address: m.Addr.String(),
84
+
Meta: m.Tags,
85
+
}
86
+
} else if len(out.Services) > 0 {
87
+
// Node has gossipped services but serf hasn't surfaced it yet.
88
+
out.Node = &api.Node{Node: name}
89
+
} else {
90
+
http.Error(w, fmt.Sprintf("unknown node %q", name), http.StatusNotFound)
91
+
return
92
+
}
93
+
s.writeJSON(w, r, out)
94
+
}
95
+
96
+
// writeJSON is the standard 200 + application/json response helper.
97
+
func (s *HTTPHandlers) writeJSON(w http.ResponseWriter, r *http.Request, obj interface{}) {
98
+
buf, err := s.marshalJSON(r, obj)
99
+
if err != nil {
100
+
s.agent.logger.Error("marshal response", err)
101
+
w.WriteHeader(http.StatusInternalServerError)
102
+
return
103
+
}
104
+
w.Header().Set("Content-Type", "application/json")
105
+
w.Write(buf)
106
+
}
+8
-5
internal/agent/http.go
+8
-5
internal/agent/http.go
···
110
110
111
111
mux.HandleFunc("/", s.Index)
112
112
endpoints := map[string]func(resp http.ResponseWriter, req *http.Request){
113
-
"/v1/agent/members": s.agentMembers,
114
-
"/v1/agent/self": s.agentSelf,
115
-
"/v1/agent/services": s.agentServices,
116
-
"/v1/agent/service/register": s.agentServiceRegister,
113
+
"/v1/agent/members": s.agentMembers,
114
+
"/v1/agent/self": s.agentSelf,
115
+
"/v1/agent/services": s.agentServices,
116
+
"/v1/agent/service/register": s.agentServiceRegister,
117
117
"/v1/agent/service/deregister/": s.agentServiceDeregister,
118
+
"/v1/catalog/services": s.catalogServices,
119
+
"/v1/catalog/service/": s.catalogService,
120
+
"/v1/catalog/nodes": s.catalogNodes,
121
+
"/v1/catalog/node/": s.catalogNode,
118
122
}
119
123
120
124
for pattern, fn := range endpoints {
···
246
250
// Give them something helpful if there's no UI so they at least know
247
251
// what this server is.
248
252
fmt.Fprint(resp, "cascade agent\n")
249
-
return
250
253
}
251
254
252
255
func decodeBody(body io.Reader, out interface{}) error {
+87
internal/agent/services.go
+87
internal/agent/services.go
···
3
3
import (
4
4
"encoding/json"
5
5
"fmt"
6
+
"sort"
6
7
"sync"
7
8
8
9
"git.j3s.sh/cascade/api"
···
171
172
cp := *svc
172
173
out[id] = &cp
173
174
}
175
+
return out
176
+
}
177
+
178
+
// NodeService pairs a service with the node that owns it.
179
+
type NodeService struct {
180
+
Node string
181
+
Service *api.AgentService
182
+
}
183
+
184
+
// AllServices returns the cluster-wide map of service name to the union of
185
+
// tags seen across all instances. The empty slice means the service exists
186
+
// with no tags somewhere.
187
+
func (s *ServiceStore) AllServices() map[string][]string {
188
+
s.mu.RLock()
189
+
defer s.mu.RUnlock()
190
+
tagsByName := map[string]map[string]struct{}{}
191
+
for _, byID := range s.services {
192
+
for _, svc := range byID {
193
+
set, ok := tagsByName[svc.Service]
194
+
if !ok {
195
+
set = map[string]struct{}{}
196
+
tagsByName[svc.Service] = set
197
+
}
198
+
for _, t := range svc.Tags {
199
+
set[t] = struct{}{}
200
+
}
201
+
}
202
+
}
203
+
out := make(map[string][]string, len(tagsByName))
204
+
for name, set := range tagsByName {
205
+
tags := make([]string, 0, len(set))
206
+
for t := range set {
207
+
tags = append(tags, t)
208
+
}
209
+
sort.Strings(tags)
210
+
out[name] = tags
211
+
}
212
+
return out
213
+
}
214
+
215
+
// ServiceInstances returns every gossipped instance of the named service,
216
+
// across all nodes.
217
+
func (s *ServiceStore) ServiceInstances(name string) []NodeService {
218
+
s.mu.RLock()
219
+
defer s.mu.RUnlock()
220
+
var out []NodeService
221
+
for node, byID := range s.services {
222
+
for _, svc := range byID {
223
+
if svc.Service != name {
224
+
continue
225
+
}
226
+
cp := *svc
227
+
out = append(out, NodeService{Node: node, Service: &cp})
228
+
}
229
+
}
230
+
sort.Slice(out, func(i, j int) bool {
231
+
if out[i].Node != out[j].Node {
232
+
return out[i].Node < out[j].Node
233
+
}
234
+
return out[i].Service.ID < out[j].Service.ID
235
+
})
236
+
return out
237
+
}
238
+
239
+
// NodeServices returns every service owned by the named node.
240
+
func (s *ServiceStore) NodeServices(node string) map[string]*api.AgentService {
241
+
s.mu.RLock()
242
+
defer s.mu.RUnlock()
243
+
out := map[string]*api.AgentService{}
244
+
for id, svc := range s.services[node] {
245
+
cp := *svc
246
+
out[id] = &cp
247
+
}
248
+
return out
249
+
}
250
+
251
+
// Nodes returns the set of node names that currently have at least one
252
+
// gossipped service.
253
+
func (s *ServiceStore) Nodes() []string {
254
+
s.mu.RLock()
255
+
defer s.mu.RUnlock()
256
+
out := make([]string, 0, len(s.services))
257
+
for node := range s.services {
258
+
out = append(out, node)
259
+
}
260
+
sort.Strings(out)
174
261
return out
175
262
}
176
263
+21
-17
internal/cli/members.go
internal/cli/nodes.go
+21
-17
internal/cli/members.go
internal/cli/nodes.go
···
11
11
"git.j3s.sh/cascade/api"
12
12
)
13
13
14
-
type membersCommand struct {
15
-
usage string
16
-
14
+
type nodesCommand struct {
17
15
flagAPIAddr string
18
16
flagDetails bool
19
17
}
20
18
21
-
func (c membersCommand) Usage() {
22
-
fmt.Printf(`usage: cascade members [flags]
23
-
list the members of the cascade serf cluster
24
-
19
+
func (c nodesCommand) Usage() {
20
+
fmt.Printf(`usage: cascade nodes [flags]
21
+
list the nodes in the cascade cluster.
22
+
25
23
flags:
26
24
-api
27
25
address of the cascade http api to target (default = 127.0.0.1:8500)
28
26
`)
29
27
}
30
28
31
-
func (c *membersCommand) Init(args []string) {
29
+
func (c *nodesCommand) Init(args []string) {
32
30
flags := flag.NewFlagSet("", flag.ContinueOnError)
33
31
flags.Usage = c.Usage
34
32
flags.StringVar(&c.flagAPIAddr, "api", "", "")
···
40
38
}
41
39
}
42
40
43
-
func RunMembers(args []string) {
44
-
c := membersCommand{}
41
+
func RunNodes(args []string) {
42
+
c := nodesCommand{}
45
43
c.Init(args)
46
44
47
45
cfg := api.DefaultConfig()
46
+
if c.flagAPIAddr != "" {
47
+
cfg.Address = c.flagAPIAddr
48
+
}
48
49
client, err := api.NewClient(cfg)
49
50
if err != nil {
50
51
fmt.Println(err)
51
52
os.Exit(1)
52
53
}
53
-
members, err := client.Agent().Members()
54
+
nodes, err := client.Agent().Members()
54
55
if err != nil {
55
56
fmt.Println(err)
56
57
os.Exit(1)
···
59
60
var b bytes.Buffer
60
61
tw := tabwriter.NewWriter(&b, 0, 2, 2, ' ', 0)
61
62
fmt.Fprintf(tw, "node\taddr\tstatus\ttags\n")
62
-
for _, m := range members {
63
-
fmt.Fprintf(tw, "%s\t", m.Name)
64
-
fmt.Fprintf(tw, "%s:%d\t", m.Addr, m.Port)
65
-
fmt.Fprintf(tw, "%s\t", m.StatusPretty())
66
-
fmt.Fprintf(tw, "%s\t", c.printTags(m.Tags))
63
+
for _, n := range nodes {
64
+
fmt.Fprintf(tw, "%s\t", n.Name)
65
+
fmt.Fprintf(tw, "%s:%d\t", n.Addr, n.Port)
66
+
fmt.Fprintf(tw, "%s\t", n.StatusPretty())
67
+
fmt.Fprintf(tw, "%s\t", c.printTags(n.Tags))
67
68
fmt.Fprintln(tw)
68
69
}
69
70
if err := tw.Flush(); err != nil {
···
73
74
fmt.Print(b.String())
74
75
}
75
76
76
-
func (c membersCommand) printTags(tags map[string]string) string {
77
+
func (c nodesCommand) printTags(tags map[string]string) string {
77
78
var results []string
78
79
for k, v := range tags {
79
80
results = append(results, fmt.Sprintf("%s=%s", k, v))
81
+
}
82
+
if len(results) == 0 {
83
+
return "<none>"
80
84
}
81
85
return strings.Join(results, ",")
82
86
}
+307
-22
internal/cli/services.go
+307
-22
internal/cli/services.go
···
1
1
package cli
2
2
3
3
import (
4
+
"bufio"
4
5
"bytes"
6
+
"encoding/json"
5
7
"flag"
6
8
"fmt"
9
+
"io"
7
10
"net"
8
11
"os"
9
12
"sort"
···
19
22
}
20
23
21
24
func (c servicesCommand) Usage() {
22
-
fmt.Printf(`usage: cascade services [flags]
23
-
list the services registered to a cascade agent.
25
+
fmt.Printf(`usage: cascade services [name] [flags]
26
+
with no name, list every service in the cluster.
27
+
with a name, list every instance of that service.
24
28
25
29
flags:
26
30
-api
···
28
32
`)
29
33
}
30
34
31
-
func (c *servicesCommand) Init(args []string) {
35
+
func (c *servicesCommand) Init(args []string) []string {
32
36
flags := flag.NewFlagSet("", flag.ContinueOnError)
33
37
flags.Usage = c.Usage
34
38
flags.StringVar(&c.flagAPIAddr, "api", "", "")
···
36
40
fmt.Println(err)
37
41
os.Exit(1)
38
42
}
43
+
return flags.Args()
39
44
}
40
45
41
46
func RunServices(args []string) {
42
47
c := servicesCommand{}
43
-
c.Init(args)
48
+
rest := c.Init(args)
44
49
45
50
cfg := api.DefaultConfig()
46
51
if c.flagAPIAddr != "" {
···
52
57
os.Exit(1)
53
58
}
54
59
55
-
services, err := client.Agent().Services()
60
+
if len(rest) == 0 {
61
+
listAllServices(client)
62
+
return
63
+
}
64
+
listServiceInstances(client, rest[0])
65
+
}
66
+
67
+
func listAllServices(client *api.Client) {
68
+
services, err := client.Catalog().Services()
56
69
if err != nil {
57
70
fmt.Println(err)
58
71
os.Exit(1)
59
72
}
60
-
61
73
if len(services) == 0 {
62
-
fmt.Printf("no services registered on %s\n", cfg.Address)
74
+
fmt.Println("no services known in the cluster")
63
75
return
64
76
}
65
77
66
-
ids := make([]string, 0, len(services))
67
-
for id := range services {
68
-
ids = append(ids, id)
78
+
names := make([]string, 0, len(services))
79
+
for n := range services {
80
+
names = append(names, n)
81
+
}
82
+
sort.Strings(names)
83
+
84
+
var b bytes.Buffer
85
+
tw := tabwriter.NewWriter(&b, 0, 2, 2, ' ', 0)
86
+
fmt.Fprintln(tw, "service\ttags")
87
+
for _, n := range names {
88
+
fmt.Fprintf(tw, "%s\t%s\n", n, joinTagsOrNone(services[n]))
69
89
}
70
-
sort.Strings(ids)
90
+
tw.Flush()
91
+
fmt.Print(b.String())
92
+
}
71
93
72
-
nodeName, err := client.Agent().NodeName()
94
+
func listServiceInstances(client *api.Client, name string) {
95
+
instances, err := client.Catalog().Service(name)
73
96
if err != nil {
74
97
fmt.Println(err)
75
98
os.Exit(1)
76
99
}
100
+
if len(instances) == 0 {
101
+
fmt.Printf("no instances of %q known in the cluster\n", name)
102
+
return
103
+
}
104
+
77
105
var b bytes.Buffer
78
106
tw := tabwriter.NewWriter(&b, 0, 2, 2, ' ', 0)
79
-
fmt.Fprintf(tw, "node\tid\tname\taddr\ttags\n")
80
-
for _, id := range ids {
81
-
s := services[id]
82
-
addr := s.Address
83
-
if s.Port != 0 {
84
-
addr = net.JoinHostPort(addr, strconv.Itoa(s.Port))
107
+
fmt.Fprintln(tw, "node\tid\taddr\ttags")
108
+
for _, inst := range instances {
109
+
addr := inst.ServiceAddress
110
+
if addr == "" {
111
+
addr = inst.Address
112
+
}
113
+
if inst.ServicePort != 0 {
114
+
addr = net.JoinHostPort(addr, strconv.Itoa(inst.ServicePort))
85
115
}
86
-
fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\n", nodeName, s.ID, s.Service, addr, strings.Join(s.Tags, ","))
116
+
fmt.Fprintf(tw, "%s\t%s\t%s\t%s\n", inst.Node, inst.ServiceID, addr, joinTagsOrNone(inst.ServiceTags))
117
+
}
118
+
tw.Flush()
119
+
fmt.Print(b.String())
120
+
}
121
+
122
+
type registerCmd struct {
123
+
flagAPIAddr string
124
+
flagName string
125
+
flagID string
126
+
flagAddr string
127
+
flagPort int
128
+
flagTags string
129
+
flagFile string
130
+
}
131
+
132
+
func (c registerCmd) Usage() {
133
+
fmt.Printf(`usage: cascade register [flags]
134
+
register a service on the local cascade agent. with no flags, prompts
135
+
interactively. with -f, reads a full AgentService JSON document from a
136
+
file (use '-' for stdin).
137
+
138
+
flags:
139
+
-api address of the cascade http api to target (default = 127.0.0.1:8500)
140
+
-name service name (required)
141
+
-port service port (required)
142
+
-id service id (default = name)
143
+
-addr service address (default = empty; consumers use the node address)
144
+
-tags comma-separated tags
145
+
-f read AgentService JSON from file or '-' for stdin
146
+
`)
147
+
}
148
+
149
+
func RunRegister(args []string) {
150
+
c := registerCmd{}
151
+
flags := flag.NewFlagSet("", flag.ContinueOnError)
152
+
flags.Usage = c.Usage
153
+
flags.StringVar(&c.flagAPIAddr, "api", "", "")
154
+
flags.StringVar(&c.flagName, "name", "", "")
155
+
flags.StringVar(&c.flagID, "id", "", "")
156
+
flags.StringVar(&c.flagAddr, "addr", "", "")
157
+
flags.IntVar(&c.flagPort, "port", 0, "")
158
+
flags.StringVar(&c.flagTags, "tags", "", "")
159
+
flags.StringVar(&c.flagFile, "f", "", "")
160
+
if err := flags.Parse(args); err != nil {
161
+
os.Exit(1)
162
+
}
163
+
164
+
var svc *api.AgentService
165
+
var err error
166
+
switch {
167
+
case c.flagFile != "":
168
+
svc, err = readServiceFile(c.flagFile)
169
+
case c.flagName != "" && c.flagPort != 0:
170
+
svc = c.fromFlags()
171
+
default:
172
+
svc, err = c.prompt()
173
+
}
174
+
if err != nil {
175
+
fmt.Println(err)
176
+
os.Exit(1)
177
+
}
178
+
if svc.Service == "" {
179
+
fmt.Println("service name required")
180
+
os.Exit(1)
181
+
}
182
+
if svc.ID == "" {
183
+
svc.ID = svc.Service
184
+
}
185
+
186
+
cfg := api.DefaultConfig()
187
+
if c.flagAPIAddr != "" {
188
+
cfg.Address = c.flagAPIAddr
189
+
}
190
+
client, err := api.NewClient(cfg)
191
+
if err != nil {
192
+
fmt.Println(err)
193
+
os.Exit(1)
194
+
}
195
+
if err := client.Agent().ServiceRegister(svc); err != nil {
196
+
fmt.Printf("register failed: %s\n", err)
197
+
os.Exit(1)
198
+
}
199
+
200
+
addrDesc := svc.Address
201
+
if addrDesc == "" {
202
+
addrDesc = "<node addr>"
203
+
}
204
+
if svc.Port != 0 {
205
+
addrDesc = net.JoinHostPort(addrDesc, strconv.Itoa(svc.Port))
206
+
}
207
+
fmt.Printf("service registered: %s (%s)", svc.Service, addrDesc)
208
+
if len(svc.Tags) > 0 {
209
+
fmt.Printf(" tags=%s", strings.Join(svc.Tags, ","))
210
+
}
211
+
fmt.Printf("\n")
212
+
}
213
+
214
+
func (c registerCmd) fromFlags() *api.AgentService {
215
+
svc := &api.AgentService{
216
+
ID: c.flagID,
217
+
Service: c.flagName,
218
+
Address: c.flagAddr,
219
+
Port: c.flagPort,
220
+
}
221
+
if c.flagTags != "" {
222
+
svc.Tags = splitTags(c.flagTags)
223
+
}
224
+
return svc
225
+
}
226
+
227
+
func (c registerCmd) prompt() (*api.AgentService, error) {
228
+
if !isTerminal(os.Stdin) {
229
+
return nil, fmt.Errorf("register requires -name and -port (or -f) when stdin is not a tty")
230
+
}
231
+
r := bufio.NewReader(os.Stdin)
232
+
233
+
name := promptLine(r, "service name", c.flagName)
234
+
if name == "" {
235
+
return nil, fmt.Errorf("service name required")
87
236
}
88
-
if err := tw.Flush(); err != nil {
89
-
fmt.Printf("error flushing tabwriter: %s", err)
237
+
238
+
portDef := ""
239
+
if c.flagPort != 0 {
240
+
portDef = strconv.Itoa(c.flagPort)
241
+
}
242
+
portStr := promptLine(r, "port", portDef)
243
+
port, err := strconv.Atoi(portStr)
244
+
if err != nil || port <= 0 {
245
+
return nil, fmt.Errorf("port must be a positive integer")
246
+
}
247
+
248
+
id := promptLine(r, "id", firstNonEmpty(c.flagID, name))
249
+
addr := promptLine(r, "address (empty = use node addr)", c.flagAddr)
250
+
tags := promptLine(r, "tags (comma-separated)", c.flagTags)
251
+
252
+
svc := &api.AgentService{
253
+
ID: id,
254
+
Service: name,
255
+
Address: addr,
256
+
Port: port,
257
+
}
258
+
if tags != "" {
259
+
svc.Tags = splitTags(tags)
260
+
}
261
+
return svc, nil
262
+
}
263
+
264
+
type deregisterCmd struct {
265
+
flagAPIAddr string
266
+
}
267
+
268
+
func (c deregisterCmd) Usage() {
269
+
fmt.Printf(`usage: cascade deregister <id> [flags]
270
+
remove a service from the local cascade agent.
271
+
272
+
flags:
273
+
-api address of the cascade http api to target (default = 127.0.0.1:8500)
274
+
`)
275
+
}
276
+
277
+
func RunDeregister(args []string) {
278
+
c := deregisterCmd{}
279
+
flags := flag.NewFlagSet("", flag.ContinueOnError)
280
+
flags.Usage = c.Usage
281
+
flags.StringVar(&c.flagAPIAddr, "api", "", "")
282
+
if err := flags.Parse(args); err != nil {
90
283
os.Exit(1)
91
284
}
92
-
fmt.Print(b.String())
285
+
rest := flags.Args()
286
+
if len(rest) == 0 {
287
+
c.Usage()
288
+
os.Exit(1)
289
+
}
290
+
id := rest[0]
291
+
292
+
cfg := api.DefaultConfig()
293
+
if c.flagAPIAddr != "" {
294
+
cfg.Address = c.flagAPIAddr
295
+
}
296
+
client, err := api.NewClient(cfg)
297
+
if err != nil {
298
+
fmt.Println(err)
299
+
os.Exit(1)
300
+
}
301
+
if err := client.Agent().ServiceDeregister(id); err != nil {
302
+
fmt.Printf("deregister failed: %s\n", err)
303
+
os.Exit(1)
304
+
}
305
+
fmt.Printf("deregistered %s\n", id)
306
+
}
307
+
308
+
func readServiceFile(path string) (*api.AgentService, error) {
309
+
var r io.Reader
310
+
if path == "-" {
311
+
r = os.Stdin
312
+
} else {
313
+
f, err := os.Open(path)
314
+
if err != nil {
315
+
return nil, err
316
+
}
317
+
defer f.Close()
318
+
r = f
319
+
}
320
+
var svc api.AgentService
321
+
if err := json.NewDecoder(r).Decode(&svc); err != nil {
322
+
return nil, fmt.Errorf("decode %s: %w", path, err)
323
+
}
324
+
return &svc, nil
325
+
}
326
+
327
+
func promptLine(r *bufio.Reader, label, def string) string {
328
+
if def != "" {
329
+
fmt.Printf("%s [%s]: ", label, def)
330
+
} else {
331
+
fmt.Printf("%s: ", label)
332
+
}
333
+
line, err := r.ReadString('\n')
334
+
if err != nil && err != io.EOF {
335
+
return def
336
+
}
337
+
line = strings.TrimRight(line, "\r\n")
338
+
if line == "" {
339
+
return def
340
+
}
341
+
return line
342
+
}
343
+
344
+
func joinTagsOrNone(tags []string) string {
345
+
if len(tags) == 0 {
346
+
return "<none>"
347
+
}
348
+
return strings.Join(tags, ",")
349
+
}
350
+
351
+
func splitTags(s string) []string {
352
+
parts := strings.Split(s, ",")
353
+
out := make([]string, 0, len(parts))
354
+
for _, p := range parts {
355
+
p = strings.TrimSpace(p)
356
+
if p != "" {
357
+
out = append(out, p)
358
+
}
359
+
}
360
+
return out
361
+
}
362
+
363
+
func firstNonEmpty(vals ...string) string {
364
+
for _, v := range vals {
365
+
if v != "" {
366
+
return v
367
+
}
368
+
}
369
+
return ""
370
+
}
371
+
372
+
func isTerminal(f *os.File) bool {
373
+
fi, err := f.Stat()
374
+
if err != nil {
375
+
return false
376
+
}
377
+
return (fi.Mode() & os.ModeCharDevice) != 0
93
378
}
+35
-7
internal/cli/status.go
+35
-7
internal/cli/status.go
···
6
6
"fmt"
7
7
"os"
8
8
"sort"
9
+
"strings"
9
10
"text/tabwriter"
10
11
"time"
11
12
···
54
55
agent := client.Agent()
55
56
56
57
start := time.Now()
57
-
members, err := agent.Members()
58
+
nodes, err := agent.Members()
58
59
apiRTT := time.Since(start)
59
60
if err != nil {
60
-
fmt.Printf("error fetching members from %s: %s\n", cfg.Address, err)
61
+
fmt.Printf("error fetching nodes from %s: %s\n", cfg.Address, err)
61
62
os.Exit(1)
62
63
}
63
64
···
73
74
74
75
statusCounts := map[string]int{}
75
76
tagKeyCounts := map[string]int{}
76
-
for _, m := range members {
77
-
statusCounts[m.StatusPretty()]++
78
-
for k := range m.Tags {
77
+
for _, n := range nodes {
78
+
statusCounts[n.StatusPretty()]++
79
+
for k := range n.Tags {
79
80
tagKeyCounts[k]++
80
81
}
81
82
}
82
83
83
-
total := len(members)
84
+
localServices, _ := agent.Services()
85
+
catalogServices, _ := client.Catalog().Services()
86
+
87
+
total := len(nodes)
84
88
alive := statusCounts["alive"]
85
89
healthPct := 0.0
86
90
if total > 0 {
···
94
98
fmt.Fprintf(tw, " agent\t%s\n", nodeName)
95
99
fmt.Fprintf(tw, " api\t%s\n", cfg.Address)
96
100
fmt.Fprintf(tw, " api rtt\t%s\n", apiRTT.Round(time.Microsecond))
97
-
fmt.Fprintf(tw, " members\t%d\n", total)
101
+
fmt.Fprintf(tw, " nodes\t%d\n", total)
98
102
fmt.Fprintf(tw, " health\t%.1f%% alive (%d/%d)\n", healthPct, alive, total)
99
103
fmt.Fprintln(tw)
100
104
···
106
110
fmt.Fprintln(tw)
107
111
}
108
112
113
+
fmt.Fprintln(tw, "services")
114
+
fmt.Fprintf(tw, " local agent\t%s\n", summarizeLocalServices(localServices))
115
+
fmt.Fprintf(tw, " cluster\t%s\n", summarizeCatalog(catalogServices))
116
+
fmt.Fprintln(tw)
117
+
109
118
fmt.Fprintln(tw, "status")
110
119
for _, s := range sortedKeys(statusCounts) {
111
120
fmt.Fprintf(tw, " %s\t%d\n", s, statusCounts[s])
···
138
147
sort.Strings(out)
139
148
return out
140
149
}
150
+
151
+
func summarizeLocalServices(services map[string]*api.AgentService) string {
152
+
if len(services) == 0 {
153
+
return "none"
154
+
}
155
+
names := make([]string, 0, len(services))
156
+
for _, s := range services {
157
+
names = append(names, s.Service)
158
+
}
159
+
sort.Strings(names)
160
+
return fmt.Sprintf("%d (%s)", len(names), strings.Join(names, ", "))
161
+
}
162
+
163
+
func summarizeCatalog(services map[string][]string) string {
164
+
if len(services) == 0 {
165
+
return "none"
166
+
}
167
+
return fmt.Sprintf("%d unique services", len(services))
168
+
}
+14
-10
main.go
+14
-10
main.go
···
9
9
10
10
const usage = ` usage: cascade [command] [flags] [-h]
11
11
12
-
common commands:
13
-
cascade agent start a cascade agent
14
-
cascade members list cascade cluster members
15
-
cascade services list services on a particular agent
16
-
cascade status overview of the cascade cluster
17
-
18
-
advanced:
19
-
cascade rtt estimate latency between nodes
12
+
commands:
13
+
cascade agent start a cascade agent
14
+
cascade status overview of the cascade cluster
15
+
cascade nodes list nodes
16
+
cascade services [name] list services
17
+
cascade register register a service to an agent
18
+
cascade deregister <id> remove a service from an agent
19
+
cascade rtt estimate latency between nodes
20
20
`
21
21
22
22
// TODO: rename agent to something cooler
···
31
31
switch os.Args[1] {
32
32
case "agent":
33
33
cli.RunAgent(args)
34
-
case "members":
35
-
cli.RunMembers(args)
34
+
case "nodes":
35
+
cli.RunNodes(args)
36
36
case "status":
37
37
cli.RunStatus(args)
38
38
case "services":
39
39
cli.RunServices(args)
40
+
case "register":
41
+
cli.RunRegister(args)
42
+
case "deregister":
43
+
cli.RunDeregister(args)
40
44
default:
41
45
fmt.Fprintf(os.Stderr, "'%s' is not a valid command\n\n%s", command, usage)
42
46
os.Exit(1)