From c1a16eefaa1d07ad75be81819db10c6e6483fa79 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Sat, 21 Nov 2015 12:33:43 -0800 Subject: [PATCH 01/13] Removing a hot debug line --- client/consul.go | 1 - 1 file changed, 1 deletion(-) diff --git a/client/consul.go b/client/consul.go index 03e9497a8..8250d0a1f 100644 --- a/client/consul.go +++ b/client/consul.go @@ -122,7 +122,6 @@ func (c *ConsulClient) SyncWithConsul() { // 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) continue } From 5549c1d2dca2760ca0d56d9861714fbe9e8028be Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Sat, 21 Nov 2015 12:34:01 -0800 Subject: [PATCH 02/13] Generating the sha1 for a check --- nomad/structs/structs.go | 15 +++++++++++++++ nomad/structs/structs_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/nomad/structs/structs.go b/nomad/structs/structs.go index df7aa5f9d..721740404 100644 --- a/nomad/structs/structs.go +++ b/nomad/structs/structs.go @@ -2,8 +2,10 @@ package structs import ( "bytes" + "crypto/sha1" "errors" "fmt" + "io" "reflect" "regexp" "strings" @@ -1038,6 +1040,19 @@ func (sc *ServiceCheck) Validate() error { return nil } +func (sc *ServiceCheck) Hash() string { + h := sha1.New() + io.WriteString(h, sc.Name) + io.WriteString(h, sc.Type) + io.WriteString(h, sc.Script) + io.WriteString(h, sc.Path) + io.WriteString(h, sc.Path) + io.WriteString(h, sc.Protocol) + io.WriteString(h, sc.Interval.String()) + io.WriteString(h, sc.Timeout.String()) + return fmt.Sprintf("%x", h.Sum(nil)) +} + // The Service model represents a Consul service defintion type Service struct { Id string // Id of the service, this needs to be unique on a local machine diff --git a/nomad/structs/structs_test.go b/nomad/structs/structs_test.go index 84af2a198..e224a60c2 100644 --- a/nomad/structs/structs_test.go +++ b/nomad/structs/structs_test.go @@ -1,6 +1,7 @@ package structs import ( + "fmt" "github.com/hashicorp/go-multierror" "reflect" "strings" @@ -375,3 +376,33 @@ func TestInvalidServiceCheck(t *testing.T) { t.Fatalf("Service should be invalid") } } + +func TestDistinctCheckId(t *testing.T) { + c1 := ServiceCheck{ + Name: "web-health", + Type: "http", + Path: "/health", + Interval: 2 * time.Second, + Timeout: 3 * time.Second, + } + c2 := ServiceCheck{ + Name: "web-health", + Type: "http", + Path: "/health1", + Interval: 2 * time.Second, + Timeout: 3 * time.Second, + } + + c3 := ServiceCheck{ + Name: "web-health", + Type: "http", + Path: "/health", + Interval: 4 * time.Second, + Timeout: 3 * time.Second, + } + + if c1.Hash() == c2.Hash() || c1.Hash() == c3.Hash() || c3.Hash() == c2.Hash() { + t.Fatalf("Checks need to be uniq c1: %s, c2: %s, c3: %s", c1.Hash(), c2.Hash(), c3.Hash()) + } + +} From f9c3d16b9f29cb1881352b8d008796cca4070fdc Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Sat, 21 Nov 2015 12:35:49 -0800 Subject: [PATCH 03/13] Removing un-used imports --- nomad/structs/structs_test.go | 1 - 1 file changed, 1 deletion(-) diff --git a/nomad/structs/structs_test.go b/nomad/structs/structs_test.go index e224a60c2..a7df0d523 100644 --- a/nomad/structs/structs_test.go +++ b/nomad/structs/structs_test.go @@ -1,7 +1,6 @@ package structs import ( - "fmt" "github.com/hashicorp/go-multierror" "reflect" "strings" From a3d5b266a03af259e419ab139e9de25fc555e227 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Sun, 22 Nov 2015 23:27:59 -0800 Subject: [PATCH 04/13] Registering Checks independently --- client/consul.go | 44 ++++++++++++++++++++++++++-------------- client/task_runner.go | 4 ---- nomad/structs/structs.go | 8 ++++++++ 3 files changed, 37 insertions(+), 19 deletions(-) diff --git a/client/consul.go b/client/consul.go index 8250d0a1f..4b6d1a939 100644 --- a/client/consul.go +++ b/client/consul.go @@ -38,7 +38,9 @@ type ConsulClient struct { shutdownCh chan struct{} trackedServices map[string]*trackedService // Service ID to Tracked Service Map + trackedChecks map[string]bool // List of check ids that is being tracked trackedSrvLock sync.Mutex + trackedChkLock sync.Mutex } func NewConsulClient(logger *log.Logger, consulAddr string) (*ConsulClient, error) { @@ -61,9 +63,6 @@ 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) @@ -156,19 +155,17 @@ func (c *ConsulClient) SyncWithConsul() { func (c *ConsulClient) registerService(service *structs.Service, task *structs.Task, allocID string) error { var mErr multierror.Error - service.Id = fmt.Sprintf("%s-%s", allocID, service.Name) + service.Id = service.Hash() host, port := c.findPortAndHostForLabel(service.PortLabel, task) if host == "" || port == 0 { return fmt.Errorf("consul: The port:%s marked for registration of service: %s couldn't be found", service.PortLabel, service.Name) } - checks := c.makeChecks(service, host, port) asr := &consul.AgentServiceRegistration{ ID: service.Id, Name: service.Name, Tags: service.Tags, Port: port, Address: host, - Checks: checks, } ts := &trackedService{ allocId: allocID, @@ -183,6 +180,16 @@ func (c *ConsulClient) registerService(service *structs.Service, task *structs.T c.logger.Printf("[ERROR] consul: Error while registering service %v with Consul: %v", service.Name, err) mErr.Errors = append(mErr.Errors, err) } + checks := c.makeChecks(service, host, port) + for _, check := range checks { + if err := c.client.Agent().CheckRegister(check); err != nil { + c.logger.Printf("[ERROR] consul: Error while registerting check %v with Consul: %v", check.Name, err) + mErr.Errors = append(mErr.Errors, err) + } + c.trackedChkLock.Lock() + c.trackedChecks[check.ID] = true + c.trackedChkLock.Unlock() + } return mErr.ErrorOrNil() } @@ -197,13 +204,19 @@ func (c *ConsulClient) deregisterService(serviceId string) error { return nil } -func (c *ConsulClient) makeChecks(service *structs.Service, ip string, port int) []*consul.AgentServiceCheck { - var checks []*consul.AgentServiceCheck +func (c *ConsulClient) makeChecks(service *structs.Service, ip string, port int) []*consul.AgentCheckRegistration { + var checks []*consul.AgentCheckRegistration for _, check := range service.Checks { - c := &consul.AgentServiceCheck{ - Interval: check.Interval.String(), - Timeout: check.Timeout.String(), + if check.Name == "" { + check.Name = fmt.Sprintf("service: '%s' check", service.Name) } + cr := &consul.AgentCheckRegistration{ + ID: check.Hash(), + Name: check.Name, + ServiceID: service.Id, + } + cr.Interval = check.Interval.String() + cr.Timeout = check.Timeout.String() switch check.Type { case structs.ServiceCheckHTTP: if check.Protocol == "" { @@ -214,13 +227,14 @@ func (c *ConsulClient) makeChecks(service *structs.Service, ip string, port int) Host: fmt.Sprintf("%s:%d", ip, port), Path: check.Path, } - c.HTTP = url.String() + cr.HTTP = url.String() case structs.ServiceCheckTCP: - c.TCP = fmt.Sprintf("%s:%d", ip, port) + cr.TCP = fmt.Sprintf("%s:%d", ip, port) case structs.ServiceCheckScript: - c.Script = check.Script // TODO This needs to include the path of the alloc dir and based on driver types + cr.Script = check.Script // TODO This needs to include the path of the alloc dir and based on driver types } - checks = append(checks, c) + + checks = append(checks, cr) } return checks } diff --git a/client/task_runner.go b/client/task_runner.go index fd01b1f96..749b4d6d4 100644 --- a/client/task_runner.go +++ b/client/task_runner.go @@ -251,10 +251,6 @@ func (r *TaskRunner) run() { if err := r.handle.Update(update); err != nil { r.logger.Printf("[ERR] client: failed to update task '%s' for alloc '%s': %v", r.task.Name, r.allocID, err) } - - if err := r.consulClient.Register(update, r.allocID); err != nil { - r.logger.Printf("[ERR] client: failed to update service definition: %v", err) - } case <-r.destroyCh: // Avoid destroying twice if destroyed { diff --git a/nomad/structs/structs.go b/nomad/structs/structs.go index 721740404..f7a3a1b04 100644 --- a/nomad/structs/structs.go +++ b/nomad/structs/structs.go @@ -1072,6 +1072,14 @@ func (s *Service) Validate() error { return mErr.ErrorOrNil() } +func (s *Service) Hash() string { + h := sha1.New() + io.WriteString(h, s.Name) + io.WriteString(h, strings.Join(s.Tags, "")) + io.WriteString(h, s.PortLabel) + return fmt.Sprintf("%x", h.Sum(nil)) +} + // Task is a single process typically that is executed as part of a task group. type Task struct { // Name of the task From 135006699b446722ad01cafe457c11d4d1ebc549 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Tue, 24 Nov 2015 12:34:26 -0800 Subject: [PATCH 05/13] Renamed consul client to service --- client/alloc_runner.go | 36 ++++++++++++++++++------------------ client/alloc_runner_test.go | 4 ++-- client/client.go | 32 ++++++++++++++++---------------- client/consul.go | 32 ++++++++++++++++---------------- client/consul_test.go | 16 ++++++++++------ client/task_runner.go | 10 +++++----- client/task_runner_test.go | 4 ++-- 7 files changed, 69 insertions(+), 65 deletions(-) diff --git a/client/alloc_runner.go b/client/alloc_runner.go index ba85f58cb..d48aaf166 100644 --- a/client/alloc_runner.go +++ b/client/alloc_runner.go @@ -33,10 +33,10 @@ type AllocStateUpdater func(alloc *structs.Allocation) error // AllocRunner is used to wrap an allocation and provide the execution context. type AllocRunner struct { - config *config.Config - updater AllocStateUpdater - logger *log.Logger - consulClient *ConsulClient + config *config.Config + updater AllocStateUpdater + logger *log.Logger + consulService *ConsulService alloc *structs.Allocation @@ -68,19 +68,19 @@ type allocRunnerState struct { // NewAllocRunner is used to create a new allocation context func NewAllocRunner(logger *log.Logger, config *config.Config, updater AllocStateUpdater, - alloc *structs.Allocation, consulClient *ConsulClient) *AllocRunner { + alloc *structs.Allocation, consulService *ConsulService) *AllocRunner { ar := &AllocRunner{ - config: config, - updater: updater, - logger: logger, - alloc: alloc, - consulClient: consulClient, - dirtyCh: make(chan struct{}, 1), - tasks: make(map[string]*TaskRunner), - restored: make(map[string]struct{}), - updateCh: make(chan *structs.Allocation, 8), - destroyCh: make(chan struct{}), - waitCh: make(chan struct{}), + config: config, + updater: updater, + logger: logger, + alloc: alloc, + consulService: consulService, + dirtyCh: make(chan struct{}, 1), + tasks: make(map[string]*TaskRunner), + restored: make(map[string]struct{}), + updateCh: make(chan *structs.Allocation, 8), + destroyCh: make(chan struct{}), + waitCh: make(chan struct{}), } return ar } @@ -113,7 +113,7 @@ func (r *AllocRunner) RestoreState() error { restartTracker := newRestartTracker(r.alloc.Job.Type, r.RestartPolicy) tr := NewTaskRunner(r.logger, r.config, r.setTaskState, r.ctx, r.alloc.ID, task, r.alloc.TaskStates[task.Name], restartTracker, - r.consulClient) + r.consulService) r.tasks[name] = tr // Skip tasks in terminal states. @@ -325,7 +325,7 @@ func (r *AllocRunner) Run() { restartTracker := newRestartTracker(r.alloc.Job.Type, r.RestartPolicy) tr := NewTaskRunner(r.logger, r.config, r.setTaskState, r.ctx, r.alloc.ID, task, r.alloc.TaskStates[task.Name], restartTracker, - r.consulClient) + r.consulService) r.tasks[task.Name] = tr go tr.Run() } diff --git a/client/alloc_runner_test.go b/client/alloc_runner_test.go index bc4a7aa4f..d52283865 100644 --- a/client/alloc_runner_test.go +++ b/client/alloc_runner_test.go @@ -31,7 +31,7 @@ func testAllocRunner(restarts bool) (*MockAllocStateUpdater, *AllocRunner) { conf.AllocDir = os.TempDir() upd := &MockAllocStateUpdater{} alloc := mock.Alloc() - consulClient, _ := NewConsulClient(logger, "127.0.0.1:8500") + consulClient, _ := NewConsulService(logger, "127.0.0.1:8500") if !restarts { alloc.Job.Type = structs.JobTypeBatch *alloc.Job.LookupTaskGroup(alloc.TaskGroup).RestartPolicy = structs.RestartPolicy{Attempts: 0} @@ -142,7 +142,7 @@ func TestAllocRunner_SaveRestoreState(t *testing.T) { } // Create a new alloc runner - consulClient, err := NewConsulClient(ar.logger, "127.0.0.1:8500") + consulClient, err := NewConsulService(ar.logger, "127.0.0.1:8500") ar2 := NewAllocRunner(ar.logger, ar.config, upd.Update, &structs.Allocation{ID: ar.alloc.ID}, consulClient) err = ar2.RestoreState() diff --git a/client/client.go b/client/client.go index 44c12e14f..b95168d12 100644 --- a/client/client.go +++ b/client/client.go @@ -71,7 +71,7 @@ type Client struct { logger *log.Logger - consulClient *ConsulClient + consulService *ConsulService lastServer net.Addr lastRPCTime time.Time @@ -99,22 +99,22 @@ func NewClient(cfg *config.Config) (*Client, error) { // Create a logger logger := log.New(cfg.LogOutput, "", log.LstdFlags) - // Create the consul client + // Create the consul service consulAddr := cfg.ReadDefault("consul.address", "127.0.0.1:8500") - consulClient, err := NewConsulClient(logger, consulAddr) + consulService, err := NewConsulService(logger, consulAddr) if err != nil { return nil, fmt.Errorf("failed to create the consul client: %v", err) } // Create the client c := &Client{ - config: cfg, - start: time.Now(), - consulClient: consulClient, - connPool: nomad.NewPool(cfg.LogOutput, clientRPCCache, clientMaxStreams, nil), - logger: logger, - allocs: make(map[string]*AllocRunner), - shutdownCh: make(chan struct{}), + config: cfg, + start: time.Now(), + consulService: consulService, + connPool: nomad.NewPool(cfg.LogOutput, clientRPCCache, clientMaxStreams, nil), + logger: logger, + allocs: make(map[string]*AllocRunner), + shutdownCh: make(chan struct{}), } // Initialize the client @@ -148,8 +148,8 @@ func NewClient(cfg *config.Config) (*Client, error) { // Start the client! go c.run() - // Start the consul client - go c.consulClient.SyncWithConsul() + // Start the consul service + go c.consulService.SyncWithConsul() return c, nil } @@ -214,8 +214,8 @@ func (c *Client) Shutdown() error { } } - // Stop the consul client - c.consulClient.ShutDown() + // Stop the consul service + c.consulService.ShutDown() c.shutdown = true close(c.shutdownCh) @@ -352,7 +352,7 @@ func (c *Client) restoreState() error { for _, entry := range list { id := entry.Name() alloc := &structs.Allocation{ID: id} - ar := NewAllocRunner(c.logger, c.config, c.updateAllocStatus, alloc, c.consulClient) + ar := NewAllocRunner(c.logger, c.config, c.updateAllocStatus, alloc, c.consulService) c.allocs[id] = ar if err := ar.RestoreState(); err != nil { c.logger.Printf("[ERR] client: failed to restore state for alloc %s: %v", id, err) @@ -791,7 +791,7 @@ func (c *Client) updateAlloc(exist, update *structs.Allocation) error { func (c *Client) addAlloc(alloc *structs.Allocation) error { c.allocLock.Lock() defer c.allocLock.Unlock() - ar := NewAllocRunner(c.logger, c.config, c.updateAllocStatus, alloc, c.consulClient) + ar := NewAllocRunner(c.logger, c.config, c.updateAllocStatus, alloc, c.consulService) c.allocs[alloc.ID] = ar go ar.Run() return nil diff --git a/client/consul.go b/client/consul.go index 4b6d1a939..12ef8ae69 100644 --- a/client/consul.go +++ b/client/consul.go @@ -24,7 +24,7 @@ type trackedService struct { func (t *trackedService) IsServiceValid() bool { for _, service := range t.task.Services { - if service.Id == t.service.Id { + if service.Hash() == t.service.Hash() { return true } } @@ -32,7 +32,7 @@ func (t *trackedService) IsServiceValid() bool { return false } -type ConsulClient struct { +type ConsulService struct { client *consul.Client logger *log.Logger shutdownCh chan struct{} @@ -43,7 +43,7 @@ type ConsulClient struct { trackedChkLock sync.Mutex } -func NewConsulClient(logger *log.Logger, consulAddr string) (*ConsulClient, error) { +func NewConsulService(logger *log.Logger, consulAddr string) (*ConsulService, error) { var err error var c *consul.Client cfg := consul.DefaultConfig() @@ -52,17 +52,17 @@ func NewConsulClient(logger *log.Logger, consulAddr string) (*ConsulClient, erro return nil, err } - consulClient := ConsulClient{ + consulService := ConsulService{ client: c, logger: logger, trackedServices: make(map[string]*trackedService), shutdownCh: make(chan struct{}), } - return &consulClient, nil + return &consulService, nil } -func (c *ConsulClient) Register(task *structs.Task, allocID string) error { +func (c *ConsulService) Register(task *structs.Task, allocID string) error { var mErr multierror.Error for _, service := range task.Services { c.logger.Printf("[INFO] consul: Registering service %s with Consul.", service.Name) @@ -74,7 +74,7 @@ func (c *ConsulClient) Register(task *structs.Task, allocID string) error { return mErr.ErrorOrNil() } -func (c *ConsulClient) Deregister(task *structs.Task) error { +func (c *ConsulService) Deregister(task *structs.Task) error { var mErr multierror.Error for _, service := range task.Services { if service.Id == "" { @@ -82,18 +82,18 @@ func (c *ConsulClient) Deregister(task *structs.Task) error { } 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) + c.logger.Printf("[DEBUG] consul: Error in de-registering service %v from Consul", service.Name) mErr.Errors = append(mErr.Errors, err) } } return mErr.ErrorOrNil() } -func (c *ConsulClient) ShutDown() { +func (c *ConsulService) ShutDown() { close(c.shutdownCh) } -func (c *ConsulClient) findPortAndHostForLabel(portLabel string, task *structs.Task) (string, int) { +func (c *ConsulService) findPortAndHostForLabel(portLabel string, task *structs.Task) (string, int) { for _, network := range task.Resources.Networks { if p, ok := network.MapLabelToValues(nil)[portLabel]; ok { return network.IP, p @@ -102,7 +102,7 @@ func (c *ConsulClient) findPortAndHostForLabel(portLabel string, task *structs.T return "", 0 } -func (c *ConsulClient) SyncWithConsul() { +func (c *ConsulService) SyncWithConsul() { sync := time.After(syncInterval) agent := c.client.Agent() @@ -153,9 +153,9 @@ func (c *ConsulClient) SyncWithConsul() { } } -func (c *ConsulClient) registerService(service *structs.Service, task *structs.Task, allocID string) error { +func (c *ConsulService) registerService(service *structs.Service, task *structs.Task, allocID string) error { var mErr multierror.Error - service.Id = service.Hash() + service.Id = fmt.Sprintf("%s-%s", allocID, service.Name) host, port := c.findPortAndHostForLabel(service.PortLabel, task) if host == "" || port == 0 { return fmt.Errorf("consul: The port:%s marked for registration of service: %s couldn't be found", service.PortLabel, service.Name) @@ -177,7 +177,7 @@ func (c *ConsulClient) registerService(service *structs.Service, task *structs.T 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) + c.logger.Printf("[DEBUG] consul: Error while registering service %v with Consul: %v", service.Name, err) mErr.Errors = append(mErr.Errors, err) } checks := c.makeChecks(service, host, port) @@ -193,7 +193,7 @@ func (c *ConsulClient) registerService(service *structs.Service, task *structs.T return mErr.ErrorOrNil() } -func (c *ConsulClient) deregisterService(serviceId string) error { +func (c *ConsulService) deregisterService(serviceId string) error { c.trackedSrvLock.Lock() delete(c.trackedServices, serviceId) c.trackedSrvLock.Unlock() @@ -204,7 +204,7 @@ func (c *ConsulClient) deregisterService(serviceId string) error { return nil } -func (c *ConsulClient) makeChecks(service *structs.Service, ip string, port int) []*consul.AgentCheckRegistration { +func (c *ConsulService) makeChecks(service *structs.Service, ip string, port int) []*consul.AgentCheckRegistration { var checks []*consul.AgentCheckRegistration for _, check := range service.Checks { if check.Name == "" { diff --git a/client/consul_test.go b/client/consul_test.go index bd497e871..bad1e7c16 100644 --- a/client/consul_test.go +++ b/client/consul_test.go @@ -8,13 +8,13 @@ import ( "time" ) -func newConsulClient() *ConsulClient { +func newConsulService() *ConsulService { logger := log.New(os.Stdout, "logger: ", log.Lshortfile) - c, _ := NewConsulClient(logger, "") + c, _ := NewConsulService(logger, "") return c } -func TestMakeChecks(t *testing.T) { +func TestConsul_MakeChecks(t *testing.T) { service := &structs.Service{ Id: "Foo", Name: "Bar", @@ -40,7 +40,7 @@ func TestMakeChecks(t *testing.T) { }, } - c := newConsulClient() + c := newConsulService() checks := c.makeChecks(service, "10.10.0.1", 8090) @@ -57,7 +57,7 @@ func TestMakeChecks(t *testing.T) { } } -func TestInvalidPortLabelForService(t *testing.T) { +func TestConsul_InvalidPortLabelForService(t *testing.T) { task := &structs.Task{ Name: "foo", Driver: "docker", @@ -93,8 +93,12 @@ func TestInvalidPortLabelForService(t *testing.T) { Checks: make([]structs.ServiceCheck, 0), } - c := newConsulClient() + c := newConsulService() if err := c.registerService(service, task, "allocid"); err == nil { t.Fatalf("Service should be invalid") } } + +func TestSyncWithConsul_Services_Deleted_From_Task(t *testing.T) { + +} diff --git a/client/task_runner.go b/client/task_runner.go index 749b4d6d4..c515abac5 100644 --- a/client/task_runner.go +++ b/client/task_runner.go @@ -25,7 +25,7 @@ type TaskRunner struct { ctx *driver.ExecContext allocID string restartTracker restartTracker - consulClient *ConsulClient + consulService *ConsulService task *structs.Task state *structs.TaskState @@ -53,14 +53,14 @@ type TaskStateUpdater func(taskName string) func NewTaskRunner(logger *log.Logger, config *config.Config, updater TaskStateUpdater, ctx *driver.ExecContext, allocID string, task *structs.Task, state *structs.TaskState, - restartTracker restartTracker, consulClient *ConsulClient) *TaskRunner { + restartTracker restartTracker, consulService *ConsulService) *TaskRunner { tc := &TaskRunner{ config: config, updater: updater, logger: logger, restartTracker: restartTracker, - consulClient: consulClient, + consulService: consulService, ctx: ctx, allocID: allocID, task: task, @@ -234,10 +234,10 @@ func (r *TaskRunner) run() { destroyed := false // Register the services defined by the task with Consil - r.consulClient.Register(r.task, r.allocID) + r.consulService.Register(r.task, r.allocID) // De-Register the services belonging to the task from consul - defer r.consulClient.Deregister(r.task) + defer r.consulService.Deregister(r.task) OUTER: // Wait for updates diff --git a/client/task_runner_test.go b/client/task_runner_test.go index 1ada5060b..ae9a2c4c5 100644 --- a/client/task_runner_test.go +++ b/client/task_runner_test.go @@ -32,7 +32,7 @@ func testTaskRunner(restarts bool) (*MockTaskStateUpdater, *TaskRunner) { upd := &MockTaskStateUpdater{} alloc := mock.Alloc() task := alloc.Job.TaskGroups[0].Tasks[0] - consulClient, _ := NewConsulClient(logger, "127.0.0.1:8500") + consulClient, _ := NewConsulService(logger, "127.0.0.1:8500") // Initialize the port listing. This should be done by the offer process but // we have a mock so that doesn't happen. task.Resources.Networks[0].ReservedPorts = []structs.Port{{"", 80}} @@ -164,7 +164,7 @@ func TestTaskRunner_SaveRestoreState(t *testing.T) { } // Create a new task runner - consulClient, _ := NewConsulClient(tr.logger, "127.0.0.1:8500") + consulClient, _ := NewConsulService(tr.logger, "127.0.0.1:8500") tr2 := NewTaskRunner(tr.logger, tr.config, upd.Update, tr.ctx, tr.allocID, &structs.Task{Name: tr.task.Name}, tr.state, tr.restartTracker, consulClient) From 919f763ca8adf20144dda86744a3ab0c08051e06 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Tue, 24 Nov 2015 14:37:14 -0800 Subject: [PATCH 06/13] Added a test to re-sync services --- client/consul.go | 77 +++++++++++++++++++++++-------------------- client/consul_test.go | 61 +++++++++++++++++++++++++++++++++- 2 files changed, 101 insertions(+), 37 deletions(-) diff --git a/client/consul.go b/client/consul.go index 12ef8ae69..738a08c71 100644 --- a/client/consul.go +++ b/client/consul.go @@ -109,42 +109,7 @@ func (c *ConsulService) SyncWithConsul() { for { select { case <-sync: - 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 { - continue - } - - // See if we have services that Consul doesn't know about yet. - // Register with Consul the services which are not registered - for serviceId := range c.trackedServices { - if _, ok := consulServices[serviceId]; !ok { - ts := c.trackedServices[serviceId] - c.registerService(ts.service, ts.task, ts.allocId) - } - } - - // See if consul thinks we have some services which are not running - // anymore on the node. We de-register those services - for serviceId := range consulServices { - if serviceId == "consul" { - continue - } - if _, ok := c.trackedServices[serviceId]; !ok { - if err := c.deregisterService(serviceId); err != nil { - c.logger.Printf("[DEBUG] consul: Error while de-registering service with ID: %s", serviceId) - } - } - } + c.performSync(agent) sync = time.After(syncInterval) case <-c.shutdownCh: c.logger.Printf("[INFO] Shutting down Consul Client") @@ -153,6 +118,46 @@ func (c *ConsulService) SyncWithConsul() { } } +func (c *ConsulService) performSync(agent *consul.Agent) { + 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 { + return + } + + // See if we have services that Consul doesn't know about yet. + // Register with Consul the services which are not registered + for serviceId := range c.trackedServices { + if _, ok := consulServices[serviceId]; !ok { + ts := c.trackedServices[serviceId] + c.registerService(ts.service, ts.task, ts.allocId) + } + } + + // See if consul thinks we have some services which are not running + // anymore on the node. We de-register those services + for serviceId := range consulServices { + if serviceId == "consul" { + continue + } + if _, ok := c.trackedServices[serviceId]; !ok { + if err := c.deregisterService(serviceId); err != nil { + c.logger.Printf("[DEBUG] consul: Error while de-registering service with ID: %s", serviceId) + } + } + } + +} + func (c *ConsulService) registerService(service *structs.Service, task *structs.Task, allocID string) error { var mErr multierror.Error service.Id = fmt.Sprintf("%s-%s", allocID, service.Name) diff --git a/client/consul_test.go b/client/consul_test.go index bad1e7c16..87254ee65 100644 --- a/client/consul_test.go +++ b/client/consul_test.go @@ -99,6 +99,65 @@ func TestConsul_InvalidPortLabelForService(t *testing.T) { } } -func TestSyncWithConsul_Services_Deleted_From_Task(t *testing.T) { +func TestConsul_Services_Deleted_From_Task(t *testing.T) { + c := newConsulService() + task := structs.Task{ + Name: "redis", + Services: make([]*structs.Service, 0), + } + s1 := structs.Service{ + Id: "1-example-cache-redis", + Name: "example-cache-redis", + Tags: []string{"global"}, + PortLabel: "db", + } + ts := trackedService{ + allocId: "1", + task: &task, + service: &s1, + } + c.trackedServices = map[string]*trackedService{ + "1-example-cache-redis": &ts, + } + c.performSync(c.client.Agent()) + if len(c.trackedServices) != 0 { + t.Fatal("All services should have been de-registered") + } +} + +func TestConsul_Service_Should_Be_Re_Reregistered_On_Change(t *testing.T) { + c := newConsulService() + var services []*structs.Service + task := structs.Task{ + Name: "redis", + Services: services, + } + s1 := structs.Service{ + Id: "1-example-cache-redis", + Name: "example-cache-redis", + Tags: []string{"global"}, + PortLabel: "db", + } + task.Services = append(task.Services, &s1) + ts := trackedService{ + allocId: "1", + task: &task, + service: &s1, + } + c.trackedServices = map[string]*trackedService{ + "1-example-cache-redis": &ts, + } + + s1.Tags = []string{"frontcache"} + + c.performSync(c.client.Agent()) + + if len(c.trackedServices) != 1 { + t.Fatal("We should be tracking one service") + } + + if c.trackedServices[s1.Id].service.Tags[0] != "frontcache" { + t.Fatalf("Tag is %v, expected %v", c.trackedServices[s1.Id].service.Tags[0], "frontcache") + } } From 2c0822284bc622f3e46087f4275d2a295a9d0fed Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Tue, 24 Nov 2015 17:26:30 -0800 Subject: [PATCH 07/13] Tracking the tasks too --- client/consul.go | 41 ++++++++++++++++++++++++++++++++++------- client/consul_test.go | 17 +++++++++-------- client/task_runner.go | 2 +- 3 files changed, 44 insertions(+), 16 deletions(-) diff --git a/client/consul.go b/client/consul.go index 738a08c71..c216f1095 100644 --- a/client/consul.go +++ b/client/consul.go @@ -17,14 +17,20 @@ const ( ) type trackedService struct { - allocId string + allocId string + task *structs.Task + serviceHash string + service *structs.Service +} + +type trackedTask struct { + allocID string task *structs.Task - service *structs.Service } func (t *trackedService) IsServiceValid() bool { for _, service := range t.task.Services { - if service.Hash() == t.service.Hash() { + if service.Id == t.service.Id && service.Hash() == t.serviceHash { return true } } @@ -39,8 +45,10 @@ type ConsulService struct { trackedServices map[string]*trackedService // Service ID to Tracked Service Map trackedChecks map[string]bool // List of check ids that is being tracked + trackedTasks map[string]*trackedTask trackedSrvLock sync.Mutex trackedChkLock sync.Mutex + trackedTskLock sync.Mutex } func NewConsulService(logger *log.Logger, consulAddr string) (*ConsulService, error) { @@ -56,6 +64,7 @@ func NewConsulService(logger *log.Logger, consulAddr string) (*ConsulService, er client: c, logger: logger, trackedServices: make(map[string]*trackedService), + trackedTasks: make(map[string]*trackedTask), shutdownCh: make(chan struct{}), } @@ -64,6 +73,10 @@ func NewConsulService(logger *log.Logger, consulAddr string) (*ConsulService, er func (c *ConsulService) Register(task *structs.Task, allocID string) error { var mErr multierror.Error + c.trackedTskLock.Lock() + tt := &trackedTask{allocID: allocID, task: task} + c.trackedTasks[fmt.Sprintf("%s-%s", allocID, task.Name)] = tt + c.trackedTskLock.Unlock() 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 { @@ -74,8 +87,11 @@ func (c *ConsulService) Register(task *structs.Task, allocID string) error { return mErr.ErrorOrNil() } -func (c *ConsulService) Deregister(task *structs.Task) error { +func (c *ConsulService) Deregister(task *structs.Task, allocID string) error { var mErr multierror.Error + c.trackedTskLock.Lock() + delete(c.trackedTasks, fmt.Sprintf("%s-%s", allocID, task.Name)) + c.trackedTskLock.Unlock() for _, service := range task.Services { if service.Id == "" { continue @@ -122,6 +138,7 @@ func (c *ConsulService) performSync(agent *consul.Agent) { var consulServices map[string]*consul.AgentService var err error + // Remove the tracked services which tasks no longer references 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) @@ -129,6 +146,15 @@ func (c *ConsulService) performSync(agent *consul.Agent) { } } + // Add additional tasks that we might not have added from tasks + for _, trackedTask := range c.trackedTasks { + for _, service := range trackedTask.task.Services { + if _, ok := c.trackedServices[service.Id]; !ok { + c.registerService(service, trackedTask.task, trackedTask.allocID) + } + } + } + // Get the list of the services that Consul knows about if consulServices, err = agent.Services(); err != nil { return @@ -173,9 +199,10 @@ func (c *ConsulService) registerService(service *structs.Service, task *structs. Address: host, } ts := &trackedService{ - allocId: allocID, - task: task, - service: service, + allocId: allocID, + task: task, + serviceHash: service.Hash(), + service: service, } c.trackedSrvLock.Lock() c.trackedServices[service.Id] = ts diff --git a/client/consul_test.go b/client/consul_test.go index 87254ee65..002b89f09 100644 --- a/client/consul_test.go +++ b/client/consul_test.go @@ -132,6 +132,14 @@ func TestConsul_Service_Should_Be_Re_Reregistered_On_Change(t *testing.T) { task := structs.Task{ Name: "redis", Services: services, + Resources: &structs.Resources{ + Networks: []*structs.NetworkResource{ + { + IP: "10.10.0.1", + DynamicPorts: []structs.Port{{"db", 20413}}, + }, + }, + }, } s1 := structs.Service{ Id: "1-example-cache-redis", @@ -140,14 +148,7 @@ func TestConsul_Service_Should_Be_Re_Reregistered_On_Change(t *testing.T) { PortLabel: "db", } task.Services = append(task.Services, &s1) - ts := trackedService{ - allocId: "1", - task: &task, - service: &s1, - } - c.trackedServices = map[string]*trackedService{ - "1-example-cache-redis": &ts, - } + c.Register(&task, "1") s1.Tags = []string{"frontcache"} diff --git a/client/task_runner.go b/client/task_runner.go index c515abac5..7f6cc40ff 100644 --- a/client/task_runner.go +++ b/client/task_runner.go @@ -237,7 +237,7 @@ func (r *TaskRunner) run() { r.consulService.Register(r.task, r.allocID) // De-Register the services belonging to the task from consul - defer r.consulService.Deregister(r.task) + defer r.consulService.Deregister(r.task, r.allocID) OUTER: // Wait for updates From f72027277d0975e08013705e1e91e8f448786cc9 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Tue, 24 Nov 2015 17:33:04 -0800 Subject: [PATCH 08/13] Dried the tests --- client/consul_test.go | 34 ++++++++++++++++++---------------- 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/client/consul_test.go b/client/consul_test.go index 002b89f09..eba619622 100644 --- a/client/consul_test.go +++ b/client/consul_test.go @@ -102,23 +102,25 @@ func TestConsul_InvalidPortLabelForService(t *testing.T) { func TestConsul_Services_Deleted_From_Task(t *testing.T) { c := newConsulService() task := structs.Task{ - Name: "redis", - Services: make([]*structs.Service, 0), - } - s1 := structs.Service{ - Id: "1-example-cache-redis", - Name: "example-cache-redis", - Tags: []string{"global"}, - PortLabel: "db", - } - ts := trackedService{ - allocId: "1", - task: &task, - service: &s1, - } - c.trackedServices = map[string]*trackedService{ - "1-example-cache-redis": &ts, + Name: "redis", + Services: []*structs.Service{ + &structs.Service{ + Name: "example-cache-redis", + Tags: []string{"global"}, + PortLabel: "db", + }, + }, + Resources: &structs.Resources{ + Networks: []*structs.NetworkResource{ + { + IP: "10.10.0.1", + DynamicPorts: []structs.Port{{"db", 20413}}, + }, + }, + }, } + c.Register(&task, "1") + task.Services = []*structs.Service{} c.performSync(c.client.Agent()) if len(c.trackedServices) != 0 { From 28a21061ce49b69951cc72ac6df47e88cc7e406b Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Tue, 24 Nov 2015 18:39:38 -0800 Subject: [PATCH 09/13] Implemented syncing of checks --- client/consul.go | 63 ++++++++++++++++++++++++++++++---------- nomad/structs/structs.go | 9 ++++++ 2 files changed, 56 insertions(+), 16 deletions(-) diff --git a/client/consul.go b/client/consul.go index c216f1095..68077e8e7 100644 --- a/client/consul.go +++ b/client/consul.go @@ -43,8 +43,8 @@ type ConsulService struct { logger *log.Logger shutdownCh chan struct{} - trackedServices map[string]*trackedService // Service ID to Tracked Service Map - trackedChecks map[string]bool // List of check ids that is being tracked + trackedServices map[string]*trackedService // Service ID to Tracked Service Map + trackedChecks map[string]*consul.AgentCheckRegistration // List of check ids that is being tracked trackedTasks map[string]*trackedTask trackedSrvLock sync.Mutex trackedChkLock sync.Mutex @@ -65,6 +65,7 @@ func NewConsulService(logger *log.Logger, consulAddr string) (*ConsulService, er logger: logger, trackedServices: make(map[string]*trackedService), trackedTasks: make(map[string]*trackedTask), + trackedChecks: make(map[string]*consul.AgentCheckRegistration), shutdownCh: make(chan struct{}), } @@ -109,15 +110,6 @@ func (c *ConsulService) ShutDown() { close(c.shutdownCh) } -func (c *ConsulService) findPortAndHostForLabel(portLabel string, task *structs.Task) (string, int) { - for _, network := range task.Resources.Networks { - if p, ok := network.MapLabelToValues(nil)[portLabel]; ok { - return network.IP, p - } - } - return "", 0 -} - func (c *ConsulService) SyncWithConsul() { sync := time.After(syncInterval) agent := c.client.Agent() @@ -136,6 +128,7 @@ func (c *ConsulService) SyncWithConsul() { func (c *ConsulService) performSync(agent *consul.Agent) { var consulServices map[string]*consul.AgentService + var consulChecks map[string]*consul.AgentCheck var err error // Remove the tracked services which tasks no longer references @@ -182,12 +175,37 @@ func (c *ConsulService) performSync(agent *consul.Agent) { } } + if consulChecks, err = agent.Checks(); err != nil { + return + } + + // Remove checks that Consul knows about but we don't + for checkID := range consulChecks { + if _, ok := c.trackedChecks[checkID]; !ok { + c.deregisterCheck(checkID) + } + } + + // Add checks that might not be present + for _, trackedService := range c.trackedServices { + host, port := trackedService.task.FindHostAndPortFor(trackedService.service.PortLabel) + if host == "" || port == 0 { + continue + } + checks := c.makeChecks(trackedService.service, host, port) + for _, check := range checks { + if _, ok := consulChecks[check.ID]; !ok { + c.registerCheck(check) + } + } + + } } func (c *ConsulService) registerService(service *structs.Service, task *structs.Task, allocID string) error { var mErr multierror.Error service.Id = fmt.Sprintf("%s-%s", allocID, service.Name) - host, port := c.findPortAndHostForLabel(service.PortLabel, task) + host, port := task.FindHostAndPortFor(service.PortLabel) if host == "" || port == 0 { return fmt.Errorf("consul: The port:%s marked for registration of service: %s couldn't be found", service.PortLabel, service.Name) } @@ -214,17 +232,30 @@ func (c *ConsulService) registerService(service *structs.Service, task *structs. } checks := c.makeChecks(service, host, port) for _, check := range checks { - if err := c.client.Agent().CheckRegister(check); err != nil { + if err := c.registerCheck(check); err != nil { c.logger.Printf("[ERROR] consul: Error while registerting check %v with Consul: %v", check.Name, err) mErr.Errors = append(mErr.Errors, err) } - c.trackedChkLock.Lock() - c.trackedChecks[check.ID] = true - c.trackedChkLock.Unlock() } return mErr.ErrorOrNil() } +func (c *ConsulService) registerCheck(check *consul.AgentCheckRegistration) error { + c.logger.Printf("[DEBUG] Registering Check with ID: %v for Service: %v", check.ID, check.ServiceID) + c.trackedChkLock.Lock() + c.trackedChecks[check.ID] = check + c.trackedChkLock.Unlock() + return c.client.Agent().CheckRegister(check) +} + +func (c *ConsulService) deregisterCheck(checkID string) error { + c.logger.Printf("[DEBUG] Removing check with ID: %v", checkID + c.trackedChkLock.Lock() + delete(c.trackedChecks, checkID) + c.trackedChkLock.Unlock() + return c.client.Agent().CheckDeregister(checkID) +} + func (c *ConsulService) deregisterService(serviceId string) error { c.trackedSrvLock.Lock() delete(c.trackedServices, serviceId) diff --git a/nomad/structs/structs.go b/nomad/structs/structs.go index f7a3a1b04..8aaf0e7ef 100644 --- a/nomad/structs/structs.go +++ b/nomad/structs/structs.go @@ -1113,6 +1113,15 @@ func (t *Task) GoString() string { return fmt.Sprintf("*%#v", *t) } +func (t *Task) FindHostAndPortFor(portLabel string) (string, int) { + for _, network := range t.Resources.Networks { + if p, ok := network.MapLabelToValues(nil)[portLabel]; ok { + return network.IP, p + } + } + return "", 0 +} + // Set of possible states for a task. const ( TaskStatePending = "pending" // The task is waiting to be run. From 6353dde236577dd9b11ccd86d46d77ee726a9f43 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Tue, 24 Nov 2015 18:43:23 -0800 Subject: [PATCH 10/13] Fixing comment and syntax --- client/consul.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/client/consul.go b/client/consul.go index 68077e8e7..ed27de0c2 100644 --- a/client/consul.go +++ b/client/consul.go @@ -139,7 +139,7 @@ func (c *ConsulService) performSync(agent *consul.Agent) { } } - // Add additional tasks that we might not have added from tasks + // Add additional services that we might not have added from tasks for _, trackedTask := range c.trackedTasks { for _, service := range trackedTask.task.Services { if _, ok := c.trackedServices[service.Id]; !ok { @@ -249,7 +249,7 @@ func (c *ConsulService) registerCheck(check *consul.AgentCheckRegistration) erro } func (c *ConsulService) deregisterCheck(checkID string) error { - c.logger.Printf("[DEBUG] Removing check with ID: %v", checkID + c.logger.Printf("[DEBUG] Removing check with ID: %v", checkID) c.trackedChkLock.Lock() delete(c.trackedChecks, checkID) c.trackedChkLock.Unlock() From 5480a2bd0f2f4750385aad1b8e2780c7c11eac8b Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Tue, 24 Nov 2015 18:58:53 -0800 Subject: [PATCH 11/13] Saving the host and port in tracked service --- client/consul.go | 31 ++++++++++++++++--------------- 1 file changed, 16 insertions(+), 15 deletions(-) diff --git a/client/consul.go b/client/consul.go index ed27de0c2..fd220eaf9 100644 --- a/client/consul.go +++ b/client/consul.go @@ -21,6 +21,8 @@ type trackedService struct { task *structs.Task serviceHash string service *structs.Service + host string + port int } type trackedTask struct { @@ -187,12 +189,8 @@ func (c *ConsulService) performSync(agent *consul.Agent) { } // Add checks that might not be present - for _, trackedService := range c.trackedServices { - host, port := trackedService.task.FindHostAndPortFor(trackedService.service.PortLabel) - if host == "" || port == 0 { - continue - } - checks := c.makeChecks(trackedService.service, host, port) + for _, ts := range c.trackedServices { + checks := c.makeChecks(ts.service, ts.host, ts.port) for _, check := range checks { if _, ok := consulChecks[check.ID]; !ok { c.registerCheck(check) @@ -209,6 +207,18 @@ func (c *ConsulService) registerService(service *structs.Service, task *structs. if host == "" || port == 0 { return fmt.Errorf("consul: The port:%s marked for registration of service: %s couldn't be found", service.PortLabel, service.Name) } + ts := &trackedService{ + allocId: allocID, + task: task, + serviceHash: service.Hash(), + service: service, + host: host, + port: port, + } + c.trackedSrvLock.Lock() + c.trackedServices[service.Id] = ts + c.trackedSrvLock.Unlock() + asr := &consul.AgentServiceRegistration{ ID: service.Id, Name: service.Name, @@ -216,15 +226,6 @@ func (c *ConsulService) registerService(service *structs.Service, task *structs. Port: port, Address: host, } - ts := &trackedService{ - allocId: allocID, - task: task, - serviceHash: service.Hash(), - 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("[DEBUG] consul: Error while registering service %v with Consul: %v", service.Name, err) From 576a6f402e035a66424ba0e6acc53f624dc3bf88 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Wed, 25 Nov 2015 11:20:36 -0800 Subject: [PATCH 12/13] Added a test to make sure we are adding a check to a service --- client/consul.go | 9 ++------- client/consul_test.go | 45 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 7 deletions(-) diff --git a/client/consul.go b/client/consul.go index fd220eaf9..753511d53 100644 --- a/client/consul.go +++ b/client/consul.go @@ -131,7 +131,6 @@ func (c *ConsulService) SyncWithConsul() { func (c *ConsulService) performSync(agent *consul.Agent) { var consulServices map[string]*consul.AgentService var consulChecks map[string]*consul.AgentCheck - var err error // Remove the tracked services which tasks no longer references for serviceId, ts := range c.trackedServices { @@ -151,9 +150,7 @@ func (c *ConsulService) performSync(agent *consul.Agent) { } // Get the list of the services that Consul knows about - if consulServices, err = agent.Services(); err != nil { - return - } + consulServices, _ = agent.Services() // See if we have services that Consul doesn't know about yet. // Register with Consul the services which are not registered @@ -177,9 +174,7 @@ func (c *ConsulService) performSync(agent *consul.Agent) { } } - if consulChecks, err = agent.Checks(); err != nil { - return - } + consulChecks, _ = agent.Checks() // Remove checks that Consul knows about but we don't for checkID := range consulChecks { diff --git a/client/consul_test.go b/client/consul_test.go index eba619622..b205fe705 100644 --- a/client/consul_test.go +++ b/client/consul_test.go @@ -14,6 +14,22 @@ func newConsulService() *ConsulService { return c } +func newTask() *structs.Task { + var services []*structs.Service + return &structs.Task{ + Name: "redis", + Services: services, + Resources: &structs.Resources{ + Networks: []*structs.NetworkResource{ + { + IP: "10.10.0.1", + DynamicPorts: []structs.Port{{"db", 20413}}, + }, + }, + }, + } +} + func TestConsul_MakeChecks(t *testing.T) { service := &structs.Service{ Id: "Foo", @@ -164,3 +180,32 @@ func TestConsul_Service_Should_Be_Re_Reregistered_On_Change(t *testing.T) { t.Fatalf("Tag is %v, expected %v", c.trackedServices[s1.Id].service.Tags[0], "frontcache") } } + +func TestConsul_AddCheck_To_Service(t *testing.T) { + c := newConsulService() + task := newTask() + var checks []structs.ServiceCheck + s1 := structs.Service{ + Id: "1-example-cache-redis", + Name: "example-cache-redis", + Tags: []string{"global"}, + PortLabel: "db", + Checks: checks, + } + task.Services = append(task.Services, &s1) + c.Register(task, "1") + + check1 := structs.ServiceCheck{ + Name: "alive", + Type: "tcp", + Interval: 10 * time.Second, + Timeout: 5 * time.Second, + } + + s1.Checks = append(s1.Checks, check1) + + c.performSync(c.client.Agent()) + if len(c.trackedChecks) != 1 { + t.Fatalf("Expected tracked checks: %v, actual: %v", 1, len(c.trackedChecks)) + } +} From 95942b4179ba651c2bf53b0dc2a6ed185526abc1 Mon Sep 17 00:00:00 2001 From: Diptanu Choudhury Date: Wed, 25 Nov 2015 12:06:14 -0800 Subject: [PATCH 13/13] Improved tests --- client/consul.go | 2 +- client/consul_test.go | 21 ++++++--------------- 2 files changed, 7 insertions(+), 16 deletions(-) diff --git a/client/consul.go b/client/consul.go index 753511d53..d60c0d744 100644 --- a/client/consul.go +++ b/client/consul.go @@ -135,7 +135,7 @@ func (c *ConsulService) performSync(agent *consul.Agent) { // Remove the tracked services which tasks no longer references 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.logger.Printf("[DEBUG] consul: Removing service: %s since the task doesn't have it anymore", ts.service.Name) c.deregisterService(serviceId) } } diff --git a/client/consul_test.go b/client/consul_test.go index b205fe705..901b655f1 100644 --- a/client/consul_test.go +++ b/client/consul_test.go @@ -136,29 +136,20 @@ func TestConsul_Services_Deleted_From_Task(t *testing.T) { }, } c.Register(&task, "1") + if len(c.trackedServices) != 1 { + t.Fatalf("Expected tracked services: %v, Actual: %v", 1, len(c.trackedServices)) + } task.Services = []*structs.Service{} c.performSync(c.client.Agent()) if len(c.trackedServices) != 0 { - t.Fatal("All services should have been de-registered") + t.Fatalf("Expected tracked services: %v, Actual: %v", 0, len(c.trackedServices)) } } func TestConsul_Service_Should_Be_Re_Reregistered_On_Change(t *testing.T) { c := newConsulService() - var services []*structs.Service - task := structs.Task{ - Name: "redis", - Services: services, - Resources: &structs.Resources{ - Networks: []*structs.NetworkResource{ - { - IP: "10.10.0.1", - DynamicPorts: []structs.Port{{"db", 20413}}, - }, - }, - }, - } + task := newTask() s1 := structs.Service{ Id: "1-example-cache-redis", Name: "example-cache-redis", @@ -166,7 +157,7 @@ func TestConsul_Service_Should_Be_Re_Reregistered_On_Change(t *testing.T) { PortLabel: "db", } task.Services = append(task.Services, &s1) - c.Register(&task, "1") + c.Register(task, "1") s1.Tags = []string{"frontcache"}