health

Documentation

Index

• Variables
• type Condition
  • func NewCondition(condType ConditionType, ts time.Time, status *bool, frequency stats.ByWindow, ...) *Condition
• type ConditionType
• type Core
  • func NewCore(opts CoreOptions, consumer event.StreamConsumer) *Core
  • func (c *Core) GetPastEvents(manifestID string, from, to *time.Time) ([]data.Event, error)
  • func (c *Core) GetStatus(manifestID string) (Status, error)
  • func (c *Core) HandleMessage(msg event.StreamMessage)
  • func (c *Core) IsHealthy() bool
  • func (c *Core) Start(ctx context.Context) error
  • func (c *Core) SubscribeEvents(ctx context.Context, manifestID string, lastEvtID *uuid.UUID, from *time.Time) ([]data.Event, <-chan data.Event, error)
  • func (c *Core) Use(reducers ...Reducer) *Core
• type CoreOptions
• type Record
  • func NewRecord(id string, conditions []ConditionType) *Record
  • func (r *Record) SubscribeLocked(ctx context.Context, subs chan data.Event) chan data.Event
• type RecordStorage
  • func (s *RecordStorage) Get(id string) (*Record, bool)
  • func (s *RecordStorage) GetOrCreate(id string, conditions []ConditionType) *Record
  • func (s *RecordStorage) StartCleanupLoop(ctx context.Context, ttl time.Duration)
• type Reducer
• type ReducerFunc
  • func (f ReducerFunc) Bindings() []event.BindingArgs
  • func (f ReducerFunc) Conditions() []ConditionType
  • func (f ReducerFunc) Reduce(current Status, state interface{}, evt data.Event) (Status, interface{})
• type Status
  • func (s Status) GetCondition(condType ConditionType) *Condition
• type StreamingOptions

Variables

var ErrEventNotFound = errors.New("event not found")

var ErrStreamNotFound = errors.New("stream not found")

Types

type Condition

type Condition struct {
	Type               ConditionType  `json:"type,omitempty"`
	Status             *bool          `json:"status"`
	Frequency          stats.ByWindow `json:"frequency,omitempty"`
	LastProbeTime      *time.Time     `json:"lastProbeTime"`
	LastTransitionTime *time.Time     `json:"lastTransitionsTime"`
}

func NewCondition

func NewCondition(condType ConditionType, ts time.Time, status *bool, frequency stats.ByWindow, last *Condition) *Condition

type ConditionType

type ConditionType string

type Core

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

func NewCore

func NewCore(opts CoreOptions, consumer event.StreamConsumer) *Core

func (*Core) GetPastEvents

func (c *Core) GetPastEvents(manifestID string, from, to *time.Time) ([]data.Event, error)

func (*Core) GetStatus

func (c *Core) GetStatus(manifestID string) (Status, error)

func (*Core) HandleMessage

func (c *Core) HandleMessage(msg event.StreamMessage)

func (*Core) IsHealthy

func (c *Core) IsHealthy() bool

func (*Core) Start

func (c *Core) Start(ctx context.Context) error

func (*Core) SubscribeEvents

func (c *Core) SubscribeEvents(ctx context.Context, manifestID string, lastEvtID *uuid.UUID, from *time.Time) ([]data.Event, <-chan data.Event, error)

func (*Core) Use

func (c *Core) Use(reducers ...Reducer) *Core

type CoreOptions

type CoreOptions struct {
	Streaming        StreamingOptions
	StartTimeOffset  time.Duration
	MemoryRecordsTtl time.Duration
}

type Record

type Record struct {
	ID         string
	Conditions []ConditionType

	sync.RWMutex

	PastEvents []data.Event
	EventsByID map[uuid.UUID]data.Event
	EventSubs  []chan<- data.Event

	ReducersState map[int]interface{}
	LastStatus    Status
	// contains filtered or unexported fields
}

func NewRecord

func NewRecord(id string, conditions []ConditionType) *Record

func (*Record) SubscribeLocked

func (r *Record) SubscribeLocked(ctx context.Context, subs chan data.Event) chan data.Event

type RecordStorage

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

func (*RecordStorage) Get

func (s *RecordStorage) Get(id string) (*Record, bool)

func (*RecordStorage) GetOrCreate

func (s *RecordStorage) GetOrCreate(id string, conditions []ConditionType) *Record

func (*RecordStorage) StartCleanupLoop

func (s *RecordStorage) StartCleanupLoop(ctx context.Context, ttl time.Duration)

type Reducer

type Reducer interface {
	Bindings() []event.BindingArgs
	Conditions() []ConditionType
	Reduce(current Status, state interface{}, evt data.Event) (Status, interface{})
}

type ReducerFunc

type ReducerFunc func(Status, interface{}, data.Event) (Status, interface{})

func (ReducerFunc) Bindings

func (f ReducerFunc) Bindings() []event.BindingArgs

func (ReducerFunc) Conditions

func (f ReducerFunc) Conditions() []ConditionType

func (ReducerFunc) Reduce

func (f ReducerFunc) Reduce(current Status, state interface{}, evt data.Event) (Status, interface{})

type Status

type Status struct {
	ID         string       `json:"id"`
	Healthy    Condition    `json:"healthy"`
	Conditions []*Condition `json:"conditions"`
}

func (Status) GetCondition

func (s Status) GetCondition(condType ConditionType) *Condition

type StreamingOptions

type StreamingOptions struct {
	Stream, ConsumerName string

	event.RawStreamOptions
}

Purposedly made of built-in types only to bind directly to cli flags.