repository

package
v0.2.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 30, 2021 License: Apache-2.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source 
const JobNotFound = "no job found with provided Id"

Variables

View Source 
var ErrQueueAlreadyExists = errors.New("Queue already exists")
View Source 
var ErrQueueNotFound = errors.New("Queue does not exist")

Functions

This section is empty.

Types

type EventRepository 

type EventRepository interface {
	ReadEvents(queue, jobSetId string, lastId string, limit int64, block time.Duration) ([]*api.EventStreamMessage, error)
	GetLastMessageId(queue, jobSetId string) (string, error)
}

type EventStore added in v0.1.6 

type EventStore interface {
	ReportEvents(message []*api.EventMessage) error
}

type JobRepository 

type JobRepository interface {
	PeekQueue(queue string, limit int64) ([]*api.Job, error)
	TryLeaseJobs(clusterId string, queue string, jobs []*api.Job) ([]*api.Job, error)
	CreateJobs(request *api.JobSubmitRequest, owner string, ownershipGroups []string) ([]*api.Job, error)
	AddJobs(job []*api.Job) ([]*SubmitJobResult, error)
	GetExistingJobsByIds(ids []string) ([]*api.Job, error)
	FilterActiveQueues(queues []*api.Queue) ([]*api.Queue, error)
	GetQueueSizes(queues []*api.Queue) (sizes []int64, e error)
	IterateQueueJobs(queueName string, action func(*api.Job)) error
	GetQueueJobIds(queueName string) ([]string, error)
	RenewLease(clusterId string, jobIds []string) (renewed []string, e error)
	ExpireLeases(queue string, deadline time.Time) (expired []*api.Job, e error)
	ReturnLease(clusterId string, jobId string) (returnedJob *api.Job, err error)
	DeleteJobs(jobs []*api.Job) map[*api.Job]error
	GetActiveJobIds(queue string, jobSetId string) ([]string, error)
	GetLeasedJobIds(queue string) ([]string, error)
	UpdateStartTime(jobId string, clusterId string, startTime time.Time) error
	UpdatePriority(jobs []*api.Job, newPriority float64) (map[string]string, error)
	GetJobRunInfos(jobIds []string) (map[string]*RunInfo, error)
	GetQueueActiveJobSets(queue string) ([]*api.JobSetInfo, error)
	AddRetryAttempt(jobId string) error
	GetNumberOfRetryAttempts(jobId string) (int, error)
}

type KafkaEventJobStatusProcessor added in v0.1.32 

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

func NewKafkaJobStatusProcessor added in v0.1.32 

func NewKafkaJobStatusProcessor(reader *kafka.Reader, jobRepository JobRepository) *KafkaEventJobStatusProcessor

func (*KafkaEventJobStatusProcessor) ProcessEvents added in v0.1.32 

func (p *KafkaEventJobStatusProcessor) ProcessEvents()

type KafkaEventRedisProcessor added in v0.1.6 

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

func NewKafkaEventRedisProcessor added in v0.1.6 

func NewKafkaEventRedisProcessor(reader *kafka.Reader, repository EventStore) *KafkaEventRedisProcessor

func (*KafkaEventRedisProcessor) ProcessEvents added in v0.1.6 

func (p *KafkaEventRedisProcessor) ProcessEvents()

type KafkaEventStore added in v0.1.6 

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

func NewKafkaEventStore added in v0.1.6 

func NewKafkaEventStore(writer *kafka.Writer) *KafkaEventStore

func (*KafkaEventStore) ReportEvents added in v0.1.6 

func (k *KafkaEventStore) ReportEvents(messages []*api.EventMessage) error

type NatsEventJobStatusProcessor added in v0.1.32 

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

func NewNatsEventJobStatusProcessor added in v0.1.32 

func NewNatsEventJobStatusProcessor(connection *stanUtil.DurableConnection, jobRepository JobRepository, subject string, group string) *NatsEventJobStatusProcessor

func (*NatsEventJobStatusProcessor) Start added in v0.1.32 

func (p *NatsEventJobStatusProcessor) Start()

type NatsEventRedisProcessor added in v0.1.6 

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

func NewNatsEventRedisProcessor added in v0.1.6 

func NewNatsEventRedisProcessor(connection *stanUtil.DurableConnection, repository EventStore, subject string, group string) *NatsEventRedisProcessor

func (*NatsEventRedisProcessor) Start added in v0.1.6 

func (p *NatsEventRedisProcessor) Start()

type NatsEventStore added in v0.1.6 

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

func NewNatsEventStore added in v0.1.6 

func NewNatsEventStore(connection *stanUtil.DurableConnection, subject string) *NatsEventStore

func (*NatsEventStore) ReportEvents added in v0.1.6 

func (n *NatsEventStore) ReportEvents(messages []*api.EventMessage) error

type QueueRepository 

type QueueRepository interface {
	GetAllQueues() ([]*api.Queue, error)
	GetQueue(name string) (*api.Queue, error)
	CreateQueue(queue *api.Queue) error
	UpdateQueue(queue *api.Queue) error
	DeleteQueue(name string) error
}

type RedisEventRepository 

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

func NewRedisEventRepository 

func NewRedisEventRepository(db redis.UniversalClient, eventRetention configuration.EventRetentionPolicy) *RedisEventRepository

func (*RedisEventRepository) GetLastMessageId added in v0.0.5 

func (repo *RedisEventRepository) GetLastMessageId(queue, jobSetId string) (string, error)

func (*RedisEventRepository) ReadEvents 

func (repo *RedisEventRepository) ReadEvents(queue, jobSetId string, lastId string, limit int64, block time.Duration) ([]*api.EventStreamMessage, error)

func (*RedisEventRepository) ReportEvent 

func (repo *RedisEventRepository) ReportEvent(message *api.EventMessage) error

func (*RedisEventRepository) ReportEvents 

func (repo *RedisEventRepository) ReportEvents(messages []*api.EventMessage) error

type RedisHealth added in v0.2.3 

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

func NewRedisHealth added in v0.2.3 

func NewRedisHealth(db redis.UniversalClient) *RedisHealth

func (*RedisHealth) Check added in v0.2.3 

func (r *RedisHealth) Check() error

type RedisJobRepository 

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

func NewRedisJobRepository 

func NewRedisJobRepository(db redis.UniversalClient, defaultJobLimits common.ComputeResources) *RedisJobRepository

func (*RedisJobRepository) AddJobs added in v0.0.3 

func (repo *RedisJobRepository) AddJobs(jobs []*api.Job) ([]*SubmitJobResult, error)

func (*RedisJobRepository) AddRetryAttempt added in v0.1.16 

func (repo *RedisJobRepository) AddRetryAttempt(jobId string) error

func (*RedisJobRepository) CreateJobs added in v0.0.3 

func (repo *RedisJobRepository) CreateJobs(request *api.JobSubmitRequest, owner string, ownershipGroups []string) ([]*api.Job, error)

func (*RedisJobRepository) DeleteJobs added in v0.0.10 

func (repo *RedisJobRepository) DeleteJobs(jobs []*api.Job) map[*api.Job]error

func (*RedisJobRepository) ExpireLeases 

func (repo *RedisJobRepository) ExpireLeases(queue string, deadline time.Time) ([]*api.Job, error)

func (*RedisJobRepository) FilterActiveQueues 

func (repo *RedisJobRepository) FilterActiveQueues(queues []*api.Queue) ([]*api.Queue, error)

func (*RedisJobRepository) GetActiveJobIds 

func (repo *RedisJobRepository) GetActiveJobIds(queue string, jobSetId string) ([]string, error)

func (*RedisJobRepository) GetExistingJobsByIds added in v0.0.10 

func (repo *RedisJobRepository) GetExistingJobsByIds(ids []string) ([]*api.Job, error)

Returns existing jobs by Id If an Id is supplied that no longer exists, that job will simply be omitted from the result. No error will be thrown for missing jobs

func (*RedisJobRepository) GetJobRunInfos added in v0.1.33 

func (repo *RedisJobRepository) GetJobRunInfos(jobIds []string) (map[string]*RunInfo, error)

Returns the run info of each job id for the cluster they are currently associated with (leased by) Jobs with no value will be omitted from the results, which happens in the following cases: - The job is not associated with a cluster - The job has does not have a start time for the cluster it is associated with

func (*RedisJobRepository) GetLeasedJobIds added in v0.1.32 

func (repo *RedisJobRepository) GetLeasedJobIds(queue string) ([]string, error)

func (*RedisJobRepository) GetNumberOfRetryAttempts added in v0.1.16 

func (repo *RedisJobRepository) GetNumberOfRetryAttempts(jobId string) (int, error)

func (*RedisJobRepository) GetQueueActiveJobSets added in v0.1.2 

func (repo *RedisJobRepository) GetQueueActiveJobSets(queue string) ([]*api.JobSetInfo, error)

func (*RedisJobRepository) GetQueueJobIds added in v0.1.26 

func (repo *RedisJobRepository) GetQueueJobIds(queueName string) ([]string, error)

func (*RedisJobRepository) GetQueueSizes 

func (repo *RedisJobRepository) GetQueueSizes(queues []*api.Queue) (sizes []int64, err error)

func (*RedisJobRepository) IterateQueueJobs added in v0.1.24 

func (repo *RedisJobRepository) IterateQueueJobs(queueName string, action func(*api.Job)) error

func (*RedisJobRepository) PeekQueue 

func (repo *RedisJobRepository) PeekQueue(queue string, limit int64) ([]*api.Job, error)

func (*RedisJobRepository) RenewLease 

func (repo *RedisJobRepository) RenewLease(clusterId string, jobIds []string) (renewedJobIds []string, e error)

func (*RedisJobRepository) ReturnLease 

func (repo *RedisJobRepository) ReturnLease(clusterId string, jobId string) (returnedJob *api.Job, err error)

func (*RedisJobRepository) TryLeaseJobs 

func (repo *RedisJobRepository) TryLeaseJobs(clusterId string, queue string, jobs []*api.Job) ([]*api.Job, error)

returns list of jobs which are successfully leased

func (*RedisJobRepository) UpdatePriority added in v0.1.39 

func (repo *RedisJobRepository) UpdatePriority(jobs []*api.Job, newPriority float64) (map[string]string, error)

Updates priority if job exists, does not error if job doesn't exist

func (*RedisJobRepository) UpdateStartTime added in v0.1.32 

func (repo *RedisJobRepository) UpdateStartTime(jobId string, clusterId string, startTime time.Time) error

type RedisQueueRepository 

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

func NewRedisQueueRepository 

func NewRedisQueueRepository(db redis.UniversalClient) *RedisQueueRepository

func (*RedisQueueRepository) CreateQueue 

func (r *RedisQueueRepository) CreateQueue(queue *api.Queue) error

func (*RedisQueueRepository) DeleteQueue added in v0.1.14 

func (r *RedisQueueRepository) DeleteQueue(name string) error

func (*RedisQueueRepository) GetAllQueues 

func (r *RedisQueueRepository) GetAllQueues() ([]*api.Queue, error)

func (*RedisQueueRepository) GetQueue 

func (r *RedisQueueRepository) GetQueue(name string) (*api.Queue, error)

func (*RedisQueueRepository) UpdateQueue added in v0.2.3 

func (r *RedisQueueRepository) UpdateQueue(queue *api.Queue) error

type RedisSchedulingInfoRepository added in v0.1.6 

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

func NewRedisSchedulingInfoRepository added in v0.1.6 

func NewRedisSchedulingInfoRepository(db redis.UniversalClient) *RedisSchedulingInfoRepository

func (*RedisSchedulingInfoRepository) GetClusterSchedulingInfo added in v0.1.6 

func (r *RedisSchedulingInfoRepository) GetClusterSchedulingInfo() (map[string]*api.ClusterSchedulingInfoReport, error)

func (*RedisSchedulingInfoRepository) UpdateClusterSchedulingInfo added in v0.1.6 

func (r *RedisSchedulingInfoRepository) UpdateClusterSchedulingInfo(report *api.ClusterSchedulingInfoReport) error

type RedisUsageRepository 

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

func NewRedisUsageRepository 

func NewRedisUsageRepository(db redis.UniversalClient) *RedisUsageRepository

func (*RedisUsageRepository) GetClusterLeasedReports added in v0.1.6 

func (r *RedisUsageRepository) GetClusterLeasedReports() (map[string]*api.ClusterLeasedReport, error)

func (*RedisUsageRepository) GetClusterPriorities 

func (r *RedisUsageRepository) GetClusterPriorities(clusterIds []string) (map[string]map[string]float64, error)

func (*RedisUsageRepository) GetClusterPriority 

func (r *RedisUsageRepository) GetClusterPriority(clusterId string) (map[string]float64, error)

func (*RedisUsageRepository) GetClusterUsageReports 

func (r *RedisUsageRepository) GetClusterUsageReports() (map[string]*api.ClusterUsageReport, error)

func (*RedisUsageRepository) UpdateCluster 

func (r *RedisUsageRepository) UpdateCluster(report *api.ClusterUsageReport, priorities map[string]float64) error

func (*RedisUsageRepository) UpdateClusterLeased added in v0.1.6 

func (r *RedisUsageRepository) UpdateClusterLeased(report *api.ClusterLeasedReport) error

type RunInfo added in v0.1.33 

type RunInfo struct {
	StartTime        time.Time
	CurrentClusterId string
}

type SchedulingInfoRepository added in v0.1.6 

type SchedulingInfoRepository interface {
	GetClusterSchedulingInfo() (map[string]*api.ClusterSchedulingInfoReport, error)
	UpdateClusterSchedulingInfo(report *api.ClusterSchedulingInfoReport) error
}

type SubmitJobResult added in v0.0.3 

type SubmitJobResult struct {
	JobId             string
	SubmittedJob      *api.Job
	DuplicateDetected bool
	Error             error
}

type Usage 

type Usage struct {
	PriorityPerQueue     map[string]float64
	CurrentUsagePerQueue map[string]float64
}

type UsageRepository 

type UsageRepository interface {
	GetClusterUsageReports() (map[string]*api.ClusterUsageReport, error)
	GetClusterPriority(clusterId string) (map[string]float64, error)
	GetClusterPriorities(clusterIds []string) (map[string]map[string]float64, error)
	GetClusterLeasedReports() (map[string]*api.ClusterLeasedReport, error)

	UpdateCluster(report *api.ClusterUsageReport, priorities map[string]float64) error
	UpdateClusterLeased(report *api.ClusterLeasedReport) error
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.