j3s.sh / cascade
leaderless gossip-based service discovery w/ consul compatibility
5
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

add bbolt, datadir, service endpoints, services cmd

Jes Olson (May 23, 2026, 9:08 AM -0500) f96d10a3 f0826821

+576 -18
+1
.gitignore
··· 1 1 cascade 2 2 *.prof 3 + cascade-data
+68
api/service.go
··· 1 + package api 2 + 3 + import ( 4 + "bytes" 5 + "encoding/json" 6 + "fmt" 7 + "io" 8 + ) 9 + 10 + // AgentService describes a service registered on an agent. The JSON shape 11 + // mirrors Consul's so existing SDKs round-trip cleanly. 12 + type AgentService struct { 13 + ID string 14 + Service string 15 + Tags []string 16 + Address string 17 + Port int 18 + Meta map[string]string 19 + } 20 + 21 + // ServiceRegister registers a service with the local agent. The agent owns 22 + // the entry and gossips it to the rest of the cluster. 23 + func (a *Agent) ServiceRegister(svc *AgentService) error { 24 + body, err := json.Marshal(svc) 25 + if err != nil { 26 + return err 27 + } 28 + r := a.c.newRequest("PUT", "/v1/agent/service/register") 29 + r.body = bytes.NewReader(body) 30 + r.header.Set("Content-Type", "application/json") 31 + _, resp, err := requireOK(a.c.doRequest(r)) 32 + if err != nil { 33 + return err 34 + } 35 + defer closeResponseBody(resp) 36 + return nil 37 + } 38 + 39 + // ServiceDeregister removes a service the local agent owns. 40 + func (a *Agent) ServiceDeregister(id string) error { 41 + r := a.c.newRequest("PUT", fmt.Sprintf("/v1/agent/service/deregister/%s", id)) 42 + _, resp, err := requireOK(a.c.doRequest(r)) 43 + if err != nil { 44 + return err 45 + } 46 + defer closeResponseBody(resp) 47 + return nil 48 + } 49 + 50 + // Services lists services registered on the local agent. 51 + func (a *Agent) Services() (map[string]*AgentService, error) { 52 + r := a.c.newRequest("GET", "/v1/agent/services") 53 + _, resp, err := requireOK(a.c.doRequest(r)) 54 + if err != nil { 55 + return nil, err 56 + } 57 + defer closeResponseBody(resp) 58 + 59 + var out map[string]*AgentService 60 + if err := decodeJSONBody(resp.Body, &out); err != nil { 61 + return nil, err 62 + } 63 + return out, nil 64 + } 65 + 66 + func decodeJSONBody(body io.Reader, out interface{}) error { 67 + return json.NewDecoder(body).Decode(out) 68 + }
+3 -2
go.mod
··· 1 1 module git.j3s.sh/cascade 2 2 3 - go 1.20 3 + go 1.23 4 4 5 5 require ( 6 6 github.com/hashicorp/serf v0.10.1 ··· 19 19 github.com/hashicorp/memberlist v0.5.0 // indirect 20 20 github.com/miekg/dns v1.1.41 // indirect 21 21 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect 22 + go.etcd.io/bbolt v1.4.3 // indirect 22 23 golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1 // indirect 23 - golang.org/x/sys v0.1.0 // indirect 24 + golang.org/x/sys v0.29.0 // indirect 24 25 )
+6
go.sum
··· 59 59 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= 60 60 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk= 61 61 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= 62 + github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= 63 + go.etcd.io/bbolt v1.4.3 h1:dEadXpI6G79deX5prL3QRNP6JB8UxVkqo4UPnHaNXJo= 64 + go.etcd.io/bbolt v1.4.3/go.mod h1:tKQlpPaYCVFctUIgFKFnAlvbmB3tpy1vkTnDWohtc0E= 62 65 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= 63 66 golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392/go.mod h1:/lpIB1dKB+9EgE3H3cr1v9wB50oz8l4C4h62xy7jSTY= 64 67 golang.org/x/exp v0.0.0-20230213192124-5e25df0256eb h1:PaBZQdo+iSDyHT053FjUCgZQ/9uqVwPOcl7KSWhKn6w= ··· 72 75 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= 73 76 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ= 74 77 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= 78 + golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ= 75 79 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= 76 80 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= 77 81 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= ··· 86 90 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 87 91 golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U= 88 92 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 93 + golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU= 94 + golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= 89 95 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= 90 96 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= 91 97 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+130 -3
internal/agent/agent.go
··· 55 55 package agent 56 56 57 57 import ( 58 + "encoding/json" 58 59 "fmt" 59 60 "net/http" 60 61 "os" 62 + "path/filepath" 61 63 "sync" 62 64 "time" 63 65 66 + "git.j3s.sh/cascade/api" 64 67 "github.com/hashicorp/serf/serf" 65 68 "golang.org/x/exp/slog" 66 69 ) 67 70 71 + // antiEntropyInterval is how often each agent re-broadcasts its owned 72 + // service state so peers that missed the original user event catch up. 73 + const antiEntropyInterval = 60 * time.Second 74 + 68 75 // Agent starts and manages a Serf & adds service discovery (TODO) 69 76 type Agent struct { 70 77 // this is the agent's configuration ··· 76 83 77 84 // This is the http api handler - see http.go and *_endpoint.go 78 85 httpHandlers *HTTPHandlers 86 + 87 + // services holds the agent's own + gossipped service catalog. 88 + services *ServiceStore 79 89 80 90 // We receive serf events on this channel 81 91 eventCh chan serf.Event ··· 135 145 for { 136 146 select { 137 147 case e := <-a.eventCh: 138 - a.logger.Info("receive event", "event", e.String()) 148 + a.handleSerfEvent(e) 139 149 140 150 case <-serfShutdownCh: 141 151 a.logger.Info("detect serf shutdown, turning off event loop") ··· 148 158 } 149 159 } 150 160 161 + func (a *Agent) handleSerfEvent(e serf.Event) { 162 + switch ev := e.(type) { 163 + case serf.UserEvent: 164 + a.handleUserEvent(ev) 165 + case serf.MemberEvent: 166 + if ev.Type == serf.EventMemberFailed || ev.Type == serf.EventMemberLeave || ev.Type == serf.EventMemberReap { 167 + for _, m := range ev.Members { 168 + a.logger.Info("drop services for departed member", "node", m.Name, "event", ev.Type.String()) 169 + a.services.DropNode(m.Name) 170 + } 171 + } 172 + default: 173 + a.logger.Debug("receive event", "event", e.String()) 174 + } 175 + } 176 + 177 + func (a *Agent) handleUserEvent(ev serf.UserEvent) { 178 + switch ev.Name { 179 + case serfRegisterEvent: 180 + var p registerPayload 181 + if err := json.Unmarshal(ev.Payload, &p); err != nil { 182 + a.logger.Warn("decode svc-register event", "err", err) 183 + return 184 + } 185 + if p.Service == nil { 186 + return 187 + } 188 + a.services.ApplyRegister(p.Node, p.Service) 189 + case serfDeregisterEvent: 190 + var p deregisterPayload 191 + if err := json.Unmarshal(ev.Payload, &p); err != nil { 192 + a.logger.Warn("decode svc-deregister event", "err", err) 193 + return 194 + } 195 + a.services.ApplyDeregister(p.Node, p.ID) 196 + } 197 + } 198 + 199 + // RegisterService stores a service in the local agent and gossips the new 200 + // state to the cluster. 201 + func (a *Agent) RegisterService(svc *api.AgentService) error { 202 + if err := a.services.Register(svc); err != nil { 203 + return err 204 + } 205 + return a.broadcastRegister(svc) 206 + } 207 + 208 + // DeregisterService removes a service from the local agent and gossips the 209 + // removal to the cluster. 210 + func (a *Agent) DeregisterService(id string) error { 211 + if err := a.services.Deregister(id); err != nil { 212 + return err 213 + } 214 + payload, err := json.Marshal(deregisterPayload{Node: a.Config.NodeName, ID: id}) 215 + if err != nil { 216 + return err 217 + } 218 + return a.serf.UserEvent(serfDeregisterEvent, payload, false) 219 + } 220 + 221 + func (a *Agent) broadcastRegister(svc *api.AgentService) error { 222 + payload, err := json.Marshal(registerPayload{Node: a.Config.NodeName, Service: svc}) 223 + if err != nil { 224 + return err 225 + } 226 + return a.serf.UserEvent(serfRegisterEvent, payload, false) 227 + } 228 + 229 + // OwnedServices returns this agent's currently-owned service registrations. 230 + func (a *Agent) OwnedServices() map[string]*api.AgentService { 231 + return a.services.Owned() 232 + } 233 + 234 + // antiEntropyLoop periodically re-broadcasts this agent's owned services so 235 + // peers that missed the original user event eventually catch up. 236 + func (a *Agent) antiEntropyLoop() { 237 + t := time.NewTicker(antiEntropyInterval) 238 + defer t.Stop() 239 + for { 240 + select { 241 + case <-t.C: 242 + for _, svc := range a.services.Owned() { 243 + if err := a.broadcastRegister(svc); err != nil { 244 + a.logger.Warn("anti-entropy rebroadcast", "id", svc.ID, "err", err) 245 + } 246 + } 247 + case <-a.shutdownCh: 248 + return 249 + } 250 + } 251 + } 252 + 151 253 // Start creates and starts an agent's serf client, eventloop, api, and more! 152 254 // The agent config should not be modified after an agent has been started. 153 255 func (a *Agent) Start() error { 256 + // open the service store before serf so we never gossip without it 257 + if err := os.MkdirAll(a.Config.DataDir, 0o700); err != nil { 258 + return fmt.Errorf("create data dir %q: %w", a.Config.DataDir, err) 259 + } 260 + dbPath := filepath.Join(a.Config.DataDir, a.Config.NodeName+".db") 261 + store, err := NewServiceStore(a.Config.NodeName, dbPath) 262 + if err != nil { 263 + return err 264 + } 265 + a.services = store 266 + 154 267 // Create serf first 155 - serf, err := serf.Create(a.serfConfig) 268 + srf, err := serf.Create(a.serfConfig) 156 269 if err != nil { 157 270 return err 158 271 } 159 - a.serf = serf 272 + a.serf = srf 160 273 161 274 // Start event loop 162 275 go a.eventLoop() 276 + go a.antiEntropyLoop() 277 + 278 + // Re-broadcast any services that survived restart so peers catch up 279 + // without waiting for the first anti-entropy tick. 280 + for _, svc := range a.services.Owned() { 281 + if err := a.broadcastRegister(svc); err != nil { 282 + a.logger.Warn("startup rebroadcast", "id", svc.ID, "err", err) 283 + } 284 + } 163 285 164 286 // Start API Server 165 287 apiHandler := a.httpHandlers.handler() ··· 196 318 197 319 EXIT: 198 320 a.logger.Info("complete serf shutdown") 321 + if a.services != nil { 322 + if err := a.services.Close(); err != nil { 323 + a.logger.Warn("close service store", "err", err) 324 + } 325 + } 199 326 a.shutdown = true 200 327 close(a.shutdownCh) 201 328 return nil
+64
internal/agent/agent_endpoint.go
··· 1 1 package agent 2 2 3 3 import ( 4 + "encoding/json" 4 5 "fmt" 5 6 "net/http" 7 + "strings" 6 8 9 + "git.j3s.sh/cascade/api" 7 10 "github.com/hashicorp/serf/coordinate" 8 11 "github.com/hashicorp/serf/serf" 9 12 ) ··· 58 61 } 59 62 fmt.Fprintf(w, string(json)) 60 63 } 64 + 65 + func (s *HTTPHandlers) agentServices(w http.ResponseWriter, r *http.Request) { 66 + services := s.agent.OwnedServices() 67 + buf, err := s.marshalJSON(r, services) 68 + if err != nil { 69 + s.agent.logger.Error("marshal services", err) 70 + w.WriteHeader(http.StatusInternalServerError) 71 + return 72 + } 73 + w.Header().Set("Content-Type", "application/json") 74 + w.Write(buf) 75 + } 76 + 77 + func (s *HTTPHandlers) agentServiceRegister(w http.ResponseWriter, r *http.Request) { 78 + if r.Method != http.MethodPut && r.Method != http.MethodPost { 79 + w.Header().Set("Allow", "PUT, POST") 80 + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) 81 + return 82 + } 83 + 84 + var svc api.AgentService 85 + if err := json.NewDecoder(r.Body).Decode(&svc); err != nil { 86 + http.Error(w, fmt.Sprintf("bad request: %s", err), http.StatusBadRequest) 87 + return 88 + } 89 + if svc.ID == "" { 90 + svc.ID = svc.Service 91 + } 92 + if svc.Service == "" { 93 + http.Error(w, "Service name required", http.StatusBadRequest) 94 + return 95 + } 96 + 97 + if err := s.agent.RegisterService(&svc); err != nil { 98 + s.agent.logger.Error("register service", err, "id", svc.ID) 99 + http.Error(w, err.Error(), http.StatusInternalServerError) 100 + return 101 + } 102 + w.WriteHeader(http.StatusOK) 103 + } 104 + 105 + func (s *HTTPHandlers) agentServiceDeregister(w http.ResponseWriter, r *http.Request) { 106 + if r.Method != http.MethodPut && r.Method != http.MethodPost { 107 + w.Header().Set("Allow", "PUT, POST") 108 + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) 109 + return 110 + } 111 + 112 + id := strings.TrimPrefix(r.URL.Path, "/v1/agent/service/deregister/") 113 + if id == "" || strings.Contains(id, "/") { 114 + http.Error(w, "service ID required in path", http.StatusBadRequest) 115 + return 116 + } 117 + 118 + if err := s.agent.DeregisterService(id); err != nil { 119 + s.agent.logger.Error("deregister service", err, "id", id) 120 + http.Error(w, err.Error(), http.StatusInternalServerError) 121 + return 122 + } 123 + w.WriteHeader(http.StatusOK) 124 + }
+6 -3
internal/agent/config.go
··· 6 6 ) 7 7 8 8 const ( 9 - DefaultDNSPort int = 8600 10 - DefaultHTTPPort int = 8500 11 - DefaultSerfPort int = 8301 9 + DefaultDNSPort int = 8600 10 + DefaultHTTPPort int = 8500 11 + DefaultSerfPort int = 8301 12 + DefaultDataDir string = "./cascade-data" 12 13 ) 13 14 14 15 func DefaultConfig() *Config { ··· 22 23 cfg.HTTPBindAddr = &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: DefaultHTTPPort} 23 24 cfg.SerfBindAddr = &net.TCPAddr{IP: net.ParseIP("0.0.0.0"), Port: DefaultSerfPort} 24 25 cfg.NodeName = hostname 26 + cfg.DataDir = DefaultDataDir 25 27 26 28 return &cfg 27 29 } ··· 33 35 SerfBindAddr *net.TCPAddr 34 36 NodeName string 35 37 StartJoin []string 38 + DataDir string 36 39 37 40 //// non user-configurable 38 41 // HTTPResponseHeaders are used to add HTTP header response fields to the HTTP API responses.
+5 -2
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, 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 + "/v1/agent/service/deregister/": s.agentServiceDeregister, 115 118 } 116 119 117 120 for pattern, fn := range endpoints {
+193
internal/agent/services.go
··· 1 + package agent 2 + 3 + import ( 4 + "encoding/json" 5 + "fmt" 6 + "sync" 7 + 8 + "git.j3s.sh/cascade/api" 9 + "go.etcd.io/bbolt" 10 + ) 11 + 12 + // serfRegisterEvent / serfDeregisterEvent are the user event names we 13 + // gossip when an agent's owned service set changes. 14 + const ( 15 + serfRegisterEvent = "cascade:svc-register" 16 + serfDeregisterEvent = "cascade:svc-deregister" 17 + ) 18 + 19 + // servicesBucket is the bbolt bucket holding this agent's owned services, 20 + // keyed by service ID. 21 + var servicesBucket = []byte("services") 22 + 23 + // registerPayload is the wire format for svc-register events. 24 + type registerPayload struct { 25 + Node string `json:"node"` 26 + Service *api.AgentService `json:"service"` 27 + } 28 + 29 + // deregisterPayload is the wire format for svc-deregister events. 30 + type deregisterPayload struct { 31 + Node string `json:"node"` 32 + ID string `json:"id"` 33 + } 34 + 35 + // ServiceStore tracks services known to this agent — both services this 36 + // agent owns (which are persisted) and services owned by peers (which 37 + // are kept in memory only and refreshed via gossip). 38 + type ServiceStore struct { 39 + mu sync.RWMutex 40 + nodeName string 41 + db *bbolt.DB 42 + 43 + // services indexed by node name then service ID. 44 + services map[string]map[string]*api.AgentService 45 + } 46 + 47 + // NewServiceStore opens the on-disk store and loads any previously-owned 48 + // services for this node into memory. 49 + func NewServiceStore(nodeName, dbPath string) (*ServiceStore, error) { 50 + db, err := bbolt.Open(dbPath, 0o600, nil) 51 + if err != nil { 52 + return nil, fmt.Errorf("open service db %q: %w", dbPath, err) 53 + } 54 + 55 + s := &ServiceStore{ 56 + nodeName: nodeName, 57 + db: db, 58 + services: map[string]map[string]*api.AgentService{}, 59 + } 60 + 61 + if err := db.Update(func(tx *bbolt.Tx) error { 62 + b, err := tx.CreateBucketIfNotExists(servicesBucket) 63 + if err != nil { 64 + return err 65 + } 66 + return b.ForEach(func(k, v []byte) error { 67 + var svc api.AgentService 68 + if err := json.Unmarshal(v, &svc); err != nil { 69 + return fmt.Errorf("decode persisted service %q: %w", string(k), err) 70 + } 71 + s.putLocked(nodeName, &svc) 72 + return nil 73 + }) 74 + }); err != nil { 75 + _ = db.Close() 76 + return nil, err 77 + } 78 + 79 + return s, nil 80 + } 81 + 82 + // Close releases the on-disk handle. Safe to call multiple times. 83 + func (s *ServiceStore) Close() error { 84 + if s.db == nil { 85 + return nil 86 + } 87 + err := s.db.Close() 88 + s.db = nil 89 + return err 90 + } 91 + 92 + // Register adds or replaces a service owned by this agent. Persists to disk 93 + // before returning so a crash after the response can't lose the write. 94 + func (s *ServiceStore) Register(svc *api.AgentService) error { 95 + if svc.ID == "" { 96 + return fmt.Errorf("service ID required") 97 + } 98 + if svc.Service == "" { 99 + return fmt.Errorf("service name required") 100 + } 101 + 102 + buf, err := json.Marshal(svc) 103 + if err != nil { 104 + return err 105 + } 106 + 107 + if err := s.db.Update(func(tx *bbolt.Tx) error { 108 + return tx.Bucket(servicesBucket).Put([]byte(svc.ID), buf) 109 + }); err != nil { 110 + return err 111 + } 112 + 113 + s.mu.Lock() 114 + s.putLocked(s.nodeName, svc) 115 + s.mu.Unlock() 116 + return nil 117 + } 118 + 119 + // Deregister removes a service owned by this agent. 120 + func (s *ServiceStore) Deregister(id string) error { 121 + if err := s.db.Update(func(tx *bbolt.Tx) error { 122 + return tx.Bucket(servicesBucket).Delete([]byte(id)) 123 + }); err != nil { 124 + return err 125 + } 126 + 127 + s.mu.Lock() 128 + s.deleteLocked(s.nodeName, id) 129 + s.mu.Unlock() 130 + return nil 131 + } 132 + 133 + // ApplyRegister handles a gossipped service registration from a peer. 134 + // No-op if the event is one we originated (already in our store). 135 + func (s *ServiceStore) ApplyRegister(node string, svc *api.AgentService) { 136 + if node == s.nodeName { 137 + return 138 + } 139 + s.mu.Lock() 140 + s.putLocked(node, svc) 141 + s.mu.Unlock() 142 + } 143 + 144 + // ApplyDeregister handles a gossipped service deregistration from a peer. 145 + func (s *ServiceStore) ApplyDeregister(node, id string) { 146 + if node == s.nodeName { 147 + return 148 + } 149 + s.mu.Lock() 150 + s.deleteLocked(node, id) 151 + s.mu.Unlock() 152 + } 153 + 154 + // DropNode forgets every service owned by the given node. Called when a 155 + // peer leaves or is declared failed. 156 + func (s *ServiceStore) DropNode(node string) { 157 + if node == s.nodeName { 158 + return 159 + } 160 + s.mu.Lock() 161 + delete(s.services, node) 162 + s.mu.Unlock() 163 + } 164 + 165 + // Owned returns the services this agent owns. Copy, safe to mutate. 166 + func (s *ServiceStore) Owned() map[string]*api.AgentService { 167 + s.mu.RLock() 168 + defer s.mu.RUnlock() 169 + out := map[string]*api.AgentService{} 170 + for id, svc := range s.services[s.nodeName] { 171 + cp := *svc 172 + out[id] = &cp 173 + } 174 + return out 175 + } 176 + 177 + // putLocked must be called with s.mu held. 178 + func (s *ServiceStore) putLocked(node string, svc *api.AgentService) { 179 + if s.services[node] == nil { 180 + s.services[node] = map[string]*api.AgentService{} 181 + } 182 + s.services[node][svc.ID] = svc 183 + } 184 + 185 + // deleteLocked must be called with s.mu held. 186 + func (s *ServiceStore) deleteLocked(node, id string) { 187 + if m, ok := s.services[node]; ok { 188 + delete(m, id) 189 + if len(m) == 0 { 190 + delete(s.services, node) 191 + } 192 + } 193 + }
+8
internal/cli/agent.go
··· 23 23 flagBindSerf string 24 24 flagJoin string 25 25 flagNode string 26 + flagDataDir string 26 27 } 27 28 28 29 func (c *agentCommand) Usage() { ··· 48 49 49 50 -node=<name> 50 51 name of this node, must be globally unique (default = hostname) 52 + 53 + -data-dir=<path> 54 + directory for cascade's local state (default = ./cascade-data) 51 55 `) 52 56 } 53 57 ··· 61 65 flags.StringVar(&c.flagBindSerf, "bind-serf", "", "") 62 66 flags.StringVar(&c.flagJoin, "join", "", "") 63 67 flags.StringVar(&c.flagNode, "node", "", "") 68 + flags.StringVar(&c.flagDataDir, "data-dir", "", "") 64 69 65 70 if err := flags.Parse(args); err != nil { 66 71 fmt.Println(err) ··· 201 206 202 207 if c.flagNode != "" { 203 208 config.NodeName = c.flagNode 209 + } 210 + if c.flagDataDir != "" { 211 + config.DataDir = c.flagDataDir 204 212 } 205 213 return config, nil 206 214 }
+85 -6
internal/cli/services.go
··· 1 1 package cli 2 2 3 3 import ( 4 + "bytes" 5 + "flag" 4 6 "fmt" 7 + "net" 8 + "os" 9 + "sort" 10 + "strconv" 11 + "strings" 12 + "text/tabwriter" 13 + 14 + "git.j3s.sh/cascade/api" 5 15 ) 6 16 7 - const usage = `cascade list|ls nodes 8 - cascade list|ls members 9 - cascade list|ls services 10 - ` 17 + type servicesCommand struct { 18 + flagAPIAddr string 19 + } 20 + 21 + func (c servicesCommand) Usage() { 22 + fmt.Printf(`usage: cascade services [flags] 23 + list the services registered to a cascade agent. 24 + 25 + flags: 26 + -api 27 + address of the cascade http api to target (default = 127.0.0.1:8500) 28 + `) 29 + } 30 + 31 + func (c *servicesCommand) Init(args []string) { 32 + flags := flag.NewFlagSet("", flag.ContinueOnError) 33 + flags.Usage = c.Usage 34 + flags.StringVar(&c.flagAPIAddr, "api", "", "") 35 + if err := flags.Parse(args); err != nil { 36 + fmt.Println(err) 37 + os.Exit(1) 38 + } 39 + } 40 + 41 + func RunServices(args []string) { 42 + c := servicesCommand{} 43 + c.Init(args) 44 + 45 + cfg := api.DefaultConfig() 46 + if c.flagAPIAddr != "" { 47 + cfg.Address = c.flagAPIAddr 48 + } 49 + client, err := api.NewClient(cfg) 50 + if err != nil { 51 + fmt.Println(err) 52 + os.Exit(1) 53 + } 54 + 55 + services, err := client.Agent().Services() 56 + if err != nil { 57 + fmt.Println(err) 58 + os.Exit(1) 59 + } 60 + 61 + if len(services) == 0 { 62 + fmt.Printf("no services registered on %s\n", cfg.Address) 63 + return 64 + } 65 + 66 + ids := make([]string, 0, len(services)) 67 + for id := range services { 68 + ids = append(ids, id) 69 + } 70 + sort.Strings(ids) 11 71 12 - func Run(args []string) { 13 - fmt.Printf("%+v", usage) 72 + nodeName, err := client.Agent().NodeName() 73 + if err != nil { 74 + fmt.Println(err) 75 + os.Exit(1) 76 + } 77 + var b bytes.Buffer 78 + 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)) 85 + } 86 + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\n", nodeName, s.ID, s.Service, addr, strings.Join(s.Tags, ",")) 87 + } 88 + if err := tw.Flush(); err != nil { 89 + fmt.Printf("error flushing tabwriter: %s", err) 90 + os.Exit(1) 91 + } 92 + fmt.Print(b.String()) 14 93 }
+7 -2
main.go
··· 7 7 "git.j3s.sh/cascade/internal/cli" 8 8 ) 9 9 10 - const usage = ` usage: cascade [command] [flags] 10 + const usage = ` usage: cascade [command] [flags] [-h] 11 11 12 - commands: 12 + common commands: 13 13 cascade agent start a cascade agent 14 14 cascade members list cascade cluster members 15 + cascade services list services on a particular agent 15 16 cascade status overview of the cascade cluster 17 + 18 + advanced: 16 19 cascade rtt estimate latency between nodes 17 20 ` 18 21 ··· 32 35 cli.RunMembers(args) 33 36 case "status": 34 37 cli.RunStatus(args) 38 + case "services": 39 + cli.RunServices(args) 35 40 default: 36 41 fmt.Fprintf(os.Stderr, "'%s' is not a valid command\n\n%s", command, usage) 37 42 os.Exit(1)