queue

Documentation

Index

• Variables
• type Config
  • func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet)
  • func (cfg *Config) Validate() error
• type Index
• type Limits
• type Metrics
  • func NewMetrics(registerer prometheus.Registerer, metricsNamespace string, subsystem string) *Metrics
• type Queue
  • func NewQueue(logger log.Logger, cfg Config, limits Limits, metrics *Metrics, ...) (*Queue, error)
  • func (q *Queue) Dequeue(ctx context.Context, last Index, consumerID string) (*protos.ProtoTask, any, Index, error)
  • func (q *Queue) Enqueue(task *protos.ProtoTask, metadata any, successFn func()) error
  • func (q *Queue) GetConnectedConsumersMetric() float64
  • func (q *Queue) NotifyConsumerShutdown(consumer string)
  • func (q *Queue) RegisterConsumerConnection(consumer string)
  • func (q *Queue) Release(task *protos.ProtoTask)
  • func (q *Queue) TotalPending() (total int)
  • func (q *Queue) UnregisterConsumerConnection(consumer string)

Variables

var ErrStopped = queue.ErrStopped

var StartIndex = queue.StartIndex

Types

type Config

type Config struct {
	MaxQueuedTasksPerTenant int    `yaml:"max_queued_tasks_per_tenant"`
	StoreTasksOnDisk        bool   `yaml:"store_tasks_on_disk"`
	TasksDiskDirectory      string `yaml:"tasks_disk_directory"`
	CleanTasksDirectory     bool   `yaml:"clean_tasks_directory"`
}

func (*Config) RegisterFlagsWithPrefix

func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet)

RegisterFlagsWithPrefix registers flags for the bloom-planner configuration.

func (*Config) Validate

func (cfg *Config) Validate() error

type Index

type Index = queue.QueueIndex

type Limits

type Limits = queue.Limits

type Metrics

type Metrics = queue.Metrics

func NewMetrics

func NewMetrics(registerer prometheus.Registerer, metricsNamespace string, subsystem string) *Metrics

type Queue

type Queue struct {
	services.Service
	// contains filtered or unexported fields
}

Queue is a wrapper of queue.RequestQueue that uses the file system to store the pending tasks. The queue also allows to store metadata with the task. This metadata can be anything. Metadata is stored in memory. When a task is enqueued (Enqueue), it's stored in the file system and recorded as pending. When it's dequeued (Dequeue), it's removed from the queue but kept in FS until released (Release).

TODO(salvacorts): In the future we may reuse this queue to store any proto message. We would need to use generics for that.

func NewQueue

func NewQueue(
	logger log.Logger,
	cfg Config,
	limits Limits,
	metrics *Metrics,
	storeMetrics storage.ClientMetrics,
) (*Queue, error)

func (*Queue) Dequeue

func (q *Queue) Dequeue(ctx context.Context, last Index, consumerID string) (*protos.ProtoTask, any, Index, error)

Dequeue takes a task from the queue. The task is not removed from the filesystem until Release is called.

func (*Queue) Enqueue

func (q *Queue) Enqueue(task *protos.ProtoTask, metadata any, successFn func()) error

Enqueue adds a task to the queue. The task is enqueued only if it doesn't already exist in the queue.

func (*Queue) GetConnectedConsumersMetric

func (q *Queue) GetConnectedConsumersMetric() float64

func (*Queue) NotifyConsumerShutdown

func (q *Queue) NotifyConsumerShutdown(consumer string)

func (*Queue) RegisterConsumerConnection

func (q *Queue) RegisterConsumerConnection(consumer string)

func (*Queue) Release

func (q *Queue) Release(task *protos.ProtoTask)

Release removes a task from the filesystem. Dequeue should be called before Remove.

func (*Queue) TotalPending

func (q *Queue) TotalPending() (total int)

func (*Queue) UnregisterConsumerConnection

func (q *Queue) UnregisterConsumerConnection(consumer string)