balanceworker

Documentation

Index

• Constants
• func WithEventAt(eventAt time.Time) handleOption
• func WithRawIngestedEvents(rawIngestedEvents []serializer.CloudEventsKafkaPayload) handleOption
• func WithSource(source string) handleOption
• func WithSourceOperation(sourceOperation snapshot.ValueOperationType) handleOption
• type BalanceWorkerRepository
• type BatchedIngestEventHandler
• type EntitlementFilters
  • func NewEntitlementFilters(cfg EntitlementFiltersConfig) (*EntitlementFilters, error)
  • func (f EntitlementFilters) IsEntitlementInScope(ctx context.Context, req filters.EntitlementFilterRequest) (bool, error)
  • func (f EntitlementFilters) IsNamespaceInScope(ctx context.Context, namespace string) (bool, error)
  • func (f EntitlementFilters) RecordLastCalculation(ctx context.Context, req filters.RecordLastCalculationRequest) error
  • func (f EntitlementFilters) WithMetrics(meter metric.Meter) (*EntitlementFilters, error)
• type EntitlementFiltersConfig
  • func (c EntitlementFiltersConfig) Validate() error
• type FilterScope
• type FilterStateStorage
  • func NewFilterStateStorage[T FilterStateStorageRedis | FilterStateStorageInMemory](storage T) (FilterStateStorage, error)
  • func (c FilterStateStorage) Driver() FilterStateStorageDriver
  • func (c FilterStateStorage) Redis() (FilterStateStorageRedis, error)
  • func (c FilterStateStorage) Validate() error
• type FilterStateStorageDriver
• type FilterStateStorageInMemory
  • func (c FilterStateStorageInMemory) Validate() error
• type FilterStateStorageRedis
  • func (c FilterStateStorageRedis) Validate() error
• type IngestEventDataResponse
• type IngestEventQueryFilter
• type Recalculator
  • func NewRecalculator(opts RecalculatorOptions) (*Recalculator, error)
  • func (r *Recalculator) GetEntitlementFilters() *EntitlementFilters
  • func (r *Recalculator) Recalculate(ctx context.Context, ns string, recalculationStartedAt time.Time) error
• type RecalculatorOptions
  • func (o RecalculatorOptions) Validate() error
• type Worker
  • func New(opts WorkerOptions) (*Worker, error)
  • func (w *Worker) AddHandler(handler grouphandler.GroupEventHandler)
  • func (w *Worker) Close() error
  • func (w *Worker) GetEntitlementsAffectedByMeterSubject(ctx context.Context, namespace string, meterSlugs []string, subject string) ([]pkgmodels.NamespacedID, error)
  • func (w *Worker) Run(ctx context.Context) error
• type WorkerOptions
  • func (o *WorkerOptions) Validate() error

Constants

const (
	// DefaultIncludeDeletedDuration is the default duration for which deleted entitlements are included in recalculation.
	// This ensures that the recent deleted snapshot events are also resent.
	DefaultIncludeDeletedDuration = 24 * time.Hour
)

Functions

func WithEventAt

func WithEventAt(eventAt time.Time) handleOption

func WithRawIngestedEvents

func WithRawIngestedEvents(rawIngestedEvents []serializer.CloudEventsKafkaPayload) handleOption

func WithSource

func WithSource(source string) handleOption

func WithSourceOperation

func WithSourceOperation(sourceOperation snapshot.ValueOperationType) handleOption

Types

type BalanceWorkerRepository

type BalanceWorkerRepository interface {
	ListAffectedEntitlements(ctx context.Context, filterPairs []IngestEventQueryFilter) ([]IngestEventDataResponse, error)
}

type BatchedIngestEventHandler

type BatchedIngestEventHandler = func(ctx context.Context, event ingestevents.EventBatchedIngest) error

type EntitlementFilters

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

func NewEntitlementFilters

func NewEntitlementFilters(cfg EntitlementFiltersConfig) (*EntitlementFilters, error)

func (EntitlementFilters) IsEntitlementInScope

func (f EntitlementFilters) IsEntitlementInScope(ctx context.Context, req filters.EntitlementFilterRequest) (bool, error)

func (EntitlementFilters) IsNamespaceInScope

func (f EntitlementFilters) IsNamespaceInScope(ctx context.Context, namespace string) (bool, error)

func (EntitlementFilters) RecordLastCalculation

func (f EntitlementFilters) RecordLastCalculation(ctx context.Context, req filters.RecordLastCalculationRequest) error

func (EntitlementFilters) WithMetrics

func (f EntitlementFilters) WithMetrics(meter metric.Meter) (*EntitlementFilters, error)

type EntitlementFiltersConfig

type EntitlementFiltersConfig struct {
	NotificationService notification.Service
	MetricMeter         metric.Meter
	StateStorage        FilterStateStorage
	Logger              *slog.Logger
}

func (EntitlementFiltersConfig) Validate

func (c EntitlementFiltersConfig) Validate() error

type FilterScope

type FilterScope string

const (
	FilterScopeNamespace   FilterScope = "namespace"
	FilterScopeEntitlement FilterScope = "entitlement"
)

type FilterStateStorage

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

func NewFilterStateStorage

func NewFilterStateStorage[T FilterStateStorageRedis | FilterStateStorageInMemory](storage T) (FilterStateStorage, error)

func (FilterStateStorage) Driver

func (c FilterStateStorage) Driver() FilterStateStorageDriver

func (FilterStateStorage) Redis

func (c FilterStateStorage) Redis() (FilterStateStorageRedis, error)

func (FilterStateStorage) Validate

func (c FilterStateStorage) Validate() error

type FilterStateStorageDriver

type FilterStateStorageDriver string

const (
	FilterStateStorageDriverRedis    FilterStateStorageDriver = "redis"
	FilterStateStorageDriverInMemory FilterStateStorageDriver = "in-memory"
)

type FilterStateStorageInMemory

type FilterStateStorageInMemory struct{}

func (FilterStateStorageInMemory) Validate

func (c FilterStateStorageInMemory) Validate() error

type FilterStateStorageRedis

type FilterStateStorageRedis struct {
	Client     *redis.Client
	Expiration time.Duration
}

func (FilterStateStorageRedis) Validate

func (c FilterStateStorageRedis) Validate() error

type IngestEventDataResponse

type IngestEventDataResponse struct {
	Namespace     string
	EntitlementID string
	SubjectKey    string
	// not all entitlements have a meter associated
	MeterSlug *string
}

type IngestEventQueryFilter

type IngestEventQueryFilter struct {
	Namespace  string
	SubjectKey string
	MeterSlugs []string
}

type Recalculator

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

func NewRecalculator

func NewRecalculator(opts RecalculatorOptions) (*Recalculator, error)

func (*Recalculator) GetEntitlementFilters

func (r *Recalculator) GetEntitlementFilters() *EntitlementFilters

func (*Recalculator) Recalculate

func (r *Recalculator) Recalculate(ctx context.Context, ns string, recalculationStartedAt time.Time) error

type RecalculatorOptions

type RecalculatorOptions struct {
	Entitlement *registry.Entitlement
	Subject     subject.Service
	EventBus    eventbus.Publisher
	MetricMeter metric.Meter

	NotificationService notification.Service
	FilterStateStorage  FilterStateStorage
	Logger              *slog.Logger
}

func (RecalculatorOptions) Validate

func (o RecalculatorOptions) Validate() error

type Worker

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

func New

func New(opts WorkerOptions) (*Worker, error)

func (*Worker) AddHandler

func (w *Worker) AddHandler(handler grouphandler.GroupEventHandler)

AddHandler adds an additional handler to the list of batched ingest event handlers. Handlers are called in the order they are added and run after the riginal balance worker handler. In the case of any handler returning an error, the event will be retried so it is important that all handlers are idempotent.

func (*Worker) Close

func (w *Worker) Close() error

func (*Worker) GetEntitlementsAffectedByMeterSubject

func (w *Worker) GetEntitlementsAffectedByMeterSubject(ctx context.Context, namespace string, meterSlugs []string, subject string) ([]pkgmodels.NamespacedID, error)

func (*Worker) Run

func (w *Worker) Run(ctx context.Context) error

type WorkerOptions

type WorkerOptions struct {
	SystemEventsTopic string
	IngestEventsTopic string

	Router   router.Options
	EventBus eventbus.Publisher

	Entitlement *registry.Entitlement
	Repo        BalanceWorkerRepository

	// External connectors
	NotificationService notification.Service
	Subject             subject.Service

	MetricMeter metric.Meter

	Logger *slog.Logger

	FilterStateStorage FilterStateStorage
}

func (*WorkerOptions) Validate

func (o *WorkerOptions) Validate() error