backend

Documentation

Index

• Constants
• func CalculateAndChooseBestNode(cfg *config.NodeConfig, nodeLoad map[int]int) (int, map[int]int, error)
• func Gcd(ary []int) int
• type BackendConn
  • func (p *BackendConn) Close()
  • func (p *BackendConn) CloseForExtendedProtocol()
• type Conn
  • func (c *Conn) Begin() error
  • func (c *Conn) Close() error
  • func (c *Conn) ClosePrepare(id uint32) error
  • func (c *Conn) Commit() error
  • func (c *Conn) Connect(addr string, user string, password string, db string, parseNode bool) error
  • func (c *Conn) Execute(command string, args ...interface{}) (*mysql.Result, error)
  • func (c *Conn) FieldList(table string, wildcard string) ([]*mysql.Field, error)
  • func (c *Conn) GetAddr() string
  • func (c *Conn) GetCharset() string
  • func (c *Conn) GetDB() string
  • func (c *Conn) IsAutoCommit() bool
  • func (c *Conn) IsInTransaction() bool
  • func (c *Conn) Ping(parseNode bool) error
  • func (c *Conn) Prepare(query string) (*Stmt, error)
  • func (c *Conn) ReConnect() error
  • func (c *Conn) ReConnectPg(parseNode bool) error
  • func (c *Conn) ReadPgAllPacket() ([]byte, error)
  • func (c *Conn) ReadPgPacket(reader netpoll.Reader) ([]byte, error)
  • func (c *Conn) Rollback() error
  • func (c *Conn) SetAutoCommit(n uint8) error
  • func (c *Conn) SetCharset(charset string, collation mysql.CollationId) error
  • func (c *Conn) UseDB(dbName string) error
• type DB
  • func Open(addr string, user string, password string, dbName string, maxConnNum int, ...) (*DB, error)
  • func (db *DB) Addr() string
  • func (db *DB) Close() error
  • func (db *DB) ConnCount() (int, int, int64, int64)
  • func (db *DB) GetConn() (*BackendConn, error)
  • func (db *DB) GetConnFromCache(cacheConns chan *Conn) *Conn
  • func (db *DB) GetConnFromIdle(cacheConns, idleConns chan *Conn) (*Conn, error)
  • func (db *DB) GetConnPg(dbname string, dbuser string) (*BackendConn, error)
  • func (db *DB) GetLastPing() int64
  • func (db *DB) InitConnPoolPg(dbname string, dbuser string) (cacheConns chan *Conn, err error)
  • func (db *DB) IsExceedMaxConns() bool
  • func (db *DB) Ping(parseNode bool) error
  • func (db *DB) PopConn() (*Conn, error)
  • func (db *DB) PopConnPg(dbname string, dbuser string) (*Conn, error)
  • func (db *DB) PushConn(co *Conn, err error)
  • func (db *DB) PushConnForExtendedProtocol(co *Conn, err error)
  • func (db *DB) SetLastPing()
  • func (db *DB) State() string
• type Data
• type Metric
• type Node
  • func (n *Node) AddSlave(addr string) error
  • func (n *Node) CheckConnsCache()
  • func (n *Node) CheckNode()
  • func (n *Node) DeleteSlave(addr string) error
  • func (n *Node) DownMaster(addr string, state int32) error
  • func (n *Node) DownSlave(addr string, state int32) error
  • func (n *Node) GetMasterConn() (*BackendConn, error)
  • func (n *Node) GetMasterConnPg(dbname string, dbuser string) (*BackendConn, error)
  • func (n *Node) GetNextSlave() (*DB, error)
  • func (n *Node) GetSlaveConn() (*BackendConn, error)
  • func (n *Node) GetSlaveConnPg(dbname string, dbuser string, tablename string) (*BackendConn, error)
  • func (n *Node) InitBalancer()
  • func (n *Node) OpenDB(addr string, parseNode bool) (*DB, error)
  • func (n *Node) ParseMaster(masterStr string) error
  • func (n *Node) ParseSlave(slaveStr string) error
  • func (n *Node) String() string
  • func (n *Node) UpDB(addr string) (*DB, error)
  • func (n *Node) UpMaster(addr string) error
  • func (n *Node) UpSlave(addr string) error
• type Prometheus
• type Result
• type Stmt
  • func (s *Stmt) Close() error
  • func (s *Stmt) ColumnNum() int
  • func (s *Stmt) Execute(args ...interface{}) (*mysql.Result, error)
  • func (s *Stmt) GetId() uint32
  • func (s *Stmt) ParamNum() int

Constants

const (
	AuthTypeOk                = 0
	AuthTypeCleartextPassword = 3
	AuthTypeMD5Password       = 5
	AuthTypeSCMCreds          = 6
	AuthTypeGSS               = 7
	AuthTypeGSSCont           = 8
	AuthTypeSSPI              = 9
	AuthTypeSASL              = 10
	AuthTypeSASLContinue      = 11
	AuthTypeSASLFinal         = 12
)

const (
	Up = iota
	Down
	ManualDown
	Unknown

	InitConnCount           = 16
	DefaultMaxConnNum       = 1024
	PingPeroid        int64 = 4
	MaxPoolNum              = 4
)

const (
	Master      = "master"
	Slave       = "slave"
	SlaveSplit  = ","
	WeightSplit = "@"
)

const (
	CPU_WEIGHT        = 1
	MEM_WEIGHT        = 3
	DISK_IO_WEIGHT    = 3
	CONNECTION_WEIGHT = 3
)

Functions

func CalculateAndChooseBestNode

func CalculateAndChooseBestNode(cfg *config.NodeConfig, nodeLoad map[int]int) (int, map[int]int, error)

func Gcd

func Gcd(ary []int) int

Types

type BackendConn

type BackendConn struct {
	*Conn

	IsInTransaction bool // 是否在事务中
	// contains filtered or unexported fields
}

func (*BackendConn) Close

func (p *BackendConn) Close()

func (*BackendConn) CloseForExtendedProtocol

func (p *BackendConn) CloseForExtendedProtocol()

type Conn

type Conn struct {

	// pg param
	ConnPg    netpoll.Connection
	ConnPgx   *pgx.Conn
	ProcessID uint32
	SecretKey uint32
	// contains filtered or unexported fields
}

proxy <-> mysql server

func (*Conn) Begin

func (c *Conn) Begin() error

func (*Conn) Close

func (c *Conn) Close() error

func (*Conn) ClosePrepare

func (c *Conn) ClosePrepare(id uint32) error

func (*Conn) Commit

func (c *Conn) Commit() error

func (*Conn) Connect

func (c *Conn) Connect(addr string, user string, password string, db string, parseNode bool) error

func (*Conn) Execute

func (c *Conn) Execute(command string, args ...interface{}) (*mysql.Result, error)

func (*Conn) FieldList

func (c *Conn) FieldList(table string, wildcard string) ([]*mysql.Field, error)

func (*Conn) GetAddr

func (c *Conn) GetAddr() string

func (*Conn) GetCharset

func (c *Conn) GetCharset() string

func (*Conn) GetDB

func (c *Conn) GetDB() string

func (*Conn) IsAutoCommit

func (c *Conn) IsAutoCommit() bool

func (*Conn) IsInTransaction

func (c *Conn) IsInTransaction() bool

func (*Conn) Ping

func (c *Conn) Ping(parseNode bool) error

func (*Conn) Prepare

func (c *Conn) Prepare(query string) (*Stmt, error)

func (*Conn) ReConnect

func (c *Conn) ReConnect() error

func (*Conn) ReConnectPg

func (c *Conn) ReConnectPg(parseNode bool) error

func (*Conn) ReadPgAllPacket

func (c *Conn) ReadPgAllPacket() ([]byte, error)

pg msg streaming can not through this way to get all msg

func (*Conn) ReadPgPacket

func (c *Conn) ReadPgPacket(reader netpoll.Reader) ([]byte, error)

func (*Conn) Rollback

func (c *Conn) Rollback() error

func (*Conn) SetAutoCommit

func (c *Conn) SetAutoCommit(n uint8) error

func (*Conn) SetCharset

func (c *Conn) SetCharset(charset string, collation mysql.CollationId) error

func (*Conn) UseDB

func (c *Conn) UseDB(dbName string) error

type DB

type DB struct {
	sync.RWMutex

	InitConnNum int
	// contains filtered or unexported fields
}

func Open

func Open(addr string, user string, password string, dbName string, maxConnNum int, parseNode bool) (*DB, error)

func (*DB) Addr

func (db *DB) Addr() string

func (*DB) Close

func (db *DB) Close() error

func (*DB) ConnCount

func (db *DB) ConnCount() (int, int, int64, int64)

func (*DB) GetConn

func (db *DB) GetConn() (*BackendConn, error)

func (*DB) GetConnFromCache

func (db *DB) GetConnFromCache(cacheConns chan *Conn) *Conn

func (*DB) GetConnFromIdle

func (db *DB) GetConnFromIdle(cacheConns, idleConns chan *Conn) (*Conn, error)

func (*DB) GetConnPg

func (db *DB) GetConnPg(dbname string, dbuser string) (*BackendConn, error)

func (*DB) GetLastPing

func (db *DB) GetLastPing() int64

func (*DB) InitConnPoolPg

func (db *DB) InitConnPoolPg(dbname string, dbuser string) (cacheConns chan *Conn, err error)

func (*DB) IsExceedMaxConns

func (db *DB) IsExceedMaxConns() bool

TODO 如何判断是否超过最大链接数

func (*DB) Ping

func (db *DB) Ping(parseNode bool) error

func (*DB) PopConn

func (db *DB) PopConn() (*Conn, error)

func (*DB) PopConnPg

func (db *DB) PopConnPg(dbname string, dbuser string) (*Conn, error)

func (*DB) PushConn

func (db *DB) PushConn(co *Conn, err error)

func (*DB) PushConnForExtendedProtocol

func (db *DB) PushConnForExtendedProtocol(co *Conn, err error)

func (*DB) SetLastPing

func (db *DB) SetLastPing()

func (*DB) State

func (db *DB) State() string

type Data

type Data struct {
	ResultType string   `json:"resultType"`
	Result     []Result `json:"result"`
}

type Metric

type Metric struct {
	Device   string `json:"device"`
	Instance string `json:"instance"`
	Job      string `json:"job"`
}

type Node

type Node struct {
	Cfg config.NodeConfig

	sync.RWMutex
	Master *DB

	Slave          []*DB
	LastSlaveIndex int
	RoundRobinQ    []int
	SlaveWeights   []int

	DownAfterNoAlive time.Duration

	Online bool

	// Select the subscript value of the optimal node through the index
	BestNodeIndexByMetric int

	// save node-lsn relationship
	NodeLsn sync.Map

	// Node load score
	NodeLoad map[int]int

	// save node-cached table relationship
	NodeCache sync.Map

	// db_table
	NodeCacheKey string
}

func (*Node) AddSlave

func (n *Node) AddSlave(addr string) error

func (*Node) CheckConnsCache

func (n *Node) CheckConnsCache()

检验链接是否超时

func (*Node) CheckNode

func (n *Node) CheckNode()

func (*Node) DeleteSlave

func (n *Node) DeleteSlave(addr string) error

func (*Node) DownMaster

func (n *Node) DownMaster(addr string, state int32) error

func (*Node) DownSlave

func (n *Node) DownSlave(addr string, state int32) error

func (*Node) GetMasterConn

func (n *Node) GetMasterConn() (*BackendConn, error)

func (*Node) GetMasterConnPg

func (n *Node) GetMasterConnPg(dbname string, dbuser string) (*BackendConn, error)

func (*Node) GetNextSlave

func (n *Node) GetNextSlave() (*DB, error)

func (*Node) GetSlaveConn

func (n *Node) GetSlaveConn() (*BackendConn, error)

func (*Node) GetSlaveConnPg

func (n *Node) GetSlaveConnPg(dbname string, dbuser string, tablename string) (*BackendConn, error)

func (*Node) InitBalancer

func (n *Node) InitBalancer()

func (*Node) OpenDB

func (n *Node) OpenDB(addr string, parseNode bool) (*DB, error)

func (*Node) ParseMaster

func (n *Node) ParseMaster(masterStr string) error

func (*Node) ParseSlave

func (n *Node) ParseSlave(slaveStr string) error

slaveStr(127.0.0.1:3306@2,192.168.0.12:3306@3)

func (*Node) String

func (n *Node) String() string

func (*Node) UpDB

func (n *Node) UpDB(addr string) (*DB, error)

func (*Node) UpMaster

func (n *Node) UpMaster(addr string) error

func (*Node) UpSlave

func (n *Node) UpSlave(addr string) error

type Prometheus

type Prometheus struct {
	Status string `json:"status"`
	Data   Data   `json:"data"`
}

type Result

type Result struct {
	Metric Metric        `json:"metric"`
	Values []interface{} `json:"values"`
	Value  []interface{} `json:"value"`
}

type Stmt

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

func (*Stmt) Close

func (s *Stmt) Close() error

func (*Stmt) ColumnNum

func (s *Stmt) ColumnNum() int

func (*Stmt) Execute

func (s *Stmt) Execute(args ...interface{}) (*mysql.Result, error)

func (*Stmt) GetId

func (s *Stmt) GetId() uint32

func (*Stmt) ParamNum

func (s *Stmt) ParamNum() int