capture

Documentation

Index

• func IsHTTPBadRequestError(err error) bool
• func SetLogLevel(c *gin.Context)
• type Capture
  • func NewCapture(pdClient pd.Client, kvStorage tikv.Storage, ...) *Capture
  • func NewCapture4Test() *Capture
  • func (c *Capture) AsyncClose()
  • func (c *Capture) CreateChangefeedInfo(ctx context.Context, info *model.ChangeFeedInfo, cfID string) error
  • func (c *Capture) GetOwner(ctx context.Context) (*model.CaptureInfo, error)
  • func (c *Capture) Info() model.CaptureInfo
  • func (c *Capture) IsOwner() bool
  • func (c *Capture) OperateOwnerUnderLock(fn func(*owner.Owner) error) error
  • func (c *Capture) Run(ctx context.Context) error
  • func (c *Capture) SaveChangeFeedInfo(ctx context.Context, info *model.ChangeFeedInfo, cfID string) error
  • func (c *Capture) WriteDebugInfo(w io.Writer)
• type HTTPHandler
  • func NewHTTPHandler(capture *Capture) HTTPHandler
  • func (h *HTTPHandler) CreateChangefeed(c *gin.Context)
  • func (h *HTTPHandler) GetChangefeed(c *gin.Context)
  • func (h *HTTPHandler) GetProcessor(c *gin.Context)
  • func (h *HTTPHandler) Health(c *gin.Context)
  • func (h *HTTPHandler) ListCapture(c *gin.Context)
  • func (h *HTTPHandler) ListChangefeed(c *gin.Context)
  • func (h *HTTPHandler) ListProcessor(c *gin.Context)
  • func (h *HTTPHandler) MoveKeySpan(c *gin.Context)
  • func (h *HTTPHandler) PauseChangefeed(c *gin.Context)
  • func (h *HTTPHandler) RebalanceKeySpan(c *gin.Context)
  • func (h *HTTPHandler) RemoveChangefeed(c *gin.Context)
  • func (h *HTTPHandler) ResignOwner(c *gin.Context)
  • func (h *HTTPHandler) ResumeChangefeed(c *gin.Context)
  • func (h *HTTPHandler) ServerStatus(c *gin.Context)
  • func (h *HTTPHandler) UpdateChangefeed(c *gin.Context)

Functions

func IsHTTPBadRequestError

func IsHTTPBadRequestError(err error) bool

IsHTTPBadRequestError check if a error is a http bad request error

func SetLogLevel

func SetLogLevel(c *gin.Context)

SetLogLevel changes TiCDC log level dynamically. @Summary Change TiCDC log level @Description change TiCDC log level dynamically @Tags common @Accept json @Produce json @Param log_level body string true "log level" @Success 200 @Failure 400 {object} model.HTTPError @Router /api/v1/log [post]

Types

type Capture

type Capture struct {
	TimeAcquirer pdtime.TimeAcquirer
	// contains filtered or unexported fields
}

Capture represents a Capture server, it monitors the changefeed information in etcd and schedules Task on it.

func NewCapture

func NewCapture(pdClient pd.Client, kvStorage tikv.Storage, createEtcdClient createEtcdClientFunc) *Capture

NewCapture returns a new Capture instance

func NewCapture4Test

func NewCapture4Test() *Capture

func (*Capture) AsyncClose

func (c *Capture) AsyncClose()

AsyncClose closes the capture by unregistering it from etcd Note: this function should be reentrant

func (*Capture) CreateChangefeedInfo

func (c *Capture) CreateChangefeedInfo(ctx context.Context, info *model.ChangeFeedInfo, cfID string) error

CreateChangefeddInfo put new changefeed info in etcd

func (*Capture) GetOwner

func (c *Capture) GetOwner(ctx context.Context) (*model.CaptureInfo, error)

GetOwner return the owner of current TiCDC cluster

func (*Capture) Info

func (c *Capture) Info() model.CaptureInfo

Info gets the capture info

func (*Capture) IsOwner

func (c *Capture) IsOwner() bool

IsOwner returns whether the capture is an owner

func (*Capture) OperateOwnerUnderLock

func (c *Capture) OperateOwnerUnderLock(fn func(*owner.Owner) error) error

OperateOwnerUnderLock operates the owner with lock

func (*Capture) Run

func (c *Capture) Run(ctx context.Context) error

Run runs the capture

func (*Capture) SaveChangeFeedInfo

func (c *Capture) SaveChangeFeedInfo(ctx context.Context, info *model.ChangeFeedInfo, cfID string) error

SaveChangFeedInfo update changefeed info in etcd

func (*Capture) WriteDebugInfo

func (c *Capture) WriteDebugInfo(w io.Writer)

WriteDebugInfo writes the debug info into writer.

type HTTPHandler

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

HTTPHandler is a HTTPHandler of capture

func NewHTTPHandler

func NewHTTPHandler(capture *Capture) HTTPHandler

NewHTTPHandler return a HTTPHandler for OpenAPI

func (*HTTPHandler) CreateChangefeed

func (h *HTTPHandler) CreateChangefeed(c *gin.Context)

CreateChangefeed creates a changefeed @Summary Create changefeed @Description create a new changefeed @Tags changefeed @Accept json @Produce json @Param changefeed body model.ChangefeedConfig true "changefeed config" @Success 202 @Failure 500,400 {object} model.HTTPError @Router /api/v1/changefeeds [post]

func (*HTTPHandler) GetChangefeed

func (h *HTTPHandler) GetChangefeed(c *gin.Context)

GetChangefeed get detailed info of a changefeed @Summary Get changefeed @Description get detail information of a changefeed @Tags changefeed @Accept json @Produce json @Param changefeed_id path string true "changefeed_id" @Success 200 {object} model.ChangefeedDetail @Failure 500,400 {object} model.HTTPError @Router /api/v1/changefeeds/{changefeed_id} [get]

func (*HTTPHandler) GetProcessor

func (h *HTTPHandler) GetProcessor(c *gin.Context)

GetProcessor gets the detailed info of a processor @Summary Get processor detail information @Description get the detail information of a processor @Tags processor @Accept json @Produce json @Success 200 {object} model.ProcessorDetail @Failure 500,400 {object} model.HTTPError @Router /api/v1/processors/{changefeed_id}/{capture_id} [get]

func (*HTTPHandler) Health

func (h *HTTPHandler) Health(c *gin.Context)

Health check if cdc cluster is health @Summary Check if CDC cluster is health @Description check if CDC cluster is health @Tags common @Accept json @Produce json @Success 200 @Failure 500 {object} model.HTTPError @Router /api/v1/health [get]

func (*HTTPHandler) ListCapture

func (h *HTTPHandler) ListCapture(c *gin.Context)

ListCapture lists all captures @Summary List captures @Description list all captures in cdc cluster @Tags capture @Accept json @Produce json @Success 200 {array} model.Capture @Failure 500,400 {object} model.HTTPError @Router /api/v1/captures [get]

func (*HTTPHandler) ListChangefeed

func (h *HTTPHandler) ListChangefeed(c *gin.Context)

ListChangefeed lists all changgefeeds in cdc cluster @Summary List changefeed @Description list all changefeeds in cdc cluster @Tags changefeed @Accept json @Produce json @Param state query string false "state" @Success 200 {array} model.ChangefeedCommonInfo @Failure 500 {object} model.HTTPError @Router /api/v1/changefeeds [get]

func (*HTTPHandler) ListProcessor

func (h *HTTPHandler) ListProcessor(c *gin.Context)

ListProcessor lists all processors in the TiCDC cluster @Summary List processors @Description list all processors in the TiCDC cluster @Tags processor @Accept json @Produce json @Success 200 {array} model.ProcessorCommonInfo @Failure 500,400 {object} model.HTTPError @Router /api/v1/processors [get]

func (*HTTPHandler) MoveKeySpan

func (h *HTTPHandler) MoveKeySpan(c *gin.Context)

MoveKeySpan moves a keyspan to target capture @Summary move keyspan @Description move one keyspan to the target capture @Tags changefeed @Accept json @Produce json @Param changefeed_id path string true "changefeed_id" @Param keyspan_id body integer true "keyspan_id" @Param capture_id body string true "capture_id" @Success 202 @Failure 500,400 {object} model.HTTPError @Router /api/v1/changefeeds/{changefeed_id}/keyspans/move_keyspan [post]

func (*HTTPHandler) PauseChangefeed

func (h *HTTPHandler) PauseChangefeed(c *gin.Context)

PauseChangefeed pauses a changefeed @Summary Pause a changefeed @Description Pause a changefeed @Tags changefeed @Accept json @Produce json @Param changefeed_id path string true "changefeed_id" @Success 202 @Failure 500,400 {object} model.HTTPError @Router /api/v1/changefeeds/{changefeed_id}/pause [post]

func (*HTTPHandler) RebalanceKeySpan

func (h *HTTPHandler) RebalanceKeySpan(c *gin.Context)

RebalanceKeySpan rebalances keyspans @Summary rebalance keyspans @Description rebalance all keyspans of a changefeed @Tags changefeed @Accept json @Produce json @Param changefeed_id path string true "changefeed_id" @Success 202 @Failure 500,400 {object} model.HTTPError @Router /api/v1/changefeeds/{changefeed_id}/keyspans/rebalance_keyspan [post]

func (*HTTPHandler) RemoveChangefeed

func (h *HTTPHandler) RemoveChangefeed(c *gin.Context)

RemoveChangefeed removes a changefeed @Summary Remove a changefeed @Description Remove a changefeed @Tags changefeed @Accept json @Produce json @Param changefeed_id path string true "changefeed_id" @Success 202 @Failure 500,400 {object} model.HTTPError @Router /api/v1/changefeeds/{changefeed_id} [delete]

func (*HTTPHandler) ResignOwner

func (h *HTTPHandler) ResignOwner(c *gin.Context)

ResignOwner makes the current owner resign @Summary notify the owner to resign @Description notify the current owner to resign @Tags owner @Accept json @Produce json @Success 202 @Failure 500,400 {object} model.HTTPError @Router /api/v1/owner/resign [post]

func (*HTTPHandler) ResumeChangefeed

func (h *HTTPHandler) ResumeChangefeed(c *gin.Context)

ResumeChangefeed resumes a changefeed @Summary Resume a changefeed @Description Resume a changefeed @Tags changefeed @Accept json @Produce json @Param changefeed-id path string true "changefeed_id" @Success 202 @Failure 500,400 {object} model.HTTPError @Router /api/v1/changefeeds/{changefeed_id}/resume [post]

func (*HTTPHandler) ServerStatus

func (h *HTTPHandler) ServerStatus(c *gin.Context)

ServerStatus gets the status of server(capture) @Summary Get server status @Description get the status of a server(capture) @Tags common @Accept json @Produce json @Success 200 {object} model.ServerStatus @Failure 500,400 {object} model.HTTPError @Router /api/v1/status [get]

func (*HTTPHandler) UpdateChangefeed

func (h *HTTPHandler) UpdateChangefeed(c *gin.Context)

UpdateChangefeed updates a changefeed @Summary Update a changefeed @Description Update a changefeed @Tags changefeed @Accept json @Produce json @Param changefeed_id path string true "changefeed_id" @Param target_ts body integer false "changefeed target ts" @Param sink_uri body string false "sink uri" @Param filter_rules body []string false "filter rules" @Param ignore_txn_start_ts body integer false "ignore transaction start ts" @Param mounter_worker_num body integer false "mounter worker nums" @Param sink_config body config.SinkConfig false "sink config" @Success 202 @Failure 500,400 {object} model.HTTPError @Router /api/v1/changefeeds/{changefeed_id} [put]