stats

Documentation

Index

• func NewProtoFromSegmentStat(stat *SegmentStats) *streamingpb.SegmentAssignmentStat
• type FlushOperationMetrics
• type InsertMetrics
  • func (m *InsertMetrics) Collect(other InsertMetrics)
  • func (m *InsertMetrics) Subtract(other InsertMetrics)
• type SealSignalNotifier
  • func NewSealSignalNotifier() *SealSignalNotifier
  • func (n *SealSignalNotifier) AddAndNotify(belongs SegmentBelongs)
  • func (n *SealSignalNotifier) Get() typeutil.Set[SegmentBelongs]
  • func (n *SealSignalNotifier) WaitChan() <-chan struct{}
• type SegmentBelongs
• type SegmentStats
  • func NewSegmentStatFromProto(statProto *streamingpb.SegmentAssignmentStat) *SegmentStats
  • func (s *SegmentStats) AllocRows(m InsertMetrics) bool
  • func (s *SegmentStats) BinaryCanBeAssign() uint64
  • func (s *SegmentStats) Copy() *SegmentStats
  • func (s *SegmentStats) UpdateOnFlush(f FlushOperationMetrics)
• type StatsManager
  • func NewStatsManager() *StatsManager
  • func (m *StatsManager) AllocRows(segmentID int64, insert InsertMetrics) bool
  • func (m *StatsManager) GetStatsOfSegment(segmentID int64) *SegmentStats
  • func (m *StatsManager) RegisterNewGrowingSegment(belongs SegmentBelongs, segmentID int64, stats *SegmentStats)
  • func (m *StatsManager) SealNotifier() *SealSignalNotifier
  • func (m *StatsManager) UnregisterSealedSegment(segmentID int64) *SegmentStats
  • func (m *StatsManager) UpdateOnFlush(segmentID int64, flush FlushOperationMetrics)

Functions

func NewProtoFromSegmentStat

func NewProtoFromSegmentStat(stat *SegmentStats) *streamingpb.SegmentAssignmentStat

NewProtoFromSegmentStat creates a new proto from segment assignment stat.

Types

type FlushOperationMetrics

type FlushOperationMetrics struct {
	BinLogCounter uint64
}

FlushOperationMetrics is the metrics of flush operation.

type InsertMetrics

type InsertMetrics struct {
	Rows       uint64
	BinarySize uint64
}

InsertOpeatationMetrics is the metrics of insert operation.

func (*InsertMetrics) Collect

func (m *InsertMetrics) Collect(other InsertMetrics)

Collect collects other metrics.

func (*InsertMetrics) Subtract

func (m *InsertMetrics) Subtract(other InsertMetrics)

Subtract subtract by other metrics.

type SealSignalNotifier

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

SealSignalNotifier is a notifier for seal signal.

func NewSealSignalNotifier

func NewSealSignalNotifier() *SealSignalNotifier

NewSealSignalNotifier creates a new seal signal notifier.

func (*SealSignalNotifier) AddAndNotify

func (n *SealSignalNotifier) AddAndNotify(belongs SegmentBelongs)

AddAndNotify adds a signal and notifies the waiter.

func (*SealSignalNotifier) Get

func (n *SealSignalNotifier) Get() typeutil.Set[SegmentBelongs]

Get gets the signal.

func (*SealSignalNotifier) WaitChan

func (n *SealSignalNotifier) WaitChan() <-chan struct{}

type SegmentBelongs

type SegmentBelongs struct {
	PChannel     string
	VChannel     string
	CollectionID int64
	PartitionID  int64
}

type SegmentStats

type SegmentStats struct {
	Insert           InsertMetrics
	MaxBinarySize    uint64    // MaxBinarySize of current segment should be assigned, it's a fixed value when segment is transfer int growing.
	CreateTime       time.Time // created timestamp of this segment, it's a fixed value when segment is created, not a tso.
	LastModifiedTime time.Time // LastWriteTime is the last write time of this segment, it's not a tso, just a local time.
	BinLogCounter    uint64    // BinLogCounter is the counter of binlog, it's an async stat not real time.
	ReachLimit       bool      // ReachLimit is a flag to indicate the segment reach the limit once.
}

SegmentStats is the usage stats of a segment.

func NewSegmentStatFromProto

func NewSegmentStatFromProto(statProto *streamingpb.SegmentAssignmentStat) *SegmentStats

NewSegmentStatFromProto creates a new segment assignment stat from proto.

func (*SegmentStats) AllocRows

func (s *SegmentStats) AllocRows(m InsertMetrics) bool

AllocRows alloc space of rows on current segment. Return true if the segment is assigned.

func (*SegmentStats) BinaryCanBeAssign

func (s *SegmentStats) BinaryCanBeAssign() uint64

BinaryCanBeAssign returns the capacity of binary size can be inserted.

func (*SegmentStats) Copy

func (s *SegmentStats) Copy() *SegmentStats

Copy copies the segment stats.

func (*SegmentStats) UpdateOnFlush

func (s *SegmentStats) UpdateOnFlush(f FlushOperationMetrics)

UpdateOnFlush updates the stats of segment on flush.

type StatsManager

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

StatsManager is the manager of stats. It manages the insert stats of all segments, used to check if a segment has enough space to insert or should be sealed. If there will be a lock contention, we can optimize it by apply lock per segment.

func NewStatsManager

func NewStatsManager() *StatsManager

NewStatsManager creates a new stats manager.

func (*StatsManager) AllocRows

func (m *StatsManager) AllocRows(segmentID int64, insert InsertMetrics) bool

AllocRows alloc number of rows on current segment.

func (*StatsManager) GetStatsOfSegment

func (m *StatsManager) GetStatsOfSegment(segmentID int64) *SegmentStats

GetStatsOfSegment gets the stats of segment.

func (*StatsManager) RegisterNewGrowingSegment

func (m *StatsManager) RegisterNewGrowingSegment(belongs SegmentBelongs, segmentID int64, stats *SegmentStats)

RegisterNewGrowingSegment registers a new growing segment. delegate the stats management to stats manager.

func (*StatsManager) SealNotifier

func (m *StatsManager) SealNotifier() *SealSignalNotifier

SealNotifier returns the seal notifier.

func (*StatsManager) UnregisterSealedSegment

func (m *StatsManager) UnregisterSealedSegment(segmentID int64) *SegmentStats

UnregisterSealedSegment unregisters the sealed segment.

func (*StatsManager) UpdateOnFlush

func (m *StatsManager) UpdateOnFlush(segmentID int64, flush FlushOperationMetrics)

UpdateOnFlush updates the stats of segment on flush. It's an async update operation, so it's not necessary to do success.