service

Documentation

Index

• Constants
• func ExtractEssentialJobMetadata(jobRun *executorapi.JobRunLease) (*job.RunMeta, error)
• type ClusterAllocationService
  • func NewClusterAllocationService(clusterId executorContext.ClusterIdentity, ...) *ClusterAllocationService
  • func (allocationService *ClusterAllocationService) AllocateSpareClusterCapacity()
• type ClusterAllocator
• type IssueHandler
• type JobLeaseRequester
  • func NewJobLeaseRequester(executorApiClient executorapi.ExecutorApiClient, ...) *JobLeaseRequester
  • func (requester *JobLeaseRequester) LeaseJobRuns(ctx *armadacontext.Context, request *LeaseRequest) (*LeaseResponse, error)
• type JobRequester
  • func NewJobRequester(clusterId executorContext.ClusterIdentity, ...) *JobRequester
  • func (r *JobRequester) RequestJobsRuns()
• type JobStateReporter
  • func NewJobStateReporter(clusterContext clusterContext.ClusterContext, ...) (*JobStateReporter, error)
  • func (stateReporter *JobStateReporter) ReportMissingJobEvents()
• type LeaseRequest
• type LeaseRequester
• type LeaseResponse
• type PodIssueHandler
  • func NewPodIssuerHandler(jobRunState job.RunStateStore, clusterContext executorContext.ClusterContext, ...) (*PodIssueHandler, error)
  • func (p *PodIssueHandler) DetectAndRegisterFailedPodIssue(pod *v1.Pod) (bool, error)
  • func (p *PodIssueHandler) HandlePodIssues()
  • func (p *PodIssueHandler) HasIssue(runId string) bool
• type ResourceCleanupService
  • func NewResourceCleanupService(clusterContext clusterContext.ClusterContext, ...) (*ResourceCleanupService, error)
  • func (r *ResourceCleanupService) CleanupResources()

Constants

const (
	UnableToSchedule podIssueType = iota
	StuckStartingUp
	StuckTerminating
	ActiveDeadlineExceeded
	ExternallyDeleted
	ErrorDuringIssueHandling
	FailedStartingUp
)

Functions

func ExtractEssentialJobMetadata

func ExtractEssentialJobMetadata(jobRun *executorapi.JobRunLease) (*job.RunMeta, error)

Types

type ClusterAllocationService

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

func NewClusterAllocationService

func NewClusterAllocationService(
	clusterId executorContext.ClusterIdentity,
	eventReporter reporter.EventReporter,
	jobRunStateManager job.RunStateStore,
	submitter job.Submitter,
	clusterHealthMonitor healthmonitor.HealthMonitor,
) *ClusterAllocationService

func (*ClusterAllocationService) AllocateSpareClusterCapacity

func (allocationService *ClusterAllocationService) AllocateSpareClusterCapacity()

type ClusterAllocator

type ClusterAllocator interface {
	AllocateSpareClusterCapacity()
}

type IssueHandler

type IssueHandler interface {
	HasIssue(runId string) bool
	DetectAndRegisterFailedPodIssue(pod *v1.Pod) (bool, error)
}

type JobLeaseRequester

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

func NewJobLeaseRequester

func NewJobLeaseRequester(
	executorApiClient executorapi.ExecutorApiClient,
	clusterIdentity clusterContext.ClusterIdentity,
) *JobLeaseRequester

func (*JobLeaseRequester) LeaseJobRuns

func (requester *JobLeaseRequester) LeaseJobRuns(ctx *armadacontext.Context, request *LeaseRequest) (*LeaseResponse, error)

type JobRequester

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

func NewJobRequester

func NewJobRequester(
	clusterId executorContext.ClusterIdentity,
	eventReporter reporter.EventReporter,
	leaseRequester LeaseRequester,
	jobRunStateStore job.RunStateStore,
	utilisationService utilisation.UtilisationService,
	podDefaults *configuration.PodDefaults,
	maxLeasedJobs int,
	maxRequestDuration time.Duration,
) *JobRequester

func (*JobRequester) RequestJobsRuns

func (r *JobRequester) RequestJobsRuns()

type JobStateReporter

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

func NewJobStateReporter

func NewJobStateReporter(
	clusterContext clusterContext.ClusterContext,
	eventReporter reporter.EventReporter,
	podIssueHandler IssueHandler,
) (*JobStateReporter, error)

func (*JobStateReporter) ReportMissingJobEvents

func (stateReporter *JobStateReporter) ReportMissingJobEvents()

type LeaseRequest

type LeaseRequest struct {
	AvailableResource   armadaresource.ComputeResources
	Nodes               []*executorapi.NodeInfo
	UnassignedJobRunIds []string
	MaxJobsToLease      uint32
}

type LeaseRequester

type LeaseRequester interface {
	LeaseJobRuns(ctx *armadacontext.Context, request *LeaseRequest) (*LeaseResponse, error)
}

type LeaseResponse

type LeaseResponse struct {
	LeasedRuns      []*executorapi.JobRunLease
	RunIdsToCancel  []string
	RunIdsToPreempt []string
}

type PodIssueHandler

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

func NewPodIssuerHandler

func NewPodIssuerHandler(
	jobRunState job.RunStateStore,
	clusterContext executorContext.ClusterContext,
	eventReporter reporter.EventReporter,
	stateChecksConfig configuration.StateChecksConfiguration,
	pendingPodChecker podchecks.PodChecker,
	failedPodChecker failedpodchecks.RetryChecker,
	stuckTerminatingPodExpiry time.Duration,
) (*PodIssueHandler, error)

func (*PodIssueHandler) DetectAndRegisterFailedPodIssue

func (p *PodIssueHandler) DetectAndRegisterFailedPodIssue(pod *v1.Pod) (bool, error)

func (*PodIssueHandler) HandlePodIssues

func (p *PodIssueHandler) HandlePodIssues()

func (*PodIssueHandler) HasIssue

func (p *PodIssueHandler) HasIssue(runId string) bool

type ResourceCleanupService

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

func NewResourceCleanupService

func NewResourceCleanupService(
	clusterContext clusterContext.ClusterContext,
	kubernetesConfiguration configuration.KubernetesConfiguration,
) (*ResourceCleanupService, error)

func (*ResourceCleanupService) CleanupResources

func (r *ResourceCleanupService) CleanupResources()

CleanupResources

• This function finds and delete old resources. It does this in two ways:
• - By deleting all expired terminated pods
• - Deleting non-expired terminated pods when then MaxTerminatedPods limit is exceeded