// computeJobAllocs is used to reconcile differences between the job,
// existing allocations and node status to update the allocations.
328014
func (s *SystemScheduler) computeJobAllocs() error {
	// Lookup the allocations by JobID
	ws := memdb.NewWatchSet()
	allocs, err := s.state.AllocsByJob(ws, s.eval.Namespace, s.eval.JobID, true)
	if err != nil {
		return fmt.Errorf("failed to get allocs for job '%s': %v",
			s.eval.JobID, err)
	}

	// 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)
	}

	// Update the allocations which are in pending/running state on tainted
	// nodes to lost
	updateNonTerminalAllocsToLost(s.plan, tainted, allocs)

	// Filter out the allocations in a terminal state
	allocs, terminalAllocs := structs.FilterTerminalAllocs(allocs)

	// Diff the required and existing allocations
	diff := diffSystemAllocs(s.job, s.nodes, tainted, allocs, terminalAllocs)
	s.logger.Debug("reconciled current state with desired state",
		"place", len(diff.place), "update", len(diff.update),
		"migrate", len(diff.migrate), "stop", len(diff.stop),
		"ignore", len(diff.ignore), "lost", len(diff.lost))

	// Add all the allocs to stop
	for _, e := range diff.stop {
		s.plan.AppendStoppedAlloc(e.Alloc, allocNotNeeded, "")
	}

	// Add all the allocs to migrate
	for _, e := range diff.migrate {
		s.plan.AppendStoppedAlloc(e.Alloc, allocNodeTainted, "")
	}

	// Lost allocations should be transitioned to desired status stop and client
	// status lost.
	for _, e := range diff.lost {
		s.plan.AppendStoppedAlloc(e.Alloc, allocLost, structs.AllocClientStatusLost)
	}

	// Attempt to do the upgrades in place
	destructiveUpdates, inplaceUpdates := inplaceUpdate(s.ctx, s.eval, s.job, s.stack, diff.update)
	diff.update = destructiveUpdates

	if s.eval.AnnotatePlan {
		s.plan.Annotations = &structs.PlanAnnotations{
			DesiredTGUpdates: desiredUpdates(diff, inplaceUpdates, destructiveUpdates),
		}
	}

	// Check if a rolling upgrade strategy is being used
	limit := len(diff.update)
	if !s.job.Stopped() && 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 {
		if !s.job.Stopped() {
			for _, tg := range s.job.TaskGroups {
				s.queuedAllocs[tg.Name] = 0
			}
		}
		return nil
	}

	// Record the number of allocations that needs to be placed per Task Group
	for _, allocTuple := range diff.place {
		s.queuedAllocs[allocTuple.TaskGroup.Name] += 1
	}

	// Compute the placements
	return s.computePlacements(diff.place)
}
1052956
func buildJobStatusHandler(ctx context.Context) (JobStatusHandler, error) {

	jobStatus := JobStatusHandler{}

	pfconfigdriver.PfconfigPool.AddStruct(ctx, &redisclient.Config)
	var network string
	if redisclient.Config.RedisArgs.Server[0] == '/' {
		network = "unix"
	} else {
		network = "tcp"
	}

	jobStatus.redis = redis.NewClient(&redis.Options{
		Addr:    redisclient.Config.RedisArgs.Server,
		Network: network,
	})

	router := httprouter.New()
	router.GET("/api/v1/pfqueue/task/:job_id/status", jobStatus.handleStatus)
	router.GET("/api/v1/pfqueue/task/:job_id/status/poll", jobStatus.handleStatusPoll)

	jobStatus.router = router

	return jobStatus, nil
}
524378
func (d *Daemon) makeLoggingJobFunc(f jobFunc) jobFunc {
	return func(ctx context.Context, id job.ID, logger log.Logger) (job.Result, error) {
		started := time.Now().UTC()
		result, err := f(ctx, id, logger)
		if err != nil {
			return result, err
		}
		logger.Log("revision", result.Revision)
		if result.Revision != "" {
			var workloadIDs []flux.ResourceID
			for id, result := range result.Result {
				if result.Status == update.ReleaseStatusSuccess {
					workloadIDs = append(workloadIDs, id)
				}
			}

			metadata := &event.CommitEventMetadata{
				Revision: result.Revision,
				Spec:     result.Spec,
				Result:   result.Result,
			}

			return result, d.LogEvent(event.Event{
				ServiceIDs: workloadIDs,
				Type:       event.EventCommit,
				StartedAt:  started,
				EndedAt:    started,
				LogLevel:   event.LogLevelInfo,
				Metadata:   metadata,
			})
		}
		return result, nil
	}
}
176767
func Convert_v1_Job_To_batch_Job(in *v1.Job, out *batch.Job, s conversion.Scope) error {
	return autoConvert_v1_Job_To_batch_Job(in, out, s)
}
900976
func (n *Node) batchUpdate(future *structs.BatchFuture, updates []*structs.Allocation, evals []*structs.Evaluation) {
	// Group pending evals by jobID to prevent creating unnecessary evals
	evalsByJobId := make(map[structs.NamespacedID]struct{})
	var trimmedEvals []*structs.Evaluation
	for _, eval := range evals {
		namespacedID := structs.NamespacedID{
			ID:        eval.JobID,
			Namespace: eval.Namespace,
		}
		_, exists := evalsByJobId[namespacedID]
		if !exists {
			trimmedEvals = append(trimmedEvals, eval)
			evalsByJobId[namespacedID] = struct{}{}
		}
	}

	if len(trimmedEvals) > 0 {
		n.logger.Debug("adding evaluations for rescheduling failed allocations", "num_evals", len(trimmedEvals))
	}
	// Prepare the batch update
	batch := &structs.AllocUpdateRequest{
		Alloc:        updates,
		Evals:        trimmedEvals,
		WriteRequest: structs.WriteRequest{Region: n.srv.config.Region},
	}

	// Commit this update via Raft
	var mErr multierror.Error
	_, index, err := n.srv.raftApply(structs.AllocClientUpdateRequestType, batch)
	if err != nil {
		n.logger.Error("alloc update failed", "error", err)
		mErr.Errors = append(mErr.Errors, err)
	}

	// For each allocation we are updating check if we should revoke any
	// Vault Accessors
	var revoke []*structs.VaultAccessor
	for _, alloc := range updates {
		// Skip any allocation that isn't dead on the client
		if !alloc.Terminated() {
			continue
		}

		// Determine if there are any Vault accessors for the allocation
		ws := memdb.NewWatchSet()
		accessors, err := n.srv.State().VaultAccessorsByAlloc(ws, alloc.ID)
		if err != nil {
			n.logger.Error("looking up Vault accessors for alloc failed", "alloc_id", alloc.ID, "error", err)
			mErr.Errors = append(mErr.Errors, err)
		}

		revoke = append(revoke, accessors...)
	}

	if l := len(revoke); l != 0 {
		n.logger.Debug("revoking accessors due to terminal allocations", "num_accessors", l)
		if err := n.srv.vault.RevokeTokens(context.Background(), revoke, true); err != nil {
			n.logger.Error("batched Vault accessor revocation failed", "error", err)
			mErr.Errors = append(mErr.Errors, err)
		}
	}

	// Respond to the future
	future.Respond(index, mErr.ErrorOrNil())
}
1088507
func (s *Service) SheduleJob(handlerFunc interface{}, priority int, time time.Time, data ...interface{}) (*jobs.Job, error) {
	if jobsPerName, ok := jobsPerService[s.Name]; ok {
		if jobType, ok := jobsPerName[s.getShortFuncPath(handlerFunc)]; ok {
			return jobType.Schedule(priority, time, data...)
		}
		panic("SheduleJob: Job type not found")
	}
	panic("SheduleJob: Service pool not found")
}
215938
func (c *Client) updateAlloc(update *structs.Allocation) {
	ar, err := c.getAllocRunner(update.ID)
	if err != nil {
		c.logger.Warn("cannot update nonexistent alloc", "alloc_id", update.ID)
		return
	}

	// Update local copy of alloc
	if err := c.stateDB.PutAllocation(update); err != nil {
		c.logger.Error("error persisting updated alloc locally", "error", err, "alloc_id", update.ID)
	}

	// Update alloc runner
	ar.Update(update)
}
1024913
func (j *Job) Canonicalize() (warnings error) {
	if j == nil {
		return nil
	}

	var mErr multierror.Error
	// Ensure that an empty and nil map are treated the same to avoid scheduling
	// problems since we use reflect DeepEquals.
	if len(j.Meta) == 0 {
		j.Meta = nil
	}

	// Ensure the job is in a namespace.
	if j.Namespace == "" {
		j.Namespace = DefaultNamespace
	}

	for _, tg := range j.TaskGroups {
		tg.Canonicalize(j)
	}

	if j.ParameterizedJob != nil {
		j.ParameterizedJob.Canonicalize()
	}

	if j.Periodic != nil {
		j.Periodic.Canonicalize()
	}

	return mErr.ErrorOrNil()
}
1121066
func appendNamespacedJobID(jobIDs map[structs.NamespacedID]struct{}, alloc *structs.Allocation) {
	id := structs.NamespacedID{Namespace: alloc.Namespace, ID: alloc.JobID}
	if _, ok := jobIDs[id]; !ok {
		jobIDs[id] = struct{}{}
	}
}
126215
func checkJobReadiness(obj runtime.Object) (bool, bool, error) {
	var (
		hasCompletionTime bool
		isJobFailed       bool
	)
	switch j := obj.(type) {
	case *batchv1.Job:
		hasCompletionTime = j.Status.CompletionTime != nil
		isJobFailed = j.Status.Failed > 0
	default:
		return false, false, fmt.Errorf("unsupported job version: %T", j)
	}
	return hasCompletionTime, isJobFailed, nil
}
71452
func (s *Scheduler) getRunnableJobs() (running_jobs [MAXJOBNUM]*Job, n int) {
	runnableJobs := [MAXJOBNUM]*Job{}
	n = 0
	sort.Sort(s)
	for i := 0; i < s.size; i++ {
		if s.jobs[i].shouldRun() {

			runnableJobs[n] = s.jobs[i]
			//fmt.Println(runnableJobs)
			n++
		} else {
			break
		}
	}
	return runnableJobs, n
}
727313
func (a *allocReconciler) computePlacements(group *structs.TaskGroup,
	nameIndex *allocNameIndex, untainted, migrate allocSet, reschedule allocSet) []allocPlaceResult {

	// Add rescheduled placement results
	var place []allocPlaceResult
	for _, alloc := range reschedule {
		place = append(place, allocPlaceResult{
			name:          alloc.Name,
			taskGroup:     group,
			previousAlloc: alloc,
			reschedule:    true,
			canary:        alloc.DeploymentStatus.IsCanary(),
		})
	}

	// Hot path the nothing to do case
	existing := len(untainted) + len(migrate) + len(reschedule)
	if existing >= group.Count {
		return place
	}

	// Add remaining placement results
	if existing < group.Count {
		for _, name := range nameIndex.Next(uint(group.Count - existing)) {
			place = append(place, allocPlaceResult{
				name:      name,
				taskGroup: group,
			})
		}
	}

	return place
}
553644
func (p Pool) ParallelDo(tasks ...func() error) error {
	jobs := make(chan func() error, len(tasks))
	errs := make(chan error, len(tasks))
	wg := &sync.WaitGroup{}

	wg.Add(p.Count)
	for i := 0; i < p.Count; i++ {
		p.spawnWorker(jobs, errs, wg)
	}

	for _, task := range tasks {
		jobs <- task
	}

	close(jobs)

	wg.Wait()

	close(errs)

	var combinedErrors []error
	for e := range errs {
		combinedErrors = append(combinedErrors, e)
	}

	if len(combinedErrors) > 0 {
		return bosherr.NewMultiError(combinedErrors...)
	}

	return nil
}
504160
func olderVersionTerminalAllocs(allocs []*structs.Allocation, job *structs.Job) []string {
	var ret []string
	for _, alloc := range allocs {
		if alloc.Job != nil && alloc.Job.CreateIndex < job.CreateIndex && alloc.TerminalStatus() {
			ret = append(ret, alloc.ID)
		}
	}
	return ret
}
736545
func getNextJobs(n int, poolId string) ([]*Job, error) {
	// Start a new transaction
	t := newTransaction()
	// Invoke a script to get all the jobs which are ready to execute based on their
	// time parameter and whether or not they are in the queued set.
	jobs := []*Job{}
	t.popNextJobs(n, poolId, newScanJobsHandler(&jobs))

	// Execute the transaction
	if err := t.exec(); err != nil {
		return nil, err
	}
	return jobs, nil
}
31617
func (s *Scheduler) findJobToStop(f *Formation, typ string) (*Job, error) {
	var found *Job
	for _, job := range s.jobs.WithFormationAndType(f, typ) {
		switch job.State {
		case JobStatePending:
			return job, nil
		case JobStateStarting, JobStateRunning:
			// if the job is on a host which is shutting down,
			// return it (it is about to stop anyway, and this
			// avoids a race where modifying the omni counts to
			// remove a shut down host could cause a subsequent
			// rectify to stop a job on an active host before the
			// shutting down host's job has stopped)
			if host, ok := s.hosts[job.HostID]; ok && host.Shutdown {
				return job, nil
			}

			// return the most recent job (which is the first in
			// the slice we are iterating over) if none of the
			// above cases match, preferring starting jobs to
			// running ones
			if found == nil || found.State == JobStateRunning && job.State == JobStateStarting {
				found = job
			}
		}
	}
	if found == nil {
		return nil, fmt.Errorf("no %s jobs running", typ)
	}
	return found, nil
}
259723
func awaitJob(ctx context.Context, client api.Server, jobID job.ID) (job.Result, error) {
	var result job.Result
	err := backoff(100*time.Millisecond, 2, 50, 1*time.Minute, func() (bool, error) {
		j, err := client.JobStatus(ctx, jobID)
		if err != nil {
			return false, err
		}
		switch j.StatusString {
		case job.StatusFailed:
			return false, j
		case job.StatusSucceeded:
			if j.Err != "" {
				// How did we succeed but still get an error!?
				return false, j
			}
			result = j.Result
			return true, nil
		}
		return false, nil
	})
	return result, err
}