stats

Documentation

Index

• Variables
• func NewProtoFromSegmentStat(stat *SegmentStats) *streamingpb.SegmentAssignmentStat
• 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) IsEmpty() bool
  • func (s *SegmentStats) ShouldBeSealed() bool
  • func (s *SegmentStats) UpdateOnSync(f SyncOperationMetrics)
• type StatsManager
  • func NewStatsManager() *StatsManager
  • func (m *StatsManager) AllocRows(segmentID int64, insert InsertMetrics) error
  • func (m *StatsManager) GetStatsOfSegment(segmentID int64) *SegmentStats
  • func (m *StatsManager) RegisterNewGrowingSegment(belongs SegmentBelongs, segmentID int64, stats *SegmentStats)
  • func (m *StatsManager) SealByTotalGrowingSegmentsSize() SegmentBelongs
  • func (m *StatsManager) SealNotifier() *SealSignalNotifier
  • func (m *StatsManager) UnregisterSealedSegment(segmentID int64) *SegmentStats
  • func (m *StatsManager) UpdateOnSync(segmentID int64, syncMetric SyncOperationMetrics)
• type SyncOperationMetrics

Variables

var (
	ErrNotEnoughSpace = errors.New("not enough space")
	ErrTooLargeInsert = errors.New("insert too large")
)

Functions

func NewProtoFromSegmentStat

func NewProtoFromSegmentStat(stat *SegmentStats) *streamingpb.SegmentAssignmentStat

NewProtoFromSegmentStat creates a new proto from segment assignment stat.

Types

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
	SegmentID    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 (equal to the binlog file count of primary key), it's an async stat not real time.
	BinLogFileCounter uint64    // BinLogFileCounter is the counter of binlog files, 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. The SegmentStats is imprecise, so it is not promised to be recoverable for performance.

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) IsEmpty

func (s *SegmentStats) IsEmpty() bool

IsEmpty returns if the segment is empty.

func (*SegmentStats) ShouldBeSealed

func (s *SegmentStats) ShouldBeSealed() bool

ShouldBeSealed returns if the segment should be sealed.

func (*SegmentStats) UpdateOnSync

func (s *SegmentStats) UpdateOnSync(f SyncOperationMetrics)

UpdateOnSync updates the stats of segment on sync.

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) error

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) SealByTotalGrowingSegmentsSize

func (m *StatsManager) SealByTotalGrowingSegmentsSize() SegmentBelongs

SealByTotalGrowingSegmentsSize seals the largest growing segment if the total size of growing segments in ANY vchannel exceeds the threshold.

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) UpdateOnSync

func (m *StatsManager) UpdateOnSync(segmentID int64, syncMetric SyncOperationMetrics)

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

type SyncOperationMetrics

type SyncOperationMetrics struct {
	BinLogCounterIncr     uint64 // the counter increment of bin log
	BinLogFileCounterIncr uint64 // the counter increment of bin log file
}

SyncOperationMetrics is the metrics of sync operation.