persister

Documentation

Index

• Variables
• type Option
  • func WithBatchSize(size int) Option
  • func WithCheckInterval(interval time.Duration) Option
  • func WithFlushInterval(i time.Duration) Option
  • func WithFlushSize(s int) Option
  • func WithFlushTimeout(timeout time.Duration) Option
  • func WithLogger(l *zap.Logger) Option
  • func WithMaxMPS(mps int) Option
  • func WithMetrics(r metrics.Registerer) Option
  • func WithNumWorkers(n int) Option
• type PersisterDWH
  • func NewPersisterDWH(client *pubsub.Client, r metrics.Registerer, writer Writer, ...) *PersisterDWH
  • func (p *PersisterDWH) Check(ctx context.Context) health.Status
  • func (p *PersisterDWH) IsRunning() bool
  • func (p *PersisterDWH) Run() error
  • func (p *PersisterDWH) Stop()
• type Writer
  • func NewEvalEventWriter(ctx context.Context, r metrics.Registerer, l *zap.Logger, exClient ec.Client, ...) (Writer, error)
  • func NewGoalEventWriter(ctx context.Context, r metrics.Registerer, l *zap.Logger, exClient ec.Client, ...) (Writer, error)

Variables

var (
	ErrUnexpectedMessageType                     = errors.New("eventpersister: unexpected message type")
	ErrAutoOpsRulesNotFound                      = errors.New("eventpersister: auto ops rules not found")
	ErrEvaluationsAreEmpty                       = errors.New("eventpersister: evaluations are empty")
	ErrEvaluationEventIssuedAfterExperimentEnded = errors.New("eventpersister: evaluation event issued after experiment ended") //nolint:lll
	ErrExperimentNotFound                        = errors.New("eventpersister: experiment not found")
	ErrFailedToEvaluateUser                      = errors.New("eventpersister: failed to evaluate user")
	ErrNoAutoOpsRules                            = errors.New("eventpersister: no auto ops rules")
	ErrNothingToLink                             = errors.New("eventpersister: nothing to link")
	ErrInvalidEventTimestamp                     = errors.New("eventpersister: invalid event timestamp")
)

Types

type Option

type Option func(*options)

func WithBatchSize

func WithBatchSize(size int) Option

func WithCheckInterval

func WithCheckInterval(interval time.Duration) Option

func WithFlushInterval

func WithFlushInterval(i time.Duration) Option

func WithFlushSize

func WithFlushSize(s int) Option

func WithFlushTimeout

func WithFlushTimeout(timeout time.Duration) Option

func WithLogger

func WithLogger(l *zap.Logger) Option

func WithMaxMPS

func WithMaxMPS(mps int) Option

func WithMetrics

func WithMetrics(r metrics.Registerer) Option

func WithNumWorkers

func WithNumWorkers(n int) Option

type PersisterDWH

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

func NewPersisterDWH

func NewPersisterDWH(
	client *pubsub.Client,
	r metrics.Registerer,
	writer Writer,
	mysqlClient mysql.Client,
	subscription string,
	topic string,
	pullerNumGoroutines int,
	pullerMaxOutstandingMessages int,
	pullerMaxOutstandingBytes int,
	opts ...Option,
) *PersisterDWH

func (*PersisterDWH) Check

func (p *PersisterDWH) Check(ctx context.Context) health.Status

func (*PersisterDWH) IsRunning

func (p *PersisterDWH) IsRunning() bool

func (*PersisterDWH) Run

func (p *PersisterDWH) Run() error

func (*PersisterDWH) Stop

func (p *PersisterDWH) Stop()

type Writer

type Writer interface {
	Write(ctx context.Context, evt environmentEventMap) map[string]bool
}

func NewEvalEventWriter

func NewEvalEventWriter(
	ctx context.Context,
	r metrics.Registerer,
	l *zap.Logger,
	exClient ec.Client,
	cache cachev3.ExperimentsCache,
	project, ds string,
	size int,
	location *time.Location,
) (Writer, error)

func NewGoalEventWriter

func NewGoalEventWriter(
	ctx context.Context,
	r metrics.Registerer,
	l *zap.Logger,
	exClient ec.Client,
	ftClient ft.Client,
	cache cachev3.ExperimentsCache,
	project, ds string,
	size int,
	location *time.Location,
) (Writer, error)