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 MultistreamStatus
• 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 NewMergedStatus(base *Status, values Status) *Status
  • func (s Status) ConditionsCopy() []*Condition
  • func (s Status) GetCondition(condType ConditionType) *Condition
  • func (s Status) MultistreamCopy() []*MultistreamStatus
• 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 MultistreamStatus

type MultistreamStatus struct {
	Target    data.MultistreamTargetInfo `json:"target"`
	Connected *Condition                 `json:"connected"`
}

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"`
	Multistream []*MultistreamStatus `json:"multistream,omitempty"`
}

Status is a soft-immutable struct. It should never be modified inline or a lot of things could become inconsistent. Create a new instance/copy for any mutation to be performed and beware of the internal slices and pointers.

Use NewMergedStatus below to facilitate the creation of new status objects with mutated fields. Notice that you still need to clone the internal slices if you want to do any mutations to them.

func NewMergedStatus

func NewMergedStatus(base *Status, values Status) *Status

func (Status) ConditionsCopy

func (s Status) ConditionsCopy() []*Condition

func (Status) GetCondition

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

func (Status) MultistreamCopy

func (s Status) MultistreamCopy() []*MultistreamStatus

type StreamingOptions

type StreamingOptions struct {
	Stream, ConsumerName string

	event.RawStreamOptions
}

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