pump

Documentation

Index

• Variables
• func FormatNodeID(nodeID string) string
• func NewPumpNode(cfg *Config, getMaxCommitTs func() int64) (node.Node, error)
• type ChangeStateReq
• type Config
  • func NewConfig() *Config
  • func (cfg *Config) Parse(arguments []string) error
• type HTTPStatus
  • func (s *HTTPStatus) Status(w http.ResponseWriter, r *http.Request)
• type Server
  • func NewServer(cfg *Config) (*Server, error)
  • func (s *Server) AllDrainers(w http.ResponseWriter, r *http.Request)
  • func (s *Server) ApplyAction(w http.ResponseWriter, r *http.Request)
  • func (s *Server) BinlogByTS(w http.ResponseWriter, r *http.Request)
  • func (s *Server) Close()
  • func (s *Server) PullBinlogs(in *binlog.PullBinlogReq, stream binlog.Pump_PullBinlogsServer) error
  • func (s *Server) PumpStatus() *HTTPStatus
  • func (s *Server) Start() error
  • func (s *Server) Status(w http.ResponseWriter, r *http.Request)
  • func (s *Server) TriggerGC(w http.ResponseWriter, r *http.Request)
  • func (s *Server) WriteBinlog(ctx context.Context, in *binlog.WriteBinlogReq) (*binlog.WriteBinlogResp, error)

Variables

var (

	// GlobalConfig is global config of pump
	GlobalConfig *globalConfig
)

Functions

func FormatNodeID

func FormatNodeID(nodeID string) string

FormatNodeID formats the nodeID, the nodeID should looks like "host:port"

func NewPumpNode

func NewPumpNode(cfg *Config, getMaxCommitTs func() int64) (node.Node, error)

NewPumpNode returns a pumpNode obj that initialized by server config

Types

type ChangeStateReq

type ChangeStateReq struct {
	NodeID string `json:"nodeID"`
	State  string `json:"state"`
}

ChangeStateReq is the request struct for change state.

type Config

type Config struct {
	*flag.FlagSet     `json:"-"`
	LogLevel          string `toml:"log-level" json:"log-level"`
	NodeID            string `toml:"node-id" json:"node-id"`
	ListenAddr        string `toml:"addr" json:"addr"`
	AdvertiseAddr     string `toml:"advertise-addr" json:"advertise-addr"`
	Socket            string `toml:"socket" json:"socket"`
	EtcdURLs          string `toml:"pd-urls" json:"pd-urls"`
	EtcdDialTimeout   time.Duration
	DataDir           string `toml:"data-dir" json:"data-dir"`
	HeartbeatInterval int    `toml:"heartbeat-interval" json:"heartbeat-interval"`
	// pump only stores binlog events whose ts >= current time - GC Time. The default unit is day
	GC       util.Duration   `toml:"gc" json:"gc"`
	LogFile  string          `toml:"log-file" json:"log-file"`
	Security security.Config `toml:"security" json:"security"`

	GenFakeBinlogInterval int `toml:"gen-binlog-interval" json:"gen-binlog-interval"`

	MetricsAddr     string
	MetricsInterval int

	Storage storage.Config `toml:"storage" json:"storage"`
	// contains filtered or unexported fields
}

Config holds the configuration of pump

func NewConfig

func NewConfig() *Config

NewConfig return an instance of configuration

func (*Config) Parse

func (cfg *Config) Parse(arguments []string) error

Parse parses all config from command-line flags, environment vars or configuration file

type HTTPStatus

type HTTPStatus struct {
	StatusMap  map[string]*node.Status `json:"status"`
	CommitTS   int64                   `json:"CommitTS"`
	CheckPoint pb.Pos                  `json:"Checkpoint"`
	ErrMsg     string                  `json:"ErrMsg"`
}

HTTPStatus exposes current status of all pumps via HTTP

func (*HTTPStatus) Status

func (s *HTTPStatus) Status(w http.ResponseWriter, r *http.Request)

Status implements http.ServeHTTP interface

type Server

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

Server implements the gRPC interface, and maintains pump's status at run time.

func NewServer

func NewServer(cfg *Config) (*Server, error)

NewServer returns a instance of pump server

func (*Server) AllDrainers

func (s *Server) AllDrainers(w http.ResponseWriter, r *http.Request)

AllDrainers exposes drainers' status to HTTP handler.

func (*Server) ApplyAction

func (s *Server) ApplyAction(w http.ResponseWriter, r *http.Request)

ApplyAction change the pump's state, now can be pause or close.

func (*Server) BinlogByTS

func (s *Server) BinlogByTS(w http.ResponseWriter, r *http.Request)

BinlogByTS exposes api get get binlog by ts

func (*Server) Close

func (s *Server) Close()

Close gracefully releases resource of pump server

func (*Server) PullBinlogs

func (s *Server) PullBinlogs(in *binlog.PullBinlogReq, stream binlog.Pump_PullBinlogsServer) error

PullBinlogs sends binlogs in the streaming way

func (*Server) PumpStatus

func (s *Server) PumpStatus() *HTTPStatus

PumpStatus returns all pumps' status.

func (*Server) Start

func (s *Server) Start() error

Start runs Pump Server to serve the listening addr, and maintains heartbeat to Etcd

func (*Server) Status

func (s *Server) Status(w http.ResponseWriter, r *http.Request)

Status exposes pumps' status to HTTP handler.

func (*Server) TriggerGC

func (s *Server) TriggerGC(w http.ResponseWriter, r *http.Request)

TriggerGC trigger pump to gc now

func (*Server) WriteBinlog

func (s *Server) WriteBinlog(ctx context.Context, in *binlog.WriteBinlogReq) (*binlog.WriteBinlogResp, error)

WriteBinlog implements the gRPC interface of pump server