leaderless gossip-based service discovery w/ consul compatibility
moar ux
+85
-32
+17
api/catalog.go
+17
api/catalog.go
···
70
70
return out, nil
71
71
}
72
72
73
+
// Instances returns every gossipped service instance across the cluster.
74
+
// One round trip — useful when you want to render a per-node view rather
75
+
// than per-service.
76
+
func (cat *Catalog) Instances() ([]*CatalogService, error) {
77
+
r := cat.c.newRequest("GET", "/v1/catalog/instances")
78
+
_, resp, err := requireOK(cat.c.doRequest(r))
79
+
if err != nil {
80
+
return nil, err
81
+
}
82
+
defer closeResponseBody(resp)
83
+
var out []*CatalogService
84
+
if err := decodeJSONBody(resp.Body, &out); err != nil {
85
+
return nil, err
86
+
}
87
+
return out, nil
88
+
}
89
+
73
90
// Nodes returns every node serf knows about.
74
91
func (cat *Catalog) Nodes() ([]*Node, error) {
75
92
r := cat.c.newRequest("GET", "/v1/catalog/nodes")
+6
internal/agent/agent.go
+6
internal/agent/agent.go
···
245
245
return a.services.ServiceInstances(name)
246
246
}
247
247
248
+
// CatalogAllInstances returns every gossipped service instance across the
249
+
// whole cluster as a flat slice.
250
+
func (a *Agent) CatalogAllInstances() []NodeService {
251
+
return a.services.AllInstances()
252
+
}
253
+
248
254
// CatalogNodeServices returns the services owned by the named node.
249
255
func (a *Agent) CatalogNodeServices(node string) map[string]*api.AgentService {
250
256
return a.services.NodeServices(node)
+23
internal/agent/catalog_endpoint.go
+23
internal/agent/catalog_endpoint.go
···
8
8
"j3s.sh/cascade/api"
9
9
)
10
10
11
+
// catalogInstances serves GET /v1/catalog/instances. Returns one entry
12
+
// per (node, service) pair across the whole cluster.
13
+
func (s *HTTPHandlers) catalogInstances(w http.ResponseWriter, r *http.Request) {
14
+
instances := s.agent.CatalogAllInstances()
15
+
out := make([]api.CatalogService, 0, len(instances))
16
+
for _, inst := range instances {
17
+
entry := api.CatalogService{
18
+
Node: inst.Node,
19
+
ServiceID: inst.Service.ID,
20
+
ServiceName: inst.Service.Service,
21
+
ServiceAddress: inst.Service.Address,
22
+
ServicePort: inst.Service.Port,
23
+
ServiceTags: inst.Service.Tags,
24
+
ServiceMeta: inst.Service.Meta,
25
+
}
26
+
if m, ok := s.agent.MemberByName(inst.Node); ok {
27
+
entry.Address = m.Addr.String()
28
+
}
29
+
out = append(out, entry)
30
+
}
31
+
s.writeJSON(w, r, out)
32
+
}
33
+
11
34
// catalogServices serves GET /v1/catalog/services. Returns the cluster-wide
12
35
// map of service name -> union of tags seen on any instance.
13
36
func (s *HTTPHandlers) catalogServices(w http.ResponseWriter, r *http.Request) {
+1
internal/agent/http.go
+1
internal/agent/http.go
···
117
117
"/v1/agent/service/deregister/": s.agentServiceDeregister,
118
118
"/v1/catalog/services": s.catalogServices,
119
119
"/v1/catalog/service/": s.catalogService,
120
+
"/v1/catalog/instances": s.catalogInstances,
120
121
"/v1/catalog/nodes": s.catalogNodes,
121
122
"/v1/catalog/node/": s.catalogNode,
122
123
}
+21
internal/agent/services.go
+21
internal/agent/services.go
···
212
212
return out
213
213
}
214
214
215
+
// AllInstances returns every gossipped service instance across the entire
216
+
// cluster, one entry per (node, service) pair.
217
+
func (s *ServiceStore) AllInstances() []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
+
cp := *svc
224
+
out = append(out, NodeService{Node: node, Service: &cp})
225
+
}
226
+
}
227
+
sort.Slice(out, func(i, j int) bool {
228
+
if out[i].Node != out[j].Node {
229
+
return out[i].Node < out[j].Node
230
+
}
231
+
return out[i].Service.ID < out[j].Service.ID
232
+
})
233
+
return out
234
+
}
235
+
215
236
// ServiceInstances returns every gossipped instance of the named service,
216
237
// across all nodes.
217
238
func (s *ServiceStore) ServiceInstances(name string) []NodeService {
+17
-20
internal/cli/nodes.go
+17
-20
internal/cli/nodes.go
···
5
5
"flag"
6
6
"fmt"
7
7
"os"
8
+
"sort"
8
9
"strings"
9
10
"text/tabwriter"
10
11
···
13
14
14
15
type nodesCommand struct {
15
16
flagAPIAddr string
16
-
flagDetails bool
17
17
}
18
18
19
19
func (c nodesCommand) Usage() {
···
30
30
flags := flag.NewFlagSet("", flag.ContinueOnError)
31
31
flags.Usage = c.Usage
32
32
flags.StringVar(&c.flagAPIAddr, "api", "", "")
33
-
flags.BoolVar(&c.flagDetails, "details", false, "")
34
-
flags.BoolVar(&c.flagDetails, "l", false, "")
35
33
if err := flags.Parse(args); err != nil {
36
34
fmt.Println(err)
37
35
os.Exit(1)
···
57
55
os.Exit(1)
58
56
}
59
57
58
+
// Fetch the full instance list once; group by owning node.
59
+
instances, _ := client.Catalog().Instances()
60
+
servicesByNode := map[string][]string{}
61
+
for _, inst := range instances {
62
+
servicesByNode[inst.Node] = append(servicesByNode[inst.Node], inst.ServiceName)
63
+
}
64
+
for n := range servicesByNode {
65
+
sort.Strings(servicesByNode[n])
66
+
}
67
+
60
68
var b bytes.Buffer
61
69
tw := tabwriter.NewWriter(&b, 0, 2, 2, ' ', 0)
62
-
fmt.Fprintf(tw, "node\taddr\tstatus\ttags\n")
70
+
fmt.Fprintf(tw, "node\taddr\tstatus\tservices\n")
63
71
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))
68
-
fmt.Fprintln(tw)
72
+
services := "<none>"
73
+
if names := servicesByNode[n.Name]; len(names) > 0 {
74
+
services = strings.Join(names, ", ")
75
+
}
76
+
fmt.Fprintf(tw, "%s\t%s:%d\t%s\t%s\n", n.Name, n.Addr, n.Port, n.StatusPretty(), services)
69
77
}
70
78
if err := tw.Flush(); err != nil {
71
79
fmt.Printf("error flushing tabwriter: %s", err)
···
73
81
}
74
82
fmt.Print(b.String())
75
83
}
76
-
77
-
func (c nodesCommand) printTags(tags map[string]string) string {
78
-
var results []string
79
-
for k, v := range tags {
80
-
results = append(results, fmt.Sprintf("%s=%s", k, v))
81
-
}
82
-
if len(results) == 0 {
83
-
return "<none>"
84
-
}
85
-
return strings.Join(results, ",")
86
-
}
-12
internal/cli/status.go
-12
internal/cli/status.go
···
74
74
}
75
75
76
76
statusCounts := map[string]int{}
77
-
tagKeyCounts := map[string]int{}
78
77
for _, n := range nodes {
79
78
statusCounts[n.StatusPretty()]++
80
-
for k := range n.Tags {
81
-
tagKeyCounts[k]++
82
-
}
83
79
}
84
80
85
81
localServices, _ := agent.Services()
···
122
118
fmt.Fprintln(tw, "status")
123
119
for _, s := range sortedKeys(statusCounts) {
124
120
fmt.Fprintf(tw, " %s\t%d\n", s, statusCounts[s])
125
-
}
126
-
127
-
if len(tagKeyCounts) > 0 {
128
-
fmt.Fprintln(tw)
129
-
fmt.Fprintln(tw, "tags")
130
-
for _, k := range sortedKeys(tagKeyCounts) {
131
-
fmt.Fprintf(tw, " %s\t%d nodes\n", k, tagKeyCounts[k])
132
-
}
133
121
}
134
122
135
123
if err := tw.Flush(); err != nil {