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

func FilterActiveClusters

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

func FilterClusterLeasedReports

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

func FilterPoolClusters

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

func GetClusterReportIds

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

func GroupByPool

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

func GroupSchedulingInfoByPool

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

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