store

Documentation

Index

• Constants
• Variables
• func InitRegistry()
• type Backend
  • func NewBackend(parent *badger.DB, storeSecret *memguard.LockedBuffer) (b *Backend, err error)
  • func (b *Backend) GetPartition(qtype QueueType, dtype conf.DestinationType) db.Partition
• type Destinations
  • func (dests *Destinations) Has(one conf.DestinationType) bool
  • func (dests *Destinations) Load() (res []conf.DestinationType)
  • func (dests *Destinations) Store(ds conf.DestinationType)
• type Forwarder
  • func NewForwarder(desttype conf.DestinationType, st *MessageStore, bc conf.BaseConfig, ...) *Forwarder
  • func (fwder *Forwarder) CreateDestination(ctx context.Context) (err error)
  • func (fwder *Forwarder) Forward(ctx context.Context) (err error)
• type MessageStore
  • func NewStore(ctx context.Context, cfg conf.StoreConfig, r kring.Ring, ...) (*MessageStore, error)
  • func (s *MessageStore) ACK(uid utils.MyULID, dest conf.DestinationType)
  • func (s *MessageStore) Confined() bool
  • func (s *MessageStore) Destinations() []conf.DestinationType
  • func (s *MessageStore) Errors() chan struct{}
  • func (s *MessageStore) GetSyslogConfig(confID utils.MyULID) (*conf.FilterSubConfig, error)
  • func (s *MessageStore) Ingest(m map[utils.MyULID]string) (int, error)
  • func (s *MessageStore) NACK(uid utils.MyULID, dest conf.DestinationType)
  • func (s *MessageStore) Outputs(dest conf.DestinationType) chan []*model.FullMessage
  • func (s *MessageStore) PermError(uid utils.MyULID, dest conf.DestinationType)
  • func (s *MessageStore) PurgeBadger() (err error)
  • func (s *MessageStore) ReadAllBadgers() (map[string]string, map[string]string, map[string]string)
  • func (s *MessageStore) SetDestinations(dests conf.DestinationType)
  • func (s *MessageStore) StoreAllSyslogConfigs(c conf.BaseConfig) (err error)
  • func (s *MessageStore) StoreSyslogConfig(confID utils.MyULID, config conf.FilterSubConfig) (err error)
  • func (s *MessageStore) WaitFinished()
• type QueueType

Constants

const (
	Messages = iota
	Ready
	Sent
	Failed
	PermErrors
)

Variables

var Queues = map[QueueType]string{
	Ready:      "r",
	Sent:       "s",
	Failed:     "f",
	PermErrors: "p",
}

var Registry *prometheus.Registry

Functions

func InitRegistry

func InitRegistry()

Types

type Backend

type Backend struct {
	Partitions map[QueueType]map[conf.DestinationType]db.Partition
	Messages   db.Partition
	Configs    db.Partition
	Whole      db.Partition
}

func NewBackend

func NewBackend(parent *badger.DB, storeSecret *memguard.LockedBuffer) (b *Backend, err error)

func (*Backend) GetPartition

func (b *Backend) GetPartition(qtype QueueType, dtype conf.DestinationType) db.Partition

type Destinations

type Destinations struct {
	atomic.Uint64
}

func (*Destinations) Has

func (dests *Destinations) Has(one conf.DestinationType) bool

func (*Destinations) Load

func (dests *Destinations) Load() (res []conf.DestinationType)

func (*Destinations) Store

func (dests *Destinations) Store(ds conf.DestinationType)

type Forwarder

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

func NewForwarder

func NewForwarder(desttype conf.DestinationType, st *MessageStore, bc conf.BaseConfig, logger log15.Logger, bindr binder.Client) *Forwarder

func (*Forwarder) CreateDestination

func (fwder *Forwarder) CreateDestination(ctx context.Context) (err error)

func (*Forwarder) Forward

func (fwder *Forwarder) Forward(ctx context.Context) (err error)

type MessageStore

type MessageStore struct {
	FatalErrorChan chan struct{}
	OutputsChans   map[conf.DestinationType]chan []*model.FullMessage

	BatchSize uint32
	// contains filtered or unexported fields
}

func NewStore

func NewStore(ctx context.Context, cfg conf.StoreConfig, r kring.Ring, dests conf.DestinationType, cfnd bool, l log15.Logger) (*MessageStore, error)

func (*MessageStore) ACK

func (s *MessageStore) ACK(uid utils.MyULID, dest conf.DestinationType)

func (*MessageStore) Confined

func (s *MessageStore) Confined() bool

func (*MessageStore) Destinations

func (s *MessageStore) Destinations() []conf.DestinationType

func (*MessageStore) Errors

func (s *MessageStore) Errors() chan struct{}

func (*MessageStore) GetSyslogConfig

func (s *MessageStore) GetSyslogConfig(confID utils.MyULID) (*conf.FilterSubConfig, error)

func (*MessageStore) Ingest

func (s *MessageStore) Ingest(m map[utils.MyULID]string) (int, error)

func (*MessageStore) NACK

func (s *MessageStore) NACK(uid utils.MyULID, dest conf.DestinationType)

func (*MessageStore) Outputs

func (s *MessageStore) Outputs(dest conf.DestinationType) chan []*model.FullMessage

func (*MessageStore) PermError

func (s *MessageStore) PermError(uid utils.MyULID, dest conf.DestinationType)

func (*MessageStore) PurgeBadger

func (s *MessageStore) PurgeBadger() (err error)

func (*MessageStore) ReadAllBadgers

func (s *MessageStore) ReadAllBadgers() (map[string]string, map[string]string, map[string]string)

func (*MessageStore) SetDestinations

func (s *MessageStore) SetDestinations(dests conf.DestinationType)

func (*MessageStore) StoreAllSyslogConfigs

func (s *MessageStore) StoreAllSyslogConfigs(c conf.BaseConfig) (err error)

func (*MessageStore) StoreSyslogConfig

func (s *MessageStore) StoreSyslogConfig(confID utils.MyULID, config conf.FilterSubConfig) (err error)

func (*MessageStore) WaitFinished

func (s *MessageStore) WaitFinished()

type QueueType

type QueueType uint8