package infra import ( "context" "encoding/json" "errors" "fmt" "io" "math" "net" "net/http" "sort" "strconv" "strings" "sync" "time" "github.com/ipipdotnet/ipdb-go" "google.golang.org/grpc/metadata" "github.com/tidbcloud/pkgv2/clusterproxy" "github.com/tidbcloud/pkgv2/secret/provider" "github.com/tidbcloud/pkgv2/secret/vault" "github.com/tidbcloud/pkgv2/secret/encrypt" "github.com/duke-git/lancet/v2/convertor" "github.com/duke-git/lancet/v2/slice" "github.com/spf13/viper" "github.com/patrickmn/go-cache" "github.com/tidbcloud/account/internel/dao/entity" "github.com/tidbcloud/account/commons/config" "github.com/tidbcloud/core-svc/idl/pbgen/proto/core" "github.com/tidbcloud/tiinf/errs" "github.com/tidbcloud/tiinf/monitor/tilog" tigrpc "github.com/tidbcloud/tiinf/transport/tigrpc/client" "go.buf.build/protocolbuffers/go/linuxgit/tidbcloud/base" "go.uber.org/zap" ) const ( CLUSTER_STATUS_ERROR = "cluster status error" CLUSTER_NOT_EXIST = "cluster is not exist" CLUSTER_NOT_SUPPORT_DBUSER = "cluster is not support dbuser" contextTimeoutMsg = "context deadline exceeded" accessDeniedMsg = "Access denied" // cluster version is not support dbuser ErrCodeTokenBasedAuthRequired = 60100039 // cluster status is not support dbuser ErrCodeUnexpectedClusterStatus = 60100040 // cluster is deleted ErrCodeDeletedCluster = 60100041 // ErrCodeCannotCreateClusterInAws can not create cluster in aws ErrCodeCannotCreateClusterInAws = 49900001 ) var ( //call cm timeout defaultTimeout = 10 * time.Second corePortalCli core.CoreServiceClient coreSvcCli core.CoreServiceClient coreSvcOnce sync.Once corePortalOnce sync.Once notActiveStatus = []core.ClusterStatus{ core.ClusterStatus_CLUSTER_STATUS_DELETED, core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE, core.ClusterStatus_CLUSTER_STATUS_CLEARED, } SupportDBUserStatus = []core.ClusterStatus{ core.ClusterStatus_CLUSTER_STATUS_NORMAL, core.ClusterStatus_CLUSTER_STATUS_HIBERNATED, core.ClusterStatus_CLUSTER_STATUS_SCALING, } CacheClusterStatus = []core.ClusterStatus{ core.ClusterStatus_CLUSTER_STATUS_NORMAL, core.ClusterStatus_CLUSTER_STATUS_DELETED, core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE, core.ClusterStatus_CLUSTER_STATUS_CLEARED, } PushKeyStatus = []core.ClusterStatus{ core.ClusterStatus_CLUSTER_STATUS_PREPARING, core.ClusterStatus_CLUSTER_STATUS_LAUNCHING, core.ClusterStatus_CLUSTER_STATUS_NORMAL, core.ClusterStatus_CLUSTER_STATUS_HIBERNATED, core.ClusterStatus_CLUSTER_STATUS_PAUSING, core.ClusterStatus_CLUSTER_STATUS_PAUSED, core.ClusterStatus_CLUSTER_STATUS_RESUMING, core.ClusterStatus_CLUSTER_STATUS_SCALING, core.ClusterStatus_CLUSTER_STATUS_RECOVERING, } ) var ( dbReadWriteRoleGrantSql = []string{ // query tables&users "GRANT SELECT ON information_schema.* TO '%s'@'%%';", "GRANT SELECT ON performance_schema.* TO '%s'@'%%';", "GRANT SELECT ON mysql.user TO '%s'@'%%';", // query all databases "GRANT SHOW DATABASES on *.* TO '%s'@'%%';", // create&drop table "GRANT CREATE ON *.* TO '%s'@'%%';", "GRANT CREATE TEMPORARY TABLES ON *.* TO '%s'@'%%';", "GRANT DROP ON *.* TO '%s'@'%%';", // modify table "GRANT INDEX ON *.* TO '%s'@'%%';", "GRANT REFERENCES ON *.* TO '%s'@'%%';", "GRANT ALTER ON *.* TO '%s'@'%%';", "GRANT CREATE VIEW ON *.* TO '%s'@'%%';", // process data "GRANT INSERT ON *.* TO '%s'@'%%';", "GRANT UPDATE ON *.* TO '%s'@'%%';", "GRANT DELETE ON *.* TO '%s'@'%%';", // query data "GRANT SELECT ON *.* TO '%s'@'%%';", "GRANT SHOW VIEW ON *.* TO '%s'@'%%';", } dbReadOnlyRoleGrantSql = []string{ // query tables&users "GRANT SELECT ON information_schema.tables TO '%s'@'%%';", "GRANT SELECT ON mysql.user TO '%s'@'%%';", // query all databases "GRANT SHOW DATABASES on *.* TO '%s'@'%%';", // query data "GRANT SELECT ON *.* TO '%s'@'%%';", "GRANT SHOW VIEW ON *.* TO '%s'@'%%';", } ) type InfraHandlerImpl struct{} func parseInfraResp(baseResp *base.BaseResp) error { if baseResp.ErrCode == errs.ErrCodeOK { return nil } return errs.NewInternalServerError("infra server error, "+fmt.Sprintf("code= %d", baseResp.ErrCode), nil) } func isLogError(err error, method string) bool { if err == nil { return false } if strings.Contains(err.Error(), contextTimeoutMsg) { return false } if strings.Contains(err.Error(), accessDeniedMsg) { return false } return true } func GetCorePortalClient() core.CoreServiceClient { addr := config.Global().GrpcConfig.CorePortalAddr //addr := "127.0.0.1:8888" ctx := context.Background() tilog.Info(ctx, "will Initializing core portal service client", zap.String("addr", addr)) corePortalOnce.Do(func() { conn, err := tigrpc.NewClientConn(addr, tigrpc.WithDefault()) if err != nil { panic(fmt.Sprintf("dial core-portal-svc: %s", err)) } corePortalCli = core.NewCoreServiceClient(conn) tilog.Info(ctx, "Initializing core portal service client", zap.String("addr", addr), zap.Any("err", err)) }) return corePortalCli } func GetCoreSvcClient() core.CoreServiceClient { addr := config.Global().GrpcConfig.CoreSvcAddr //addr = "127.0.0.1:8888" coreSvcOnce.Do(func() { conn, err := tigrpc.NewClientConn(addr, tigrpc.WithDefault()) if err != nil { panic(fmt.Sprintf("dial core-svc: %s", err)) } coreSvcCli = core.NewCoreServiceClient(conn) tilog.Info(context.Background(), "Initializing core service client", zap.String("addr", addr), zap.Any("err", err)) }) return coreSvcCli } func (*InfraHandlerImpl) CleanClusters(ctx context.Context, tenantID uint64, disableFinalBackup bool) error { client := GetCoreSvcClient() req := &core.CleanClustersReq{ TenantId: tenantID, IsDevTier: false, DisableFinalBackup: disableFinalBackup, } key := config.Global().GrpcConfig.CleanClusterKey //tilog.Info(ctx, "CleanClusters get api key", zap.String("key", key)) ctx = metadata.AppendToOutgoingContext(ctx, "api-key", key) resp, e := client.CleanClusters(ctx, req) if e != nil { return errs.NewInternalGRPC("CleanClusters", e) } return parseInfraResp(resp.BaseResp) } func (*InfraHandlerImpl) CleanFreeClusters(ctx context.Context, tenantID uint64, disableFinalBackup bool) error { client := GetCoreSvcClient() req := &core.CleanClustersReq{ TenantId: tenantID, IsDevTier: true, DisableFinalBackup: disableFinalBackup, } ctx = metadata.AppendToOutgoingContext(ctx, "api-key", config.Global().GrpcConfig.CleanClusterKey) resp, e := client.CleanClusters(ctx, req) if e != nil { return errs.NewInternalGRPC("CleanFreeClusters", e) } return parseInfraResp(resp.BaseResp) } func (*InfraHandlerImpl) CleanProjectResource(ctx context.Context, projectID uint64) error { client := GetCorePortalClient() req := &core.CleanProjectResourceReq{ ProjectId: projectID, } resp, e := client.CleanProjectResource(ctx, req) if e != nil { return errs.NewInternalGRPC("CleanProjectResource", e) } return parseInfraResp(resp.BaseResp) } // In the old architecture, vpc peering and network cr were deleted. // In the new architecture, vpc peering and vpc are deleted. func (*InfraHandlerImpl) CleanTenantResource(ctx context.Context, tenantID uint64) error { client := GetCorePortalClient() req := &core.CleanTenantResourceReq{ TenantId: tenantID, } resp, e := client.CleanTenantResource(ctx, req) if e != nil { return errs.NewInternalGRPC("CleanTenantResource", e) } return parseInfraResp(resp.BaseResp) } func (*InfraHandlerImpl) GetProjectClustersInStatus(ctx context.Context, projectID uint64, status ...core.ClusterStatus) ([]*core.Cluster, error) { client := GetCoreSvcClient() req := &core.ListClustersReq{ Filter: &core.ClusterFilter{ ProjectIds: []uint64{projectID}, Statuses: status, }, } resp, e := client.ListClusters(ctx, req) if e != nil { return nil, errs.NewInternalGRPC("GetProjectClustersInStatus", e) } if err := parseInfraResp(resp.BaseResp); err != nil { return nil, err } clusters := []*core.Cluster{} for _, clusterID := range resp.OrderedClusterIds { clusters = append(clusters, resp.Clusters[clusterID]) } return clusters, nil } func (*InfraHandlerImpl) GetProjectClustersNotInStatus(ctx context.Context, projectID uint64, status ...core.ClusterStatus) ([]*core.Cluster, error) { client := GetCoreSvcClient() req := &core.ListClustersReq{ Filter: &core.ClusterFilter{ ProjectIds: []uint64{projectID}, ExcludeStatuses: status, }, } resp, e := client.ListClusters(ctx, req) if e != nil { return nil, errs.NewInternalGRPC("GetProjectClustersNotInStatus", e) } if err := parseInfraResp(resp.BaseResp); err != nil { return nil, err } clusters := []*core.Cluster{} for _, clusterID := range resp.OrderedClusterIds { clusters = append(clusters, resp.Clusters[clusterID]) } tilog.Info(ctx, "query ListClusters :"+fmt.Sprintf("size=%d", len(clusters))) return clusters, nil } func (*InfraHandlerImpl) GetOrgClusterSupportDbUser(ctx context.Context, orgID uint64) ([]*core.Cluster, error) { req := &core.ListClustersReq{ Filter: &core.ClusterFilter{ TenantIds: []uint64{orgID}, ExcludeStatuses: notActiveStatus, }, } return listCluster(ctx, req) } func (*InfraHandlerImpl) GetProjectClusterSupportDbUser(ctx context.Context, projectIds []uint64) ([]*core.Cluster, error) { req := &core.ListClustersReq{ Filter: &core.ClusterFilter{ ProjectIds: projectIds, ExcludeStatuses: notActiveStatus, }, } return listCluster(ctx, req) } func (*InfraHandlerImpl) GetNormalStatusClusters(ctx context.Context, orgID uint64, projectIds []uint64) ([]*core.Cluster, error) { filter := &core.ClusterFilter{ Statuses: []core.ClusterStatus{ core.ClusterStatus_CLUSTER_STATUS_NORMAL, }, } if orgID != 0 { filter.TenantIds = []uint64{orgID} } if len(projectIds) > 0 { filter.ProjectIds = projectIds } req := &core.ListClustersReq{ Filter: filter, } return listCluster(ctx, req) } func (*InfraHandlerImpl) CountClustersByProject(ctx context.Context, projectIds []uint64) (int64, error) { req := &core.CountClustersReq{ Filter: &core.ClusterFilter{ ProjectIds: projectIds, ExcludeStatuses: notActiveStatus, }, } return contClusters(ctx, req) } func (*InfraHandlerImpl) CountClustersByOrg(ctx context.Context, orgIds []uint64) (int64, error) { req := &core.CountClustersReq{ Filter: &core.ClusterFilter{ TenantIds: orgIds, ExcludeStatuses: notActiveStatus, }, } return contClusters(ctx, req) } func (*InfraHandlerImpl) CountClusters(ctx context.Context, filter *core.ClusterFilter) (int64, error) { req := &core.CountClustersReq{ Filter: filter, } return contClusters(ctx, req) } func contClusters(ctx context.Context, req *core.CountClustersReq) (int64, error) { tilog.Info(ctx, "contClusters ", zap.Any("filer", req.Filter)) client := GetCoreSvcClient() resp, e := client.CountClusters(ctx, req) if e != nil { return 0, errs.NewInternalGRPC("contClusters", e) } if err := parseInfraResp(resp.BaseResp); err != nil { return 0, err } count := resp.Count tilog.Info(ctx, "contClusters "+fmt.Sprintf("count=%d", count)) return count, nil } func (*InfraHandlerImpl) CountNetworkPeers(ctx context.Context, filter *core.NetworkPeerFilter) (int64, error) { tilog.Info(ctx, "CountNetworkPeer ", zap.Any("filer", filter)) req := &core.ListNetworkPeerReq{ Filter: filter, } client := GetCoreSvcClient() resp, e := client.ListNetworkPeer(ctx, req) if e != nil { return 0, errs.NewInternalGRPC("CountNetworkPeers", e) } if err := parseInfraResp(resp.BaseResp); err != nil { return 0, err } var count int64 if resp.PeeringList != nil { count = int64(len(resp.PeeringList)) tilog.Info(ctx, "resp.PeeringList ", zap.Any("resp.PeeringList", resp.PeeringList)) } tilog.Info(ctx, "CountNetworkPeer "+fmt.Sprintf("count=%d", count)) return count, nil } func listCluster(ctx context.Context, req *core.ListClustersReq) ([]*core.Cluster, error) { client := GetCoreSvcClient() resp, e := client.ListClusters(ctx, req) tilog.Info(ctx, "listCluster result", zap.Any("req", req), zap.Any("resp", resp), zap.Any("err", e)) if e != nil { tilog.Warn(ctx, "listCluster", zap.Any("req", req), zap.Error(e)) return nil, errs.NewInternalGRPC("ListClusters", e) } if err := parseInfraResp(resp.BaseResp); err != nil { return nil, err } //tilog.Info(ctx, "listCluster ", zap.Any("req", req), zap.Any("resp", resp)) clusters := []*core.Cluster{} for _, clusterID := range resp.OrderedClusterIds { c := resp.Clusters[clusterID] features := c.Features for _, f := range features { // the cluster support dbuser if f == core.AvailableFeature_DELEGATE_USER { clusters = append(clusters, c) continue } } } return clusters, nil } // IsDedicatedCluster // // @Description: call CoreSvc API by clusterId // @Author wangletian 2022-11-25 15:00:39 // @param clusterId // @return bool // @return error func (*InfraHandlerImpl) IsDedicatedCluster(ctx context.Context, clusterId uint64) (bool, error) { client := GetCoreSvcClient() req := &core.ListClustersReq{ Filter: &core.ClusterFilter{ ClusterIds: []uint64{clusterId}, }, } resp, err := client.ListClusters(ctx, req) tilog.Info(ctx, "query cluster ", zap.Any("clusterId", clusterId), zap.Any("resp", resp)) if err != nil { tilog.Error(ctx, "IsDedicatedCluster err", zap.Error(err), zap.Int64("ErrCode", resp.BaseResp.ErrCode), zap.String("ErrMsg", resp.BaseResp.ErrMsg)) return false, err } tilog.Info(ctx, "query IsDedicatedCluster", zap.Any("clusters", resp), zap.Any("IsDedicatedCluster", false)) if len(resp.OrderedClusterIds) > 0 { c := resp.Clusters[clusterId] for _, s := range notActiveStatus { if c.GetStatus() == s { return false, errors.New(CLUSTER_NOT_EXIST) } } statusFlag := false for _, s := range SupportDBUserStatus { if c.GetStatus() == s { statusFlag = true } } if !statusFlag { tilog.Info(ctx, "query IsDedicatedCluster", zap.Any("clusterId", clusterId), zap.Any("status", c.Status)) return false, errors.New(CLUSTER_STATUS_ERROR) } if c.IsDevTier { tilog.Info(ctx, "query IsDedicatedCluster", zap.Any("clusterId", clusterId), zap.Any("IsDedicatedCluster", false)) return false, nil } } else { tilog.Info(ctx, "query IsDedicatedCluster result , the cluster is not exist", zap.Any("clusterId", clusterId)) return false, errors.New(CLUSTER_NOT_EXIST) } return true, nil } func (*InfraHandlerImpl) GetClusterById(ctx context.Context, clusterId uint64, useCache bool, status ...core.ClusterStatus) (*core.Cluster, error) { tilog.Info(ctx, "get cluster by id request", zap.Any("ClusterId", clusterId)) key := strconv.FormatUint(clusterId, 10) if useCache { c := GetClusterCache(ctx) if x, found := c.Get(key); found { cluster := x.(*core.Cluster) tilog.Info(ctx, "get cluster form cache", zap.Any("ClusterId", key), zap.Any("cluster", cluster)) return cluster, nil } } client := GetCoreSvcClient() req := &core.ListClustersReq{ Filter: &core.ClusterFilter{ ClusterIds: []uint64{clusterId}, ExcludeStatuses: notActiveStatus, }, } tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() resp, err := client.ListClusters(tctx, req) tilog.Info(ctx, "call CM GetClusterById result ", zap.Any("clusterId", clusterId), zap.Any("req", req), zap.Any("resp", resp)) if err != nil { if isLogError(err, "ListClusters") { tilog.Warn(ctx, "GetClusterById, query cm cluster", zap.Error(err), zap.Any("req", req)) } return nil, err } //tilog.Info(ctx, "GetClusterById", zap.Any("clusters", resp), zap.Any("IsDedicatedCluster", false)) if len(resp.OrderedClusterIds) > 0 { cluster := resp.Clusters[clusterId] //if status param is not null if len(status) > 0 { statusFlag := false for _, s := range status { if cluster.GetStatus() == s { statusFlag = true } } if !statusFlag { tilog.Info(ctx, "GetClusterById the status is not in param status", zap.Any("clusterId", clusterId), zap.Any("status", cluster.GetStatus()), zap.Any("filterStatus", status)) return nil, nil } } if cluster != nil && cluster.Id != 0 && slice.Contain(CacheClusterStatus, cluster.Status) { // set cache c := GetClusterCache(ctx) c.Set(key, cluster, cache.DefaultExpiration) tilog.Info(ctx, "GetClusterById set cache", zap.Any("clusterId", clusterId), zap.Any("cluster", cluster)) } return cluster, nil } tilog.Info(ctx, "GetClusterById result , the cluster is not exist", zap.Any("clusterId", clusterId)) return nil, nil } var isGoProxy = true func isPrintErrLog(command *entity.CommandDBUserCluster) bool { if command == nil { return false } failNum := command.FailNum if failNum >= 3 && failNum%3 == 0 { return true } return false } func (i *InfraHandlerImpl) CreateDBUser(ctx context.Context, clusterId uint64, OperatorUserName string, operatorAuthMethod clusterproxy.AuthMethod, jwtToken string, userEmail string, dbUser string, ignoreExists bool, command *entity.CommandDBUserCluster, dbRole string) entity.CommandStatus { if isGoProxy { cluster, err := i.GetClusterById(ctx, clusterId, true) if err != nil { if isPrintErrLog(command) { tilog.Error(ctx, "CreateDBUser GetClusterById error", zap.Any("clusterId", clusterId), zap.Any("err", err)) } else { tilog.Warn(ctx, "CreateDBUser GetClusterById error", zap.Any("clusterId", clusterId), zap.Any("err", err)) } return entity.CommandStatusFail } operator := &clusterproxy.ClusterOperator{ Username: OperatorUserName, AuthMethod: operatorAuthMethod, Credential: jwtToken, } var builtinRole *string = nil var customRoles []string if dbRole == entity.DbRoleAdmin { var tRole = BuiltinRoleNameAdmin builtinRole = &tRole } else { err = CheckAndCreateCustomRoleWithProxy(ctx, cluster, operator, dbRole) if err != nil { if isPrintErrLog(command) { tilog.Error(ctx, "CreateDBUser CheckAndCreateCustomRoleWithProxy error", zap.Error(err), zap.Any("clusterId", clusterId), zap.Any("dbRole", dbRole), zap.Any("operator", operator.Username), zap.Any("err", err)) } else { tilog.Warn(ctx, "CreateDBUser CheckAndCreateCustomRoleWithProxy error", zap.Error(err), zap.Any("clusterId", clusterId), zap.Any("dbRole", dbRole), zap.Any("operator", operator.Username), zap.Any("err", err)) } return entity.CommandStatusFail } customRoles = []string{dbRole} } params := &CreateUserParams{ Cluster: cluster, Operator: operator, Settings: &core.CreateDBUserReq_Settings{IgnoreExists: ignoreExists}, CreateUserInfo: &CreateUserInfo{ //UserType: DBUserDelegate, UserName: dbUser, UserAttribute: map[string]string{ "email": userEmail, }, AuthMethod: clusterproxy.AuthMethodToken, BuiltinRole: builtinRole, CustomRoles: customRoles, }, } err = CreateUser(ctx, params) if err != nil { if isPrintErrLog(command) { tilog.Error(ctx, "CreateDBUser CreateUser error", zap.Any("clusterId", cluster.Id), zap.Any("operator", operator.Username), zap.Any("username", dbUser), zap.Any("err", err)) } else { tilog.Warn(ctx, "CreateDBUser CreateUser error", zap.Any("clusterId", cluster.Id), zap.Any("operator", operator.Username), zap.Any("username", dbUser), zap.Any("err", err)) } return entity.CommandStatusFail } tilog.Info(ctx, "CreateDBUser CreateUser success", zap.Any("clusterId", cluster.Id), zap.Any("operator", operator.Username), zap.Any("username", dbUser), zap.Any("email", userEmail), zap.Any("builtinRole", builtinRole), zap.Any("customRoles", customRoles)) return entity.CommandStatusComplete } client := GetCoreSvcClient() operator := &core.DBUserOperator{ Username: OperatorUserName, AuthMethod: core.DBUser_AUTH_METHOD_TOKEN, Credential: jwtToken, } var builtinRole = core.DBUser_BUILTIN_ROLE_UNSPECIFIED var customRoles []string if dbRole == entity.DbRoleAdmin { builtinRole = core.DBUser_BUILTIN_ROLE_USER_ADMIN } else { err := CheckAndCreateCustomRole(ctx, clusterId, dbRole, operator) if err != nil { tilog.Error(ctx, "CreateDBUser CheckAndCreateCustomRole error", zap.Error(err), zap.Any("clusterId", clusterId), zap.Any("dbRole", dbRole), zap.Any("operator", operator)) return entity.CommandStatusFail } customRoles = []string{dbRole} } // check the user whether exists and ignore if already exists before trying to create settings := &core.CreateDBUserReq_Settings{IgnoreExists: ignoreExists} req := &core.CreateDBUserReq{ ClusterId: clusterId, Email: &userEmail, Username: dbUser, AuthMethod: core.DBUser_AUTH_METHOD_TOKEN, Operator: operator, UserType: core.DBUser_TYPE_DELEGATE, BuiltinRole: builtinRole, CustomRoles: customRoles, Settings: settings, } tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() resp, e := client.CreateDBUser(tctx, req) tilog.Info(ctx, "CreateDBUser resp", zap.Any("resp", resp)) if e != nil { tilog.Error(ctx, "CreateDBUser error", zap.Error(e), zap.Any("CreateDBUserReq", req)) return entity.CommandStatusFail } var methodName string if ignoreExists { methodName = "Check-CreateDBUser" } else { methodName = "CreateDBUser" } param := parseReq{ methodName: methodName, clusterId: req.ClusterId, req: req, baseResp: resp.BaseResp, proxyError: resp.ProxyError, } if command != nil { param.command = *command } result := parseDBuserResp(ctx, param) tilog.Info(ctx, "call CM CreateDBUser result", zap.Any("result", result), zap.Any("email", req.Email), zap.Any("dbUser", dbUser), zap.Any("clusterId", clusterId), zap.Any("resp.BaseResp", resp.BaseResp)) return result } func (i *InfraHandlerImpl) UpdateSqlUser(ctx context.Context, req *core.UpdateDBUserReq, cloudBuiltinRole string) error { if isGoProxy { cluster, err := i.GetClusterById(ctx, req.ClusterId, true) if err != nil { tilog.Error(ctx, "UpdateSqlUser GetClusterById error", zap.Any("clusterId", req.ClusterId), zap.Any("err", err)) return err } operator := &clusterproxy.ClusterOperator{ Username: req.Operator.Username, AuthMethod: clusterproxy.AuthMethodToken, Credential: req.Operator.Credential, } if req.Operator.AuthMethod == core.DBUser_AUTH_METHOD_PASSWORD { operator.AuthMethod = clusterproxy.AuthMethodPassword } if cloudBuiltinRole != "" && cloudBuiltinRole != entity.DbRoleAdmin { // check cloudBuiltinRole is exits in cluster err = CheckAndCreateCustomRoleWithProxy(ctx, cluster, operator, cloudBuiltinRole) if err != nil { if isLogError(err, "UpdateSqlUser") { tilog.Error(ctx, "UpdateSqlUser CheckAndCreateCustomRoleWithProxy error", zap.Error(err), zap.Any("clusterId", req.ClusterId), zap.Any("cloudBuiltinRole", cloudBuiltinRole), zap.Any("operator", req.Operator.Username)) } return err } } var builtinRole = "" if req.BuiltinRole != nil { if *req.BuiltinRole == core.DBUser_BUILTIN_ROLE_USER_ADMIN { var tRole = BuiltinRoleNameAdmin builtinRole = tRole } else if *req.BuiltinRole == core.DBUser_BUILTIN_ROLE_CDC_ADMIN { var tRole = BuiltinRoleNameCDCAdmin builtinRole = tRole } } err = UpdateUser(ctx, &UpdateUserParams{ Cluster: cluster, Operator: operator, Username: req.Username, BuiltinRole: &builtinRole, CustomRoles: req.CustomRoles, Password: req.Password, ClearCustomRoles: req.ClearCustomRoles, }) if err != nil { if isLogError(err, "UpdateSqlUser") { tilog.Error(ctx, "UpdateSqlUser UpdateUser error", zap.Error(err), zap.Any("clusterId", req.ClusterId), zap.Any("username", req.Username), zap.Any("operator", req.Operator), zap.Any("err", err)) } return err } return nil } tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() client := GetCoreSvcClient() reqParam := fmt.Sprintf("Username=%s,ClusterId=%d,BuiltinRole=%s,CustomRoles=%s", req.Username, req.ClusterId, req.BuiltinRole, req.CustomRoles) tilog.Info(ctx, "UpdateSQLUser param:"+reqParam) if cloudBuiltinRole != "" && cloudBuiltinRole != entity.DbRoleAdmin { // check cloudBuiltinRole is exits in cluster err := CheckAndCreateCustomRole(ctx, req.ClusterId, cloudBuiltinRole, req.Operator) if err != nil { tilog.Error(ctx, "CreateDBUser CheckAndCreateCustomRole error", zap.Error(err), zap.Any("clusterId", req.ClusterId), zap.Any("cloudBuiltinRole", cloudBuiltinRole), zap.Any("operator", req.Operator)) return err } } resp, err := client.UpdateDBUser(tctx, req) tilog.Info(ctx, "UpdateSQLUser resp", zap.Any("resp", resp)) if err != nil { tilog.Error(ctx, "UpdateSQLUser error", zap.Any("err", err.Error()), zap.Any("UpdateSQLUser", reqParam)) return err } param := parseReq{ methodName: "UpdateSqlUser", clusterId: req.ClusterId, req: req, proxyError: resp.ProxyError, baseResp: resp.BaseResp} return parseBaseResp(ctx, param) } func (i *InfraHandlerImpl) CreateGeneralDBUser(ctx context.Context, req *core.CreateDBUserReq, checkCloudBuiltinRole string) error { if isGoProxy { cluster, err := i.GetClusterById(ctx, req.ClusterId, true) if err != nil { tilog.Error(ctx, "CreateGeneralDBUser GetClusterById error", zap.Any("clusterId", req.ClusterId), zap.Any("err", err)) return err } operator := &clusterproxy.ClusterOperator{ Username: req.Operator.Username, AuthMethod: clusterproxy.AuthMethodToken, Credential: req.Operator.Credential, } if req.Operator.AuthMethod == core.DBUser_AUTH_METHOD_PASSWORD { operator.AuthMethod = clusterproxy.AuthMethodPassword } if checkCloudBuiltinRole != "" && checkCloudBuiltinRole != entity.DbRoleAdmin { // check cloudBuiltinRole is exits in cluster err = CheckAndCreateCustomRoleWithProxy(ctx, cluster, operator, checkCloudBuiltinRole) if err != nil { if isLogError(err, "CreateDBUser") { tilog.Error(ctx, "CreateGeneralDBUser CheckAndCreateCustomRoleWithProxy error", zap.Error(err), zap.Any("clusterId", req.ClusterId), zap.Any("cloudBuiltinRole", checkCloudBuiltinRole), zap.Any("operator", operator.Username), zap.Any("err", err)) } return err } } credential := "" if req.Credential != nil { password := getBase64Password(ctx, *req.Credential) credential = *password } userEmail := "" if req.Email != nil { userEmail = *req.Email } userInfo := &CreateUserInfo{ UserName: req.Username, UserAttribute: map[string]string{ "email": userEmail, }, EncryptCredential: credential, CustomRoles: req.CustomRoles, } /*if req.UserType == core.DBUser_TYPE_USER_DEFINED { userInfo.UserType = DBUserUserDefined } else if req.UserType == core.DBUser_TYPE_DELEGATE { userInfo.UserType = DBUserDelegate }*/ if req.AuthMethod == core.DBUser_AUTH_METHOD_PASSWORD { userInfo.AuthMethod = clusterproxy.AuthMethodPassword } else if req.AuthMethod == core.DBUser_AUTH_METHOD_TOKEN { userInfo.AuthMethod = clusterproxy.AuthMethodToken } if req.BuiltinRole == core.DBUser_BUILTIN_ROLE_USER_ADMIN { var tRole = BuiltinRoleNameAdmin userInfo.BuiltinRole = &tRole } else if req.BuiltinRole == core.DBUser_BUILTIN_ROLE_CDC_ADMIN { var tRole = BuiltinRoleNameCDCAdmin userInfo.BuiltinRole = &tRole } params := CreateUserParams{ Cluster: cluster, Operator: operator, Settings: req.Settings, CreateUserInfo: userInfo, } err = CreateUser(ctx, ¶ms) if err != nil { if isLogError(err, "CreateDBUser") { tilog.Error(ctx, "CreateGeneralDBUser CreateUser error", zap.Error(err), zap.Any("clusterId", req.ClusterId), zap.Any("cloudBuiltinRole", checkCloudBuiltinRole), zap.Any("operator", operator.Username), zap.Any("err", err)) } return err } return nil } //tilog.Info(ctx, "CreateGeneralDBUser", zap.Any("req", req)) reqParam := fmt.Sprintf("Username=%s,ClusterId=%d,BuiltinRole=%s,CustomRoles=%s", req.Username, req.ClusterId, req.BuiltinRole, req.CustomRoles) tilog.Info(ctx, "CreateGeneralDBUser param: "+reqParam) tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() client := GetCoreSvcClient() if checkCloudBuiltinRole != "" && checkCloudBuiltinRole != entity.DbRoleAdmin { // check cloudBuiltinRole is exits in cluster err := CheckAndCreateCustomRole(ctx, req.ClusterId, checkCloudBuiltinRole, req.Operator) if err != nil { tilog.Error(ctx, "CreateGeneralDBUser CheckAndCreateCustomRole error", zap.Error(err), zap.Any("clusterId", req.ClusterId), zap.Any("cloudBuiltinRole", checkCloudBuiltinRole), zap.Any("operator", req.Operator)) return err } } req.Credential = getBase64Password(ctx, *req.Credential) resp, err := client.CreateDBUser(tctx, req) tilog.Info(ctx, "CreateDBUser resp", zap.Any("resp", resp), zap.Any("err", err)) if err != nil { alertMsg := fmt.Sprintf("ERROR %d: %s", resp.ProxyError.SqlErrNo, resp.ProxyError.SqlErrMsg) tilog.Error(ctx, "CreateGeneralDBUser error "+alertMsg, zap.Any("reqParam", req.Username)) return err } param := parseReq{ proxyError: resp.ProxyError, methodName: "CreateSqlUser", clusterId: req.ClusterId, req: req, baseResp: resp.BaseResp} return parseBaseResp(ctx, param) } func (i *InfraHandlerImpl) ListSqlUser(ctx context.Context, req *core.ListDBUsersReq) (*core.ListDBUsersResp, error) { if req.ClusterId == 0 { return nil, errs.NewBadRequest("cluster_id invalid", nil) } if isGoProxy { cluster, err := i.GetClusterById(ctx, req.ClusterId, true) if err != nil { tilog.Error(ctx, "ListSqlUser GetClusterById error", zap.Any("clusterId", req.ClusterId), zap.Any("err", err)) return nil, err } operator := &clusterproxy.ClusterOperator{ Username: req.Operator.Username, AuthMethod: clusterproxy.AuthMethodToken, Credential: req.Operator.Credential, } if req.Operator.AuthMethod == core.DBUser_AUTH_METHOD_PASSWORD { operator.AuthMethod = clusterproxy.AuthMethodPassword } params := ListUsersParams{ Cluster: cluster, Operator: operator, Filter: req.Filter, } sqlVals, dbUsers, err := ListUsers(ctx, ¶ms) if err != nil { if errs.IsError(err, errs.TiErr) { if e, ok := errs.AsTiError(err); ok && e.BizCode() == errs.ErrCodeForbidden { tilog.Warn(ctx, "ListSqlUser ListUsers error", zap.Any("clusterId", req.ClusterId), zap.Any("operator", operator.Username), zap.Any("err", err)) rsp := &core.ListDBUsersResp{ ResultMap: map[string]*core.SqlValues{}, DbUsers: []*core.DBUser{}, ProxyError: &core.ClusterProxyError{}, BaseResp: &base.BaseResp{}, } return rsp, nil } } if isLogError(err, "ListSqlUser") { tilog.Error(ctx, "ListSqlUser ListUsers error", zap.Any("clusterId", req.ClusterId), zap.Any("operator", operator.Username), zap.Any("err", err)) } return nil, err } var coreDBUsers = []*core.DBUser{} for _, u := range dbUsers { coreUser := &core.DBUser{ Username: u.Username, CustomRoles: u.CustomRoles, Plugin: u.Plugin, AccountLocked: u.AccountLocked, } if u.BuiltinRole != "" { if u.BuiltinRole == BuiltinRoleNameAdmin { coreUser.BuiltinRole = core.DBUser_BUILTIN_ROLE_USER_ADMIN } else if u.BuiltinRole == BuiltinRoleNameCDCAdmin { coreUser.BuiltinRole = core.DBUser_BUILTIN_ROLE_CDC_ADMIN } } coreDBUsers = append(coreDBUsers, coreUser) } rsp := &core.ListDBUsersResp{ ResultMap: sqlVals, DbUsers: coreDBUsers, ProxyError: &core.ClusterProxyError{}, BaseResp: &base.BaseResp{}, } tilog.Info(ctx, "ListSqlUser result", zap.Any("clusterId", req.ClusterId), zap.Any("operator", operator.Username), zap.Any("rsp", rsp)) return rsp, nil } tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() client := GetCoreSvcClient() tilog.Info(ctx, "Call CM ListDBUsers req", zap.Any("req", req)) resp, err := client.ListDBUsers(tctx, req) tilog.Info(ctx, "Call CM ListDBUsers resp", zap.Any("resp", resp)) if err != nil { tilog.Error(ctx, "ListSqlUser error", zap.Error(err), zap.Any("CreateDBUserReq", req)) return resp, errs.NewInternalGRPC("", err) } tilog.Info(ctx, "ListSqlUser result", zap.Any("clusterId", req.ClusterId), zap.Any("resp", resp)) //str := fmt.Sprintf("Call CM %s result:", "ListDBUsers") param := parseReq{ methodName: "ListSqlUser", clusterId: req.ClusterId, req: req, baseResp: resp.BaseResp} return resp, parseBaseResp(ctx, param) } func (*InfraHandlerImpl) GetClusterConfidentialInfo(ctx context.Context, req *core.GetClusterConfidentialInfoReq) (*core.GetClusterConfidentialInfoResp, error) { tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() client := GetCoreSvcClient() resp, err := client.GetClusterConfidentialInfo(tctx, req) if err != nil { tilog.Warn(ctx, "GetClusterConfidentialInfo error", zap.Error(err), zap.Any("req", req)) return resp, errs.NewInternalGRPC("", err) } param := parseReq{ methodName: "GetClusterConfidentialInfo", clusterId: req.ClusterId, req: req, baseResp: resp.BaseResp} return resp, parseBaseResp(ctx, param) } func getDbRoleGrantRevokeSql(dbRole string) (grant, revoke string, err error) { if strings.HasSuffix(dbRole, entity.DbRoleReadWrite) { var grantSqls []string for _, sql := range dbReadWriteRoleGrantSql { oneSql := fmt.Sprintf(sql, dbRole) grantSqls = append(grantSqls, oneSql) } return strings.Join(grantSqls, " "), "", nil } if strings.HasSuffix(dbRole, entity.DbRoleReadOnly) { var grantSqls []string for _, sql := range dbReadOnlyRoleGrantSql { oneSql := fmt.Sprintf(sql, dbRole) grantSqls = append(grantSqls, oneSql) } return strings.Join(grantSqls, " "), "", nil } return "", "", fmt.Errorf("unknown db role: %s", dbRole) } func CheckAndCreateCustomRoleWithProxy(ctx context.Context, cluster *core.Cluster, operator *clusterproxy.ClusterOperator, dbRole string) error { existing, err := IsUserExisting(ctx, cluster, operator, dbRole) tilog.Info(ctx, "CheckAndCreateCustomRoleWithProxy IsUserExisting result", zap.Any("clusterId", cluster.Id), zap.Any("operator", operator.Username), zap.Any("dbRole", dbRole), zap.Any("existing", existing), zap.Any("err", err)) if err != nil { return err } if existing { return nil } grant, revoke, err := getDbRoleGrantRevokeSql(dbRole) if err != nil { return err } params := &OperateRoleParams{ Cluster: cluster, Operator: operator, RoleName: dbRole, Grants: grant, Revocation: revoke, } err = CreateRole(ctx, params) tilog.Info(ctx, "CheckAndCreateCustomRole CreateRole result", zap.Any("params", params), zap.Any("err", err)) return err } func CheckAndCreateCustomRole(ctx context.Context, clusterId uint64, dbRole string, operator *core.DBUserOperator) error { client := GetCoreSvcClient() existingRoleReq := &core.IsDBUserExistingReq{ Operator: operator, ClusterId: clusterId, Username: dbRole, } existingRoleRsp, err := client.IsDBUserExisting(ctx, existingRoleReq) tilog.Info(ctx, "CheckAndCreateCustomRole IsDBUserExisting result", zap.Any("req", existingRoleReq), zap.Any("rsp", existingRoleRsp), zap.Any("err", err)) if err != nil { return err } if existingRoleRsp.Existing { return nil } grant, revoke, err := getDbRoleGrantRevokeSql(dbRole) if err != nil { return err } createRoleReq := &core.CreateDBRoleReq{ Operator: operator, ClusterId: clusterId, RoleName: dbRole, Grants: grant, Revocation: revoke, } createRoleRsp, err := client.CreateDBRole(ctx, createRoleReq) tilog.Info(ctx, "CheckAndCreateCustomRole CreateDBRole result", zap.Any("req", createRoleReq), zap.Any("rsp", createRoleRsp), zap.Any("err", err)) if err != nil { return err } param := parseReq{ methodName: "CreateDBRole, " + dbRole, clusterId: createRoleReq.ClusterId, req: createRoleReq, baseResp: createRoleRsp.BaseResp, proxyError: createRoleRsp.ProxyError, } result := parseDBuserResp(ctx, param) if result != entity.CommandStatusComplete { return errors.New("create db role return error code") } return nil } func (i *InfraHandlerImpl) DeleteDBUser(ctx context.Context, clusterId uint64, OperatorUserName string, jwtToken string, dbUser string, command *entity.CommandDBUserCluster) entity.CommandStatus { if isGoProxy { cluster, err := i.GetClusterById(ctx, clusterId, true) if err != nil { if isPrintErrLog(command) { tilog.Error(ctx, "CreateDBUser GetClusterById error", zap.Any("clusterId", clusterId), zap.Any("err", err)) } else { tilog.Warn(ctx, "CreateDBUser GetClusterById error", zap.Any("clusterId", clusterId), zap.Any("err", err)) } return entity.CommandStatusFail } operator := &clusterproxy.ClusterOperator{ Username: OperatorUserName, AuthMethod: clusterproxy.AuthMethodToken, Credential: jwtToken, } params := &DeleteUserParams{ Cluster: cluster, Operator: operator, Username: dbUser, } err = DeleteUser(ctx, params) if err != nil { if isPrintErrLog(command) { tilog.Error(ctx, "DeleteDBUser DeleteUser error", zap.Any("clusterId", clusterId), zap.Any("operator", operator.Username), zap.Any("username", dbUser), zap.Any("err", err)) } else { tilog.Warn(ctx, "DeleteDBUser DeleteUser error", zap.Any("clusterId", clusterId), zap.Any("operator", operator.Username), zap.Any("username", dbUser), zap.Any("err", err)) } return entity.CommandStatusFail } tilog.Info(ctx, "DeleteDBUser DeleteUser success", zap.Any("clusterId", clusterId), zap.Any("operator", operator.Username), zap.Any("username", dbUser)) return entity.CommandStatusComplete } client := GetCoreSvcClient() operator := &core.DBUserOperator{ Username: OperatorUserName, AuthMethod: core.DBUser_AUTH_METHOD_TOKEN, Credential: jwtToken, } req := &core.DeleteDBUserReq{ ClusterId: clusterId, Username: dbUser, Operator: operator, } tilog.Info(ctx, "Call CM DeleteDBUser", zap.Any("DeleteDBUserReq", req)) tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() resp, e := client.DeleteDBUser(tctx, req) tilog.Info(ctx, "DeleteDBUser resp", zap.Any("resp", resp)) if e != nil { tilog.Error(ctx, "call CM DeleteDBUser err", zap.Error(e), zap.Any("DeleteDBUser", req)) return entity.CommandStatusFail } param := parseReq{ //methodName: "DeleteDBUser, " + dbUser, methodName: fmt.Sprintf("DeleteDBUser, %s", dbUser), clusterId: req.ClusterId, req: req, baseResp: resp.BaseResp, proxyError: resp.ProxyError} if command != nil { param.command = *command } //tilog.Info(ctx, "parseDBuserResp req"+fmt.Sprintf("param=%+v,param.command=%+v", param, *command), zap.Any("param", param)) result := parseDBuserResp(ctx, param) //tilog.Info(ctx, "DeleteDBUser result", zap.Any("result", result), // zap.Any("clusterId", clusterId), zap.Any("dbUser", dbUser), // zap.Any("req", req), zap.Any("resp.BaseResp", resp.BaseResp)) return result } func (i *InfraHandlerImpl) DeleteSqlUser(ctx context.Context, operator *core.DBUserOperator, dbUser string, clusterId uint64) error { if isGoProxy { cluster, err := i.GetClusterById(ctx, clusterId, true) if err != nil { tilog.Error(ctx, "DeleteSqlUser GetClusterById error", zap.Any("clusterId", clusterId), zap.Any("err", err)) return err } cpOperator := &clusterproxy.ClusterOperator{ Username: operator.Username, AuthMethod: clusterproxy.AuthMethodToken, Credential: operator.Credential, } if operator.AuthMethod == core.DBUser_AUTH_METHOD_PASSWORD { cpOperator.AuthMethod = clusterproxy.AuthMethodPassword } params := &DeleteUserParams{ Cluster: cluster, Operator: cpOperator, Username: dbUser, } err = DeleteUser(ctx, params) if err != nil { tilog.Error(ctx, "DeleteDBUser DeleteUser error", zap.Any("clusterId", clusterId), zap.Any("operator", operator.Username), zap.Any("username", dbUser), zap.Any("err", err)) return err } return nil } client := GetCoreSvcClient() req := &core.DeleteDBUserReq{ ClusterId: clusterId, Username: dbUser, Operator: operator, } tilog.Info(ctx, "Call CM DeleteDBUser", zap.Any("DeleteDBUserReq", req)) tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() resp, err := client.DeleteDBUser(tctx, req) tilog.Info(ctx, "DeleteDBUser resp", zap.Any("resp", resp)) if err != nil { tilog.Error(ctx, "call CM DeleteDBUser err", zap.Any("DeleteDBUser", req), zap.Error(err)) return err } param := parseReq{ methodName: "UpdateSqlUser", clusterId: req.ClusterId, req: req, baseResp: resp.BaseResp} //tilog.Info(ctx, "parseDBuserResp req"+fmt.Sprintf("param=%+v,param.command=%+v", param, *command), zap.Any("param", param)) return parseBaseResp(ctx, param) } func getTimeout(ctx context.Context) time.Duration { timeout := viper.GetDuration("cm.timeout") if timeout != 0 { defaultTimeout = timeout * time.Second } tilog.Info(ctx, fmt.Sprintf("get timeout =%v", defaultTimeout)) return defaultTimeout } func (*InfraHandlerImpl) DBUserISExisting(ctx context.Context, clusterId uint64, OperatorDBUser string, jwtToken string, dbUser string) (bool, entity.CommandStatus) { client := GetCoreSvcClient() operator := &core.DBUserOperator{ Username: OperatorDBUser, AuthMethod: core.DBUser_AUTH_METHOD_TOKEN, Credential: jwtToken, } req := &core.IsDBUserExistingReq{ ClusterId: clusterId, Username: dbUser, Operator: operator, } tilog.Info(ctx, "IsDBUserExisting", zap.Any("dbUser", dbUser), zap.Any("clusterId", clusterId), zap.Any("OperatorDBUser", OperatorDBUser)) tctx, cf := context.WithTimeout(ctx, getTimeout(ctx)) defer cf() resp, e := client.IsDBUserExisting(tctx, req) tilog.Info(ctx, "IsDBUserExisting resp", zap.Any("resp", resp)) if e != nil { tilog.Error(ctx, "call CM IsDBUserExisting err", zap.Error(e), zap.Any("dbUser", dbUser), zap.Any("clusterId", clusterId), zap.Any("operator", OperatorDBUser)) return false, entity.CommandStatusFail } if resp != nil { param := parseReq{ methodName: "DBUserISExisting, " + dbUser, clusterId: req.ClusterId, req: req, baseResp: resp.BaseResp, proxyError: resp.ProxyError, } result := parseDBuserResp(ctx, param) isDBUserExisting := resp.Existing tilog.Info(ctx, "IsDBUserExisting result", zap.Any("resp.Existing", isDBUserExisting), zap.Any("clusterId", clusterId), zap.Any("dbUser", dbUser)) return resp.Existing, result } return false, entity.CommandStatusFail } func (*InfraHandlerImpl) SetClusterAuthPubKey(ctx context.Context, clusterId uint64, keys string) entity.CommandStatus { client := GetCoreSvcClient() req := &core.SetClusterAuthPubKeyReq{ ClusterId: clusterId, Jwks: keys, } tilog.Info(ctx, "will SetClusterAuthPubKey", zap.Any("SetClusterAuthPubKeyReq", req)) resp, err := client.SetClusterAuthPubKey(ctx, req) if err != nil { tilog.Error(ctx, "call CM SetClusterAuthPubKey error", zap.Error(err), zap.Any("resp", resp)) return entity.CommandStatusFail } param := parseReq{ methodName: "SetClusterAuthPubKey", clusterId: req.ClusterId, req: req, baseResp: resp.BaseResp} return parseDBuserResp(ctx, param) } func (*InfraHandlerImpl) CreateCluster(ctx context.Context, req *core.CreateClusterReq) (clusterId *uint64, err error) { client := GetCoreSvcClient() tilog.Info(ctx, "will CreateCluster", zap.Any("userId", req.CreateById)) resp, err := client.CreateCluster(ctx, req) tilog.Info(ctx, "call CM CreateCluster result", zap.Any("req", req), zap.Any("resp", resp), zap.Error(err)) if err != nil { tilog.Error(ctx, "call CM CreateCluster error", zap.Error(err), zap.Any("req", req), zap.Any("resp", resp)) return nil, err } if resp == nil { tilog.Error(ctx, "call CM CreateCluster error, return resp is nil", zap.Any("resp", resp), zap.Any("req", req)) return nil, errs.NewInternalSvcInfraAPI("get result is nil", nil) } if resp.BaseResp.ErrCode != 0 && resp.BaseResp.ErrCode != ErrCodeCannotCreateClusterInAws { tilog.Error(ctx, "call CM CreateCluster error", zap.Any("resp", resp), zap.Any("req", req)) return nil, errs.NewInternalSvcInfraAPI(resp.BaseResp.ErrMsg, nil) } tilog.Info(ctx, "Call CM CreateCluster result success", zap.Any("req", req), zap.Any("resp.BaseResp", resp.BaseResp)) clusterId = &resp.Id return clusterId, nil } func IsSupportDBUser(ctx context.Context, cluster *core.Cluster) bool { if cluster != nil { // cluster status is not exist in support dbuser status statusFlag := false for _, s := range SupportDBUserStatus { if cluster.GetStatus() == s { statusFlag = true } } if !statusFlag { tilog.Info(ctx, " this cluster status not support dbuser", zap.Any("clusterId", cluster.Id), zap.Any("cluster.Name", cluster.Name), zap.Any("cluster.Status", cluster.Status)) return false } // check the cluster.Features is supportDbuser features := cluster.Features for _, f := range features { // the cluster support dbuser if f == core.AvailableFeature_DELEGATE_USER { tilog.Info(ctx, " this cluster support dbuser", zap.Any("clusterId", cluster.Id), zap.Any("cluster.Name", cluster.Name)) return true } } tilog.Info(ctx, "this cluster not support dbuser, cluster.Status is not support or cluster.Features does not contain AvailableFeature_DELEGATE_USER", zap.Any("clusterId", cluster.Id), zap.Any("cluster.Status", cluster.Status), zap.Any("cluster.Features", cluster.Features)) return false } tilog.Info(ctx, " the param cluster is nil", zap.Any("cluster", cluster)) return false } func (*InfraHandlerImpl) GetServerlessDefaultRegion(ctx context.Context, provider string, orgId uint64, projectId uint64, clientIP string) (*string, error) { client := GetCoreSvcClient() isServerless := true deployType := getClusterDeployType(isServerless) req := &core.ListCloudProvidersReq{ SpecificTenantId: &orgId, FilterOutEmptyCloudProvider: true, FilterOutEmptyRegion: true, DeployType: &deployType, FilterOutInactiveRegion: true, FilterOutInactiveProvider: true, SpecificProjectId: &projectId, WithXaccountDeployId: true, } resp, err := client.ListCloudProviders(ctx, req) if err != nil { return nil, err } tilog.Info(ctx, "call CM ListCloudProviders result", zap.Any("req", req), zap.Any("resp", resp)) if err != nil { tilog.Error(ctx, "call CM ListCloudProviders error", zap.Error(err), zap.Any("req", req), zap.Any("resp", resp)) return nil, err } if resp == nil || resp.BaseResp.ErrCode != 0 { tilog.Error(ctx, "call CM ListCloudProviders error", zap.Any("resp", resp), zap.Any("req", req)) return nil, errs.NewInternalSvcInfraAPI(resp.BaseResp.ErrMsg, nil) } // [resp="name2cloud_provider:{key:\"aws\" value:{name:\"aws\" status:STATUS_ACTIVE name2region:{key:\"us-east-1\" // value:{name:\"us-east-1\" display_name:\"N. Virginia (us-east-1)\" cloud_provider_name:\"aws\" status:STATUS_ACTIVE priority:999999 area:\"North America\" country_code:\"US\"}}}} // base_resp:{}"] // provider := "aws" // aws CloudProvider awsCloudProvider := resp.Name2CloudProvider[provider] regions := make([]*core.Region, 0, len(awsCloudProvider.Name2Region)) for _, r := range awsCloudProvider.Name2Region { regions = append(regions, r) } tilog.Info(ctx, "regions sort before", zap.Any("regions", regions), zap.String("client_ip", clientIP)) if clientIP != "" && !net.ParseIP(clientIP).IsPrivate() { // sort regions by location city, err := IP2CityInfo(clientIP) if err == nil && city.Latitude != "" && city.Longitude != "" { for i := range regions { regions[i].Priority = calculateDistance(regionGeoLocation[regions[i].Name], city) } } tilog.Info(ctx, "regions sort rewrite priority", zap.Any("regions", regions), zap.String("client_ip", clientIP), zap.Any("city", city), zap.Error(err)) } sort.SliceStable(regions, func(i, j int) bool { left, right := regions[i], regions[j] if left.Priority != right.Priority { return left.Priority < right.Priority } return left.DisplayName < right.DisplayName }) tilog.Info(ctx, "regions sort after", zap.Any("regions", regions), zap.String("client_ip", clientIP)) if len(regions) > 0 { region := regions[0] tilog.Info(ctx, "getCloudRegion result", zap.Any("region", region), zap.String("client_ip", clientIP)) return ®ion.Name, nil } return nil, errs.NewInternalGRPC("call CM ListCloudProviders is nil", nil) } func parseDBuserResp(ctx context.Context, param parseReq) entity.CommandStatus { var status entity.CommandStatus var commandId uint64 var failNum uint32 var commandDBUserCluster entity.CommandDBUserCluster //command := param.command //tilog.Info(ctx, "param.command", zap.Any("command", param.command)) if param.command != nil { //var c interface{} = param.command var c = param.command switch v := c.(type) { case entity.CommandDBUserCluster: commandId = v.ID.Uint64() failNum = v.FailNum commandDBUserCluster = v } } str := fmt.Sprintf("Call CM %s result:", param.methodName) //tilog.Info(ctx, fmt.Sprintf("commandId=%d,failNum=%d", commandId, failNum)) if param.baseResp.ErrCode == 0 { //resultDate.Result = true status = entity.CommandStatusComplete tilog.Info(ctx, str+" success", zap.Any("clusterId", param.clusterId), zap.Any("baseResp", param.baseResp)) } else if param.baseResp.ErrCode == ErrCodeTokenBasedAuthRequired { tilog.Info(ctx, str+" fail-ignore, cluster version is not support ,ignore this cluster", zap.Any("clusterId", param.clusterId), zap.Any("baseResp", param.baseResp)) status = entity.CommandStatusIgnore } else if param.baseResp.ErrCode == ErrCodeDeletedCluster { tilog.Info(ctx, str+" fail-ignore, cluster is delete,ignore this cluster", zap.Any("clusterId", param.clusterId), zap.Any("baseResp", param.baseResp)) status = entity.CommandStatusIgnore } else if param.baseResp.ErrCode == ErrCodeUnexpectedClusterStatus { tilog.Info(ctx, str+" fail, cluster status now is not support, please try again next job", zap.Any("clusterId", param.clusterId), zap.Any("baseResp", param.baseResp)) status = entity.CommandStatusFail } else { alertMsg := "" if param.proxyError != nil { //tilog.Error(ctx, str+" fail, code="+code+" ,clusterId="+id+" "+baseResp.GetErrMsg(), zap.Any("req", req), zap.Any("CM baseResp", baseResp)) alertMsg = fmt.Sprintf(",SqlErr=%d,SqlErrMsg=%s", param.proxyError.SqlErrNo, param.proxyError.SqlErrMsg) } s := fmt.Sprintf("{id=%d, code=%v, clusterId=%d,%v %s, failNum=%d}", commandId, param.baseResp.ErrCode, param.clusterId, param.baseResp.ErrMsg, alertMsg, failNum+1) if failNum+1 >= 3 && (failNum+1)%30 == 0 { if commandDBUserCluster.ID != 0 { ignoreOrgs := viper.GetStringSlice("alert.dbUserIgnoreOrg") ignoreClusters := viper.GetStringSlice("alert.dbUserIgnoreCluster") isIgnoreOrg := slice.Contain(ignoreOrgs, convertor.ToString(commandDBUserCluster.OrgID.Uint64())) isIgnoreCluster := slice.Contain(ignoreClusters, convertor.ToString(commandDBUserCluster.ClusterID.Uint64())) tilog.Info(ctx, fmt.Sprintf(" isIgnoreOrg=%t, isIgnoreCluster=%t , {ignoreOrgs=%v,ignoreClusters=%v}", isIgnoreOrg, isIgnoreCluster, ignoreOrgs, ignoreClusters)) if isIgnoreOrg || isIgnoreCluster { tilog.Info(ctx, str+s, zap.Any("req", param.req)) return entity.CommandStatusFail } } tilog.Error(ctx, str+s, zap.Any("req", param.req)) } else { tilog.Info(ctx, str+s, zap.Any("req", param.req)) } status = entity.CommandStatusFail } return status } func parseBaseResp(ctx context.Context, param parseReq) (err error) { str := fmt.Sprintf("Call CM %s result:", param.methodName) //tilog.Info(ctx, fmt.Sprintf("commandId=%d,failNum=%d", commandId, failNum)) if param.baseResp.ErrCode == 0 { //resultDate.Result = true tilog.Info(ctx, str+" success", zap.Any("clusterId", param.clusterId), zap.Any("baseResp", param.baseResp)) } else if param.baseResp.ErrCode == ErrCodeTokenBasedAuthRequired { tilog.Info(ctx, str+" fail, cluster version is not support", zap.Any("clusterId", param.clusterId), zap.Any("baseResp", param.baseResp)) err = errs.NewBadRequest("cluster version is not support", nil) } else if param.baseResp.ErrCode == ErrCodeDeletedCluster { tilog.Info(ctx, str+" fail-ignore, cluster is delete", zap.Any("clusterId", param.clusterId), zap.Any("baseResp", param.baseResp)) err = errs.NewBadRequest("cluster is delete", nil) } else if param.baseResp.ErrCode == ErrCodeUnexpectedClusterStatus { tilog.Info(ctx, str+" fail, cluster status now is not support", zap.Any("clusterId", param.clusterId), zap.Any("baseResp", param.baseResp)) err = errs.NewBadRequest("cluster status is not support", nil) } else { alertMsg := "" if param.proxyError != nil { alertMsg = fmt.Sprintf("ERROR %d: %s", param.proxyError.SqlErrNo, param.proxyError.SqlErrMsg) } else if param.baseResp != nil { alertMsg = fmt.Sprintf("ERROR %d: %s", param.baseResp.ErrCode, param.baseResp.ErrMsg) } s := fmt.Sprintf("{ code=%v, clusterId=%d,%v %s}", param.baseResp.ErrCode, param.clusterId, param.baseResp.ErrMsg, alertMsg) tilog.Error(ctx, str+s, zap.Any("req", param.req)) //err = errs.NewInternalSvcInfraAPI(param.methodName+" fail: "+alertMsg, nil) err = errs.NewInternalSvcInfraAPI(alertMsg, nil) } return err } func getClusterDeployType(isDevTier bool) core.DeployType { if isDevTier { return core.DeployType_DEPLOY_TYPE_SHARED } return core.DeployType_DEPLOY_TYPE_DEDICATED } func GetCommandStatus(ctx context.Context, err error) entity.CommandStatus { if err != nil { if err.Error() == CLUSTER_STATUS_ERROR { tilog.Info(ctx, "cluster status is error", zap.Error(err)) return entity.CommandStatusIgnore } else if err.Error() == CLUSTER_NOT_EXIST { tilog.Info(ctx, "cluster status is not exist ,ignore", zap.Error(err)) return entity.CommandStatusIgnore } return entity.CommandStatusFail } return entity.CommandStatusComplete } func getBase64Password(ctx context.Context, password string) *string { var cmkArn string if config.IsProdEnv() { cmkArn = "arn:aws:kms:us-west-2:380838443567:key/6a3f87f3-d032-414b-870f-458979546c97" } else { cmkArn = "arn:aws:kms:us-west-2:385595570414:key/f15101b5-6aed-48bf-9b9a-fc36406efe92" } enc, err := provider.NewSecretEncryptor(cmkArn) if err != nil { tilog.Info(ctx, "NewSecretEncryptor fail", zap.Error(err)) } //password := "uuap@123" credential, err := encrypt.EncryptBase64(enc, vault.Secret(password)) if err != nil { tilog.Info(ctx, "EncryptBase64 fail", zap.Error(err)) } return &credential } type parseReq struct { methodName string clusterId uint64 req interface{} baseResp *base.BaseResp proxyError *core.ClusterProxyError command any } type CreateSqlUserReq struct { ClusterId uint64 OperatorUserName string JwtToken string UserEmail string DbUser string IgnoreExists bool Password string Command *entity.CommandDBUserCluster Operator *core.DBUserOperator DbRole string CustomRole []string } var ( regionGeoLocation = map[string]*ipdb.CityInfo{ "us-west-2": {Longitude: "-122.7011", Latitude: "45.5122"}, "us-east-1": {Longitude: "-77.4728", Latitude: "37.7664"}, "eu-central-1": {Longitude: "8.6821", Latitude: "50.1109"}, "ap-southeast-1": {Longitude: "103.852", Latitude: "1.2975"}, "ap-northeast-1": {Longitude: "139.6917", Latitude: "35.6895"}, "ap-east-1": {Longitude: "114.2723", Latitude: "22.2908"}, "ap-south-1": {Longitude: "72.8826", Latitude: "19.0728"}, } ) func IP2CityInfo(ip string) (*ipdb.CityInfo, error) { url := fmt.Sprintf("http://%s/ip/%s", config.Global().GrpcConfig.IPAPISvcAddr, ip) resp, err := http.Get(url) if err != nil { return nil, err } defer resp.Body.Close() // nolint: errcheck body, err := io.ReadAll(resp.Body) if err != nil { return nil, err } var res ipdb.CityInfo err = json.Unmarshal(body, &res) if err != nil { return nil, err } return &res, nil } // unit: km func calculateDistance(start, end *ipdb.CityInfo) int32 { if start == nil { // us-west-2 as default location start = &ipdb.CityInfo{Longitude: "-122.7011", Latitude: "45.5122"} } startLng, _ := strconv.ParseFloat(start.Longitude, 64) startLat, _ := strconv.ParseFloat(start.Latitude, 64) endLng, _ := strconv.ParseFloat(end.Longitude, 64) endLat, _ := strconv.ParseFloat(end.Latitude, 64) rad := math.Pi / 180.0 startLat *= rad startLng *= rad endLat *= rad endLng *= rad theta := endLng - startLng dist := math.Acos(math.Sin(startLat)*math.Sin(endLat) + math.Cos(startLat)*math.Cos(endLat)*math.Cos(theta)) return int32(dist * 6371000.0 /*radius*/ / 1000) }