diff --git a/client/alloc_runner.go b/client/alloc_runner.go index 0bee8103a..ba85f58cb 100644 --- a/client/alloc_runner.go +++ b/client/alloc_runner.go @@ -348,6 +348,19 @@ OUTER: // Merge in the task resources task.Resources = update.TaskResources[task.Name] + FOUND: + for _, updateGroup := range update.Job.TaskGroups { + if tg.Name != updateGroup.Name { + continue + } + for _, updateTask := range updateGroup.Tasks { + if updateTask.Name != task.Name { + continue + } + task.Services = updateTask.Services + break FOUND + } + } tr.Update(task) } r.taskLock.RUnlock() diff --git a/client/consul.go b/client/consul.go index 91cc66706..b5759223a 100644 --- a/client/consul.go +++ b/client/consul.go @@ -21,12 +21,22 @@ type trackedService struct { service *structs.Service } +func (t *trackedService) IsServiceValid() bool { + for _, service := range t.task.Services { + if service.Id == t.service.Id { + return true + } + } + + return false +} + type ConsulClient struct { client *consul.Client logger *log.Logger shutdownCh chan struct{} - trackedServices map[string]*trackedService + trackedServices map[string]*trackedService // Service ID to Tracked Service Map trackedSrvLock sync.Mutex } @@ -50,20 +60,15 @@ func NewConsulClient(logger *log.Logger, consulAddr string) (*ConsulClient, erro } func (c *ConsulClient) Register(task *structs.Task, allocID string) error { + // Removing the service first so that we can re-sync everything cleanly + c.Deregister(task) + var mErr multierror.Error for _, service := range task.Services { c.logger.Printf("[INFO] consul: Registering service %s with Consul.", service.Name) if err := c.registerService(service, task, allocID); err != nil { mErr.Errors = append(mErr.Errors, err) } - ts := &trackedService{ - allocId: allocID, - task: task, - service: service, - } - c.trackedSrvLock.Lock() - c.trackedServices[service.Id] = ts - c.trackedSrvLock.Unlock() } return mErr.ErrorOrNil() @@ -72,14 +77,14 @@ func (c *ConsulClient) Register(task *structs.Task, allocID string) error { func (c *ConsulClient) Deregister(task *structs.Task) error { var mErr multierror.Error for _, service := range task.Services { + if service.Id == "" { + continue + } c.logger.Printf("[INFO] consul: De-Registering service %v with Consul", service.Name) if err := c.deregisterService(service.Id); err != nil { c.logger.Printf("[ERROR] consul: Error in de-registering service %v from Consul", service.Name) mErr.Errors = append(mErr.Errors, err) } - c.trackedSrvLock.Lock() - delete(c.trackedServices, service.Id) - c.trackedSrvLock.Unlock() } return mErr.ErrorOrNil() } @@ -104,10 +109,16 @@ func (c *ConsulClient) SyncWithConsul() { for { select { case <-sync: - sync = time.After(syncInterval) var consulServices map[string]*consul.AgentService var err error + for serviceId, ts := range c.trackedServices { + if !ts.IsServiceValid() { + c.logger.Printf("[INFO] consul: Removing service: %s since the task doesn't have it anymore", ts.service.Name) + c.deregisterService(serviceId) + } + } + // Get the list of the services that Consul knows about if consulServices, err = agent.Services(); err != nil { c.logger.Printf("[DEBUG] consul: Error while syncing services with Consul: %v", err) @@ -135,6 +146,7 @@ func (c *ConsulClient) SyncWithConsul() { } } } + sync = time.After(syncInterval) case <-c.shutdownCh: c.logger.Printf("[INFO] Shutting down Consul Client") return @@ -158,6 +170,15 @@ func (c *ConsulClient) registerService(service *structs.Service, task *structs.T Address: host, Checks: checks, } + ts := &trackedService{ + allocId: allocID, + task: task, + service: service, + } + c.trackedSrvLock.Lock() + c.trackedServices[service.Id] = ts + c.trackedSrvLock.Unlock() + if err := c.client.Agent().ServiceRegister(asr); err != nil { c.logger.Printf("[ERROR] consul: Error while registering service %v with Consul: %v", service.Name, err) mErr.Errors = append(mErr.Errors, err) @@ -166,6 +187,10 @@ func (c *ConsulClient) registerService(service *structs.Service, task *structs.T } func (c *ConsulClient) deregisterService(serviceId string) error { + c.trackedSrvLock.Lock() + delete(c.trackedServices, serviceId) + c.trackedSrvLock.Unlock() + if err := c.client.Agent().ServiceDeregister(serviceId); err != nil { return err } diff --git a/client/fingerprint/network.go b/client/fingerprint/network.go index 23ce44554..ad9fe2944 100644 --- a/client/fingerprint/network.go +++ b/client/fingerprint/network.go @@ -64,6 +64,11 @@ func (f *NetworkFingerprint) Fingerprint(cfg *config.Config, node *structs.Node) return false, fmt.Errorf("Error while detecting network interface during fingerprinting: %v", err) } + // No interface could be found + if intf == nil { + return false, nil + } + if ip, err = f.ipAddress(intf); err != nil { return false, fmt.Errorf("Unable to find IP address of interface: %s, err: %v", intf.Name, err) } @@ -229,7 +234,7 @@ func (f *NetworkFingerprint) findInterface(deviceName string) (*net.Interface, e } if len(interfaces) == 0 { - return nil, errors.New("No network interfaces were detected") + return nil, nil } return &interfaces[0], nil } diff --git a/main.go b/main.go old mode 100644 new mode 100755