util

Documentation

Index

• func DeregisterRateCollector(label string)
• func RegisterRateCollector(label string)
• func StartTracer(msg msgstream.TsMsg, name string) (context.Context, trace.Span)
• type ChannelCheckpointUpdater
  • func NewChannelCheckpointUpdater(broker broker.Broker) *ChannelCheckpointUpdater
  • func (ccu *ChannelCheckpointUpdater) AddTask(channelPos *msgpb.MsgPosition, flush bool, callback func())
  • func (ccu *ChannelCheckpointUpdater) Close()
  • func (ccu *ChannelCheckpointUpdater) Start()
• type PipelineParams
• type RateCollector
  • func GetRateCollector() *RateCollector
  • func (r *RateCollector) GetMinFlowGraphTt() (string, typeutil.Timestamp)
  • func (r *RateCollector) RemoveFlowGraphChannel(channel string)
  • func (r *RateCollector) UpdateFlowGraphTt(channel string, t typeutil.Timestamp)
• type StatsUpdater
• type Tickler
  • func NewTickler() *Tickler
  • func (t *Tickler) Close()
  • func (t *Tickler) GetProgressSig() chan struct{}
  • func (t *Tickler) Inc()
  • func (t *Tickler) IsClosed() bool
  • func (t *Tickler) Progress() int32
  • func (t *Tickler) SetTotal(total int32)
• type TimeRange
• type TimeTickSender
  • func NewTimeTickSender(broker broker.Broker, nodeID int64, opts ...retry.Option) *TimeTickSender
  • func (m *TimeTickSender) Start()
  • func (m *TimeTickSender) Stop()
  • func (m *TimeTickSender) Update(channelName string, timestamp uint64, segStats []*commonpb.SegmentStats)

Functions

func DeregisterRateCollector

func DeregisterRateCollector(label string)

func RegisterRateCollector

func RegisterRateCollector(label string)

func StartTracer

func StartTracer(msg msgstream.TsMsg, name string) (context.Context, trace.Span)

Types

type ChannelCheckpointUpdater

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

func NewChannelCheckpointUpdater

func NewChannelCheckpointUpdater(broker broker.Broker) *ChannelCheckpointUpdater

func (*ChannelCheckpointUpdater) AddTask

func (ccu *ChannelCheckpointUpdater) AddTask(channelPos *msgpb.MsgPosition, flush bool, callback func())

func (*ChannelCheckpointUpdater) Close

func (ccu *ChannelCheckpointUpdater) Close()

func (*ChannelCheckpointUpdater) Start

func (ccu *ChannelCheckpointUpdater) Start()

type PipelineParams

type PipelineParams struct {
	Ctx                context.Context
	Broker             broker.Broker
	SyncMgr            syncmgr.SyncManager
	TimeTickSender     *TimeTickSender     // reference to TimeTickSender
	CompactionExecutor compaction.Executor // reference to compaction executor
	MsgStreamFactory   dependency.Factory
	DispClient         msgdispatcher.Client
	ChunkManager       storage.ChunkManager
	Session            *sessionutil.Session
	WriteBufferManager writebuffer.BufferManager
	CheckpointUpdater  *ChannelCheckpointUpdater
	Allocator          allocator.Interface
	FlushMsgHandler    flusher.FlushMsgHandler
}

type RateCollector

type RateCollector struct {
	*ratelimitutil.RateCollector
	// contains filtered or unexported fields
}

RateCollector helps to collect and calculate values (like rate, timeTick and etc...).

func GetRateCollector

func GetRateCollector() *RateCollector

func (*RateCollector) GetMinFlowGraphTt

func (r *RateCollector) GetMinFlowGraphTt() (string, typeutil.Timestamp)

GetMinFlowGraphTt returns the vchannel and minimal time tick of flow graphs.

func (*RateCollector) RemoveFlowGraphChannel

func (r *RateCollector) RemoveFlowGraphChannel(channel string)

RemoveFlowGraphChannel removes channel from flowGraphTt.

func (*RateCollector) UpdateFlowGraphTt

func (r *RateCollector) UpdateFlowGraphTt(channel string, t typeutil.Timestamp)

UpdateFlowGraphTt updates RateCollector's flow graph time tick.

type StatsUpdater

type StatsUpdater interface {
	Update(channel string, ts typeutil.Timestamp, stats []*commonpb.SegmentStats)
}

type Tickler

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

Tickler counts every time when called inc(),

func NewTickler

func NewTickler() *Tickler

func (*Tickler) Close

func (t *Tickler) Close()

func (*Tickler) GetProgressSig

func (t *Tickler) GetProgressSig() chan struct{}

func (*Tickler) Inc

func (t *Tickler) Inc()

func (*Tickler) IsClosed

func (t *Tickler) IsClosed() bool

func (*Tickler) Progress

func (t *Tickler) Progress() int32

progress returns the count over total if total is set else just return the count number.

func (*Tickler) SetTotal

func (t *Tickler) SetTotal(total int32)

type TimeRange

type TimeRange struct {
	TimestampMin typeutil.Timestamp
	TimestampMax typeutil.Timestamp
}

TimeRange is a range of timestamp contains the min-timestamp and max-timestamp

type TimeTickSender

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

TimeTickSender is to merge channel states updated by flow graph node and send to datacoord periodically TimeTickSender hold segmentStats cache for each channel, after send succeeds will clean the cache earlier than last sent timestamp

func NewTimeTickSender

func NewTimeTickSender(broker broker.Broker, nodeID int64, opts ...retry.Option) *TimeTickSender

func (*TimeTickSender) Start

func (m *TimeTickSender) Start()

func (*TimeTickSender) Stop

func (m *TimeTickSender) Stop()

func (*TimeTickSender) Update

func (m *TimeTickSender) Update(channelName string, timestamp uint64, segStats []*commonpb.SegmentStats)