server

Documentation

Index

• type JobqEndpoint
  • func NewEndpoint(log *zap.Logger, queues *QueueMap) *JobqEndpoint
  • func (se *JobqEndpoint) AddPlacementQueue(ctx context.Context, req *pb.JobQueueAddPlacementQueueRequest) (*pb.JobQueueAddPlacementQueueResponse, error)
  • func (se *JobqEndpoint) Clean(ctx context.Context, req *pb.JobQueueCleanRequest) (*pb.JobQueueCleanResponse, error)
  • func (se *JobqEndpoint) DestroyPlacementQueue(ctx context.Context, req *pb.JobQueueDestroyPlacementQueueRequest) (*pb.JobQueueDestroyPlacementQueueResponse, error)
  • func (se *JobqEndpoint) Inspect(ctx context.Context, req *pb.JobQueueInspectRequest) (*pb.JobQueueInspectResponse, error)
  • func (se *JobqEndpoint) Len(ctx context.Context, req *pb.JobQueueLengthRequest) (*pb.JobQueueLengthResponse, error)
  • func (se *JobqEndpoint) Peek(ctx context.Context, req *pb.JobQueuePeekRequest) (*pb.JobQueuePeekResponse, error)
  • func (se *JobqEndpoint) Pop(ctx context.Context, req *pb.JobQueuePopRequest) (*pb.JobQueuePopResponse, error)
  • func (se *JobqEndpoint) Push(ctx context.Context, req *pb.JobQueuePushRequest) (*pb.JobQueuePushResponse, error)
  • func (se *JobqEndpoint) PushBatch(ctx context.Context, req *pb.JobQueuePushBatchRequest) (*pb.JobQueuePushBatchResponse, error)
  • func (se *JobqEndpoint) Trim(ctx context.Context, req *pb.JobQueueTrimRequest) (*pb.JobQueueTrimResponse, error)
  • func (se *JobqEndpoint) Truncate(ctx context.Context, req *pb.JobQueueTruncateRequest) (*pb.JobQueueTruncateResponse, error)
• type QueueMap
  • func NewQueueMap(log *zap.Logger, ...) *QueueMap
  • func (qm *QueueMap) AddQueue(placement storj.PlacementConstraint) error
  • func (qm *QueueMap) DestroyQueue(placement storj.PlacementConstraint) error
  • func (qm *QueueMap) GetAllQueues() map[storj.PlacementConstraint]*jobqueue.Queue
  • func (qm *QueueMap) GetQueue(placement storj.PlacementConstraint) *jobqueue.Queue
  • func (qm *QueueMap) StopAll()
• type Server
  • func New(log *zap.Logger, listenAddress net.Addr, tlsOpts *tlsopts.Options, ...) (*Server, error)
  • func (s *Server) Addr() net.Addr
  • func (s *Server) Run(ctx context.Context) error
  • func (s *Server) SetTimeFunc(timeFunc func() time.Time)

Types

type JobqEndpoint

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

JobqEndpoint implements the DRPCJobQueueServer interface.

func NewEndpoint

func NewEndpoint(log *zap.Logger, queues *QueueMap) *JobqEndpoint

NewEndpoint creates a new endpoint.

func (*JobqEndpoint) AddPlacementQueue

func (se *JobqEndpoint) AddPlacementQueue(ctx context.Context, req *pb.JobQueueAddPlacementQueueRequest) (*pb.JobQueueAddPlacementQueueResponse, error)

AddPlacementQueue creates a new queue for the requested placement.

func (*JobqEndpoint) Clean

func (se *JobqEndpoint) Clean(ctx context.Context, req *pb.JobQueueCleanRequest) (*pb.JobQueueCleanResponse, error)

Clean removes all jobs from the queue that were last updated before the requested time. If the given placement is negative, all queues are cleaned.

func (*JobqEndpoint) DestroyPlacementQueue

func (se *JobqEndpoint) DestroyPlacementQueue(ctx context.Context, req *pb.JobQueueDestroyPlacementQueueRequest) (*pb.JobQueueDestroyPlacementQueueResponse, error)

DestroyPlacementQueue removes the queue for the requested placement.

func (*JobqEndpoint) Inspect

func (se *JobqEndpoint) Inspect(ctx context.Context, req *pb.JobQueueInspectRequest) (*pb.JobQueueInspectResponse, error)

Inspect finds a particular job in the queue by its placement, streamID, and position and returns all of the job information.

func (*JobqEndpoint) Len

func (se *JobqEndpoint) Len(ctx context.Context, req *pb.JobQueueLengthRequest) (*pb.JobQueueLengthResponse, error)

Len returns the number of jobs in the queues for the requested placement.

func (*JobqEndpoint) Peek

func (se *JobqEndpoint) Peek(ctx context.Context, req *pb.JobQueuePeekRequest) (*pb.JobQueuePeekResponse, error)

Peek returns the lowest-health job from the queues for the requested placements without removing the job from its queue.

func (*JobqEndpoint) Pop

func (se *JobqEndpoint) Pop(ctx context.Context, req *pb.JobQueuePopRequest) (*pb.JobQueuePopResponse, error)

Pop removes the lowest-health job from the queues for the requested placements.

func (*JobqEndpoint) Push

func (se *JobqEndpoint) Push(ctx context.Context, req *pb.JobQueuePushRequest) (*pb.JobQueuePushResponse, error)

Push inserts a job into the appropriate queue for its placement.

func (*JobqEndpoint) PushBatch

func (se *JobqEndpoint) PushBatch(ctx context.Context, req *pb.JobQueuePushBatchRequest) (*pb.JobQueuePushBatchResponse, error)

PushBatch inserts multiple jobs into the appropriate queues for their placements.

func (*JobqEndpoint) Trim

func (se *JobqEndpoint) Trim(ctx context.Context, req *pb.JobQueueTrimRequest) (*pb.JobQueueTrimResponse, error)

Trim removes all jobs from the queue with health greater than the given value. If the given placement is negative, all queues are trimmed.

func (*JobqEndpoint) Truncate

func (se *JobqEndpoint) Truncate(ctx context.Context, req *pb.JobQueueTruncateRequest) (*pb.JobQueueTruncateResponse, error)

Truncate removes all jobs from the queue for the requested placement. The queue is not destroyed.

type QueueMap

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

QueueMap is a thread-safe mapping of placement constraints to queues.

func NewQueueMap

func NewQueueMap(log *zap.Logger, queueFactory func(storj.PlacementConstraint) (*jobqueue.Queue, error)) *QueueMap

NewQueueMap creates a new QueueMap.

func (*QueueMap) AddQueue

func (qm *QueueMap) AddQueue(placement storj.PlacementConstraint) error

AddQueue creates a new queue for the given placement. If a queue already exists for the given placement, an error is returned.

func (*QueueMap) DestroyQueue

func (qm *QueueMap) DestroyQueue(placement storj.PlacementConstraint) error

DestroyQueue destroys the queue for the given placement. If no queue exists for the given placement, an error is returned.

func (*QueueMap) GetAllQueues

func (qm *QueueMap) GetAllQueues() map[storj.PlacementConstraint]*jobqueue.Queue

GetAllQueues gets a copy of the current queue map. It is possible for another caller to have destroyed one or more queues between this call and the time when the caller uses the returned map. If this happens, the affected queues will simply appear empty.

func (*QueueMap) GetQueue

func (qm *QueueMap) GetQueue(placement storj.PlacementConstraint) *jobqueue.Queue

GetQueue gets the queue for the given placement. If no queue exists for the given placement, nil is returned.

func (*QueueMap) StopAll

func (qm *QueueMap) StopAll()

StopAll stops and removes all queues.

type Server

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

Server represents a job queue DRPC server.

func New

func New(log *zap.Logger, listenAddress net.Addr, tlsOpts *tlsopts.Options, retryAfter time.Duration, initAlloc, memReleaseThreshold int) (*Server, error)

New creates a new Server instance.

func (*Server) Addr

func (s *Server) Addr() net.Addr

Addr returns the address on which the server is listening.

func (*Server) Run

func (s *Server) Run(ctx context.Context) error

Run runs the server (accepting connections on the listener) until the context is canceled.

func (*Server) SetTimeFunc

func (s *Server) SetTimeFunc(timeFunc func() time.Time)

SetTimeFunc sets the time function for all queues in the server. This is primarily used for testing to control the timestamps used in the queue.