func (s *StateStore) upsertAllocsImpl(index uint64, allocs []*structs.Allocation, txn *memdb.Txn) error {
	// Handle the allocations
	jobs := make(map[string]string, 1)
	for _, alloc := range allocs {
		existing, err := txn.First("allocs", "id", alloc.ID)
		if err != nil {
			return fmt.Errorf("alloc lookup failed: %v", err)
		}
		exist, _ := existing.(*structs.Allocation)

		if exist == nil {
			alloc.CreateIndex = index
			alloc.ModifyIndex = index
			alloc.AllocModifyIndex = index

			// Issue https://github.com/hashicorp/nomad/issues/2583 uncovered
			// the a race between a forced garbage collection and the scheduler
			// marking an allocation as terminal. The issue is that the
			// allocation from the scheduler has its job normalized and the FSM
			// will only denormalize if the allocation is not terminal.  However
			// if the allocation is garbage collected, that will result in a
			// allocation being upserted for the first time without a job
			// attached. By returning an error here, it will cause the FSM to
			// error, causing the plan_apply to error and thus causing the
			// evaluation to be failed. This will force an index refresh that
			// should solve this issue.
			if alloc.Job == nil {
				return fmt.Errorf("attempting to upsert allocation %q without a job", alloc.ID)
			}
		} else {
			alloc.CreateIndex = exist.CreateIndex
			alloc.ModifyIndex = index
			alloc.AllocModifyIndex = index

			// Keep the clients task states
			alloc.TaskStates = exist.TaskStates

			// If the scheduler is marking this allocation as lost we do not
			// want to reuse the status of the existing allocation.
			if alloc.ClientStatus != structs.AllocClientStatusLost {
				alloc.ClientStatus = exist.ClientStatus
				alloc.ClientDescription = exist.ClientDescription
			}

			// The job has been denormalized so re-attach the original job
			if alloc.Job == nil {
				alloc.Job = exist.Job
			}
		}

		if err := s.updateDeploymentWithAlloc(index, alloc, exist, txn); err != nil {
			return fmt.Errorf("error updating deployment: %v", err)
		}

		if err := s.updateSummaryWithAlloc(index, alloc, exist, txn); err != nil {
			return fmt.Errorf("error updating job summary: %v", err)
		}

		// Create the EphemeralDisk if it's nil by adding up DiskMB from task resources.
		// COMPAT 0.4.1 -> 0.5
		if alloc.Job != nil {
			s.addEphemeralDiskToTaskGroups(alloc.Job)
		}

		if err := txn.Insert("allocs", alloc); err != nil {
			return fmt.Errorf("alloc insert failed: %v", err)
		}

		// If the allocation is running, force the job to running status.
		forceStatus := ""
		if !alloc.TerminalStatus() {
			forceStatus = structs.JobStatusRunning
		}
		jobs[alloc.JobID] = forceStatus
	}

	// Update the indexes
	if err := txn.Insert("index", &IndexEntry{"allocs", index}); err != nil {
		return fmt.Errorf("index update failed: %v", err)
	}

	// Set the job's status
	if err := s.setJobStatuses(index, txn, jobs, false); err != nil {
		return fmt.Errorf("setting job status failed: %v", err)
	}

	return nil
}
// upsertAllocs is the actual implementation of UpsertAllocs so that it may be
// used with an existing transaction.
upsertAllocs is the actual implementation of UpsertAllocs so that it may be
used with an existing transaction.
// upsertAllocs is the actual implementation of UpsertAllocs so that it may be // used with an existing transaction .
func (s *StateStore) upsertAllocsImpl(index uint64, allocs []*structs.Allocation, txn *memdb.Txn) error {
	jobs := make(map[string]string, 1)
	for _, alloc := range allocs {
		existing, err := txn.First("allocs", "id", alloc.ID)
		if err != nil {
			return fmt.Errorf("alloc lookup failed: %v", err)
		}
		exist, _ := existing.(*structs.Allocation)
		if exist == nil {
			alloc.CreateIndex = index
			alloc.ModifyIndex = index
			alloc.AllocModifyIndex = index
			if alloc.Job == nil {
				return fmt.Errorf("attempting to upsert allocation %q without a job", alloc.ID)
			}
		} else {
			alloc.CreateIndex = exist.CreateIndex
			alloc.ModifyIndex = index
			alloc.AllocModifyIndex = index
			alloc.TaskStates = exist.TaskStates
			if alloc.ClientStatus != structs.AllocClientStatusLost {
				alloc.ClientStatus = exist.ClientStatus
				alloc.ClientDescription = exist.ClientDescription
			}
			if alloc.Job == nil {
				alloc.Job = exist.Job
			}
		}
		if err := s.updateDeploymentWithAlloc(index, alloc, exist, txn); err != nil {
			return fmt.Errorf("error updating deployment: %v", err)
		}
		if err := s.updateSummaryWithAlloc(index, alloc, exist, txn); err != nil {
			return fmt.Errorf("error updating job summary: %v", err)
		}
		if alloc.Job != nil {
			s.addEphemeralDiskToTaskGroups(alloc.Job)
		}
		if err := txn.Insert("allocs", alloc); err != nil {
			return fmt.Errorf("alloc insert failed: %v", err)
		}
		forceStatus := ""
		if !alloc.TerminalStatus() {
			forceStatus = structs.JobStatusRunning
		}
		jobs[alloc.JobID] = forceStatus
	}
	if err := txn.Insert("index", &IndexEntry{"allocs", index}); err != nil {
		return fmt.Errorf("index update failed: %v", err)
	}
	if err := s.setJobStatuses(index, txn, jobs, false); err != nil {
		return fmt.Errorf("setting job status failed: %v", err)
	}
	return nil
}
func ( s * StateStore ) upsertAllocsImpl ( index uint64 , allocs [ ] * structs . Allocation , txn * memdb . Txn ) error { jobs := make ( map [ string ] string , 1 ) for _ , alloc := range allocs { existing , err := txn . First ( " " , " " , alloc . ID ) if err != nil { return fmt . Errorf ( " " , err ) } exist , _ := existing . ( * structs . Allocation ) if exist == nil { alloc . CreateIndex = index alloc . ModifyIndex = index alloc . AllocModifyIndex = index if alloc . Job == nil { return fmt . Errorf ( " " , alloc . ID ) } } else { alloc . CreateIndex = exist . CreateIndex alloc . ModifyIndex = index alloc . AllocModifyIndex = index alloc . TaskStates = exist . TaskStates if alloc . ClientStatus != structs . AllocClientStatusLost { alloc . ClientStatus = exist . ClientStatus alloc . ClientDescription = exist . ClientDescription } if alloc . Job == nil { alloc . Job = exist . Job } } if err := s . updateDeploymentWithAlloc ( index , alloc , exist , txn ) ; err != nil { return fmt . Errorf ( " " , err ) } if err := s . updateSummaryWithAlloc ( index , alloc , exist , txn ) ; err != nil { return fmt . Errorf ( " " , err ) } if alloc . Job != nil { s . addEphemeralDiskToTaskGroups ( alloc . Job ) } if err := txn . Insert ( " " , alloc ) ; err != nil { return fmt . Errorf ( " " , err ) } forceStatus := " " if ! alloc . TerminalStatus ( ) { forceStatus = structs . JobStatusRunning } jobs [ alloc . JobID ] = forceStatus } if err := txn . Insert ( " " , & IndexEntry { " " , index } ) ; err != nil { return fmt . Errorf ( " " , err ) } if err := s . setJobStatuses ( index , txn , jobs , false ) ; err != nil { return fmt . Errorf ( " " , err ) } return nil }
upsertAllocs is the actual implementation of UpsertAllocs so that it may be
used with an existing transaction.
upsertAllocs is the actual implementation of UpsertAllocs so that it may be used with an existing transaction .
// Handle the allocations // Issue https://github.com/hashicorp/nomad/issues/2583 uncovered // the a race between a forced garbage collection and the scheduler // marking an allocation as terminal. The issue is that the // allocation from the scheduler has its job normalized and the FSM // will only denormalize if the allocation is not terminal.  However // if the allocation is garbage collected, that will result in a // allocation being upserted for the first time without a job // attached. By returning an error here, it will cause the FSM to // error, causing the plan_apply to error and thus causing the // evaluation to be failed. This will force an index refresh that // should solve this issue. // Keep the clients task states // If the scheduler is marking this allocation as lost we do not // want to reuse the status of the existing allocation. // The job has been denormalized so re-attach the original job // Create the EphemeralDisk if it's nil by adding up DiskMB from task resources. // COMPAT 0.4.1 -> 0.5 // If the allocation is running, force the job to running status. // Update the indexes // Set the job's status