unistack-org/micro
4
0
Fork 3
Code Issues 20 Pull Requests 1 Actions Packages Projects Releases 282 Wiki Activity

Simplified code. Small bug fix the used to lead to multi-registry loop.

Browse Source
This commit is contained in:
Milos Gajdos 2019-09-30 12:48:06 +01:00
parent 3f3f1272b3
commit 4b73ac9dc5
1 changed files with 31 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
registry/memory/memory.go
View File

@ -178,6 +178,8 @@ func (m *Registry) Register(s *registry.Service, opts ...registry.RegisterOption
m.Lock() m.Lock()
defer m.Unlock() defer m.Unlock()
log.Debugf("Registry deregistering service: %s", s.Name)
var options registry.RegisterOptions var options registry.RegisterOptions
for _, o := range opts { for _, o := range opts {
o(&options) o(&options)
@ -187,16 +189,17 @@ func (m *Registry) Register(s *registry.Service, opts ...registry.RegisterOption
m.Services[s.Name] = []*registry.Service{s} m.Services[s.Name] = []*registry.Service{s}
// add all nodes into nodes map to track their TTL // add all nodes into nodes map to track their TTL
for _, n := range s.Nodes { for _, n := range s.Nodes {
log.Debugf("Registry tracking node %s for service %s", n.Id, s.Name) log.Debugf("Registry tracking new service: %s, node %s", s.Name, n.Id)
m.nodes[nodeTrackId(s.Name, s.Version, n.Id)] = &node{ m.nodes[nodeTrackId(s.Name, s.Version, n.Id)] = &node{
lastSeen: time.Now(), lastSeen: time.Now(),
ttl: options.TTL, ttl: options.TTL,
} }
} }
go m.sendEvent(&registry.Result{Action: "update", Service: s}) go m.sendEvent(&registry.Result{Action: "update", Service: s})
return nil
} else { } else {
// svcCount essentially keep the count of all service vesions // svcCount keeps the count of all versions of particular service
svcCount := len(service) //svcCount := len(service)
// svcNodes maintains a list of node Ids per particular service version // svcNodes maintains a list of node Ids per particular service version
svcNodes := make(map[string]map[string][]string) svcNodes := make(map[string]map[string][]string)
// collect all service ids for all service versions // collect all service ids for all service versions
@ -212,20 +215,7 @@ func (m *Registry) Register(s *registry.Service, opts ...registry.RegisterOption
} }
// if merged count and original service counts changed we know we are adding a new version of the service // if merged count and original service counts changed we know we are adding a new version of the service
merged := registry.Merge(service, []*registry.Service{s}) merged := registry.Merge(service, []*registry.Service{s})
if len(merged) != svcCount { // if the node count of any service [version] changed we know we are adding a new node to the service
m.Services[s.Name] = merged
// we know s is the new [version of] service; we need to strart tracking its nodes
for _, n := range s.Nodes {
log.Debugf("Registry tracking node %s for service %s", n.Id, s.Name)
m.nodes[nodeTrackId(s.Name, s.Version, n.Id)] = &node{
lastSeen: time.Now(),
ttl: options.TTL,
}
}
go m.sendEvent(&registry.Result{Action: "update", Service: s})
return nil
}
// if the node count of any particular service [version] changed we know we are adding a new node to the service
for _, s := range merged { for _, s := range merged {
// we know that if the node counts have changed we need to track new nodes // we know that if the node counts have changed we need to track new nodes
if len(s.Nodes) != len(svcNodes[s.Name][s.Version]) { if len(s.Nodes) != len(svcNodes[s.Name][s.Version]) {
@ -238,7 +228,7 @@ func (m *Registry) Register(s *registry.Service, opts ...registry.RegisterOption
} }
} }
if !found { if !found {
log.Debugf("Registry tracking node %s for service %s", n.Id, s.Name) log.Debugf("Registry tracking new node: %s for service %s", n.Id, s.Name)
m.nodes[nodeTrackId(s.Name, s.Version, n.Id)] = &node{ m.nodes[nodeTrackId(s.Name, s.Version, n.Id)] = &node{
lastSeen: time.Now(), lastSeen: time.Now(),
ttl: options.TTL, ttl: options.TTL,
@ -268,31 +258,38 @@ func (m *Registry) Deregister(s *registry.Service) error {
m.Lock() m.Lock()
defer m.Unlock() defer m.Unlock()
log.Debugf("Registry deregistering service: %s", s.Name)
if service, ok := m.Services[s.Name]; ok { if service, ok := m.Services[s.Name]; ok {
// svcNodes maintains a list of node Ids per particular service version // svcNodes collects the list of all node Ids for each service version
svcNodes := make(map[string]map[string][]string) svcNodes := make(map[string]map[string][]string)
// collect all service ids for all service versions // collect all service node ids for all service versions
for _, s := range service { for _, svc := range service {
if _, ok := svcNodes[s.Name]; !ok { if _, ok := svcNodes[svc.Name]; !ok {
svcNodes[s.Name] = make(map[string][]string) svcNodes[svc.Name] = make(map[string][]string)
} }
if _, ok := svcNodes[s.Name][s.Version]; !ok { if _, ok := svcNodes[svc.Name][svc.Version]; !ok {
for _, n := range s.Nodes { for _, n := range svc.Nodes {
svcNodes[s.Name][s.Version] = append(svcNodes[s.Name][s.Version], n.Id) svcNodes[svc.Name][svc.Version] = append(svcNodes[svc.Name][svc.Version], n.Id)
} }
} }
} }
if service := registry.Remove(service, []*registry.Service{s}); len(service) == 0 { // if there are no more services we know we have either removed all nodes or there were no nodes
id := svcNodes[s.Name][s.Version][0] if updatedService := registry.Remove(service, []*registry.Service{s}); len(updatedService) == 0 {
for _, id := range svcNodes[s.Name][s.Version] {
log.Debugf("Registry stopped tracking node %s for service %s", id, s.Name) log.Debugf("Registry stopped tracking node %s for service %s", id, s.Name)
delete(m.nodes, nodeTrackId(s.Name, s.Version, id)) delete(m.nodes, nodeTrackId(s.Name, s.Version, id))
go m.sendEvent(&registry.Result{Action: "delete", Service: s})
}
log.Debugf("Registry deleting service %s: no service nodes", s.Name)
delete(m.Services, s.Name) delete(m.Services, s.Name)
return nil
} else { } else {
// find out which nodes have been removed // find out which nodes have been removed
for _, id := range svcNodes[s.Name][s.Version] { for _, id := range svcNodes[s.Name][s.Version] {
for _, s := range service { for _, svc := range updatedService {
var found bool var found bool
for _, n := range s.Nodes { for _, n := range svc.Nodes {
if id == n.Id { if id == n.Id {
found = true found = true
break break
@ -301,13 +298,13 @@ func (m *Registry) Deregister(s *registry.Service) error {
if !found { if !found {
log.Debugf("Registry stopped tracking node %s for service %s", id, s.Name) log.Debugf("Registry stopped tracking node %s for service %s", id, s.Name)
delete(m.nodes, nodeTrackId(s.Name, s.Version, id)) delete(m.nodes, nodeTrackId(s.Name, s.Version, id))
}
}
m.Services[s.Name] = service
}
}
go m.sendEvent(&registry.Result{Action: "delete", Service: s}) go m.sendEvent(&registry.Result{Action: "delete", Service: s})
} }
}
m.Services[s.Name] = updatedService
}
}
}
return nil return nil
} }