scheduling

Documentation

Index

• func AggregateNodeTypeAllocations(nodes []api.NodeInfo) []*nodeTypeAllocation
• func CalculatePriorityUpdate(resourceScarcity map[string]float64, previousReport *api.ClusterUsageReport, ...) map[string]float64
• func CalculateQueuesPriorityInfo(clusterPriorities map[string]map[string]float64, ...) map[*api.Queue]QueuePriorityInfo
• func CombineLeasedReportResourceByQueue(reports map[string]*api.ClusterLeasedReport) map[string]common.ComputeResources
• func CreateClusterLeasedReport(clusterId string, currentReport *api.ClusterLeasedReport, ...) *api.ClusterLeasedReport
• func CreateClusterSchedulingInfoReport(leaseRequest *api.LeaseRequest, nodeAllocations []*nodeTypeAllocation) *api.ClusterSchedulingInfoReport
• func FilterActiveClusterSchedulingInfoReports(reports map[string]*api.ClusterSchedulingInfoReport) map[string]*api.ClusterSchedulingInfoReport
• func FilterActiveClusters(reports map[string]*api.ClusterUsageReport) map[string]*api.ClusterUsageReport
• func FilterClusterLeasedReports(ids []string, reports map[string]*api.ClusterLeasedReport) map[string]*api.ClusterLeasedReport
• func FilterPoolClusters(pool string, reports map[string]*api.ClusterUsageReport) map[string]*api.ClusterUsageReport
• func GetClusterReportIds(reports map[string]*api.ClusterUsageReport) []string
• func GroupByPool(reports map[string]*api.ClusterUsageReport) map[string]map[string]*api.ClusterUsageReport
• func GroupSchedulingInfoByPool(reports map[string]*api.ClusterSchedulingInfoReport) map[string]map[string]*api.ClusterSchedulingInfoReport
• func LeaseJobs(ctx context.Context, config *configuration.SchedulingConfig, jobQueue JobQueue, ...) ([]*api.Job, error)
• func MatchSchedulingRequirements(job *api.Job, schedulingInfo *api.ClusterSchedulingInfoReport) bool
• func MatchSchedulingRequirementsOnAnyCluster(job *api.Job, ...) bool
• func QueueSlicesToShares(resourceScarcity map[string]float64, ...) map[*api.Queue]float64
• func ResourceScarcityFromReports(reports map[string]*api.ClusterUsageReport) map[string]float64
• func ResourcesAsUsage(resourceScarcity map[string]float64, resources common.ComputeResources) float64
• func ResourcesFloatAsUsage(resourceScarcity map[string]float64, resources common.ComputeResourcesFloat) float64
• func SliceResourceWithLimits(resourceScarcity map[string]float64, ...) map[*api.Queue]*QueueSchedulingInfo
• func SumRemainingResource(schedulingInfo map[*api.Queue]*QueueSchedulingInfo) common.ComputeResourcesFloat
• type JobQueue
• type LeaseManager
  • func NewLeaseManager(jobRepository repository.JobRepository, ...) *LeaseManager
  • func (l *LeaseManager) ExpireLeases()
• type LeasePayloadLimit
  • func NewLeasePayloadLimit(numberOfJobsLimit int, payloadSizeLimitBytes int, maxExpectedJobSizeBytes int) LeasePayloadLimit
  • func (s *LeasePayloadLimit) AtLimit() bool
  • func (s *LeasePayloadLimit) IsWithinLimit(job *api.Job) bool
  • func (s *LeasePayloadLimit) RemoveFromRemainingLimit(jobs ...*api.Job)
• type PodMatchingContext
  • func NewPodMatchingContext(podSpec *v1.PodSpec) *PodMatchingContext
  • func (podCtx *PodMatchingContext) Matches(nodeType *api.NodeType, availableResources common.ComputeResourcesFloat) bool
• type QueuePriorityInfo
• type QueueSchedulingInfo
  • func NewQueueSchedulingInfo(remainingSchedulingLimit common.ComputeResourcesFloat, ...) *QueueSchedulingInfo
  • func (info *QueueSchedulingInfo) UpdateLimits(resourceUsed common.ComputeResourcesFloat)

Functions

func AggregateNodeTypeAllocations

func AggregateNodeTypeAllocations(nodes []api.NodeInfo) []*nodeTypeAllocation

func CalculatePriorityUpdate

func CalculatePriorityUpdate(resourceScarcity map[string]float64, previousReport *api.ClusterUsageReport, report *api.ClusterUsageReport, previousPriority map[string]float64, halfTime time.Duration) map[string]float64

func CalculateQueuesPriorityInfo

func CalculateQueuesPriorityInfo(clusterPriorities map[string]map[string]float64, activeClusterReports map[string]*api.ClusterUsageReport, queues []*api.Queue) map[*api.Queue]QueuePriorityInfo

func CombineLeasedReportResourceByQueue

func CombineLeasedReportResourceByQueue(reports map[string]*api.ClusterLeasedReport) map[string]common.ComputeResources

func CreateClusterLeasedReport

func CreateClusterLeasedReport(clusterId string, currentReport *api.ClusterLeasedReport, additionallyLeasedJobs []*api.Job) *api.ClusterLeasedReport

func CreateClusterSchedulingInfoReport

func CreateClusterSchedulingInfoReport(leaseRequest *api.LeaseRequest, nodeAllocations []*nodeTypeAllocation) *api.ClusterSchedulingInfoReport

func FilterActiveClusterSchedulingInfoReports

func FilterActiveClusterSchedulingInfoReports(reports map[string]*api.ClusterSchedulingInfoReport) map[string]*api.ClusterSchedulingInfoReport

FilterActiveClusterSchedulingInfoReports returns the subset of reports within the expiry time.

func FilterActiveClusters

func FilterActiveClusters(reports map[string]*api.ClusterUsageReport) map[string]*api.ClusterUsageReport

FilterActiveClusters returns the subset of reports corresponding to active clusters. A cluster is considered active if the most recent ClusterUsageReport was received less than activeClusterExpiry ago.

func FilterClusterLeasedReports

func FilterClusterLeasedReports(ids []string, reports map[string]*api.ClusterLeasedReport) map[string]*api.ClusterLeasedReport

FilterClusterLeasedReports returns the subset of reports with id in the provided slice of ids. ids for which there is no corresponding report are ignored.

func FilterPoolClusters

func FilterPoolClusters(pool string, reports map[string]*api.ClusterUsageReport) map[string]*api.ClusterUsageReport

FilterPoolClusters returns the subset of reports for which the pool has a specific value.

func GetClusterReportIds

func GetClusterReportIds(reports map[string]*api.ClusterUsageReport) []string

GetClusterReportIds returns a slice composed of all unique report ids in the provided map.

func GroupByPool

func GroupByPool(reports map[string]*api.ClusterUsageReport) map[string]map[string]*api.ClusterUsageReport

GroupByPool returns a map from pool name to another map, which in turn maps report ids to reports.

func GroupSchedulingInfoByPool

func GroupSchedulingInfoByPool(reports map[string]*api.ClusterSchedulingInfoReport) map[string]map[string]*api.ClusterSchedulingInfoReport

GroupSchedulingInfoByPool returns a map from pool name to another map, which in turn maps report ids to reports.

func LeaseJobs

func LeaseJobs(ctx context.Context,
	config *configuration.SchedulingConfig,
	jobQueue JobQueue,
	onJobLease func([]*api.Job),
	request *api.LeaseRequest,
	nodeResources []*nodeTypeAllocation,
	activeClusterReports map[string]*api.ClusterUsageReport,
	activeClusterLeaseJobReports map[string]*api.ClusterLeasedReport,
	clusterPriorities map[string]map[string]float64,
	activeQueues []*api.Queue) ([]*api.Job, error)

func MatchSchedulingRequirements

func MatchSchedulingRequirements(job *api.Job, schedulingInfo *api.ClusterSchedulingInfoReport) bool

func MatchSchedulingRequirementsOnAnyCluster

func MatchSchedulingRequirementsOnAnyCluster(job *api.Job, allClusterSchedulingInfos map[string]*api.ClusterSchedulingInfoReport) bool

func QueueSlicesToShares

func QueueSlicesToShares(resourceScarcity map[string]float64, schedulingInfo map[*api.Queue]*QueueSchedulingInfo) map[*api.Queue]float64

func ResourceScarcityFromReports

func ResourceScarcityFromReports(reports map[string]*api.ClusterUsageReport) map[string]float64

func ResourcesAsUsage

func ResourcesAsUsage(resourceScarcity map[string]float64, resources common.ComputeResources) float64

func ResourcesFloatAsUsage

func ResourcesFloatAsUsage(resourceScarcity map[string]float64, resources common.ComputeResourcesFloat) float64

func SliceResourceWithLimits

func SliceResourceWithLimits(resourceScarcity map[string]float64, queueSchedulingInfo map[*api.Queue]*QueueSchedulingInfo, queuePriorities map[*api.Queue]QueuePriorityInfo, quantityToSlice common.ComputeResourcesFloat) map[*api.Queue]*QueueSchedulingInfo

func SumRemainingResource

func SumRemainingResource(schedulingInfo map[*api.Queue]*QueueSchedulingInfo) common.ComputeResourcesFloat

Types

type JobQueue

type JobQueue interface {
	PeekClusterQueue(clusterId, queue string, limit int64) ([]*api.Job, error)
	TryLeaseJobs(clusterId string, queue string, jobs []*api.Job) ([]*api.Job, error)
}

type LeaseManager

type LeaseManager struct {
	// contains filtered or unexported fields
}

func NewLeaseManager

func NewLeaseManager(
	jobRepository repository.JobRepository,
	queueRepository repository.QueueRepository,
	eventStore repository.EventStore,
	leaseExpiryDuration time.Duration) *LeaseManager

func (*LeaseManager) ExpireLeases

func (l *LeaseManager) ExpireLeases()

type LeasePayloadLimit

type LeasePayloadLimit struct {
	// contains filtered or unexported fields
}

func NewLeasePayloadLimit

func NewLeasePayloadLimit(numberOfJobsLimit int, payloadSizeLimitBytes int, maxExpectedJobSizeBytes int) LeasePayloadLimit

NewLeasePayloadLimit numberOfJobsLimit - This is the maximum number of jobs per lease payload payloadSizeLimitBytes - This is the maximum size of all the Jobs in the lease payload in bytes maxExpectedJobSizeBytes - This is the maximum expected job size in bytes

It is used as a threshold to determine when adding more jobs would exceed payloadSizeLimitBytes

func (*LeasePayloadLimit) AtLimit

func (s *LeasePayloadLimit) AtLimit() bool

AtLimit

This returns true when:
- remainingJobCount <= 0
- remainingPayloadSizeLimitBytes <= maxExpectedJobSizeBytes
  If we have less bytes left than the maximum job size, we should consider ourselves at the limit

func (*LeasePayloadLimit) IsWithinLimit

func (s *LeasePayloadLimit) IsWithinLimit(job *api.Job) bool

func (*LeasePayloadLimit) RemoveFromRemainingLimit

func (s *LeasePayloadLimit) RemoveFromRemainingLimit(jobs ...*api.Job)

type PodMatchingContext

type PodMatchingContext struct {
	// contains filtered or unexported fields
}

func NewPodMatchingContext

func NewPodMatchingContext(podSpec *v1.PodSpec) *PodMatchingContext

func (*PodMatchingContext) Matches

func (podCtx *PodMatchingContext) Matches(nodeType *api.NodeType, availableResources common.ComputeResourcesFloat) bool

type QueuePriorityInfo

type QueuePriorityInfo struct {
	Priority     float64
	CurrentUsage common.ComputeResources
}

type QueueSchedulingInfo

type QueueSchedulingInfo struct {
	// contains filtered or unexported fields
}

func NewQueueSchedulingInfo

func NewQueueSchedulingInfo(
	remainingSchedulingLimit common.ComputeResourcesFloat,
	schedulingShare common.ComputeResourcesFloat,
	adjustedShare common.ComputeResourcesFloat) *QueueSchedulingInfo

func (*QueueSchedulingInfo) UpdateLimits

func (info *QueueSchedulingInfo) UpdateLimits(resourceUsed common.ComputeResourcesFloat)