consul: Change state store to support multiple tags

This commit is contained in:
Armon Dadgar 2014-04-03 12:03:02 -07:00
parent 283c5d1be1
commit 13bff3a09b
1 changed files with 32 additions and 12 deletions

View File

@ -145,7 +145,7 @@ func (s *StateStore) initialize() error {
}, },
"service": &MDBIndex{ "service": &MDBIndex{
AllowBlank: true, AllowBlank: true,
Fields: []string{"ServiceName", "ServiceTag"}, Fields: []string{"ServiceName"},
}, },
}, },
Decoder: func(buf []byte) interface{} { Decoder: func(buf []byte) interface{} {
@ -318,7 +318,7 @@ func (s *StateStore) EnsureService(index uint64, node string, ns *structs.NodeSe
Node: node, Node: node,
ServiceID: ns.ID, ServiceID: ns.ID,
ServiceName: ns.Service, ServiceName: ns.Service,
ServiceTag: ns.Tag, ServiceTags: ns.Tags,
ServicePort: ns.Port, ServicePort: ns.Port,
} }
@ -382,7 +382,7 @@ func (s *StateStore) parseNodeServices(tables MDBTables, tx *MDBTxn, name string
srv := &structs.NodeService{ srv := &structs.NodeService{
ID: service.ServiceID, ID: service.ServiceID,
Service: service.ServiceName, Service: service.ServiceName,
Tag: service.ServiceTag, Tags: service.ServiceTags,
Port: service.ServicePort, Port: service.ServicePort,
} }
ns.Services[srv.ID] = srv ns.Services[srv.ID] = srv
@ -456,8 +456,6 @@ func (s *StateStore) DeleteNode(index uint64, node string) error {
// Services is used to return all the services with a list of associated tags // Services is used to return all the services with a list of associated tags
func (s *StateStore) Services() (uint64, map[string][]string) { func (s *StateStore) Services() (uint64, map[string][]string) {
// TODO: Optimize to not table scan.. We can do a distinct
// type of query to avoid this
services := make(map[string][]string) services := make(map[string][]string)
idx, res, err := s.serviceTable.Get("id") idx, res, err := s.serviceTable.Get("id")
if err != nil { if err != nil {
@ -466,12 +464,17 @@ func (s *StateStore) Services() (uint64, map[string][]string) {
} }
for _, r := range res { for _, r := range res {
srv := r.(*structs.ServiceNode) srv := r.(*structs.ServiceNode)
tags, ok := services[srv.ServiceName]
tags := services[srv.ServiceName] if !ok {
if !strContains(tags, srv.ServiceTag) {
tags = append(tags, srv.ServiceTag)
services[srv.ServiceName] = tags services[srv.ServiceName] = tags
} }
for _, tag := range srv.ServiceTags {
if !strContains(tags, tag) {
tags = append(tags, tag)
services[srv.ServiceName] = tags
}
}
} }
return idx, services return idx, services
} }
@ -508,10 +511,26 @@ func (s *StateStore) ServiceTagNodes(service, tag string) (uint64, structs.Servi
panic(fmt.Errorf("Failed to get last index: %v", err)) panic(fmt.Errorf("Failed to get last index: %v", err))
} }
res, err := s.serviceTable.GetTxn(tx, "service", service, tag) res, err := s.serviceTable.GetTxn(tx, "service", service)
res = serviceTagFilter(res, tag)
return idx, s.parseServiceNodes(tx, s.nodeTable, res, err) return idx, s.parseServiceNodes(tx, s.nodeTable, res, err)
} }
// serviceTagFilter is used to filter a list of *structs.ServiceNode which do
// not have the specified tag
func serviceTagFilter(l []interface{}, tag string) []interface{} {
n := len(l)
for i := 0; i < n; i++ {
srv := l[i].(*structs.ServiceNode)
if !strContains(srv.ServiceTags, tag) {
l[i], l[n-1] = l[n-1], nil
i--
n--
}
}
return l[:n]
}
// parseServiceNodes parses results ServiceNodes and ServiceTagNodes // parseServiceNodes parses results ServiceNodes and ServiceTagNodes
func (s *StateStore) parseServiceNodes(tx *MDBTxn, table *MDBTable, res []interface{}, err error) structs.ServiceNodes { func (s *StateStore) parseServiceNodes(tx *MDBTxn, table *MDBTable, res []interface{}, err error) structs.ServiceNodes {
nodes := make(structs.ServiceNodes, len(res)) nodes := make(structs.ServiceNodes, len(res))
@ -668,7 +687,8 @@ func (s *StateStore) CheckServiceTagNodes(service, tag string) (uint64, structs.
panic(fmt.Errorf("Failed to get last index: %v", err)) panic(fmt.Errorf("Failed to get last index: %v", err))
} }
res, err := s.serviceTable.GetTxn(tx, "service", service, tag) res, err := s.serviceTable.GetTxn(tx, "service", service)
res = serviceTagFilter(res, tag)
return idx, s.parseCheckServiceNodes(tx, res, err) return idx, s.parseCheckServiceNodes(tx, res, err)
} }
@ -704,7 +724,7 @@ func (s *StateStore) parseCheckServiceNodes(tx *MDBTxn, res []interface{}, err e
nodes[i].Service = structs.NodeService{ nodes[i].Service = structs.NodeService{
ID: srv.ServiceID, ID: srv.ServiceID,
Service: srv.ServiceName, Service: srv.ServiceName,
Tag: srv.ServiceTag, Tags: srv.ServiceTags,
Port: srv.ServicePort, Port: srv.ServicePort,
} }
nodes[i].Checks = checks nodes[i].Checks = checks