processor

Documentation

Index

• Constants
• Variables
• func NewAuditLogPersister(config interface{}, mysqlClient mysql.Client, logger *zap.Logger) (subscriber.Processor, error)
• func NewDomainEventInformer(environmentClient environmentclient.Client, sender sender.Sender, ...) subscriber.Processor
• func NewEvaluationCountEventPersister(ctx context.Context, config interface{}, mysqlClient mysql.Client, ...) (subscriber.Processor, error)
• func NewEventsDWHPersister(ctx context.Context, config interface{}, mysqlClient mysql.Client, ...) (subscriber.Processor, error)
• func NewEventsOPSPersister(ctx context.Context, config interface{}, mysqlClient mysql.Client, ...) (subscriber.Processor, error)
• func NewMetricsEventPersister(registerer metrics.Registerer, logger *zap.Logger) subscriber.Processor
• func NewPushSender(featureClient featureclient.Client, batchClient btclient.Client, ...) subscriber.Processor
• func NewSegmentUserPersister(config interface{}, batchClient btclient.Client, mysqlClient mysql.Client, ...) (subscriber.Processor, error)
• func NewUserEventPersister(config interface{}, mysqlClient mysql.Client, logger *zap.Logger) (subscriber.Processor, error)
• type EvaluationCountEventPersisterConfig
• type Processors
  • func NewProcessors(r metrics.Registerer) *Processors
  • func (p *Processors) GetProcessorByName(name string) (subscriber.Processor, error)
  • func (p *Processors) RegisterProcessor(name string, processor subscriber.Processor)
• type Updater
  • func NewEvalUserCountUpdater(ctx context.Context, featureClient featureclient.Client, ...) Updater
  • func NewGoalUserCountUpdater(ctx context.Context, featureClient featureclient.Client, ...) Updater
• type Writer
  • func NewEvalEventWriter(ctx context.Context, l *zap.Logger, exClient ec.Client, ...) (Writer, error)
  • func NewGoalEventWriter(ctx context.Context, l *zap.Logger, exClient ec.Client, ftClient ft.Client, ...) (Writer, error)

Constants

const (
	AuditLogPersisterName                = "auditLogPersister"
	DomainEventInformerName              = "domainEventInformer"
	EvaluationCountEventDWHPersisterName = "evaluationCountEventDWHPersister"
	EvaluationCountEventOPSPersisterName = "evaluationCountEventOPSPersister"
	EvaluationCountEventPersisterName    = "evaluationCountEventPersister"
	GoalCountEventDWHPersisterName       = "goalCountEventDWHPersister"
	GoalCountEventOPSPersisterName       = "goalCountEventOPSPersister"
	MetricsEventPersisterName            = "metricsEventPersister"
	PushSenderName                       = "pushSender"
	SegmentUserPersisterName             = "segmentUserPersister"
	UserEventPersisterName               = "userEventPersister"
)

Variables

var (
	ErrSegmentInvalidConfig                      = errors.New("segment: invalid config")
	ErrSegmentInUse                              = errors.New("segment: segment is in use")
	ErrSegmentExceededMaxUserIDLength            = fmt.Errorf("segment: max user id length allowed is %d", maxUserIDLength) //nolint:lll
	ErrUserEventInvalidConfig                    = errors.New("user event: invalid config")
	ErrEvaluationCountInvalidConfig              = errors.New("evaluation count: invalid config")
	ErrEventsDWHPersisterInvalidConfig           = errors.New("eventpersister: invalid config")
	ErrEventsOPSPersisterInvalidConfig           = errors.New("eventpersister: invalid config")
	ErrExperimentNotFound                        = errors.New("eventpersister: experiment not found")
	ErrReasonNil                                 = errors.New("eventpersister: reason is nil")
	ErrEvaluationsAreEmpty                       = errors.New("eventpersister: evaluations are empty")
	ErrEvaluationEventIssuedAfterExperimentEnded = errors.New("eventpersister: evaluation event issued after experiment ended") //nolint:lll
	ErrFailedToEvaluateUser                      = errors.New("eventpersister: failed to evaluate user")
	ErrAutoOpsRuleNotFound                       = errors.New("eventpersister: auto ops rule not found")
	ErrFeatureEmptyList                          = errors.New("eventpersister: list feature returned empty")
	ErrFeatureVersionNotFound                    = errors.New("eventpersister: feature version not found")
	ErrUnknownEvent                              = errors.New("metricsevent persister: unknown metrics event")
	ErrInvalidDuration                           = errors.New("metricsevent persister: invalid duration")
	ErrUnknownApiId                              = errors.New("metricsevent persister: unknown api id")
)

var (
	ErrUnknownSourceType = errors.New("batch-server: domain-event-informer unknown source type")
)

Functions

func NewAuditLogPersister

func NewAuditLogPersister(
	config interface{},
	mysqlClient mysql.Client,
	logger *zap.Logger,
) (subscriber.Processor, error)

func NewDomainEventInformer

func NewDomainEventInformer(
	environmentClient environmentclient.Client,
	sender sender.Sender,
	logger *zap.Logger,
) subscriber.Processor

func NewEvaluationCountEventPersister

func NewEvaluationCountEventPersister(
	ctx context.Context,
	config interface{},
	mysqlClient mysql.Client,
	evaluationCountCacher cache.MultiGetDeleteCountCache,
	logger *zap.Logger,
) (subscriber.Processor, error)

func NewEventsDWHPersister

func NewEventsDWHPersister(
	ctx context.Context,
	config interface{},
	mysqlClient mysql.Client,
	redisClient redisv3.Client,
	exClient experimentclient.Client,
	ftClient featureclient.Client,
	persisterName string,
	logger *zap.Logger,
) (subscriber.Processor, error)

func NewEventsOPSPersister

func NewEventsOPSPersister(
	ctx context.Context,
	config interface{},
	mysqlClient mysql.Client,
	redisClient redisv3.Client,
	opsClient autoopsclient.Client,
	ftClient featureclient.Client,
	persisterName string,
	logger *zap.Logger,
) (subscriber.Processor, error)

func NewMetricsEventPersister

func NewMetricsEventPersister(
	registerer metrics.Registerer,
	logger *zap.Logger,
) subscriber.Processor

func NewPushSender

func NewPushSender(
	featureClient featureclient.Client,
	batchClient btclient.Client,
	mysqlClient mysql.Client,
	logger *zap.Logger,
) subscriber.Processor

func NewSegmentUserPersister

func NewSegmentUserPersister(
	config interface{},
	batchClient btclient.Client,
	mysqlClient mysql.Client,
	logger *zap.Logger,
) (subscriber.Processor, error)

func NewUserEventPersister

func NewUserEventPersister(
	config interface{},
	mysqlClient mysql.Client,
	logger *zap.Logger,
) (subscriber.Processor, error)

Types

type EvaluationCountEventPersisterConfig

type EvaluationCountEventPersisterConfig struct {
	FlushSize          int `json:"flushSize"`
	FlushInterval      int `json:"flushInterval"`
	WriteCacheInterval int `json:"writeCacheInterval"`
}

type Processors

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

func NewProcessors

func NewProcessors(r metrics.Registerer) *Processors

func (*Processors) GetProcessorByName

func (p *Processors) GetProcessorByName(name string) (subscriber.Processor, error)

func (*Processors) RegisterProcessor

func (p *Processors) RegisterProcessor(name string, processor subscriber.Processor)

type Updater

type Updater interface {
	UpdateUserCounts(ctx context.Context, events environmentEventOPSMap) map[string]bool
}

func NewEvalUserCountUpdater

func NewEvalUserCountUpdater(
	ctx context.Context,
	featureClient featureclient.Client,
	autoOpsClient aoclient.Client,
	eventCounterCache cachev3.EventCounterCache,
	autoOpsRulesCache cachev3.AutoOpsRulesCache,
	logger *zap.Logger,
) Updater

func NewGoalUserCountUpdater

func NewGoalUserCountUpdater(
	ctx context.Context,
	featureClient featureclient.Client,
	autoOpsClient aoclient.Client,
	eventCounterCache cachev3.EventCounterCache,
	autoOpsRulesCache cachev3.AutoOpsRulesCache,
	logger *zap.Logger,
) Updater

type Writer

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

func NewEvalEventWriter

func NewEvalEventWriter(
	ctx context.Context,
	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,
	l *zap.Logger,
	exClient ec.Client,
	ftClient ft.Client,
	cache cachev3.ExperimentsCache,
	project, ds string,
	size int,
	location *time.Location,
) (Writer, error)