events

Documentation

Overview

Package events contains a set of Events sent by Publishers.

PublishingGroupBuilder can be used to Build a set of Publishers.

Publisher types: - TimeDelayPublisher - Publishes events with a configurable time delay. - ResetOnRunAttemptPublisher - TimeDelayPublisher with resettable timer (Result.RunAttempted). - RetrySyncPublisher - TimeDelayPublisher with resettable backoff (Result.ResetRetryBackoff).

Events and their Publisher: - SyncEvent - ResetOnRunAttemptPublisher (SyncPeriod) - SyncWithReimportEvent - TimeDelayPublisher (SyncWithReimportPeriod) - NamespaceResyncEvent - TimeDelayPublisher (NamespaceControllerPeriod) - RetrySyncEvent - RetrySyncPublisher (RetryBackoff) - StatusEvent - ResetOnRunAttemptPublisher (StatusUpdatePeriod)

EventResult flags: - ResetRetryBackoff - Set after a sync succeeds or the source changed (spec or commit). - DelayStatusUpdate - Set after a sync is attempted.

PublishingGroupBuilder can be used to Build a set of the above Publishers.

Index

• type Event
• type EventType
• type Funnel
  • func (f *Funnel) Start(ctx context.Context) <-chan struct{}
• type Publisher
• type PublishingGroupBuilder
  • func (t *PublishingGroupBuilder) Build() []Publisher
• type ResetOnRunAttemptPublisher
  • func NewResetOnRunAttemptPublisher(eventType EventType, c clock.Clock, period time.Duration) *ResetOnRunAttemptPublisher
  • func (s *ResetOnRunAttemptPublisher) HandleResult(result Result)
• type Result
• type RetrySyncPublisher
  • func NewRetrySyncPublisher(c clock.Clock, backoff wait.Backoff) *RetrySyncPublisher
  • func (s *RetrySyncPublisher) HandleResult(result Result)
  • func (s *RetrySyncPublisher) Publish(subscriber Subscriber) Result
  • func (s *RetrySyncPublisher) Start(ctx context.Context) reflect.Value
  • func (s *RetrySyncPublisher) Type() EventType
• type Subscriber
• type TimeDelayPublisher
  • func NewTimeDelayPublisher(eventType EventType, c clock.Clock, period time.Duration) *TimeDelayPublisher
  • func (s *TimeDelayPublisher) HandleResult(_ Result)
  • func (s *TimeDelayPublisher) Publish(subscriber Subscriber) Result
  • func (s *TimeDelayPublisher) Start(ctx context.Context) reflect.Value
  • func (s *TimeDelayPublisher) Type() EventType

Types

type Event

type Event struct {
	// Type of the event
	Type EventType
}

Event represents the cause of HandleEventFunc being called. Some event types may have special fields or methods.

type EventType

type EventType string

EventType is the name of the event

const (
	// SyncWithReimportEventType is the EventType for a sync from disk,
	// including reading and parsing resource objects from the shared volume.
	SyncWithReimportEventType EventType = "SyncWithReimportEvent"
	// SyncEventType is the EventType for a sync from the source cache,
	// only parsing objects from the shared volume if there's a new commit.
	SyncEventType EventType = "SyncEvent"
	// StatusEventType is the EventType for a periodic RSync status update.
	StatusEventType EventType = "StatusEvent"
	// NamespaceResyncEventType is the EventType for a sync triggered by an
	// update to a selected namespace.
	NamespaceResyncEventType EventType = "NamespaceResyncEvent"
	// RetrySyncEventType is the EventType for a sync triggered by an error
	// during a previous sync attempt.
	RetrySyncEventType EventType = "RetrySyncEvent"
)

type Funnel

type Funnel struct {
	Publishers []Publisher
	Subscriber Subscriber
}

Funnel events from multiple publishers to a single subscriber.

func (*Funnel) Start

func (f *Funnel) Start(ctx context.Context) <-chan struct{}

Start all the event publishers and sends events one at a time to the subscriber. Blocks until the publishers are started. Returns a channel that will be closed when the Funnel is done handling events. Stops when the context is done.

type Publisher

type Publisher interface {
	// Type of events published by this Publisher.
	Type() EventType
	// Start the event publisher and return the event channel.
	// Messages will be sent to the channel with an optional typed payload.
	// Channel will be closed when the context is done.
	Start(context.Context) reflect.Value
	// Publish an event to a specific Subscriber, and return the Result.
	// Publish should be called when the event channel sends an event.
	Publish(Subscriber) Result
	// HandleResult handles the result of all types of events.
	// This allows a Publisher to handle signals from other Publishers.
	HandleResult(Result)
}

Publisher writes events to a channel and defines how to handle them.

type PublishingGroupBuilder

type PublishingGroupBuilder struct {
	// Clock is used for time tracking, namely to simplify testing by allowing
	// a fake clock, instead of a RealClock.
	Clock clock.Clock
	// SyncPeriod is the period of time between checking the filesystem
	// for publisher updates to sync.
	SyncPeriod time.Duration
	// SyncWithReimportPeriod is the period of time between forced re-sync from
	// publisher (even without a new commit).
	SyncWithReimportPeriod time.Duration
	// StatusUpdatePeriod is how long the Parser waits between updates of the
	// sync status, to account for management conflict errors from the Remediator.
	StatusUpdatePeriod time.Duration
	// NamespaceControllerPeriod is how long to wait between checks to see if
	// the namespace-controller wants to trigger a resync.
	// TODO: Use a channel, instead of a timer checking a locked variable.
	NamespaceControllerPeriod time.Duration
	// RetryBackoff is how long the Parser waits between retries, after an error.
	RetryBackoff wait.Backoff
}

PublishingGroupBuilder oversees construction of event publishers.

For now, the publishers are driven by clock-based delay and backoff timers.

func (*PublishingGroupBuilder) Build

func (t *PublishingGroupBuilder) Build() []Publisher

Build a list of Publishers based on the PublishingGroupBuilder config.

type ResetOnRunAttemptPublisher

type ResetOnRunAttemptPublisher struct {
	TimeDelayPublisher
}

ResetOnRunAttemptPublisher is a TimeDelayPublisher that is automatically delayed when Result.RunAttempted is set by any event.

func NewResetOnRunAttemptPublisher

func NewResetOnRunAttemptPublisher(eventType EventType, c clock.Clock, period time.Duration) *ResetOnRunAttemptPublisher

NewResetOnRunAttemptPublisher constructs an ResetOnRunPublisher that generates and handles the specified events and resets the delay any time Result.RunAttempted=true.

func (*ResetOnRunAttemptPublisher) HandleResult

func (s *ResetOnRunAttemptPublisher) HandleResult(result Result)

HandleResult resets the delay timer if DelayStatusUpdate is true.

type Result

type Result struct {
	// RunAttempted triggers timer reset of any events using the
	// ResetOnRunAttemptPublisher.
	RunAttempted bool
	// ResetRetryBackoff triggers backoff reset for the RetrySyncPublisher.
	ResetRetryBackoff bool
}

Result encapsulates the result of a ConsumeFunc. This simply allows explicitly naming return values in a way that makes the implementation easier to read.

type RetrySyncPublisher

type RetrySyncPublisher struct {
	EventType EventType
	Clock     clock.Clock
	Backoff   wait.Backoff
	// contains filtered or unexported fields
}

RetrySyncPublisher sends StatusEvents periodically using a backoff timer that is incremented after each event is handled.

Unlike, TimeDelayPublisher, RetrySyncPublisher always increases the delay, unless the subscriber sets Result.ResetRetryBackoff.

func NewRetrySyncPublisher

func NewRetrySyncPublisher(c clock.Clock, backoff wait.Backoff) *RetrySyncPublisher

NewRetrySyncPublisher constructs an RetrySyncPublisher that generates and handles RetrySyncEvents with retry backoff.

func (*RetrySyncPublisher) HandleResult

func (s *RetrySyncPublisher) HandleResult(result Result)

HandleResult resets the backoff timer if ResetRetryBackoff is true.

func (*RetrySyncPublisher) Publish

func (s *RetrySyncPublisher) Publish(subscriber Subscriber) Result

Publish calls the HandleFunc with a new event, increments the backoff step, and updates the delay timer. If the maximum number of retries has been reached, the HandleFunc is NOT called and an empty Result is returned.

func (*RetrySyncPublisher) Start

func (s *RetrySyncPublisher) Start(ctx context.Context) reflect.Value

Start the timer and return the event channel.

func (*RetrySyncPublisher) Type

func (s *RetrySyncPublisher) Type() EventType

Type of events produced by this publisher.

type Subscriber

type Subscriber interface {
	// Handle an event and return the result.
	Handle(Event) Result
}

Subscriber handles an Event and returns the result.

type TimeDelayPublisher

type TimeDelayPublisher struct {
	EventType EventType
	Clock     clock.Clock
	Period    time.Duration
	// contains filtered or unexported fields
}

TimeDelayPublisher sends events periodically using a timer that is reset after each event is handled. This avoids unhandled events stacking up waiting to be handled.

func NewTimeDelayPublisher

func NewTimeDelayPublisher(eventType EventType, c clock.Clock, period time.Duration) *TimeDelayPublisher

NewTimeDelayPublisher constructs an TimeDelayPublisher that generates and handles the specified events.

func (*TimeDelayPublisher) HandleResult

func (s *TimeDelayPublisher) HandleResult(_ Result)

HandleResult is a no-op.

func (*TimeDelayPublisher) Publish

func (s *TimeDelayPublisher) Publish(subscriber Subscriber) Result

Publish calls the HandleFunc with a new event and resets the delay timer.

func (*TimeDelayPublisher) Start

func (s *TimeDelayPublisher) Start(ctx context.Context) reflect.Value

Start the timer and return the event channel.

func (*TimeDelayPublisher) Type

func (s *TimeDelayPublisher) Type() EventType

Type of events produced by this publisher.