job

Documentation

Index

• type BaseJob
  • func NewBaseJob(ctx context.Context, msgID, collectionID int64) *BaseJob
  • func (job *BaseJob) CollectionID() int64
  • func (job *BaseJob) Context() context.Context
  • func (job *BaseJob) Done()
  • func (job *BaseJob) Error() error
  • func (job *BaseJob) MsgID() int64
  • func (job *BaseJob) PostExecute()
  • func (job *BaseJob) PreExecute() error
  • func (job *BaseJob) SetError(err error)
  • func (job *BaseJob) Wait() error
• type Job
• type LoadCollectionJob
  • func NewLoadCollectionJob(ctx context.Context, req *querypb.LoadCollectionRequest, ...) *LoadCollectionJob
  • func (job *LoadCollectionJob) Execute() error
  • func (job *LoadCollectionJob) PostExecute()
  • func (job *LoadCollectionJob) PreExecute() error
• type LoadPartitionJob
  • func NewLoadPartitionJob(ctx context.Context, req *querypb.LoadPartitionsRequest, ...) *LoadPartitionJob
  • func (job *LoadPartitionJob) Execute() error
  • func (job *LoadPartitionJob) PostExecute()
  • func (job *LoadPartitionJob) PreExecute() error
• type ReleaseCollectionJob
  • func NewReleaseCollectionJob(ctx context.Context, req *querypb.ReleaseCollectionRequest, ...) *ReleaseCollectionJob
  • func (job *ReleaseCollectionJob) Execute() error
• type ReleasePartitionJob
  • func NewReleasePartitionJob(ctx context.Context, req *querypb.ReleasePartitionsRequest, ...) *ReleasePartitionJob
  • func (job *ReleasePartitionJob) Execute() error
• type Scheduler
  • func NewScheduler() *Scheduler
  • func (scheduler *Scheduler) Add(job Job)
  • func (scheduler *Scheduler) Start()
  • func (scheduler *Scheduler) Stop()
• type SyncNewCreatedPartitionJob
  • func NewSyncNewCreatedPartitionJob(ctx context.Context, req *querypb.SyncNewCreatedPartitionRequest, ...) *SyncNewCreatedPartitionJob
  • func (job *SyncNewCreatedPartitionJob) Execute() error
  • func (job *SyncNewCreatedPartitionJob) PreExecute() error
• type UndoList
  • func NewUndoList(ctx context.Context, meta *meta.Meta, cluster session.Cluster, ...) *UndoList
  • func (u *UndoList) RollBack()
• type UpdateLoadConfigJob
  • func NewUpdateLoadConfigJob(ctx context.Context, req *querypb.UpdateLoadConfigRequest, meta *meta.Meta, ...) *UpdateLoadConfigJob
  • func (job *UpdateLoadConfigJob) Execute() error

Types

type BaseJob

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

func NewBaseJob

func NewBaseJob(ctx context.Context, msgID, collectionID int64) *BaseJob

func (*BaseJob) CollectionID

func (job *BaseJob) CollectionID() int64

func (*BaseJob) Context

func (job *BaseJob) Context() context.Context

func (*BaseJob) Done

func (job *BaseJob) Done()

func (*BaseJob) Error

func (job *BaseJob) Error() error

func (*BaseJob) MsgID

func (job *BaseJob) MsgID() int64

func (*BaseJob) PostExecute

func (job *BaseJob) PostExecute()

func (*BaseJob) PreExecute

func (job *BaseJob) PreExecute() error

func (*BaseJob) SetError

func (job *BaseJob) SetError(err error)

func (*BaseJob) Wait

func (job *BaseJob) Wait() error

type Job

type Job interface {
	MsgID() int64
	CollectionID() int64
	Context() context.Context
	// PreExecute does checks, DO NOT persists any thing within this stage,
	PreExecute() error
	// Execute processes the request
	Execute() error
	// PostExecute clears resources, it will be always processed
	PostExecute()
	Error() error
	SetError(err error)
	Done()
	Wait() error
}

Job is request of loading/releasing collection/partitions, the execution flow is: 1. PreExecute() 2. Execute(), skip this step if PreExecute() failed 3. PostExecute()

type LoadCollectionJob

type LoadCollectionJob struct {
	*BaseJob
	// contains filtered or unexported fields
}

func NewLoadCollectionJob

func NewLoadCollectionJob(
	ctx context.Context,
	req *querypb.LoadCollectionRequest,
	dist *meta.DistributionManager,
	meta *meta.Meta,
	broker meta.Broker,
	cluster session.Cluster,
	targetMgr meta.TargetManagerInterface,
	targetObserver *observers.TargetObserver,
	collectionObserver *observers.CollectionObserver,
	nodeMgr *session.NodeManager,
) *LoadCollectionJob

func (*LoadCollectionJob) Execute

func (job *LoadCollectionJob) Execute() error

func (*LoadCollectionJob) PostExecute

func (job *LoadCollectionJob) PostExecute()

func (*LoadCollectionJob) PreExecute

func (job *LoadCollectionJob) PreExecute() error

type LoadPartitionJob

type LoadPartitionJob struct {
	*BaseJob
	// contains filtered or unexported fields
}

func NewLoadPartitionJob

func NewLoadPartitionJob(
	ctx context.Context,
	req *querypb.LoadPartitionsRequest,
	dist *meta.DistributionManager,
	meta *meta.Meta,
	broker meta.Broker,
	cluster session.Cluster,
	targetMgr meta.TargetManagerInterface,
	targetObserver *observers.TargetObserver,
	collectionObserver *observers.CollectionObserver,
	nodeMgr *session.NodeManager,
) *LoadPartitionJob

func (*LoadPartitionJob) Execute

func (job *LoadPartitionJob) Execute() error

func (*LoadPartitionJob) PostExecute

func (job *LoadPartitionJob) PostExecute()

func (*LoadPartitionJob) PreExecute

func (job *LoadPartitionJob) PreExecute() error

type ReleaseCollectionJob

type ReleaseCollectionJob struct {
	*BaseJob
	// contains filtered or unexported fields
}

func NewReleaseCollectionJob

func NewReleaseCollectionJob(ctx context.Context,
	req *querypb.ReleaseCollectionRequest,
	dist *meta.DistributionManager,
	meta *meta.Meta,
	broker meta.Broker,
	cluster session.Cluster,
	targetMgr meta.TargetManagerInterface,
	targetObserver *observers.TargetObserver,
	checkerController *checkers.CheckerController,
	proxyManager proxyutil.ProxyClientManagerInterface,
) *ReleaseCollectionJob

func (*ReleaseCollectionJob) Execute

func (job *ReleaseCollectionJob) Execute() error

type ReleasePartitionJob

type ReleasePartitionJob struct {
	*BaseJob
	// contains filtered or unexported fields
}

func NewReleasePartitionJob

func NewReleasePartitionJob(ctx context.Context,
	req *querypb.ReleasePartitionsRequest,
	dist *meta.DistributionManager,
	meta *meta.Meta,
	broker meta.Broker,
	cluster session.Cluster,
	targetMgr meta.TargetManagerInterface,
	targetObserver *observers.TargetObserver,
	checkerController *checkers.CheckerController,
	proxyManager proxyutil.ProxyClientManagerInterface,
) *ReleasePartitionJob

func (*ReleasePartitionJob) Execute

func (job *ReleasePartitionJob) Execute() error

type Scheduler

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

func NewScheduler

func NewScheduler() *Scheduler

func (*Scheduler) Add

func (scheduler *Scheduler) Add(job Job)

func (*Scheduler) Start

func (scheduler *Scheduler) Start()

func (*Scheduler) Stop

func (scheduler *Scheduler) Stop()

type SyncNewCreatedPartitionJob

type SyncNewCreatedPartitionJob struct {
	*BaseJob
	// contains filtered or unexported fields
}

func NewSyncNewCreatedPartitionJob

func NewSyncNewCreatedPartitionJob(
	ctx context.Context,
	req *querypb.SyncNewCreatedPartitionRequest,
	meta *meta.Meta,
	cluster session.Cluster,
	broker meta.Broker,
) *SyncNewCreatedPartitionJob

func (*SyncNewCreatedPartitionJob) Execute

func (job *SyncNewCreatedPartitionJob) Execute() error

func (*SyncNewCreatedPartitionJob) PreExecute

func (job *SyncNewCreatedPartitionJob) PreExecute() error

type UndoList

type UndoList struct {
	IsTargetUpdated  bool // indicates if target updated during loading
	IsReplicaCreated bool // indicates if created new replicas during loading
	IsNewCollection  bool // indicates if created new collection during loading

	CollectionID   int64
	LackPartitions []int64
	// contains filtered or unexported fields
}

func NewUndoList

func NewUndoList(ctx context.Context, meta *meta.Meta,
	cluster session.Cluster, targetMgr meta.TargetManagerInterface, targetObserver *observers.TargetObserver,
) *UndoList

func (*UndoList) RollBack

func (u *UndoList) RollBack()

type UpdateLoadConfigJob

type UpdateLoadConfigJob struct {
	*BaseJob
	// contains filtered or unexported fields
}

func NewUpdateLoadConfigJob

func NewUpdateLoadConfigJob(ctx context.Context,
	req *querypb.UpdateLoadConfigRequest,
	meta *meta.Meta,
	targetMgr meta.TargetManagerInterface,
	targetObserver *observers.TargetObserver,
	collectionObserver *observers.CollectionObserver,
) *UpdateLoadConfigJob

func (*UpdateLoadConfigJob) Execute

func (job *UpdateLoadConfigJob) Execute() error