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

Code before:

	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 req.IdempotencyToken != "" {
		found, err := n.state.CheckIdempotencyToken(
			req.RequestNamespace(), req.Job.ParentID, req.IdempotencyToken)
		if err != nil {
			n.logger.Error("failed to check idempotency token", "error", err)
			return err
		}
		if found != nil {
			// found a job matching the idempotency token, so bail out early
			return nil
		}
	}

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


Review comment (suggestion):
Is it possible to add a test to cover this new workflow branch? It would be nice to ensure `applyUpsertJob` is correctly calling the state functions and protect it from any future regressions.

Code after:

	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 req.IdempotencyToken != "" {
		found, err := n.state.CheckIdempotencyToken(
			req.Job.Namespace, req.Job.ParentID, req.IdempotencyToken)
		if err != nil {
			n.logger.Error("failed to check idempotency token", "error", err)
			return err
		}
		if found != nil {
			// found a job matching the idempotency token, so bail out early
			return nil
		}
	}

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