Repository: hashicorp/nomad
File: nomad/fsm.go (Go)

Code before:

func (n *nomadFSM) applyNodePoolDelete(msgType structs.MessageType, buf []byte, index uint64) interface{} {
	defer metrics.MeasureSince([]string{"nomad", "fsm", "apply_node_pool_delete"}, time.Now())
	var req structs.NodePoolDeleteRequest
	if err := structs.Decode(buf, &req); err != nil {
		panic(fmt.Errorf("failed to decode request: %v", err))
	}

	if err := n.state.DeleteNodePools(msgType, index, req.Names); err != nil {
		n.logger.Error("DeleteNodePools failed", "error", err)
		return err
	}

	return nil
}

func (n *nomadFSM) applyUpsertJob(msgType structs.MessageType, buf []byte, index uint64) interface{} {
	defer metrics.MeasureSince([]string{"nomad", "fsm", "register_job"}, time.Now())
	var req structs.JobRegisterRequest
	if err := structs.Decode(buf, &req); err != nil {
		panic(fmt.Errorf("failed to decode request: %v", err))
	}

	// Preserve the existing task group counts and resources, if requested. Do
	// this here to avoid race conditions when registering and scaling jobs.
	if req.PreserveCounts || req.PreserveResources {
		existingJobID := req.Job.ID
		if existingJobID != "" {
			existingJob, err := n.state.JobByID(nil, req.Namespace, existingJobID)
			if err != nil {
				n.logger.Error("JobByID lookup for existing job failed", "existing_job_id", existingJobID, "namespace", req.Namespace, "error", err)
				return err
			}
			if existingJob != nil {
				if req.PreserveCounts {
					prevCounts := make(map[string]int)
					for _, tg := range existingJob.TaskGroups {
						prevCounts[tg.Name] = tg.Count
					}
					for _, tg := range req.Job.TaskGroups {
						if count, ok := prevCounts[tg.Name]; ok {
							tg.Count = count
						}
					}
				}

				if req.PreserveResources {
					prevResources := make(map[string]map[string]*structs.Resources)
					for _, tg := range existingJob.TaskGroups {
						prevResources[tg.Name] = make(map[string]*structs.Resources)
						for _, task := range tg.Tasks {

Review comment (suggestion):
@tgross I moved the logic, this seemed like the place it made the most sense but let me know if you had something else in mind!

Code after:

func (n *nomadFSM) applyNodePoolDelete(msgType structs.MessageType, buf []byte, index uint64) interface{} {
	defer metrics.MeasureSince([]string{"nomad", "fsm", "apply_node_pool_delete"}, time.Now())
	var req structs.NodePoolDeleteRequest
	if err := structs.Decode(buf, &req); err != nil {
		panic(fmt.Errorf("failed to decode request: %v", err))
	}

	if err := n.state.DeleteNodePools(msgType, index, req.Names); err != nil {
		n.logger.Error("DeleteNodePools failed", "error", err)
		return err
	}

	return nil
}

func (n *nomadFSM) applyUpsertJob(msgType structs.MessageType, buf []byte, index uint64) interface{} {
	defer metrics.MeasureSince([]string{"nomad", "fsm", "register_job"}, time.Now())
	var req structs.JobRegisterRequest
	if err := structs.Decode(buf, &req); err != nil {
		panic(fmt.Errorf("failed to decode request: %v", err))
	}

	/* Handle upgrade paths:
	 * - Empty maps and slices should be treated as nil to avoid
	 *   un-intended destructive updates in scheduler since we use
	 *   reflect.DeepEqual. Starting Nomad 0.4.1, job submission sanitizes
	 *   the incoming job.
	 * - Migrate from old style upgrade block that used only a stagger.
	 */
	req.Job.Canonicalize()

	if err := n.state.UpsertJobWithRequest(msgType, index, &req); err != nil {
		n.logger.Error("UpsertJob failed", "error", err)
		return err
	}

	// We always add the job to the periodic dispatcher because there is the
	// possibility that the periodic spec was removed and then we should stop
	// tracking it.
	if err := n.periodicDispatcher.Add(req.Job); err != nil {
		n.logger.Error("periodicDispatcher.Add failed", "error", err)
		return fmt.Errorf("failed adding job to periodic dispatcher: %v", err)
	}

	// Create a watch set
	ws := memdb.NewWatchSet()

	// If it is an active periodic job, record the time it was inserted. This is
	// necessary for recovering during leader election. It is possible that from
	// the time it is added to when it was suppose to launch, leader election