providers

Documentation

Index

• func DumpSyncState(n int32, prov MetricsProvider) map[string]konsumeratorv1.InstanceState
• func LoadSyncState(mp MetricsProvider, status konsumeratorv1.ConsumerStatus)
• type DummyMP
  • func NewDummyMP(numPartitions int32) *DummyMP
  • func (l *DummyMP) GetConsumptionRate(partition int32) int64
  • func (l *DummyMP) GetLagByPartition(partition int32) time.Duration
  • func (l *DummyMP) GetMessagesBehind(partition int32) int64
  • func (l *DummyMP) GetProductionRate(partition int32) int64
  • func (l *DummyMP) Load(production map[int32]int64, consumption map[int32]int64, ...)
  • func (l *DummyMP) Update() error
• type MetricsProvider
• type PrometheusMP
  • func NewPrometheusMP(log logr.Logger, spec *konsumeratorv1.PrometheusAutoscalerSpec, ...) (*PrometheusMP, error)
  • func (l *PrometheusMP) GetConsumptionRate(partition int32) int64
  • func (l *PrometheusMP) GetLagByPartition(partition int32) time.Duration
  • func (l *PrometheusMP) GetMessagesBehind(partition int32) int64
  • func (l *PrometheusMP) GetProductionRate(partition int32) int64
  • func (l *PrometheusMP) Load(production, consumption, offset map[int32]int64)
  • func (l *PrometheusMP) Update() error

Functions

func DumpSyncState

func DumpSyncState(n int32, prov MetricsProvider) map[string]konsumeratorv1.InstanceState

func LoadSyncState

func LoadSyncState(mp MetricsProvider, status konsumeratorv1.ConsumerStatus)

Types

type DummyMP

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

func NewDummyMP

func NewDummyMP(numPartitions int32) *DummyMP

func (*DummyMP) GetConsumptionRate

func (l *DummyMP) GetConsumptionRate(partition int32) int64

func (*DummyMP) GetLagByPartition

func (l *DummyMP) GetLagByPartition(partition int32) time.Duration

func (*DummyMP) GetMessagesBehind

func (l *DummyMP) GetMessagesBehind(partition int32) int64

func (*DummyMP) GetProductionRate

func (l *DummyMP) GetProductionRate(partition int32) int64

func (*DummyMP) Load

func (l *DummyMP) Load(production map[int32]int64, consumption map[int32]int64, offset map[int32]int64)

func (*DummyMP) Update

func (l *DummyMP) Update() error

type MetricsProvider

type MetricsProvider interface {
	GetProductionRate(int32) int64
	GetConsumptionRate(int32) int64
	GetMessagesBehind(int32) int64
	GetLagByPartition(int32) time.Duration
	Update() error
	Load(map[int32]int64, map[int32]int64, map[int32]int64)
}

type PrometheusMP

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

func NewPrometheusMP

func NewPrometheusMP(log logr.Logger, spec *konsumeratorv1.PrometheusAutoscalerSpec, consumer string) (*PrometheusMP, error)

func (*PrometheusMP) GetConsumptionRate

func (l *PrometheusMP) GetConsumptionRate(partition int32) int64

GetConsumptionRate returns consumption rate. not thread-safe

func (*PrometheusMP) GetLagByPartition

func (l *PrometheusMP) GetLagByPartition(partition int32) time.Duration

GetLagByPartition calculates lag based on ProductionRate, ConsumptionRate and the number of not processed messages for partition not thread-safe

func (*PrometheusMP) GetMessagesBehind

func (l *PrometheusMP) GetMessagesBehind(partition int32) int64

GetMessagesBehind returns how many messages we're behind. not thread-safe

func (*PrometheusMP) GetProductionRate

func (l *PrometheusMP) GetProductionRate(partition int32) int64

GetProductionRate returns production rate. not thread-safe

func (*PrometheusMP) Load

func (l *PrometheusMP) Load(production, consumption, offset map[int32]int64)

Load loads given metrics into object not thread-safe

func (*PrometheusMP) Update

func (l *PrometheusMP) Update() error

Update updates metrics values by querying Prometheus not thread-safe TODO: may lead to partial update TODO: might be queried in parallel