From 94454cbb3b62679a137440c3e37ac754b8a31bf6 Mon Sep 17 00:00:00 2001 From: Armon Dadgar Date: Thu, 6 Aug 2015 17:46:14 -0700 Subject: [PATCH] scheduler: first pass at job deregister --- scheduler/scheduler.go | 3 ++ scheduler/service_sched.go | 71 +++++++++++++++++++++++++++++++++++++- 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go index 88f9316f4..165e6d79e 100644 --- a/scheduler/scheduler.go +++ b/scheduler/scheduler.go @@ -54,6 +54,9 @@ type State interface { // Nodes returns an iterator over all the nodes. // The type of each result is *structs.Node Nodes() (memdb.ResultIterator, error) + + // AllocsByJob returns the allocations by JobID + AllocsByJob(jobID string) ([]*structs.Allocation, error) } // Planner interface is used to submit a task allocation plan. diff --git a/scheduler/service_sched.go b/scheduler/service_sched.go index 2aba158d7..b64c14220 100644 --- a/scheduler/service_sched.go +++ b/scheduler/service_sched.go @@ -1,6 +1,7 @@ package scheduler import ( + "fmt" "log" "github.com/hashicorp/nomad/nomad/structs" @@ -27,6 +28,74 @@ func NewServiceScheduler(logger *log.Logger, state State, planner Planner) Sched } // Process is used to handle a single evaluation -func (s *ServiceScheduler) Process(*structs.Evaluation) error { +func (s *ServiceScheduler) Process(eval *structs.Evaluation) error { + // Use the evaluation trigger reason to determine what we need to do + switch eval.TriggeredBy { + case structs.EvalTriggerJobRegister: + return s.handleJobRegister(eval) + case structs.EvalTriggerJobDeregister: + return s.handleJobDeregister(eval) + case structs.EvalTriggerNodeUpdate: + return s.handleNodeUpdate(eval) + default: + return fmt.Errorf("service scheduler cannot handle '%s' evaluation reason", + eval.TriggeredBy) + } +} + +// handleJobRegister is used to handle a job being registered or updated +func (s *ServiceScheduler) handleJobRegister(eval *structs.Evaluation) error { + // TODO + return nil +} + +// handleJobDeregister is used to handle a job being deregistered +func (s *ServiceScheduler) handleJobDeregister(eval *structs.Evaluation) error { +START: + // Lookup the allocations by JobID + allocs, err := s.state.AllocsByJob(eval.JobID) + if err != nil { + return fmt.Errorf("failed to get allocs for job '%s': %v", + eval.JobID, err) + } + + // Nothing to do if there is no evictsion + if len(allocs) == 0 { + return nil + } + + // Create a plan to evict these + plan := &structs.Plan{ + EvalID: eval.ID, + Priority: eval.Priority, + NodeEvict: make(map[string][]string), + } + + // Add each alloc to be evicted + for _, alloc := range allocs { + nodeEvict := plan.NodeEvict[alloc.NodeID] + nodeEvict = append(nodeEvict, alloc.ID) + plan.NodeEvict[alloc.NodeID] = nodeEvict + } + + // Submit the plan + _, newState, err := s.planner.SubmitPlan(plan) + if err != nil { + return err + } + + // If we got a state refresh, try again to ensure we + // are not missing any allocations + if newState != nil { + s.state = newState + goto START + } + return nil +} + +// handleNodeUPdate is used to handle an update to a node status where +// there is an existing allocation for this job +func (s *ServiceScheduler) handleNodeUpdate(eval *structs.Evaluation) error { + // TODO return nil }