310 lines
9.9 KiB
Go
310 lines
9.9 KiB
Go
package cluster
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"linkport/internal/model"
|
|
"linkport/internal/protocol/amqpprobe"
|
|
"linkport/internal/protocol/transport"
|
|
"net"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/go-mysql-org/go-mysql/client"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
mssql "github.com/microsoft/go-mssqldb"
|
|
"github.com/microsoft/go-mssqldb/msdsn"
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
"go.mongodb.org/mongo-driver/v2/mongo"
|
|
"go.mongodb.org/mongo-driver/v2/mongo/options"
|
|
)
|
|
|
|
func probeMongo(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
|
|
tls, err := transport.TLSConfig(p)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
authDB := p.AuthSource
|
|
if authDB == "" {
|
|
authDB = "admin"
|
|
}
|
|
opts := options.Client().SetHosts([]string{node(p).Address()}).SetDirect(true).SetDialer(contextDialer{dial}).SetTLSConfig(tls).SetServerSelectionTimeout(3 * time.Second).SetConnectTimeout(3 * time.Second)
|
|
if p.TargetUser != "" {
|
|
opts.SetAuth(options.Credential{AuthSource: authDB, Username: p.TargetUser, Password: p.TargetPassword})
|
|
}
|
|
c, err := mongo.Connect(opts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() {
|
|
closeCtx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
defer cancel()
|
|
c.Disconnect(closeCtx)
|
|
}()
|
|
var hello bson.M
|
|
err = c.Database("admin").RunCommand(ctx, bson.D{{Key: "hello", Value: 1}}).Decode(&hello)
|
|
if mongoHelloUnsupported(err) {
|
|
hello = nil
|
|
err = c.Database("admin").RunCommand(ctx, bson.D{{Key: "isMaster", Value: 1}}).Decode(&hello)
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
info := singleInfo(p, "standalone", "standalone")
|
|
if hello["msg"] == "isdbgrid" {
|
|
info.Type = "mongodb-sharded"
|
|
info.Selected.Role = "router"
|
|
info.Nodes[0].Role = "router"
|
|
return info, nil
|
|
}
|
|
if name, ok := hello["setName"].(string); ok {
|
|
info.Type, info.Name = "mongodb-replica-set", name
|
|
info.Selected.Role = "replica"
|
|
if hello["isWritablePrimary"] == true || hello["ismaster"] == true {
|
|
info.Selected.Role = "primary"
|
|
}
|
|
info.Nodes[0].Role = info.Selected.Role
|
|
for _, key := range []string{"hosts", "passives"} {
|
|
if hosts, ok := hello[key].(bson.A); ok {
|
|
for _, host := range hosts {
|
|
if address, ok := host.(string); ok {
|
|
if n, ok := endpoint(address, 27017); ok {
|
|
if address == hello["primary"] {
|
|
n.Role = "primary"
|
|
}
|
|
info.Nodes = addNodes(info.Nodes, n)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return info, nil
|
|
}
|
|
|
|
func mongoHelloUnsupported(err error) bool {
|
|
var command mongo.CommandError
|
|
return errors.As(err, &command) && (command.Code == 59 || command.Name == "CommandNotFound")
|
|
}
|
|
|
|
// go-mysql GetString aliases pooled result bytes. Topology outlives Result.Close
|
|
// and subsequent queries, so retain owned strings rather than borrowed views.
|
|
func mysqlTopologyNode(host string, port int, role string) model.ClusterNode {
|
|
return model.ClusterNode{Host: strings.Clone(host), Port: port, Role: strings.ToLower(strings.Clone(role))}
|
|
}
|
|
|
|
func probeMySQL(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
|
|
tls, err := transport.TLSConfig(p)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var raw net.Conn
|
|
wrapped := func(ctx context.Context, network, address string) (net.Conn, error) {
|
|
c, e := dial(ctx, network, address)
|
|
if e == nil {
|
|
raw = c
|
|
deadline := time.Now().Add(3 * time.Second)
|
|
if d, ok := ctx.Deadline(); ok {
|
|
deadline = d
|
|
}
|
|
c.SetDeadline(deadline)
|
|
}
|
|
return c, e
|
|
}
|
|
c, err := client.ConnectWithDialer(ctx, "tcp", node(p).Address(), p.TargetUser, p.TargetPassword, "", wrapped, func(c *client.Conn) error { c.SetTLSConfig(tls); return nil })
|
|
if err != nil {
|
|
if raw != nil {
|
|
raw.Close()
|
|
}
|
|
return nil, err
|
|
}
|
|
defer c.Close()
|
|
stop := context.AfterFunc(ctx, func() { c.Close() })
|
|
defer stop()
|
|
info := singleInfo(p, "unknown", "unknown")
|
|
if r, e := c.Execute("SELECT @@global.read_only"); e == nil {
|
|
readonly, _ := r.GetInt(0, 0)
|
|
info.Selected.Role = "primary"
|
|
if readonly != 0 {
|
|
info.Selected.Role = "replica"
|
|
}
|
|
info.Nodes[0].Role = info.Selected.Role
|
|
r.Close()
|
|
}
|
|
if r, e := c.Execute("SELECT MEMBER_HOST, MEMBER_PORT, MEMBER_ROLE, MEMBER_STATE FROM performance_schema.replication_group_members"); e == nil {
|
|
if r.RowNumber() > 0 {
|
|
info.Type = "mysql-group-replication"
|
|
}
|
|
for row := 0; row < r.RowNumber(); row++ {
|
|
host, _ := r.GetString(row, 0)
|
|
port, _ := r.GetInt(row, 1)
|
|
role, _ := r.GetString(row, 2)
|
|
state, _ := r.GetString(row, 3)
|
|
if state == "ONLINE" {
|
|
info.Nodes = addNodes(info.Nodes, mysqlTopologyNode(host, int(port), role))
|
|
}
|
|
}
|
|
r.Close()
|
|
if info.Type != "unknown" {
|
|
return info, nil
|
|
}
|
|
}
|
|
r, e := c.Execute("SHOW REPLICA STATUS")
|
|
if e != nil {
|
|
r, e = c.Execute("SHOW SLAVE STATUS")
|
|
}
|
|
if e == nil {
|
|
if r.RowNumber() > 0 {
|
|
info.Type = "mysql-replication"
|
|
host, _ := r.GetStringByName(0, "Source_Host")
|
|
port, _ := r.GetIntByName(0, "Source_Port")
|
|
if host == "" {
|
|
host, _ = r.GetStringByName(0, "Master_Host")
|
|
port, _ = r.GetIntByName(0, "Master_Port")
|
|
}
|
|
info.Nodes = addNodes(info.Nodes, mysqlTopologyNode(host, int(port), "primary"))
|
|
}
|
|
r.Close()
|
|
}
|
|
if info.Type == "unknown" {
|
|
if r, e := c.Execute("SHOW REPLICAS"); e == nil {
|
|
if r.RowNumber() > 0 {
|
|
info.Type = "mysql-replication"
|
|
}
|
|
r.Close()
|
|
}
|
|
}
|
|
if info.Type == "unknown" {
|
|
info.Warning = "可确定节点只读状态，但无法完整读取复制拓扑；单机与暂无副本的主节点无法可靠区分。"
|
|
}
|
|
return info, nil
|
|
}
|
|
|
|
func probePostgres(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
|
|
cfg, err := pgconn.ParseConfig("postgres://localhost/postgres?sslmode=disable")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
cfg.Host, cfg.Port, cfg.User, cfg.Password = p.TargetHost, uint16(p.TargetPort), p.TargetUser, p.TargetPassword
|
|
if p.AuthSource != "" && p.AuthSource != "admin" {
|
|
cfg.Database = p.AuthSource
|
|
}
|
|
cfg.TLSConfig, err = transport.TLSConfig(p)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
cfg.Fallbacks = nil
|
|
cfg.DialFunc = pgconn.DialFunc(dial)
|
|
cfg.LookupFunc = func(_ context.Context, host string) ([]string, error) { return []string{host}, nil }
|
|
cfg.ConnectTimeout = 3 * time.Second
|
|
c, err := pgconn.ConnectConfig(ctx, cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer c.Close(context.Background())
|
|
r, err := c.Exec(ctx, "SELECT pg_is_in_recovery()").ReadAll()
|
|
if err != nil || len(r) == 0 || len(r[0].Rows) == 0 {
|
|
return nil, errors.New("PostgreSQL role unavailable")
|
|
}
|
|
info := singleInfo(p, "unknown", "primary")
|
|
if string(r[0].Rows[0][0]) == "t" {
|
|
info.Type = "pgsql-replication"
|
|
info.Selected.Role = "replica"
|
|
info.Nodes[0].Role = "replica"
|
|
rows, e := c.Exec(ctx, "SELECT sender_host, sender_port FROM pg_stat_wal_receiver").ReadAll()
|
|
if e == nil && len(rows) > 0 {
|
|
for _, row := range rows[0].Rows {
|
|
if len(row) < 2 {
|
|
continue
|
|
}
|
|
port, _ := strconv.Atoi(string(row[1]))
|
|
info.Nodes = addNodes(info.Nodes, model.ClusterNode{Host: string(row[0]), Port: port, Role: "primary"})
|
|
}
|
|
}
|
|
} else {
|
|
rows, e := c.Exec(ctx, "SELECT count(*) FROM pg_stat_replication").ReadAll()
|
|
if e == nil && len(rows) > 0 && len(rows[0].Rows) > 0 && string(rows[0].Rows[0][0]) != "0" {
|
|
info.Type = "pgsql-replication"
|
|
}
|
|
}
|
|
if info.Type == "unknown" {
|
|
info.Warning = "当前节点可写；没有可见复制连接时，无法区分单机与暂无副本的主节点。"
|
|
}
|
|
return info, nil
|
|
}
|
|
|
|
type sqlDialer struct {
|
|
contextDialer
|
|
host string
|
|
}
|
|
|
|
func (d sqlDialer) HostName() string { return d.host }
|
|
func probeSQLServer(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
|
|
tls, err := transport.TLSConfig(p)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
encryption := msdsn.Encryption(msdsn.EncryptionDisabled)
|
|
if p.TargetTLS {
|
|
encryption = msdsn.EncryptionRequired
|
|
}
|
|
c := mssql.NewConnectorConfig(msdsn.Config{Host: p.TargetHost, Port: uint64(p.TargetPort), User: p.TargetUser, Password: p.TargetPassword, Database: p.AuthSource, Encryption: encryption, TLSConfig: tls, Protocols: []string{"tcp"}, DisableRetry: true})
|
|
c.Dialer = sqlDialer{contextDialer{dial}, p.TargetHost}
|
|
db := sql.OpenDB(c)
|
|
defer db.Close()
|
|
var hadr, fc int
|
|
var primary sql.NullInt64
|
|
err = db.QueryRowContext(ctx, "SELECT CAST(SERVERPROPERTY('IsHadrEnabled') AS int), CAST(SERVERPROPERTY('IsClustered') AS int), sys.fn_hadr_is_primary_replica(DB_NAME())").Scan(&hadr, &fc, &primary)
|
|
if err != nil {
|
|
if err = db.PingContext(ctx); err != nil {
|
|
return nil, err
|
|
}
|
|
return singleInfo(p, "unknown", "unknown"), nil
|
|
}
|
|
info := singleInfo(p, "standalone", "standalone")
|
|
if fc == 1 {
|
|
info.Type = "sqlserver-fci"
|
|
info.Selected.Role = "primary"
|
|
}
|
|
if primary.Valid {
|
|
info.Type = "sqlserver-ag"
|
|
info.Selected.Role = "replica"
|
|
if primary.Int64 == 1 {
|
|
info.Selected.Role = "primary"
|
|
}
|
|
rows, e := db.QueryContext(ctx, "SELECT dns_name, port FROM sys.availability_group_listeners WHERE group_id IN (SELECT group_id FROM sys.availability_databases_cluster WHERE database_name = DB_NAME())")
|
|
if e == nil {
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var host string
|
|
var port int
|
|
if rows.Scan(&host, &port) == nil {
|
|
info.Nodes = addNodes(info.Nodes, model.ClusterNode{Host: host, Port: port, Role: "primary"})
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if hadr == 1 && !primary.Valid && fc == 0 {
|
|
info.Type = "unknown"
|
|
info.Warning = "已启用 Always On，但当前数据库不在可识别的可用性组中。"
|
|
}
|
|
info.Nodes[0].Role = info.Selected.Role
|
|
return info, nil
|
|
}
|
|
|
|
func probeRabbit(ctx context.Context, p model.Proxy, dial model.DialFunc) (*model.ClusterInfo, error) {
|
|
result, err := amqpprobe.Probe(ctx, p, dial)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
info := singleInfo(p, "unknown", "broker")
|
|
info.Name, info.Version = result.ClusterName, result.Version
|
|
info.Warning = "AMQP 不公布集群成员列表；请配置其他集群节点以支持新连接故障切换。"
|
|
if p.Cluster.Mode == "rabbitmq-cluster" {
|
|
info.Type = "rabbitmq-cluster"
|
|
}
|
|
return info, nil
|
|
}
|