controller

Documentation

Index

• Variables
• type Controller
  • func NewController(cfg configuration.Controller, extractor interfaces.JetDropsExtractor, ...) (*Controller, error)
  • func (c *Controller) SetJetDropData(pulse types.Pulse, jetID string)
  • func (c *Controller) Start(ctx context.Context) error
  • func (c *Controller) Stop(ctx context.Context) error
• type Metrics
  • func (s Metrics) Metrics(p *metrics.Prometheus) []prometheus.Collector
  • func (s Metrics) Refresh()
• type MissedDataManager
  • func NewMissedDataManager(ttl time.Duration, cleanPeriod time.Duration) *MissedDataManager
  • func (mdm *MissedDataManager) Add(ctx context.Context, fromPulse, toPulse int64) bool
  • func (mdm *MissedDataManager) Start()
  • func (mdm *MissedDataManager) Stop()

Variables

var (
	IncompletePulsesQueue = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "gbe_controller_data_queue",
		Help: "The number of pulses in controller's incomplete pulses queue",
	})
	CurrentSeqPulse = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "gbe_controller_current_seq_pulse",
		Help: "Current sequentual pulse rerequested from platform",
	})
	PulseCompleteCounter = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "gbe_controller_pulse_complete_counter",
		Help: "How many pulses is completed by 'pulseIsComplete' check",
	})
	PulseNotCompleteCounter = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "gbe_controller_pulse_not_complete_counter",
		Help: "How many pulses is not completed by 'pulseIsComplete' check",
	})
)

Types

type Controller

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

Controller checks pulses completeness and sends signal to reload missing data from platform

func NewController

func NewController(cfg configuration.Controller, extractor interfaces.JetDropsExtractor, storage interfaces.Storage, pv int) (*Controller, error)

NewController returns implementation of interfaces.Controller

func (*Controller) SetJetDropData

func (c *Controller) SetJetDropData(pulse types.Pulse, jetID string)

SetJetDropData stores jetID, processed at specific pulse

func (*Controller) Start

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

Start implements interfaces.Starter

func (*Controller) Stop

func (c *Controller) Stop(ctx context.Context) error

Stop implements interfaces.Stopper

type Metrics

type Metrics struct{}

func (Metrics) Metrics

func (s Metrics) Metrics(p *metrics.Prometheus) []prometheus.Collector

func (Metrics) Refresh

func (s Metrics) Refresh()

type MissedDataManager

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

MissedDataManager manages working with missed data pool It's thread safe

func NewMissedDataManager

func NewMissedDataManager(ttl time.Duration, cleanPeriod time.Duration) *MissedDataManager

NewMissedDataManager creates new missed data manager with custom params

func (*MissedDataManager) Add

func (mdm *MissedDataManager) Add(ctx context.Context, fromPulse, toPulse int64) bool

Add adds missed data to pool

func (*MissedDataManager) Start

func (mdm *MissedDataManager) Start()

func (*MissedDataManager) Stop

func (mdm *MissedDataManager) Stop()