repository

Documentation

Index

• type EventRepository
• type JobRepository
• type QueueRepository
• type RedisEventRepository
  • func NewRedisEventRepository(db redis.UniversalClient) *RedisEventRepository
  • func (repo *RedisEventRepository) ReadEvents(jobSetId string, lastId string, limit int64, block time.Duration) ([]*api.EventStreamMessage, error)
  • func (repo *RedisEventRepository) ReportEvent(message *api.EventMessage) error
  • func (repo *RedisEventRepository) ReportEvents(message []*api.EventMessage) error
• type RedisJobRepository
  • func NewRedisJobRepository(db redis.UniversalClient) *RedisJobRepository
  • func (repo *RedisJobRepository) AddJob(job *api.Job) error
  • func (repo *RedisJobRepository) Cancel(jobs []*api.Job) map[*api.Job]error
  • func (repo *RedisJobRepository) CreateJob(request *api.JobRequest) *api.Job
  • func (repo *RedisJobRepository) ExpireLeases(queue string, deadline time.Time) ([]*api.Job, error)
  • func (repo *RedisJobRepository) FilterActiveQueues(queues []*api.Queue) ([]*api.Queue, error)
  • func (repo *RedisJobRepository) GetActiveJobIds(queue string, jobSetId string) ([]string, error)
  • func (repo *RedisJobRepository) GetJobsByIds(ids []string) ([]*api.Job, error)
  • func (repo *RedisJobRepository) GetQueueSizes(queues []*api.Queue) (sizes []int64, err error)
  • func (repo *RedisJobRepository) PeekQueue(queue string, limit int64) ([]*api.Job, error)
  • func (repo *RedisJobRepository) Remove(jobIds []string) (cleanedJobIds []string, e error)
  • func (repo *RedisJobRepository) RenewLease(clusterId string, jobIds []string) (renewedJobIds []string, e error)
  • func (repo *RedisJobRepository) ReturnLease(clusterId string, jobId string) (returnedJob *api.Job, err error)
  • func (repo *RedisJobRepository) TryLeaseJobs(clusterId string, queue string, jobs []*api.Job) ([]*api.Job, error)
• type RedisQueueRepository
  • func NewRedisQueueRepository(db redis.UniversalClient) *RedisQueueRepository
  • func (r *RedisQueueRepository) CreateQueue(queue *api.Queue) error
  • func (r *RedisQueueRepository) GetAllQueues() ([]*api.Queue, error)
  • func (r *RedisQueueRepository) GetQueue(name string) (*api.Queue, error)
• type RedisUsageRepository
  • func NewRedisUsageRepository(db redis.UniversalClient) *RedisUsageRepository
  • func (r *RedisUsageRepository) GetClusterPriorities(clusterIds []string) (map[string]map[string]float64, error)
  • func (r *RedisUsageRepository) GetClusterPriority(clusterId string) (map[string]float64, error)
  • func (r *RedisUsageRepository) GetClusterUsageReports() (map[string]*api.ClusterUsageReport, error)
  • func (r *RedisUsageRepository) UpdateCluster(report *api.ClusterUsageReport, priorities map[string]float64) error
• type Usage
• type UsageRepository

Types

type EventRepository

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

type JobRepository

type JobRepository interface {
	CreateJob(request *api.JobRequest) *api.Job
	AddJob(job *api.Job) error
	GetJobsByIds(ids []string) ([]*api.Job, error)
	PeekQueue(queue string, limit int64) ([]*api.Job, error)
	FilterActiveQueues(queues []*api.Queue) ([]*api.Queue, error)
	GetQueueSizes(queues []*api.Queue) (sizes []int64, e error)
	TryLeaseJobs(clusterId string, queue string, jobs []*api.Job) ([]*api.Job, error)
	RenewLease(clusterId string, jobIds []string) (renewed []string, e error)
	ExpireLeases(queue string, deadline time.Time) (expired []*api.Job, e error)
	Remove(jobIds []string) (cleanedJobs []string, e error)
	ReturnLease(clusterId string, jobId string) (returnedJob *api.Job, err error)
	Cancel(jobs []*api.Job) map[*api.Job]error
	GetActiveJobIds(queue string, jobSetId string) ([]string, error)
}

type QueueRepository

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

type RedisEventRepository

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

func NewRedisEventRepository

func NewRedisEventRepository(db redis.UniversalClient) *RedisEventRepository

func (*RedisEventRepository) ReadEvents

func (repo *RedisEventRepository) ReadEvents(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(message []*api.EventMessage) error

type RedisJobRepository

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

func NewRedisJobRepository

func NewRedisJobRepository(db redis.UniversalClient) *RedisJobRepository

func (*RedisJobRepository) AddJob

func (repo *RedisJobRepository) AddJob(job *api.Job) error

func (*RedisJobRepository) Cancel

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

func (*RedisJobRepository) CreateJob

func (repo *RedisJobRepository) CreateJob(request *api.JobRequest) *api.Job

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

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

func (*RedisJobRepository) GetQueueSizes

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

func (*RedisJobRepository) PeekQueue

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

func (*RedisJobRepository) Remove

func (repo *RedisJobRepository) Remove(jobIds []string) (cleanedJobIds []string, e 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

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

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

func (*RedisQueueRepository) GetQueue

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

type RedisUsageRepository

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

func NewRedisUsageRepository

func NewRedisUsageRepository(db redis.UniversalClient) *RedisUsageRepository

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

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)

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