util

Documentation

Index

• func ConvertInternalImportFile(file *msgpb.ImportFile, _ int) *internalpb.ImportFile
• 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 NewChannelCheckpointUpdaterWithCallback(broker broker.Broker, updateDoneCallback func(*msgpb.MsgPosition)) *ChannelCheckpointUpdater
  • func (ccu *ChannelCheckpointUpdater) AddTask(channelPos *msgpb.MsgPosition, flush bool, callback func())
  • func (ccu *ChannelCheckpointUpdater) Close()
  • func (ccu *ChannelCheckpointUpdater) Start()
• type MsgHandler
• 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) GetLatestTimestamp(channel string) typeutil.Timestamp
  • func (m *TimeTickSender) Start()
  • func (m *TimeTickSender) Stop()
  • func (m *TimeTickSender) Update(channelName string, timestamp uint64, segStats []*commonpb.SegmentStats)

Functions

func ConvertInternalImportFile

func ConvertInternalImportFile(file *msgpb.ImportFile, _ int) *internalpb.ImportFile

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 NewChannelCheckpointUpdaterWithCallback

func NewChannelCheckpointUpdaterWithCallback(broker broker.Broker, updateDoneCallback func(*msgpb.MsgPosition)) *ChannelCheckpointUpdater

NewChannelCheckpointUpdaterWithCallback creates a ChannelCheckpointUpdater with a callback function

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 MsgHandler

type MsgHandler interface {
	HandleCreateSegment(ctx context.Context, vchannel string, createSegmentMsg message.ImmutableCreateSegmentMessageV2) error

	HandleFlush(vchannel string, flushMsg message.ImmutableFlushMessageV2) error

	HandleManualFlush(vchannel string, flushMsg message.ImmutableManualFlushMessageV2) error

	HandleImport(ctx context.Context, vchannel string, importMsg *msgpb.ImportMsg) error

	HandleSchemaChange(ctx context.Context, vchannel string, msg *adaptor.SchemaChangeMessageBody) error
}

type PipelineParams

type PipelineParams struct {
	Ctx                context.Context
	Broker             broker.Broker
	SyncMgr            syncmgr.SyncManager
	TimeTickSender     StatsUpdater       // reference to TimeTickSender
	CompactionExecutor compactor.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
	MsgHandler         MsgHandler
}

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)
	GetLatestTimestamp(channel string) typeutil.Timestamp
}

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

func (m *TimeTickSender) GetLatestTimestamp(channel string) typeutil.Timestamp

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)