server

Documentation

Index

• func EncodeMetaPosition(position *meta.PositionInfo) (string, error)
• func GetEtcdServerConfigFromMetaConfig(metaConfig CDCMetaStoreConfig) config.EtcdServerConfig
• func GetEtcdServerConfigFromSourceConfig(sourceConfig MilvusSourceConfig) config.EtcdServerConfig
• func GetMilvusToken(milvusConnectParam model.MilvusConnectParam) string
• func GetMilvusURI(milvusConnectParam model.MilvusConnectParam) string
• func GetShouldReadFunc(taskInfo *meta.TaskInfo) cdcreader.ShouldReadFunc
• type BaseCDC
  • func NewBaseCDC() *BaseCDC
  • func (b *BaseCDC) Create(request *request.CreateRequest) (*request.CreateResponse, error)
  • func (b *BaseCDC) Delete(request *request.DeleteRequest) (*request.DeleteResponse, error)
  • func (b *BaseCDC) Get(request *request.GetRequest) (*request.GetResponse, error)
  • func (b *BaseCDC) GetPosition(req *request.GetPositionRequest) (*request.GetPositionResponse, error)
  • func (b *BaseCDC) List(request *request.ListRequest) (*request.ListResponse, error)
  • func (b *BaseCDC) Maintenance(request *request.MaintenanceRequest) (*request.MaintenanceResponse, error)
  • func (b *BaseCDC) Pause(request *request.PauseRequest) (*request.PauseResponse, error)
  • func (b *BaseCDC) ReloadTask()
  • func (b *BaseCDC) Resume(request *request.ResumeRequest) (*request.ResumeResponse, error)
• type CDCMetaStoreConfig
• type CDCServer
  • func (c *CDCServer) Run(config *CDCServerConfig)
• type CDCServerConfig
• type CDCService
  • func GetCDCAPI(config *CDCServerConfig) CDCService
• type MetaCDC
  • func NewMetaCDC(serverConfig *CDCServerConfig) *MetaCDC
  • func (e *MetaCDC) Create(req *request.CreateRequest) (resp *request.CreateResponse, err error)
  • func (e *MetaCDC) Delete(req *request.DeleteRequest) (*request.DeleteResponse, error)
  • func (e *MetaCDC) Get(req *request.GetRequest) (*request.GetResponse, error)
  • func (e *MetaCDC) GetPosition(req *request.GetPositionRequest) (*request.GetPositionResponse, error)
  • func (e *MetaCDC) List(req *request.ListRequest) (*request.ListResponse, error)
  • func (e *MetaCDC) Maintenance(req *request.MaintenanceRequest) (*request.MaintenanceResponse, error)
  • func (e *MetaCDC) Pause(req *request.PauseRequest) (*request.PauseResponse, error)
  • func (e *MetaCDC) ReloadTask()
  • func (e *MetaCDC) Resume(req *request.ResumeRequest) (*request.ResumeResponse, error)
• type MilvusSourceConfig
• type ReplicateEntity
• type WriteCallback
  • func NewWriteCallback(factory api.MetaStoreFactory, rootPath string, taskID string) *WriteCallback
  • func (w *WriteCallback) UpdateTaskCollectionPosition(collectionID int64, collectionName string, pChannelName string, ...) error

Functions

func EncodeMetaPosition

func EncodeMetaPosition(position *meta.PositionInfo) (string, error)

func GetEtcdServerConfigFromMetaConfig

func GetEtcdServerConfigFromMetaConfig(metaConfig CDCMetaStoreConfig) config.EtcdServerConfig

func GetEtcdServerConfigFromSourceConfig

func GetEtcdServerConfigFromSourceConfig(sourceConfig MilvusSourceConfig) config.EtcdServerConfig

func GetMilvusToken

func GetMilvusToken(milvusConnectParam model.MilvusConnectParam) string

func GetMilvusURI

func GetMilvusURI(milvusConnectParam model.MilvusConnectParam) string

func GetShouldReadFunc

func GetShouldReadFunc(taskInfo *meta.TaskInfo) cdcreader.ShouldReadFunc

Types

type BaseCDC

type BaseCDC struct{}

func NewBaseCDC

func NewBaseCDC() *BaseCDC

func (*BaseCDC) Create

func (b *BaseCDC) Create(request *request.CreateRequest) (*request.CreateResponse, error)

func (*BaseCDC) Delete

func (b *BaseCDC) Delete(request *request.DeleteRequest) (*request.DeleteResponse, error)

func (*BaseCDC) Get

func (b *BaseCDC) Get(request *request.GetRequest) (*request.GetResponse, error)

func (*BaseCDC) GetPosition

func (b *BaseCDC) GetPosition(req *request.GetPositionRequest) (*request.GetPositionResponse, error)

func (*BaseCDC) List

func (b *BaseCDC) List(request *request.ListRequest) (*request.ListResponse, error)

func (*BaseCDC) Maintenance

func (b *BaseCDC) Maintenance(request *request.MaintenanceRequest) (*request.MaintenanceResponse, error)

func (*BaseCDC) Pause

func (b *BaseCDC) Pause(request *request.PauseRequest) (*request.PauseResponse, error)

func (*BaseCDC) ReloadTask

func (b *BaseCDC) ReloadTask()

func (*BaseCDC) Resume

func (b *BaseCDC) Resume(request *request.ResumeRequest) (*request.ResumeResponse, error)

type CDCMetaStoreConfig

type CDCMetaStoreConfig struct {
	StoreType string
	// deprecated
	EtcdEndpoints  []string
	Etcd           config.EtcdServerConfig
	MysqlSourceURL string
	RootPath       string
}

type CDCServer

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

func (*CDCServer) Run

func (c *CDCServer) Run(config *CDCServerConfig)

type CDCServerConfig

type CDCServerConfig struct {
	Address         string // like: "localhost:8080"
	MaxTaskNum      int
	MetaStoreConfig CDCMetaStoreConfig // cdc meta data save
	SourceConfig    MilvusSourceConfig // cdc source
	// deprecated
	EnableReverse  bool
	ReverseMilvus  model.MilvusConnectParam
	CurrentMilvus  model.MilvusConnectParam
	MaxNameLength  int
	Retry          config.RetrySettings
	LogLevel       string
	DetectDeadLock bool
}

type CDCService

type CDCService interface {
	ReloadTask()
	Create(request *request.CreateRequest) (*request.CreateResponse, error)
	Delete(request *request.DeleteRequest) (*request.DeleteResponse, error)
	Pause(request *request.PauseRequest) (*request.PauseResponse, error)
	Resume(request *request.ResumeRequest) (*request.ResumeResponse, error)
	Get(request *request.GetRequest) (*request.GetResponse, error)
	GetPosition(req *request.GetPositionRequest) (*request.GetPositionResponse, error)
	List(request *request.ListRequest) (*request.ListResponse, error)
	Maintenance(request *request.MaintenanceRequest) (*request.MaintenanceResponse, error)
}

func GetCDCAPI

func GetCDCAPI(config *CDCServerConfig) CDCService

type MetaCDC

type MetaCDC struct {
	BaseCDC
	// contains filtered or unexported fields
}

func NewMetaCDC

func NewMetaCDC(serverConfig *CDCServerConfig) *MetaCDC

func (*MetaCDC) Create

func (e *MetaCDC) Create(req *request.CreateRequest) (resp *request.CreateResponse, err error)

func (*MetaCDC) Delete

func (e *MetaCDC) Delete(req *request.DeleteRequest) (*request.DeleteResponse, error)

func (*MetaCDC) Get

func (e *MetaCDC) Get(req *request.GetRequest) (*request.GetResponse, error)

func (*MetaCDC) GetPosition

func (e *MetaCDC) GetPosition(req *request.GetPositionRequest) (*request.GetPositionResponse, error)

func (*MetaCDC) List

func (e *MetaCDC) List(req *request.ListRequest) (*request.ListResponse, error)

func (*MetaCDC) Maintenance

func (e *MetaCDC) Maintenance(req *request.MaintenanceRequest) (*request.MaintenanceResponse, error)

func (*MetaCDC) Pause

func (e *MetaCDC) Pause(req *request.PauseRequest) (*request.PauseResponse, error)

func (*MetaCDC) ReloadTask

func (e *MetaCDC) ReloadTask()

func (*MetaCDC) Resume

func (e *MetaCDC) Resume(req *request.ResumeRequest) (*request.ResumeResponse, error)

type MilvusSourceConfig

type MilvusSourceConfig struct {
	// deprecated
	EtcdAddress []string
	// deprecated
	EtcdRootPath string
	// deprecated
	EtcdMetaSubPath      string
	Etcd                 config.EtcdServerConfig
	ReadChanLen          int
	TimeTickInterval     int
	DefaultPartitionName string
	ReplicateChan        string
	Pulsar               config.PulsarConfig
	Kafka                config.KafkaConfig
}

type ReplicateEntity

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

type WriteCallback

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

func NewWriteCallback

func NewWriteCallback(factory api.MetaStoreFactory, rootPath string, taskID string) *WriteCallback

func (*WriteCallback) UpdateTaskCollectionPosition

func (w *WriteCallback) UpdateTaskCollectionPosition(collectionID int64, collectionName string, pChannelName string, position, opPosition, targetPosition *meta.PositionInfo) error