package scheduler import ( "fmt" "log" "github.com/hashicorp/nomad/nomad/structs" ) const ( // maxSystemScheduleAttempts is used to limit the number of times // we will attempt to schedule if we continue to hit conflicts for system // jobs. maxSystemScheduleAttempts = 5 // allocNodeTainted is the status used when stopping an alloc because it's // node is tainted. allocNodeTainted = "system alloc not needed as node is tainted" ) // SystemScheduler is used for 'system' jobs. This scheduler is // designed for services that should be run on every client. type SystemScheduler struct { logger *log.Logger state State planner Planner eval *structs.Evaluation job *structs.Job plan *structs.Plan planResult *structs.PlanResult ctx *EvalContext stack *SystemStack nodes []*structs.Node nodesByDC map[string]int limitReached bool nextEval *structs.Evaluation } // NewSystemScheduler is a factory function to instantiate a new system // scheduler. func NewSystemScheduler(logger *log.Logger, state State, planner Planner) Scheduler { return &SystemScheduler{ logger: logger, state: state, planner: planner, } } // Process is used to handle a single evaluation. func (s *SystemScheduler) Process(eval *structs.Evaluation) error { // Store the evaluation s.eval = eval // Verify the evaluation trigger reason is understood switch eval.TriggeredBy { case structs.EvalTriggerJobRegister, structs.EvalTriggerNodeUpdate, structs.EvalTriggerJobDeregister, structs.EvalTriggerRollingUpdate: default: desc := fmt.Sprintf("scheduler cannot handle '%s' evaluation reason", eval.TriggeredBy) return setStatus(s.logger, s.planner, s.eval, s.nextEval, structs.EvalStatusFailed, desc) } // Retry up to the maxSystemScheduleAttempts and reset if progress is made. progress := func() bool { return progressMade(s.planResult) } if err := retryMax(maxSystemScheduleAttempts, s.process, progress); err != nil { if statusErr, ok := err.(*SetStatusError); ok { return setStatus(s.logger, s.planner, s.eval, s.nextEval, statusErr.EvalStatus, err.Error()) } return err } // Update the status to complete return setStatus(s.logger, s.planner, s.eval, s.nextEval, structs.EvalStatusComplete, "") } // process is wrapped in retryMax to iteratively run the handler until we have no // further work or we've made the maximum number of attempts. func (s *SystemScheduler) process() (bool, error) { // Lookup the Job by ID var err error s.job, err = s.state.JobByID(s.eval.JobID) if err != nil { return false, fmt.Errorf("failed to get job '%s': %v", s.eval.JobID, err) } // Get the ready nodes in the required datacenters if s.job != nil { s.nodes, s.nodesByDC, err = readyNodesInDCs(s.state, s.job.Datacenters) if err != nil { return false, fmt.Errorf("failed to get ready nodes: %v", err) } } // Create a plan s.plan = s.eval.MakePlan(s.job) // Create an evaluation context s.ctx = NewEvalContext(s.state, s.plan, s.logger) // Construct the placement stack s.stack = NewSystemStack(s.ctx) if s.job != nil { s.stack.SetJob(s.job) } // Compute the target job allocations if err := s.computeJobAllocs(); err != nil { s.logger.Printf("[ERR] sched: %#v: %v", s.eval, err) return false, err } // If the plan is a no-op, we can bail if s.plan.IsNoOp() { return true, nil } // If the limit of placements was reached we need to create an evaluation // to pickup from here after the stagger period. if s.limitReached && s.nextEval == nil { s.nextEval = s.eval.NextRollingEval(s.job.Update.Stagger) if err := s.planner.CreateEval(s.nextEval); err != nil { s.logger.Printf("[ERR] sched: %#v failed to make next eval for rolling update: %v", s.eval, err) return false, err } s.logger.Printf("[DEBUG] sched: %#v: rolling update limit reached, next eval '%s' created", s.eval, s.nextEval.ID) } // Submit the plan result, newState, err := s.planner.SubmitPlan(s.plan) s.planResult = result if err != nil { return false, err } // If we got a state refresh, try again since we have stale data if newState != nil { s.logger.Printf("[DEBUG] sched: %#v: refresh forced", s.eval) s.state = newState return false, nil } // Try again if the plan was not fully committed, potential conflict fullCommit, expected, actual := result.FullCommit(s.plan) if !fullCommit { s.logger.Printf("[DEBUG] sched: %#v: attempted %d placements, %d placed", s.eval, expected, actual) return false, nil } // Success! return true, nil } // computeJobAllocs is used to reconcile differences between the job, // existing allocations and node status to update the allocations. func (s *SystemScheduler) computeJobAllocs() error { // Lookup the allocations by JobID allocs, err := s.state.AllocsByJob(s.eval.JobID) if err != nil { return fmt.Errorf("failed to get allocs for job '%s': %v", s.eval.JobID, err) } // Filter out the allocations in a terminal state allocs = structs.FilterTerminalAllocs(allocs) // Determine the tainted nodes containing job allocs tainted, err := taintedNodes(s.state, allocs) if err != nil { return fmt.Errorf("failed to get tainted nodes for job '%s': %v", s.eval.JobID, err) } // Diff the required and existing allocations diff := diffSystemAllocs(s.job, s.nodes, tainted, allocs) s.logger.Printf("[DEBUG] sched: %#v: %#v", s.eval, diff) // Add all the allocs to stop for _, e := range diff.stop { s.plan.AppendUpdate(e.Alloc, structs.AllocDesiredStatusStop, allocNotNeeded) } // Attempt to do the upgrades in place diff.update = inplaceUpdate(s.ctx, s.eval, s.job, s.stack, diff.update) // Check if a rolling upgrade strategy is being used limit := len(diff.update) if s.job != nil && s.job.Update.Rolling() { limit = s.job.Update.MaxParallel } // Treat non in-place updates as an eviction and new placement. s.limitReached = evictAndPlace(s.ctx, diff, diff.update, allocUpdating, &limit) // Nothing remaining to do if placement is not required if len(diff.place) == 0 { return nil } // Compute the placements return s.computePlacements(diff.place) } // computePlacements computes placements for allocations func (s *SystemScheduler) computePlacements(place []allocTuple) error { nodeByID := make(map[string]*structs.Node, len(s.nodes)) for _, node := range s.nodes { nodeByID[node.ID] = node } // Track the failed task groups so that we can coalesce // the failures together to avoid creating many failed allocs. failedTG := make(map[*structs.TaskGroup]*structs.Allocation) nodes := make([]*structs.Node, 1) for _, missing := range place { node, ok := nodeByID[missing.Alloc.NodeID] if !ok { return fmt.Errorf("could not find node %q", missing.Alloc.NodeID) } // Update the set of placement nodes nodes[0] = node s.stack.SetNodes(nodes) // Attempt to match the task group option, _ := s.stack.Select(missing.TaskGroup) if option == nil { // Check if this task group has already failed if alloc, ok := failedTG[missing.TaskGroup]; ok { alloc.Metrics.CoalescedFailures += 1 continue } } // Create an allocation for this alloc := &structs.Allocation{ ID: structs.GenerateUUID(), EvalID: s.eval.ID, Name: missing.Name, JobID: s.job.ID, TaskGroup: missing.TaskGroup.Name, Metrics: s.ctx.Metrics(), } // Store the available nodes by datacenter s.ctx.Metrics().NodesAvailable = s.nodesByDC // Set fields based on if we found an allocation option if option != nil { // Generate service IDs tasks in this allocation alloc.PopulateServiceIDs(missing.TaskGroup) alloc.NodeID = option.Node.ID alloc.TaskResources = option.TaskResources alloc.DesiredStatus = structs.AllocDesiredStatusRun alloc.ClientStatus = structs.AllocClientStatusPending s.plan.AppendAlloc(alloc) } else { alloc.DesiredStatus = structs.AllocDesiredStatusFailed alloc.DesiredDescription = "failed to find a node for placement" alloc.ClientStatus = structs.AllocClientStatusFailed s.plan.AppendFailed(alloc) failedTG[missing.TaskGroup] = alloc } } return nil }