luxolus
/
open-nomad
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

Generate check names, check id and service id on the server side

This commit is contained in:
Diptanu Choudhury 2015-12-10 16:04:04 -08:00
parent 188b495ffe
commit b3c161a7d5
4 changed files with 62 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
client/consul.go
View File

@ -139,7 +139,7 @@ func (c *ConsulService) Register(task *structs.Task, allocID string) error {
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)
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)
}
@ -159,9 +159,9 @@ func (c *ConsulService) Deregister(task *structs.Task, allocID string) error {
if service.Id == "" {
continue
}
c.logger.Printf("[INFO] consul: De-Registering service %v with consul", service.Name)
c.logger.Printf("[INFO] consul: deregistering service %v with consul", service.Name)
if err := c.deregisterService(service.Id); err != nil {
c.logger.Printf("[DEBUG] consul: Error in de-registering service %v from consul", service.Name)
c.logger.Printf("[DEBUG] consul: error in deregistering service %v from consul", service.Name)
mErr.Errors = append(mErr.Errors, err)
}
}
@ -183,7 +183,7 @@ func (c *ConsulService) SyncWithConsul() {
c.performSync()
sync = time.After(syncInterval)
case <-c.shutdownCh:
c.logger.Printf("[INFO] consul: Shutting down consul Client")
c.logger.Printf("[INFO] consul: shutting down consul service")
return
}
}
@ -207,14 +207,14 @@ func (c *ConsulService) performSync() {
// Add new services which Consul agent isn't aware of
knownServices[service.Id] = struct{}{}
if _, ok := consulServices[service.Id]; !ok {
c.logger.Printf("[INFO] consul: Registering service %s with consul.", service.Name)
c.logger.Printf("[INFO] consul: registering service %s with consul.", service.Name)
c.registerService(service, trackedTask.task, trackedTask.allocID)
continue
}
// If a service has changed, re-register it with Consul agent
if service.Hash() != c.serviceStates[service.Id] {
c.logger.Printf("[INFO] consul: Re-Registering service %s with consul.", service.Name)
c.logger.Printf("[INFO] consul: reregistering service %s with consul.", service.Name)
c.registerService(service, trackedTask.task, trackedTask.allocID)
continue
}
@ -242,7 +242,7 @@ func (c *ConsulService) performSync() {
for _, consulService := range consulServices {
if _, ok := knownServices[consulService.ID]; !ok {
delete(c.serviceStates, consulService.ID)
c.logger.Printf("[INFO] consul: De-Registering service %v with consul", consulService.Service)
c.logger.Printf("[INFO] consul: deregistering service %v with consul", consulService.Service)
c.deregisterService(consulService.ID)
}
}
@ -258,10 +258,9 @@ func (c *ConsulService) performSync() {
// registerService registers a Service with Consul
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 := 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)
return fmt.Errorf("consul: the port:%s marked for registration of service: %s couldn't be found", service.PortLabel, service.Name)
}
c.serviceStates[service.Id] = service.Hash()
@ -274,13 +273,13 @@ func (c *ConsulService) registerService(service *structs.Service, task *structs.
}
if err := c.client.ServiceRegister(asr); err != nil {
c.logger.Printf("[DEBUG] 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)
}
for _, check := range service.Checks {
cr := c.makeCheck(service, check, host, port)
if err := c.registerCheck(cr); err != nil {
c.logger.Printf("[ERROR] consul: Error while registering check %v with consul: %v", check.Name, err)
c.logger.Printf("[DEBUG] consul: error while registerting check %v with consul: %v", check.Name, err)
mErr.Errors = append(mErr.Errors, err)
}
@ -290,13 +289,13 @@ func (c *ConsulService) registerService(service *structs.Service, task *structs.
// registerCheck registers a check with Consul
func (c *ConsulService) registerCheck(check *consul.AgentCheckRegistration) error {
c.logger.Printf("[INFO] consul: Registering Check with ID: %v for Service: %v", check.ID, check.ServiceID)
c.logger.Printf("[INFO] consul: registering Check with ID: %v for service: %v", check.ID, check.ServiceID)
return c.client.CheckRegister(check)
}
// deregisterCheck de-registers a check with a specific ID from Consul
func (c *ConsulService) deregisterCheck(checkID string) error {
c.logger.Printf("[INFO] consul: Removing check with ID: %v", checkID)
c.logger.Printf("[INFO] consul: removing check with ID: %v", checkID)
return c.client.CheckDeregister(checkID)
}
@ -311,11 +310,6 @@ func (c *ConsulService) deregisterService(serviceId string) error {
// makeCheck creates a Consul Check Registration struct
func (c *ConsulService) makeCheck(service *structs.Service, check *structs.ServiceCheck, ip string, port int) *consul.AgentCheckRegistration {
if check.Name == "" {
check.Name = fmt.Sprintf("service: %s check", service.Name)
}
check.Id = check.Hash(service.Id)
cr := &consul.AgentCheckRegistration{
ID: check.Id,
Name: check.Name,

4
nomad/job_endpoint.go
View File

@ -29,8 +29,8 @@ func (j *Job) Register(args *structs.JobRegisterRequest, reply *structs.JobRegis
return err
}
// Expand the service names
args.Job.ExpandAllServiceNames()
// Initialize all the fields of services
args.Job.InitAllServiceFields()
if args.Job.Type == structs.JobTypeCore {
return fmt.Errorf("job type cannot be core")

63
nomad/structs/structs.go
View File

@ -779,11 +779,13 @@ type Job struct {
ModifyIndex uint64
}
// ExpandAllServiceNames traverses all Task Groups and makes them
// interpolate Job, Task group and Task names in all Service names
func (j *Job) ExpandAllServiceNames() {
// InitAllServiceFields traverses all Task Groups and makes them
// interpolate Job, Task group and Task names in all Service names.
// It also generates the check names if they are not set. This method also
// generates Check and Service IDs
func (j *Job) InitAllServiceFields() {
for _, tg := range j.TaskGroups {
tg.ExpandAllServiceNames(j.Name)
tg.InitAllServiceFields(j.Name)
}
}
@ -1022,11 +1024,12 @@ type TaskGroup struct {
Meta map[string]string
}
// ExpandAllServiceNames traverses over all Tasks and makes them to interpolate
// values of Job, Task Group and Task names in all Service Names
func (tg *TaskGroup) ExpandAllServiceNames(job string) {
// InitAllServiceFields traverses over all Tasks and makes them to interpolate
// values of Job, Task Group and Task names in all Service Names.
// It also generates service ids, check ids and check names
func (tg *TaskGroup) InitAllServiceFields(job string) {
for _, task := range tg.Tasks {
task.ExpandAllServiceNames(job, tg.Name)
task.InitAllServiceFields(job, tg.Name)
}
}
@ -1113,18 +1116,6 @@ type ServiceCheck struct {
Timeout time.Duration // Timeout of the response from the check before consul fails the check
}
// ExpandName interpolates values of Job, Task Group and Task in the Service
// Name
func (s *Service) ExpandName(job string, taskGroup string, task string) {
s.Name = args.ReplaceEnv(s.Name, map[string]string{
"JOB": job,
"TASKGROUP": taskGroup,
"TASK": task,
"BASE": fmt.Sprintf("%s-%s-%s", job, taskGroup, task),
},
)
}
func (sc *ServiceCheck) Validate() error {
t := strings.ToLower(sc.Type)
if t != ServiceCheckTCP && t != ServiceCheckHTTP {
@ -1163,6 +1154,27 @@ type Service struct {
Checks []*ServiceCheck // List of checks associated with the service
}
// InitFields interpolates values of Job, Task Group and Task in the Service
// Name. This also generates check names, service id and check ids.
func (s *Service) InitFields(job string, taskGroup string, task string) {
s.Id = GenerateUUID()
s.Name = args.ReplaceEnv(s.Name, map[string]string{
"JOB": job,
"TASKGROUP": taskGroup,
"TASK": task,
"BASE": fmt.Sprintf("%s-%s-%s", job, taskGroup, task),
},
)
for _, check := range s.Checks {
check.Id = check.Hash(s.Id)
if check.Name == "" {
check.Name = fmt.Sprintf("service: %s check", s.Name)
}
}
}
// Validate checks if the Check definition is valid
func (s *Service) Validate() error {
var mErr multierror.Error
for _, c := range s.Checks {
@ -1173,6 +1185,8 @@ func (s *Service) Validate() error {
return mErr.ErrorOrNil()
}
// Hash calculates the hash of the check based on it's content and the service
// which owns it
func (s *Service) Hash() string {
h := sha1.New()
io.WriteString(h, s.Name)
@ -1210,11 +1224,12 @@ type Task struct {
Meta map[string]string
}
// ExpandAllServiceNames interpolates values of Job, Task Group
// and Tasks in all the service Names of a Task
func (t *Task) ExpandAllServiceNames(job string, taskGroup string) {
// InitAllServiceFields interpolates values of Job, Task Group
// and Tasks in all the service Names of a Task. This also generates the service
// id, check id and check names.
func (t *Task) InitAllServiceFields(job string, taskGroup string) {
for _, service := range t.Services {
service.ExpandName(job, taskGroup, t.Name)
service.InitFields(job, taskGroup, t.Name)
}
}

15
nomad/structs/structs_test.go
View File

@ -442,7 +442,7 @@ func TestDistinctCheckId(t *testing.T) {
}
func TestService_Expand_Name(t *testing.T) {
func TestService_InitFiels(t *testing.T) {
job := "example"
taskGroup := "cache"
task := "redis"
@ -451,25 +451,28 @@ func TestService_Expand_Name(t *testing.T) {
Name: "${TASK}-db",
}
s.ExpandName(job, taskGroup, task)
s.InitFields(job, taskGroup, task)
if s.Name != "redis-db" {
t.Fatalf("Expected name: %v, Actual: %v", "redis-db", s.Name)
}
if s.Id == "" {
t.Fatalf("Expected a GUID for Service ID, Actual: %v", s.Id)
}
s.Name = "db"
s.ExpandName(job, taskGroup, task)
s.InitFields(job, taskGroup, task)
if s.Name != "db" {
t.Fatalf("Expected name: %v, Actual: %v", "redis-db", s.Name)
}
s.Name = "${JOB}-${TASKGROUP}-${TASK}-db"
s.ExpandName(job, taskGroup, task)
s.InitFields(job, taskGroup, task)
if s.Name != "example-cache-redis-db" {
t.Fatalf("Expected name: %v, Actual: %v", "expample-cache-redis-db", s.Name)
}
s.Name = "${BASE}-db"
s.ExpandName(job, taskGroup, task)
s.InitFields(job, taskGroup, task)
if s.Name != "example-cache-redis-db" {
t.Fatalf("Expected name: %v, Actual: %v", "expample-cache-redis-db", s.Name)
}
@ -507,7 +510,7 @@ func TestJob_ExpandServiceNames(t *testing.T) {
},
}
j.ExpandAllServiceNames()
j.InitAllServiceFields()
service1Name := j.TaskGroups[0].Tasks[0].Services[0].Name
if service1Name != "my-job-web-frontend-default" {