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, reducer Reducer) *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)
• type CoreOptions
• type Measure
  • func (m Measure) MarshalJSON() ([]byte, error)
  • func (m *Measure) UnmarshalJSON(raw []byte) error
• type Metric
  • func NewMetric(name MetricName, dimensions map[string]string, ts time.Time, value float64) *Metric
  • func (m *Metric) Matches(name MetricName, odim map[string]string) bool
• type MetricName
• type MetricsMap
  • func (m MetricsMap) Add(metric *Metric) MetricsMap
  • func (m MetricsMap) GetMetric(name MetricName, dimensions map[string]string) *Metric
• 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) MetricsCopy() MetricsMap
  • 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      *data.UnixMillisTime `json:"lastProbeTime"`
	LastTransitionTime *data.UnixMillisTime `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, reducer Reducer) *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)

type CoreOptions

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

type Measure

type Measure struct {
	Timestamp time.Time
	Value     float64
}

func (Measure) MarshalJSON

func (m Measure) MarshalJSON() ([]byte, error)

func (*Measure) UnmarshalJSON

func (m *Measure) UnmarshalJSON(raw []byte) error

type Metric

type Metric struct {
	Name       MetricName        `json:"name"`
	Dimensions map[string]string `json:"dimensions,omitempty"`
	Last       Measure           `json:"last"`
}

func NewMetric

func NewMetric(name MetricName, dimensions map[string]string, ts time.Time, value float64) *Metric

func (*Metric) Matches

func (m *Metric) Matches(name MetricName, odim map[string]string) bool

type MetricName

type MetricName string

type MetricsMap

type MetricsMap map[MetricName][]*Metric

func (MetricsMap) Add

func (m MetricsMap) Add(metric *Metric) MetricsMap

func (MetricsMap) GetMetric

func (m MetricsMap) GetMetric(name MetricName, dimensions map[string]string) *Metric

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

	ReducerState 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"`
	Metrics    MetricsMap   `json:"metrics,omitempty"`
	// TODO: Move this `multistream` field somewhere else to make this struct more
	// generic. Maybe condition dimensions/extraArgs?
	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) MetricsCopy

func (s Status) MetricsCopy() MetricsMap

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.