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)