mirror of https://github.com/status-im/consul.git
agent: refactor syncChecks
This commit is contained in:
parent
197a5a9a9a
commit
949ddefbc8
|
@ -387,11 +387,10 @@ func (l *localState) syncChanges() error {
|
||||||
} else {
|
} else {
|
||||||
l.logger.Printf("[DEBUG] agent: Check '%s' in sync", id)
|
l.logger.Printf("[DEBUG] agent: Check '%s' in sync", id)
|
||||||
}
|
}
|
||||||
|
}
|
||||||
if len(checkIDs) > 0 {
|
if len(checkIDs) > 0 {
|
||||||
if err := l.syncChecks(checkIDs); err != nil {
|
if err := l.syncChecks(checkIDs); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -471,69 +470,60 @@ func (l *localState) syncService(id string) error {
|
||||||
|
|
||||||
// syncChecks is used to sync checks to the server
|
// syncChecks is used to sync checks to the server
|
||||||
func (l *localState) syncChecks(checkIDs []string) error {
|
func (l *localState) syncChecks(checkIDs []string) error {
|
||||||
reqs := make(map[string]*structs.RegisterRequest)
|
checkMap := make(map[string]structs.HealthChecks)
|
||||||
|
|
||||||
for _, id := range checkIDs {
|
for _, id := range checkIDs {
|
||||||
if check, ok := l.checks[id]; ok {
|
if check, ok := l.checks[id]; ok {
|
||||||
// Add checks to the base request if it already exists
|
checkMap[check.ServiceID] = append(checkMap[check.ServiceID], check)
|
||||||
if req, ok := reqs[check.ServiceID]; ok {
|
|
||||||
req.Checks = append(req.Checks, check)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Pull in the associated service if any
|
|
||||||
var service *structs.NodeService
|
|
||||||
if serv, ok := l.services[check.ServiceID]; ok {
|
|
||||||
service = serv
|
|
||||||
}
|
|
||||||
|
|
||||||
// Create the base request
|
|
||||||
reqs[check.ServiceID] = &structs.RegisterRequest{
|
|
||||||
Datacenter: l.config.Datacenter,
|
|
||||||
Node: l.config.NodeName,
|
|
||||||
Address: l.config.AdvertiseAddr,
|
|
||||||
Service: service,
|
|
||||||
Checks: structs.HealthChecks{check},
|
|
||||||
WriteRequest: structs.WriteRequest{Token: l.config.ACLToken},
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, req := range reqs {
|
for serviceID, checks := range checkMap {
|
||||||
// Send check data as Check for backward compatibility if we only have a
|
service := l.services[serviceID]
|
||||||
// single check. Otherwise, send it as Checks
|
|
||||||
if len(req.Checks) == 1 {
|
// Create the sync request
|
||||||
req.Check = req.Checks[0]
|
req := structs.RegisterRequest{
|
||||||
req.Checks = nil
|
Datacenter: l.config.Datacenter,
|
||||||
|
Node: l.config.NodeName,
|
||||||
|
Address: l.config.AdvertiseAddr,
|
||||||
|
Service: service,
|
||||||
|
WriteRequest: structs.WriteRequest{Token: l.config.ACLToken},
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send single Check element for backwards compat with 0.4.x
|
||||||
|
if len(checks) == 1 {
|
||||||
|
req.Check = checks[0]
|
||||||
|
} else {
|
||||||
|
req.Checks = checks
|
||||||
}
|
}
|
||||||
|
|
||||||
// Perform the sync
|
// Perform the sync
|
||||||
var out struct{}
|
var out struct{}
|
||||||
err := l.iface.RPC("Catalog.Register", &req, &out)
|
if err := l.iface.RPC("Catalog.Register", &req, &out); err != nil {
|
||||||
if err == nil {
|
if strings.Contains(err.Error(), permissionDenied) {
|
||||||
for _, id := range checkIDs {
|
for _, check := range checks {
|
||||||
l.checkStatus[id] = syncStatus{inSync: true}
|
l.checkStatus[check.CheckID] = syncStatus{inSync: true}
|
||||||
l.logger.Printf("[INFO] agent: Synced check '%s'", id)
|
l.logger.Printf(
|
||||||
}
|
"[WARN] agent: Check '%s' registration blocked by ACLs",
|
||||||
|
check.CheckID)
|
||||||
// If the check was associated with a service and we synced it,
|
|
||||||
// then mark the service as in sync.
|
|
||||||
if svc := req.Service; svc != nil {
|
|
||||||
if status, ok := l.serviceStatus[svc.ID]; ok && status.inSync {
|
|
||||||
continue
|
|
||||||
}
|
}
|
||||||
l.serviceStatus[svc.ID] = syncStatus{inSync: true}
|
return nil
|
||||||
l.logger.Printf("[INFO] agent: Synced service '%s'", svc.ID)
|
|
||||||
}
|
}
|
||||||
} else if strings.Contains(err.Error(), permissionDenied) {
|
|
||||||
for _, id := range checkIDs {
|
|
||||||
l.checkStatus[id] = syncStatus{inSync: true}
|
|
||||||
l.logger.Printf("[WARN] agent: Check '%s' registration blocked by ACLs", id)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
} else {
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Mark the checks and services as synced
|
||||||
|
for _, check := range checks {
|
||||||
|
l.checkStatus[check.CheckID] = syncStatus{inSync: true}
|
||||||
|
l.logger.Printf("[INFO] agent: Synced check '%s'", check.CheckID)
|
||||||
|
}
|
||||||
|
if service != nil {
|
||||||
|
if status, ok := l.serviceStatus[serviceID]; ok && status.inSync {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
l.serviceStatus[serviceID] = syncStatus{inSync: true}
|
||||||
|
l.logger.Printf("[INFO] agent: Synced service '%s'", serviceID)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|
Loading…
Reference in New Issue