func (s *StateStore) updateSummaryWithAlloc(index uint64, alloc *structs.Allocation,
	existingAlloc *structs.Allocation, txn *memdb.Txn) error {

	// We don't have to update the summary if the job is missing
	if alloc.Job == nil {
		return nil
	}

	summaryRaw, err := txn.First("job_summary", "id", alloc.JobID)
	if err != nil {
		return fmt.Errorf("unable to lookup job summary for job id %q: %v", alloc.JobID, err)
	}

	if summaryRaw == nil {
		// Check if the job is de-registered
		rawJob, err := txn.First("jobs", "id", alloc.JobID)
		if err != nil {
			return fmt.Errorf("unable to query job: %v", err)
		}

		// If the job is de-registered then we skip updating it's summary
		if rawJob == nil {
			return nil
		}

		return fmt.Errorf("job summary for job %q is not present", alloc.JobID)
	}

	// Get a copy of the existing summary
	jobSummary := summaryRaw.(*structs.JobSummary).Copy()

	// Not updating the job summary because the allocation doesn't belong to the
	// currently registered job
	if jobSummary.CreateIndex != alloc.Job.CreateIndex {
		return nil
	}

	tgSummary, ok := jobSummary.Summary[alloc.TaskGroup]
	if !ok {
		return fmt.Errorf("unable to find task group in the job summary: %v", alloc.TaskGroup)
	}

	summaryChanged := false
	if existingAlloc == nil {
		switch alloc.DesiredStatus {
		case structs.AllocDesiredStatusStop, structs.AllocDesiredStatusEvict:
			s.logger.Printf("[ERR] state_store: new allocation inserted into state store with id: %v and state: %v",
				alloc.ID, alloc.DesiredStatus)
		}
		switch alloc.ClientStatus {
		case structs.AllocClientStatusPending:
			tgSummary.Starting += 1
			if tgSummary.Queued > 0 {
				tgSummary.Queued -= 1
			}
			summaryChanged = true
		case structs.AllocClientStatusRunning, structs.AllocClientStatusFailed,
			structs.AllocClientStatusComplete:
			s.logger.Printf("[ERR] state_store: new allocation inserted into state store with id: %v and state: %v",
				alloc.ID, alloc.ClientStatus)
		}
	} else if existingAlloc.ClientStatus != alloc.ClientStatus {
		// Incrementing the client of the bin of the current state
		switch alloc.ClientStatus {
		case structs.AllocClientStatusRunning:
			tgSummary.Running += 1
		case structs.AllocClientStatusFailed:
			tgSummary.Failed += 1
		case structs.AllocClientStatusPending:
			tgSummary.Starting += 1
		case structs.AllocClientStatusComplete:
			tgSummary.Complete += 1
		case structs.AllocClientStatusLost:
			tgSummary.Lost += 1
		}

		// Decrementing the count of the bin of the last state
		switch existingAlloc.ClientStatus {
		case structs.AllocClientStatusRunning:
			tgSummary.Running -= 1
		case structs.AllocClientStatusPending:
			tgSummary.Starting -= 1
		case structs.AllocClientStatusLost:
			tgSummary.Lost -= 1
		case structs.AllocClientStatusFailed, structs.AllocClientStatusComplete:
		default:
			s.logger.Printf("[ERR] state_store: invalid old state of allocation with id: %v, and state: %v",
				existingAlloc.ID, existingAlloc.ClientStatus)
		}
		summaryChanged = true
	}
	jobSummary.Summary[alloc.TaskGroup] = tgSummary

	if summaryChanged {
		jobSummary.ModifyIndex = index

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

		if err := txn.Insert("job_summary", jobSummary); err != nil {
			return fmt.Errorf("updating job summary failed: %v", err)
		}
	}

	return nil
}
// updateSummaryWithAlloc updates the job summary when allocations are updated
// or inserted
updateSummaryWithAlloc updates the job summary when allocations are updated
or inserted
// updateSummaryWithAlloc updates the job summary when allocations are updated // or inserted
func (s *StateStore) updateSummaryWithAlloc(index uint64, alloc *structs.Allocation,
	existingAlloc *structs.Allocation, txn *memdb.Txn) error {
	if alloc.Job == nil {
		return nil
	}
	summaryRaw, err := txn.First("job_summary", "id", alloc.JobID)
	if err != nil {
		return fmt.Errorf("unable to lookup job summary for job id %q: %v", alloc.JobID, err)
	}
	if summaryRaw == nil {
		rawJob, err := txn.First("jobs", "id", alloc.JobID)
		if err != nil {
			return fmt.Errorf("unable to query job: %v", err)
		}
		if rawJob == nil {
			return nil
		}
		return fmt.Errorf("job summary for job %q is not present", alloc.JobID)
	}
	jobSummary := summaryRaw.(*structs.JobSummary).Copy()
	if jobSummary.CreateIndex != alloc.Job.CreateIndex {
		return nil
	}
	tgSummary, ok := jobSummary.Summary[alloc.TaskGroup]
	if !ok {
		return fmt.Errorf("unable to find task group in the job summary: %v", alloc.TaskGroup)
	}
	summaryChanged := false
	if existingAlloc == nil {
		switch alloc.DesiredStatus {
		case structs.AllocDesiredStatusStop, structs.AllocDesiredStatusEvict:
			s.logger.Printf("[ERR] state_store: new allocation inserted into state store with id: %v and state: %v",
				alloc.ID, alloc.DesiredStatus)
		}
		switch alloc.ClientStatus {
		case structs.AllocClientStatusPending:
			tgSummary.Starting += 1
			if tgSummary.Queued > 0 {
				tgSummary.Queued -= 1
			}
			summaryChanged = true
		case structs.AllocClientStatusRunning, structs.AllocClientStatusFailed,
			structs.AllocClientStatusComplete:
			s.logger.Printf("[ERR] state_store: new allocation inserted into state store with id: %v and state: %v",
				alloc.ID, alloc.ClientStatus)
		}
	} else if existingAlloc.ClientStatus != alloc.ClientStatus {
		switch alloc.ClientStatus {
		case structs.AllocClientStatusRunning:
			tgSummary.Running += 1
		case structs.AllocClientStatusFailed:
			tgSummary.Failed += 1
		case structs.AllocClientStatusPending:
			tgSummary.Starting += 1
		case structs.AllocClientStatusComplete:
			tgSummary.Complete += 1
		case structs.AllocClientStatusLost:
			tgSummary.Lost += 1
		}
		switch existingAlloc.ClientStatus {
		case structs.AllocClientStatusRunning:
			tgSummary.Running -= 1
		case structs.AllocClientStatusPending:
			tgSummary.Starting -= 1
		case structs.AllocClientStatusLost:
			tgSummary.Lost -= 1
		case structs.AllocClientStatusFailed, structs.AllocClientStatusComplete:
		default:
			s.logger.Printf("[ERR] state_store: invalid old state of allocation with id: %v, and state: %v",
				existingAlloc.ID, existingAlloc.ClientStatus)
		}
		summaryChanged = true
	}
	jobSummary.Summary[alloc.TaskGroup] = tgSummary
	if summaryChanged {
		jobSummary.ModifyIndex = index
		if err := txn.Insert("index", &IndexEntry{"job_summary", index}); err != nil {
			return fmt.Errorf("index update failed: %v", err)
		}
		if err := txn.Insert("job_summary", jobSummary); err != nil {
			return fmt.Errorf("updating job summary failed: %v", err)
		}
	}
	return nil
}
func ( s * StateStore ) updateSummaryWithAlloc ( index uint64 , alloc * structs . Allocation , existingAlloc * structs . Allocation , txn * memdb . Txn ) error { if alloc . Job == nil { return nil } summaryRaw , err := txn . First ( " " , " " , alloc . JobID ) if err != nil { return fmt . Errorf ( " " , alloc . JobID , err ) } if summaryRaw == nil { rawJob , err := txn . First ( " " , " " , alloc . JobID ) if err != nil { return fmt . Errorf ( " " , err ) } if rawJob == nil { return nil } return fmt . Errorf ( " " , alloc . JobID ) } jobSummary := summaryRaw . ( * structs . JobSummary ) . Copy ( ) if jobSummary . CreateIndex != alloc . Job . CreateIndex { return nil } tgSummary , ok := jobSummary . Summary [ alloc . TaskGroup ] if ! ok { return fmt . Errorf ( " " , alloc . TaskGroup ) } summaryChanged := false if existingAlloc == nil { switch alloc . DesiredStatus { case structs . AllocDesiredStatusStop , structs . AllocDesiredStatusEvict : s . logger . Printf ( " " , alloc . ID , alloc . DesiredStatus ) } switch alloc . ClientStatus { case structs . AllocClientStatusPending : tgSummary . Starting += 1 if tgSummary . Queued > 0 { tgSummary . Queued -= 1 } summaryChanged = true case structs . AllocClientStatusRunning , structs . AllocClientStatusFailed , structs . AllocClientStatusComplete : s . logger . Printf ( " " , alloc . ID , alloc . ClientStatus ) } } else if existingAlloc . ClientStatus != alloc . ClientStatus { switch alloc . ClientStatus { case structs . AllocClientStatusRunning : tgSummary . Running += 1 case structs . AllocClientStatusFailed : tgSummary . Failed += 1 case structs . AllocClientStatusPending : tgSummary . Starting += 1 case structs . AllocClientStatusComplete : tgSummary . Complete += 1 case structs . AllocClientStatusLost : tgSummary . Lost += 1 } switch existingAlloc . ClientStatus { case structs . AllocClientStatusRunning : tgSummary . Running -= 1 case structs . AllocClientStatusPending : tgSummary . Starting -= 1 case structs . AllocClientStatusLost : tgSummary . Lost -= 1 case structs . AllocClientStatusFailed , structs . AllocClientStatusComplete : default : s . logger . Printf ( " " , existingAlloc . ID , existingAlloc . ClientStatus ) } summaryChanged = true } jobSummary . Summary [ alloc . TaskGroup ] = tgSummary if summaryChanged { jobSummary . ModifyIndex = index if err := txn . Insert ( " " , & IndexEntry { " " , index } ) ; err != nil { return fmt . Errorf ( " " , err ) } if err := txn . Insert ( " " , jobSummary ) ; err != nil { return fmt . Errorf ( " " , err ) } } return nil }
updateSummaryWithAlloc updates the job summary when allocations are updated
or inserted
updateSummaryWithAlloc updates the job summary when allocations are updated or inserted
// We don't have to update the summary if the job is missing // Check if the job is de-registered // If the job is de-registered then we skip updating it's summary // Get a copy of the existing summary // Not updating the job summary because the allocation doesn't belong to the // currently registered job // Incrementing the client of the bin of the current state // Decrementing the count of the bin of the last state // Update the indexes table for job summary