Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(mysql): mysql权限申请优化 #8650 #8653

Merged
merged 1 commit into from
Dec 23, 2024
Merged

feat(mysql): mysql权限申请优化 #8650 #8653

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions dbm-services/mysql/db-priv/service/add_priv_base_func.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -207,8 +207,14 @@ func GenerateBackendSQL(account TbAccounts, rule TbAccountRules, ips []string, m
}
// 备库域名只授予查询类权限
if clusterType == tendbha && tendbhaMasterDomain == false {
sql = fmt.Sprintf("GRANT SELECT, SHOW VIEW ON `%s`.* TO '%s'@'%s' %s;",
rule.Dbname, account.User, ip, identifiedByPassword)
if rule.Dbname == "%" || rule.Dbname == "*" {
sql = fmt.Sprintf("GRANT SELECT, SHOW VIEW ON *.* TO '%s'@'%s' %s;",
account.User, ip, identifiedByPassword)
} else {
sql = fmt.Sprintf("GRANT SELECT, SHOW VIEW ON `%s`.* TO '%s'@'%s' %s;",
rule.Dbname, account.User, ip, identifiedByPassword)
}

sqlTemp = append(sqlTemp, sql)
if containConnLogDBFlag {
sql = fmt.Sprintf("%s '%s'@'%s' %s;", insertConnLogPriv, account.User, ip, identifiedByPassword)
Expand All @@ -229,8 +235,14 @@ func GenerateBackendSQL(account TbAccounts, rule TbAccountRules, ips []string, m
}

if rule.DmlDdlPriv != "" {
sql = fmt.Sprintf("GRANT %s ON `%s`.* TO '%s'@'%s' %s;",
rule.DmlDdlPriv, rule.Dbname, account.User, ip, identifiedByPassword)
if rule.Dbname == "%" || rule.Dbname == "*" {
sql = fmt.Sprintf("GRANT %s ON *.* TO '%s'@'%s' %s;",
rule.DmlDdlPriv, account.User, ip, identifiedByPassword)
} else {
sql = fmt.Sprintf("GRANT %s ON `%s`.* TO '%s'@'%s' %s;",
rule.DmlDdlPriv, rule.Dbname, account.User, ip, identifiedByPassword)
}

sqlTemp = append(sqlTemp, sql)
if needInsertConnLogFlag {
sql = fmt.Sprintf("%s '%s'@'%s' %s;", insertConnLogPriv, account.User, ip, identifiedByPassword)
Expand Down
20 changes: 13 additions & 7 deletions dbm-services/mysql/db-priv/service/v2/add_priv/add_on_mysql.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,14 +64,15 @@ func (c *PrivTaskPara) addOneDtOnMySQL(
dt *service.TbAccountRules,
reports map[string][]string,
) error {
var ipStr string
var oneBatchClients []string
for idx, ip := range clientIps {
// 限长 1500 代码会比较好些, 不往极限的 2000 搞
ipStr = ipStr + "," + ip
if len(ipStr) > 1500 || idx == len(clientIps)-1 {
slog.Info("add one dt on mysql", slog.String("ipstr", ipStr))
// 限长 100 代码会比较好些, 不往极限的 2000 搞
oneBatchClients = append(oneBatchClients, ip)
if len(oneBatchClients) > 100 || idx == len(clientIps)-1 {
slog.Info("add one dt on mysql", slog.Any("one batch client", oneBatchClients))
// 一次跑一批 client
err := c.addOneDtOnMySQLForSplitClient(
strings.Trim(ipStr, ","),
strings.Join(oneBatchClients, ","),
workingInstances,
accountAndRuleDetails,
psw,
Expand All @@ -82,7 +83,7 @@ func (c *PrivTaskPara) addOneDtOnMySQL(
slog.Error("add on mysql", slog.String("err", err.Error()))
return err
}
ipStr = ""
oneBatchClients = []string{}
}
}
return nil
Expand Down Expand Up @@ -152,6 +153,7 @@ func readOneDtRes(bkCloudId int64, res []*drs.OneAddressResult, reports map[stri
slog.String("addr", r.Address),
)
reports[r.Address] = []string{r.ErrorMsg}
continue
}
readOneAddrRes(bkCloudId, r, reports)
}
Expand All @@ -163,6 +165,10 @@ func readOneAddrRes(bkCloudId int64, r *drs.OneAddressResult, reports map[string
return
}

if _, ok := reports[r.Address]; !ok {
reports[r.Address] = make([]string, 0)
}

_, sqlStat, msgText, isException := internal.ParseMySQLErrStr(errMsg)
if !isException {
reports[r.Address] = append(reports[r.Address], msgText)
Expand Down
15 changes: 12 additions & 3 deletions dbm-services/mysql/db-priv/service/v2/add_priv/add_priv.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"dbm-services/mysql/priv-service/service/v2/internal"
"encoding/json"
"log/slog"
"strings"
"time"

"github.com/pkg/errors"
Expand All @@ -17,7 +18,9 @@ func (c *PrivTaskPara) AddPriv(jsonPara, ticket string) (err error) {
slog.String("jsonPara", jsonPara),
)

if c.ClusterType == internal.ClusterTypeSqlServerHA || c.ClusterType == internal.ClusterTypeSqlServer || c.ClusterType == internal.ClusterTypeSqlServerSingle {
if c.ClusterType == internal.ClusterTypeSqlServerHA ||
c.ClusterType == internal.ClusterTypeSqlServer ||
c.ClusterType == internal.ClusterTypeSqlServerSingle {
return c.AddPrivForSqlserver(jsonPara)
}

Expand All @@ -39,6 +42,8 @@ func (c *PrivTaskPara) AddPriv(jsonPara, ticket string) (err error) {
// targetInstance 传入的其实全是域名
c.TargetInstances = internal.UniqueStringSlice(c.TargetInstances)

slog.Info("add priv", slog.String("source ips", strings.Join(c.SourceIPs, ",")))

// 写审计日志
service.AddPrivLog(
service.PrivLog{
Expand All @@ -56,6 +61,7 @@ func (c *PrivTaskPara) AddPriv(jsonPara, ticket string) (err error) {
slog.Error("add priv", slog.String("err", err.Error()))
return err
}
slog.Info("add priv", slog.Any("target meta infos", targetMetaInfos))

/*
TenDBSingle 授权是在存储实例操作
Expand All @@ -64,6 +70,7 @@ func (c *PrivTaskPara) AddPriv(jsonPara, ticket string) (err error) {
*/

// 开白名单
// proxy 白名单是前置集中开, 所有出错了直接返回
if c.ClusterType == internal.ClusterTypeTenDBHA {
err = c.addWhiteList(targetMetaInfos)
if err != nil {
Expand All @@ -80,7 +87,7 @@ func (c *PrivTaskPara) AddPriv(jsonPara, ticket string) (err error) {
clientIps, workingMySQLInstances := c.prepareMySQLPayload(targetMetaInfos)
slog.Info(
"add priv",
slog.Any("clientIps", clientIps),
slog.String("clientIps", strings.Join(clientIps, ",")),
slog.Any("workingMySQLInstances", workingMySQLInstances),
)

Expand All @@ -96,6 +103,8 @@ func (c *PrivTaskPara) AddPriv(jsonPara, ticket string) (err error) {
slog.String("accountAndRuleDetails", accountAndRuleDetails.String()),
)

// err 是调用函数出错, 直接报错返回
// reports 是实施授权的报告
reports, err := c.addOnMySQL(clientIps, workingMySQLInstances, accountAndRuleDetails)
if err != nil {
slog.Error("add priv", slog.String("err", err.Error()))
Expand All @@ -111,6 +120,6 @@ func (c *PrivTaskPara) AddPriv(jsonPara, ticket string) (err error) {
return errors.New(string(b))
}

slog.Info("add priv finish", slog.Any("reports", reports))
slog.Info("add priv finish")
return nil
}
110 changes: 54 additions & 56 deletions dbm-services/mysql/db-priv/service/v2/add_priv/add_proxy_white_list.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,71 +33,54 @@ func (c *PrivTaskPara) addWhiteList(targetMetaInfos []*service.Instance) (err er
slog.Any("proxies", workingProxies),
)

// 确实有要操作的 proxy
if len(workingProxies) > 0 {
var cmds []string
for _, clientIp := range c.SourceIPs {
//clientIp 可能是 localhost, 要忽略
if clientIp == "localhost" {
continue
}
cmds = append(
cmds,
fmt.Sprintf(`refresh_users('%s@%s', '+')`,
c.User, clientIp,
))
}
if len(workingProxies) <= 0 {
return nil
}

cmds := generateProxyCmds(c.SourceIPs, c.User)
slog.Info(
"add proxy white list",
slog.Any("cmds", cmds),
)

// drs 执行多个 sql 是循环一个一个来的
// 所以批量发送是可以的, 只是这么搞 drs 负载估计要炸
// 这里搞并发的意义不大
var errCollect error
for bkCloudId, addresses := range workingProxies {
slog.Info(
"add proxy white list",
slog.Any("cmds", cmds),
slog.Int64("bk_cloud_id", bkCloudId),
slog.Any("addresses", addresses),
)
drsRes, err := drs.RPCProxyAdmin(
bkCloudId,
addresses,
cmds,
false,
0,
)
if err != nil {
slog.Error("add proxy white list", slog.String("err", err.Error()))
return err
}

// drs 执行多个 sql 是循环一个一个来的
// 所以批量发送是可以的, 只是这么搞 drs 负载估计要炸
// 这里搞并发的意义不大
var errCollect error
for bkCloudId, addresses := range workingProxies {
// 错误要收集起来
ec := collectErrors(drsRes)
if ec != nil {
slog.Error("add proxy white list", slog.String("err collection", ec.Error()))
errCollect = errors.Join(errCollect, ec)
} else {
slog.Info(
"add proxy white list",
"add proxy white list success",
slog.Int64("bk_cloud_id", bkCloudId),
slog.Any("addresses", addresses),
)
drsRes, err := drs.RPCProxyAdmin(
bkCloudId,
addresses,
cmds,
false,
0,
)
if err != nil {
slog.Error(
"add proxy white list",
slog.Int64("bk_cloud_id", bkCloudId),
slog.String("err", err.Error()),
)
return err
}

// 错误要收集起来
ec := collectErrors(drsRes)
if ec != nil {
slog.Error(
"add proxy white list",
slog.Int64("bk_cloud_id", bkCloudId),
slog.String("err", ec.Error()),
)
errCollect = errors.Join(errCollect, ec)
} else {
slog.Info(
"add proxy white list success",
slog.Int64("bk_cloud_id", bkCloudId),
)
}
}
if errCollect != nil {
return errCollect
}
}
if errCollect != nil {
return errCollect
}

return nil
}

Expand All @@ -120,3 +103,18 @@ func collectErrors(res []*drs.OneAddressResult) (ec error) {
}
return ec
}

func generateProxyCmds(clientIps []string, username string) (cmds []string) {
for _, clientIp := range clientIps {
//clientIp 可能是 localhost, 要忽略
if clientIp == "localhost" {
continue
}
cmds = append(
cmds,
fmt.Sprintf(`refresh_users('%s@%s', '+')`,
username, clientIp,
))
}
return cmds
}
1 change: 1 addition & 0 deletions dbm-services/mysql/db-priv/service/v2/add_priv/fetch_target_dbmeta_info.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ func (c *PrivTaskPara) fetchTargetDBMetaInfo() ([]*service.Instance, error) {
)
return nil, err
}
slog.Info("fetch target db meta info", slog.Any("result", result))

res := make([]*service.Instance, 0)
err = json.Unmarshal(result.Data, &res)
Expand Down
13 changes: 11 additions & 2 deletions dbm-services/mysql/db-priv/service/v2/add_priv/prepare.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"dbm-services/mysql/priv-service/service/v2/internal"
"fmt"
"log/slog"
"strings"
)

func (c *PrivTaskPara) prepareMySQLPayload(targetMetaInfos []*service.Instance) (
Expand All @@ -27,7 +28,7 @@ func (c *PrivTaskPara) prepareMySQLPayload(targetMetaInfos []*service.Instance)
func (c *PrivTaskPara) prepareTenDBSingle(targetMetaInfos []*service.Instance) (
clientIps []string,
workingMySQLInstances map[int64][]string) {
clientIps = make([]string, 0)
clientIps = c.SourceIPs
workingMySQLInstances = make(map[int64][]string)

for _, ele := range targetMetaInfos {
Expand Down Expand Up @@ -58,13 +59,21 @@ func (c *PrivTaskPara) prepareTenDBHA(targetMetaInfos []*service.Instance) (

// 申请主域名权限要把来源替换为 proxy ip
// 如果集群有 padding proxy 属性, 则是把 proxy ip 追加到 client ip 里
slog.Info(
"prepare tendbha",
slog.String("bind to", ele.BindTo),
slog.Bool("padding proxy", ele.PaddingProxy),
)
if ele.BindTo == internal.MachineTypeProxy {
if ele.PaddingProxy {
clientIps = append(clientIps, proxyIps...)
} else {
clientIps = proxyIps
}
} else {
clientIps = c.SourceIPs
}
slog.Info("prepare tendbha", slog.String("clientIps", strings.Join(clientIps, ",")))
// TenDBHA 要在所有存储实例执行授权
for _, s := range ele.Storages {
if _, ok := workingMySQLInstances[ele.BkCloudId]; !ok {
Expand All @@ -81,7 +90,7 @@ func (c *PrivTaskPara) prepareTenDBHA(targetMetaInfos []*service.Instance) (
}

func (c *PrivTaskPara) prepareTenDBCluster(targetMetaInfos []*service.Instance) (clientIps []string, workingMySQLInstances map[int64][]string) {
clientIps = make([]string, 0)
clientIps = c.SourceIPs
workingMySQLInstances = make(map[int64][]string)

// 对应的 spider 上执行授权
Expand Down
Loading
Loading