|
| 1 | +/* |
| 2 | + * TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-DB管理系统(BlueKing-BK-DBM) available. |
| 3 | + * Copyright (C) 2017-2023 THL A29 Limited, a Tencent company. All rights reserved. |
| 4 | + * Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License. |
| 5 | + * You may obtain a copy of the License at https://opensource.org/licenses/MIT |
| 6 | + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on |
| 7 | + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the |
| 8 | + * specific language governing permissions and limitations under the License. |
| 9 | + */ |
| 10 | + |
| 11 | +package spiderctl |
| 12 | + |
| 13 | +import ( |
| 14 | + "errors" |
| 15 | + "fmt" |
| 16 | + "sync" |
| 17 | + |
| 18 | + "github.com/samber/lo" |
| 19 | + |
| 20 | + "dbm-services/common/go-pubpkg/logger" |
| 21 | + "dbm-services/mysql/db-tools/dbactuator/pkg/components" |
| 22 | + "dbm-services/mysql/db-tools/dbactuator/pkg/native" |
| 23 | +) |
| 24 | + |
| 25 | +// CheckTdbctlWithSpideRouterComp 检查spider和中控路由是否一致 |
| 26 | +type CheckTdbctlWithSpideRouterComp struct { |
| 27 | + GeneralParam *components.GeneralParam `json:"general"` |
| 28 | + Params CheckTdbctlWithSpideRouterParam `json:"extend"` |
| 29 | +} |
| 30 | + |
| 31 | +// CheckTdbctlWithSpideRouterParam 检查参数 |
| 32 | +type CheckTdbctlWithSpideRouterParam struct { |
| 33 | + Host string `json:"host" validate:"required,ip"` // 当前实例的主机地址 |
| 34 | + Port int `json:"port" validate:"required,lt=65536,gte=3306"` // 当前实例的端口 |
| 35 | +} |
| 36 | + |
| 37 | +// Example subcommand example input |
| 38 | +func (c CheckTdbctlWithSpideRouterComp) Example() interface{} { |
| 39 | + return CheckTdbctlWithSpideRouterComp{ |
| 40 | + Params: CheckTdbctlWithSpideRouterParam{ |
| 41 | + Host: "127.0.0.1", |
| 42 | + Port: 26000, |
| 43 | + }, |
| 44 | + } |
| 45 | +} |
| 46 | + |
| 47 | +// Run Run |
| 48 | +func (c *CheckTdbctlWithSpideRouterComp) Run() (err error) { |
| 49 | + user := c.GeneralParam.RuntimeAccountParam.MonitorUser |
| 50 | + pwd := c.GeneralParam.RuntimeAccountParam.MonitorPwd |
| 51 | + conn, err := native.InsObject{ |
| 52 | + Host: c.Params.Host, |
| 53 | + Port: c.Params.Port, |
| 54 | + User: user, |
| 55 | + Pwd: pwd, |
| 56 | + }.Conn() |
| 57 | + if err != nil { |
| 58 | + logger.Error("connect to tdbctl failed, err: %s", err.Error()) |
| 59 | + return err |
| 60 | + } |
| 61 | + defer conn.Close() |
| 62 | + tdbCtlConn := &native.TdbctlDbWork{DbWorker: *conn} |
| 63 | + logger.Info("开始检查SpiderMaster路由关系 ...") |
| 64 | + mspNodes, err := tdbCtlConn.GetMasterSpiderNodes() |
| 65 | + if err != nil { |
| 66 | + logger.Error("查询SpiderMaster节点信息失败: %s", err.Error()) |
| 67 | + return err |
| 68 | + } |
| 69 | + masterSptRouters, err := tdbCtlConn.GetMasterSptRouters() |
| 70 | + if err != nil { |
| 71 | + logger.Error("查询主分片节点信息失败: %s", err.Error()) |
| 72 | + return err |
| 73 | + } |
| 74 | + tdbCtlmasterSptRouters := lo.SliceToMap(masterSptRouters, func(item native.Server) (string, native.Server) { |
| 75 | + return item.ServerName, item |
| 76 | + }) |
| 77 | + err = checkRouter(mspNodes, tdbCtlmasterSptRouters) |
| 78 | + if err != nil { |
| 79 | + return err |
| 80 | + } |
| 81 | + sspNodes, err := tdbCtlConn.GetSlaveSpiderNodes() |
| 82 | + if err != nil { |
| 83 | + logger.Error("查询SpiderSlave节点信息失败: %s", err.Error()) |
| 84 | + return err |
| 85 | + } |
| 86 | + if len(sspNodes) == 0 { |
| 87 | + return err |
| 88 | + } |
| 89 | + slaveSptRouters, err := tdbCtlConn.GetSlaveSptRouters() |
| 90 | + if err != nil { |
| 91 | + logger.Error("查询Slave分片节点信息失败: %s", err.Error()) |
| 92 | + return err |
| 93 | + } |
| 94 | + tdbCtlslaveSptRouters := lo.SliceToMap(slaveSptRouters, func(item native.Server) (string, native.Server) { |
| 95 | + return item.ServerName, item |
| 96 | + }) |
| 97 | + logger.Info("检查从分片路由关系") |
| 98 | + err = checkRouter(sspNodes, tdbCtlslaveSptRouters) |
| 99 | + return err |
| 100 | +} |
| 101 | + |
| 102 | +func checkRouter(nodes []native.Server, tdbctlRouters map[string]native.Server) (err error) { |
| 103 | + var errs []error |
| 104 | + wg := sync.WaitGroup{} |
| 105 | + errChan := make(chan error) |
| 106 | + cChan := make(chan struct{}, 5) |
| 107 | + for _, node := range nodes { |
| 108 | + wg.Add(1) |
| 109 | + cChan <- struct{}{} |
| 110 | + go func(spiderNode native.Server) { |
| 111 | + logger.Info("开始检查 %s-%s的路由", spiderNode.ServerName, spiderNode.GetEndPoint()) |
| 112 | + defer func() { wg.Done(); <-cChan }() |
| 113 | + sconn, errx := spiderNode.GetConn() |
| 114 | + if errx != nil { |
| 115 | + logger.Error("connect to spider %s failed, err: %s", spiderNode.GetEndPoint(), err.Error()) |
| 116 | + errChan <- errx |
| 117 | + return |
| 118 | + } |
| 119 | + defer sconn.Close() |
| 120 | + var spiderSptRouters []native.Server |
| 121 | + if native.SvrNameIsSlaveSpiderShard(spiderNode.ServerName) { |
| 122 | + spiderSptRouters, errx = sconn.GetSlaveSptRouters() |
| 123 | + } else { |
| 124 | + spiderSptRouters, errx = sconn.GetMasterSptRouters() |
| 125 | + } |
| 126 | + if errx != nil { |
| 127 | + logger.Error("query mysql.servers failed, err: %s", err.Error()) |
| 128 | + errChan <- errx |
| 129 | + return |
| 130 | + } |
| 131 | + errx = compareRouter(tdbctlRouters, spiderSptRouters) |
| 132 | + if errx != nil { |
| 133 | + errChan <- fmt.Errorf("[%s-%s]:%w", spiderNode.ServerName, spiderNode.GetEndPoint(), errx) |
| 134 | + } |
| 135 | + |
| 136 | + }(node) |
| 137 | + } |
| 138 | + go func() { |
| 139 | + wg.Wait() |
| 140 | + close(errChan) |
| 141 | + }() |
| 142 | + for err = range errChan { |
| 143 | + errs = append(errs, err) |
| 144 | + } |
| 145 | + return errors.Join(errs...) |
| 146 | +} |
| 147 | + |
| 148 | +func compareRouter(tdbctlRouters map[string]native.Server, spiderrRouters []native.Server) (err error) { |
| 149 | + spiderrRoutersMap := lo.SliceToMap(spiderrRouters, func(item native.Server) (string, native.Server) { |
| 150 | + return item.ServerName, item |
| 151 | + }) |
| 152 | + for svrName, rt := range tdbctlRouters { |
| 153 | + spiderrRouter, ok := spiderrRoutersMap[svrName] |
| 154 | + if !ok { |
| 155 | + logger.Error("spider router not found router: %s", svrName) |
| 156 | + return err |
| 157 | + } |
| 158 | + if rt.Host != spiderrRouter.Host || rt.Port != spiderrRouter.Port { |
| 159 | + errMsg := "spider router not match tdbctl router:" |
| 160 | + errMsg += fmt.Sprintf("tdbctl router:%s %s %d\n", rt.ServerName, rt.Host, rt.Port) |
| 161 | + errMsg += fmt.Sprintf("spider router:%s %s %d\n", spiderrRouter.ServerName, spiderrRouter.Host, spiderrRouter.Port) |
| 162 | + return fmt.Errorf("%s", errMsg) |
| 163 | + } |
| 164 | + } |
| 165 | + return |
| 166 | +} |
0 commit comments