package config import ( "fmt" "regexp" "sort" "strconv" "strings" "time" "github.com/Masterminds/semver/v3" "github.com/elliotchance/pie/v2" "github.com/pingcap/errors" appentity "github.com/tidbcloud/dataflow-service/app/repo/entity" "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow" "github.com/tidbcloud/dataflow-service/internal/utils" "github.com/tidbcloud/pkgv2/gormutil" "github.com/tidbcloud/tiinf/sdk/rbac/common" "golang.org/x/exp/maps" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" "k8s.io/utils/ptr" ) var globalConf *CentralConfig = &CentralConfig{} type CentralConfig struct { Env string `mapstructure:"env"` ServerConfig ServerConfig `mapstructure:"server"` // cloud configs MysqlConfig gormutil.Config `mapstructure:"mysql"` Auth0Config Auth0Config `mapstructure:"auth0"` AwsConfig AwsConfig `mapstructure:"aws"` // infra api or aws configs GardenerConfig GardenerConfig `mapstructure:"gardener"` InfraConfig InfraConfig `mapstructure:"infra"` // internal configs ProxyConfig ProxyConfig `mapstructure:"proxy"` RBACConfig RBACConfig `mapstructure:"rbacConfig"` BillingConfig BillingConfig `mapstructure:"billing"` GrpcConfig GrpcConfig `mapstructure:"grpc"` O11yConfig O11yConfig `mapstructure:"o11y"` // our business configs TlsConfig TlsConfig `mapstructure:"tlsConfig"` MeteringConfig MeteringConfig `mapstructure:"enableBackupMetering"` ImportConfig ImportConfig `mapstructure:"import"` DMConfig DMConfig `mapstructure:"dm"` BackupConfig BackupConfig `mapstructure:"backup"` CompactConfig CompactConfig `mapstructure:"compact"` ChangefeedConfig ChangefeedConfig `mapstructure:"changefeed"` GrafanaConfig GrafanaConfig `mapstructure:"grafana"` ReplayConfig ReplayConfig `mapstructure:"replay"` DebugConfig DebugConfig `mapstructure:"debugConfig"` } type O11yConfig struct { GlobalServerDomain string `mapstructure:"globalServerDomain" validate:"required"` BizType string `mapstructure:"bizType" validate:"required"` Auth0Domain string `mapstructure:"auth0Domain" validate:"required"` Auth0ClientID string `mapstructure:"auth0ClientID" validate:"required"` Auth0Audience string `mapstructure:"auth0Audience" validate:"required"` } type DebugConfig struct { Enable bool `mapstructure:"enable" default:"false"` DisableKafka bool `mapstructure:"disableKafka" default:"false"` KafkaProxyAddress string `mapstructure:"kafkaProxyAddress" default:"127.0.0.1:9092"` InfraConfig string `mapstructure:"infraConfig" default:""` ClusterProxy string `mapstructure:"clusterProxy" default:""` // EnableConfigCenter whether to use config center to replace gitops EnableConfigCenter bool `mapstructure:"enableConfigCenter" default:"false"` TCIDMRegionalServer string `mapstructure:"tcidmRegionalServer" default:"http://127.0.0.1:20008"` // TiCDCServiceServer is the address of TiCDC service server TiCDCServiceServer string `mapstructure:"tiCDCServiceServer" default:""` } type TlsConfig struct { Enable bool `mapstructure:"enableTLS"` EnabledTenants []uint64 `mapstructure:"enabledTenants"` EnabledProjects []uint64 `mapstructure:"enabledProjects"` } func (t *TlsConfig) IsEnableTLS(tenantID uint64, projectID uint64) bool { if t.Enable { return true } for _, id := range t.EnabledTenants { if id == tenantID { return true } } for _, id := range t.EnabledProjects { if id == projectID { return true } } return false } type MeteringConfig struct { Enable bool `mapstructure:"enable"` LocalFullBackupEnable bool `mapstructure:"localFullBackupEnable" default:"false"` RefactorEnable bool `mapstructure:"refactorEnable" default:"false"` } type GrpcConfig struct { CoreSvcAddr string `mapstructure:"coreSvcAddr" default:"kubernetes:///dbaas-core-svc-server-grpc:8888"` DataflowSvcAddr string `mapstructure:"dataflowSvcAddr" default:"kubernetes:///dataflow-grpc:10000"` AccountAclSvcAddr string `mapstructure:"accountAclSvcAddr" default:"kubernetes:///account-acl-grpc:10000"` AccountProviderAddr string `mapstructure:"accountProviderSvcAddr" default:"kubernetes:///account-provider-grpc:10000"` DedicatedGlobalSvcAddr string `mapstructure:"dedicatedGlobalSvcAddr" default:"kubernetes:///dedicated-global-service.tidb-management-service:10000"` TiDBMgmtSvcAddr string `mapstructure:"tidbMgmtSvcAddr" default:"kubernetes:///tidb-mgmt-service.tidb-management-service:10001"` ConfigCenterAddr string `mapstructure:"configCenterAddr" default:"kubernetes:///config-center.devops:8080"` ServerlessSvcAddr string `mapstructure:"serverlessSvcAddr" default:"kubernetes:///serverless-svc-grpc:10000"` MgmtSvcProxyAddr string `mapstructure:"mgmtSvcProxyAddr" default:"http://tidb-mgmt-service.tidb-management-service:8080"` MgmtSvcProxyToken string `mapstructure:"mgmtSvcProxyToken" default:""` } type BucketOptionsConfig struct { // deprecated GCPBucketDualRegionOptions []GCPBucketDualRegionOptionsCandidateConfig `mapstructure:"gcpBucketRegionOptions"` // deprecated AWSBucketDualRegionOptions []AWSBucketDualRegionOptionsCandidateConfig `mapstructure:"awsBucketRegionOptions"` GCPBucketPredefinedDualRegions map[string][]string `mapstructure:"gcpBucketPredefinedDualRegions"` // predefined dual-region name to regions GCPBucketGeneralDualRegions map[string][]string `mapstructure:"gcpBucketGeneralDualRegions"` // location code to regions GCPBucketRestrictedDualRegions map[string][]string `mapstructure:"gcpBucketRestrictedDualRegions"` // region to restricted paired regions } type GCPBucketDualRegionOptionsCandidateConfig struct { Region string `mapstructure:"region"` Continent string `mapstructure:"continent"` // By default, regions on the same continent can be combined into dual-region. It's the multi-region code in gcp Description string `mapstructure:"description"` Predefined string `mapstructure:"predefined"` // It's the predefined dual-region code, when create predefined dual-region bucket, we should use it as the location doc:https://cloud.google.com/storage/docs/dual-regions#predefined_dual-regions } type AWSBucketDualRegionOptionsCandidateConfig struct { Region string `mapstructure:"region"` Description string `mapstructure:"description"` } type ServerConfig struct { // Log level: debug, info, warn, error, fatal LogLevel string `mapstructure:"logLevel" default:"info" validate:"oneof=debug info warn error fatal"` EnableDBQueryLog bool `mapstructure:"enableDBQueryLog" default:"false"` GrpcPort int `mapstructure:"grpcPort" default:"10000"` HttpPort int `mapstructure:"httpPort" default:"11000"` InnerHttpPort int `mapstructure:"innerHttpPort" default:"12000"` // the central metrics port MetricsPort uint16 `mapstructure:"metricsPort" default:"9090"` // session secret salt conf SaltSecret string `mapstructure:"saltSecret" required:"true"` SecretStore *SecretConfig `mapstructure:"secret"` PocManualBackupReservedDays *int `mapstructure:"pocManualBackupReservedDays" validate:"required"` PocAutoBackupReservedDays *int `mapstructure:"pocAutoBackupReservedDays" validate:"required"` OnDemandAutoBackupReservedDays *int `mapstructure:"onDemandAutoBackupReservedDays" validate:"required"` UsePrivilegedTiDB bool `mapstructure:"usePrivilegedTiDB" default:"false"` // bucket options BucketOptions BucketOptionsConfig `mapstructure:"bucketOptions"` // general allowlist config FeaturesSwitchers FeaturesSwitchersConfig `mapstructure:"featuresSwitchersConfig"` // FeatureGates is new version of FeaturesSwitchers, will replace it in future FeatureGates map[string]FeatureGate `mapstructure:"featureGates"` DMTenantQuota map[string]int `mapstructure:"dmTenantQuota"` EnableDMMetering bool `mapstructure:"enableDMMetering" default:"false"` CleanupBRJobIntervalMinutes int64 `mapstructure:"cleanupBRJobIntervalMinutes" default:"60"` } // FeatureGate defines the feature gate config structure. // // - Provider: aws, gcp, azure, etc. // // - Region: "us-west-1", "asia-southeast1", "eastus2", etc. (plain cloud region without provider prefix) // // Priority order: // // - RequiredXXX take priority over DisabledXXX // // - DisabledXXX take priority over EnabledXXX // // - EnabledXXX take priority over All // // - All is fallback enable flag type FeatureGate struct { RequiredProviders []string `mapstructure:"requiredProviders"` RequiredRegions []string `mapstructure:"requiredRegions"` DisabledProviders []string `mapstructure:"disabledProviders"` DisabledRegions []string `mapstructure:"disabledRegions"` DisabledTenants []uint64 `mapstructure:"disabledTenants"` DisabledProjects []uint64 `mapstructure:"disabledProjects"` DisabledClusters []uint64 `mapstructure:"disabledClusters"` EnabledProviders []string `mapstructure:"enabledProviders"` EnabledRegions []string `mapstructure:"enabledRegions"` EnabledTenants []uint64 `mapstructure:"enabledTenants"` EnabledProjects []uint64 `mapstructure:"enabledProjects"` EnabledClusters []uint64 `mapstructure:"enabledClusters"` All bool `mapstructure:"all" default:"false"` } // FeatureGateQueryParams is the parameters for querying feature gate. // empty parameter means not specified, should be ignored. type FeatureGateQueryParams struct { Provider string // aws, gcp, azure, etc. Region string // "us-west-1", "asia-southeast1", "eastus2", etc. TenantID uint64 ProjectID uint64 ClusterID uint64 } func (g *FeatureGate) Enabled(params FeatureGateQueryParams) (reason string, _ bool) { // Priority: Required > Disabled > Enabled > All // 1. Check Required conditions (if any matches, return true immediately) if params.Provider != "" && pie.Contains(g.RequiredProviders, params.Provider) { return "required-provider", true } if params.Region != "" && pie.Contains(g.RequiredRegions, params.Region) { return "required-region", true } // 2. Check Disabled conditions (if any matches, return false immediately) // DisabledProviders if params.Provider != "" && pie.Contains(g.DisabledProviders, params.Provider) { return "disabled-provider", false } if params.Region != "" && pie.Contains(g.DisabledRegions, params.Region) { return "disabled-region", false } if params.TenantID != 0 && pie.Contains(g.DisabledTenants, params.TenantID) { return "disabled-tenant", false } if params.ProjectID != 0 && pie.Contains(g.DisabledProjects, params.ProjectID) { return "disabled-project", false } if params.ClusterID != 0 && pie.Contains(g.DisabledClusters, params.ClusterID) { return "disabled-cluster", false } // 3. Check Enabled conditions (if any matches, return true) if params.Provider != "" && pie.Contains(g.EnabledProviders, params.Provider) { return "enabled-provider", true } if params.Region != "" && pie.Contains(g.EnabledRegions, params.Region) { return "enabled-region", true } if params.TenantID != 0 && pie.Contains(g.EnabledTenants, params.TenantID) { return "enabled-tenant", true } if params.ProjectID != 0 && pie.Contains(g.EnabledProjects, params.ProjectID) { return "enabled-project", true } if params.ClusterID != 0 && pie.Contains(g.EnabledClusters, params.ClusterID) { return "enabled-cluster", true } // 4. Fallback to All flag return "fallback-all", g.All } type FeaturesSwitchersConfig struct { // PublicFeatures public available feature, if a feature is defined here, same feature defined in // EnabledFeatureByDeployment and FeaturesAllowList will not take effects. PublicFeatures []string `mapstructure:"publicFeatures"` // EnabledFeatureByDeployment enable features by defined deployment, i.e. which region and which provider // if a feature is enabled by this, same feature defined in FeaturesAllowList will not take effects EnabledFeatureByDeployment map[string]DeploymentMatcher `mapstructure:"enabledFeatureByDeployment"` // FeaturesAllowList enable feature by cluster/project/tenant id, if input cluster matches any of defined // cluster/project/tenant id of some feature, the feature is enabled. FeaturesAllowList map[string]AllowListMatcher `mapstructure:"featuresAllowList"` } type AllowListMatcher struct { TenantID []string `mapstructure:"tenantID"` ProjectID []string `mapstructure:"projectID"` ClusterID []string `mapstructure:"clusterID"` DeploymentMatcher *DeploymentMatcher `mapstructure:"deploymentMatcher"` } func (m *AllowListMatcher) Match(tenantID, projectID, clusterID uint64, region, infraProvider string) bool { tenantIDStr, projectIDStr, clusterIDStr := fmt.Sprintf("%d", tenantID), fmt.Sprintf("%d", projectID), fmt.Sprintf("%d", clusterID) matched := utils.IsContain(m.TenantID, tenantIDStr) || utils.IsContain(m.ProjectID, projectIDStr) || utils.IsContain(m.ClusterID, clusterIDStr) if m.DeploymentMatcher != nil { return matched && m.DeploymentMatcher.Match(region, infraProvider) } return matched } // DeploymentMatcher match by how a feature is deployed. type DeploymentMatcher struct { Regions []string `mapstructure:"regions"` Providers []string `mapstructure:"providers"` // this is infra provider, like eks, dev-tier, aws, gcp } // Match target cluster. // todo: quite fixed, maybe support this by expression. func (m *DeploymentMatcher) Match(region string, provider string) bool { return utils.IsContain(m.Regions, region) && utils.IsContain(m.Providers, provider) } type SessionConfig struct { // A secret to encrypt or derive sessions. // Currently only Debug Portal implements session. Secret string `mapstructure:"secret" validate:"required"` } type ImportConfig struct { MachineType string `mapstructure:"machineType" default:"c5.9xlarge"` GCPMachineType string `mapstructure:"gcpMachineType" default:"c2-standard-30"` StorageRequest string `mapstructure:"storageRequest" default:"15Ti"` SkipPreCheck bool `mapstructure:"skipPreCheck"` SkipBanMyBuckets bool `mapstructure:"skipBanMyBuckets"` SortBufferSize int `mapstructure:"sortBufferSize" default:"5242880"` // import task configs for different matchers(tenant/project/cluster) TaskConfigs ImportTaskConfigs `mapstructure:"taskConfigs"` MetricsConfig ImportMeteringConfig `mapstructure:"metricsConfig"` MaxRetryTimesOnExecImportingQuerySQL uint `mapstructure:"maxRetryTimesOnExecImportingSQL" default:"3"` } // more detail: https://pingcap.feishu.cn/wiki/He6MwEbzyiOvSDkO7dncJu3pnUb#share-ZvILd5oOzoJhmrxTIzScTeMwnaf type ImportMeteringConfig struct { OverallSlowSpeed string `mapstructure:"overallSlowSpeed" default:"200g"` OverallTooSlowSpeed string `mapstructure:"overallTooSlowSpeed" default:"100g"` PrepareSlowTime time.Duration `mapstructure:"prepareSlowTime" default:"10m"` PrepareTooSlowTime time.Duration `mapstructure:"prepareTooSlowTime" default:"20m"` EncodeSlowSpeed string `mapstructure:"encodeSlowSpeed" default:"300m"` EncodeTooSlowSpeed string `mapstructure:"encodeTooSlowSpeed" default:"150m"` ImportSlowSpeed string `mapstructure:"importSlowSpeed" default:"300m"` ImportTooSlowSpeed string `mapstructure:"importTooSlowSpeed" default:"150m"` PostProcessDurationFactor float64 `mapstructure:"postProcessDurationFactor" default:"1.5"` } type ReplayConfig struct { WorkerConcurrency int `mapstructure:"workerConcurrency" default:"16"` DefaultReplayImage string `mapstructure:"defaultReplayImage"` } type ChangefeedConfig struct { SQLModeConfig []SQLModeConfig `mapstructure:"sqlModeConfig"` MultiTiCDCClustersFeatureGate MultiTiCDCClustersFeatureGate `mapstructure:"multiTiCDCClustersFeatureGate"` ClusterMaxMachineType map[uint64]*CDCMachineType `mapstructure:"clusterMaxMachineType"` RegionalInstanceTypes []*CDCRegionalInstanceType `mapstructure:"regionalInstanceTypes"` WorkerConcurrency int `mapstructure:"workerConcurrency" default:"16"` RegionalMachineTypes []*RegionalMachineType `mapstructure:"regionalMachineTypes"` MinNewkernelCompatibleTiDBVersion string `mapstructure:"minNewkernelCompatibleTiDBVersion" default:"v7.5.0"` MinNewkernelVersion string `mapstructure:"minNewkernelVersion" default:"v9.0.0-beta.1"` FeatureEnabledVersions []*ChangefeedFeatureEnabledVersions `mapstructure:"featureEnabledVersions"` } type ChangefeedFeatureEnabledVersions struct { Name string `mapstructure:"name"` // NewArch false means the feature is enabled in both new arch and old arch // true means the feature is enabled in new arch only NewArch bool `mapstructure:"newArch"` Versions []string `mapstructure:"versions"` VersionConstraint *semver.Constraints `mapstructure:"-"` } type RegionalMachineType struct { Region string `mapstructure:"region"` CpuNum uint32 `mapstructure:"cpuNum"` MemoryGi uint32 `mapstructure:"memoryGi"` StorageGi uint32 `mapstructure:"storageGi"` AWSMachineType string `mapstructure:"awsMachineType"` GCPMachineType string `mapstructure:"gcpMachineType"` AzureMachineType string `mapstructure:"azureMachineType"` AWSStorageType string `mapstructure:"awsStorageType"` GCPStorageType string `mapstructure:"gcpStorageType"` AzureStorageType string `mapstructure:"azureStorageType"` } func (c *ChangefeedConfig) Init() error { if err := c.initFeatureEnabledVersions(); err != nil { return errors.Annotate(err, "init changefeed feature enabled versions") } return nil } func (c *ChangefeedConfig) initFeatureEnabledVersions() error { for _, enabledVersions := range c.FeatureEnabledVersions { versionExpr := strings.Join(enabledVersions.Versions, "||") constraint, err := semver.NewConstraint(versionExpr) if err != nil { return errors.Annotatef( err, "invalid version constraint %s of feature %s", versionExpr, enabledVersions.Name) } enabledVersions.VersionConstraint = constraint } return nil } func (c *ChangefeedConfig) GetEnabledFeatures(version string, newArch bool) ([]string, error) { featureSet, err := c.getEnabledFeatures(version, newArch) if err != nil { return nil, err } features := maps.Keys(featureSet) sort.Strings(features) return features, nil } func (c *ChangefeedConfig) IsFeatureEnabled(feature, version string, newArch bool) bool { featureSet, err := c.getEnabledFeatures(version, newArch) if err != nil { return false } _, enabled := featureSet[feature] return enabled } func (c *ChangefeedConfig) getEnabledFeatures(version string, newArch bool) (map[string]struct{}, error) { v, err := semver.NewVersion(version) if err != nil { return nil, errors.Annotatef(err, "invalid version %s", version) } featureSet := make(map[string]struct{}) for _, enabledVersions := range c.FeatureEnabledVersions { if enabledVersions.NewArch && !newArch { // some features are only supported in new arch continue } if enabledVersions.VersionConstraint.Check(v) { featureSet[enabledVersions.Name] = struct{}{} } } return featureSet, nil } func (c *ChangefeedConfig) GetMatchedSQLMode(tenantID, projectID, clusterID uint64) string { tenantIDStr := strconv.FormatUint(tenantID, 10) projectIDStr := strconv.FormatUint(projectID, 10) clusterIDStr := strconv.FormatUint(clusterID, 10) for _, conf := range c.SQLModeConfig { if pie.Contains(conf.TenantID, tenantIDStr) { return conf.SQLMode } if pie.Contains(conf.ProjectID, projectIDStr) { return conf.SQLMode } if pie.Contains(conf.ClusterID, clusterIDStr) { return conf.SQLMode } } return "" } func (c *ChangefeedConfig) GetRegionInstanceType(region string) string { for _, regionalInstanceType := range c.RegionalInstanceTypes { if regionalInstanceType.Region == region { return regionalInstanceType.InstanceType } } return "" } func (c *ChangefeedConfig) GetRegionalMachineType(region string, cpu uint32, memory uint32) *RegionalMachineType { for _, regionalInstanceType := range c.RegionalMachineTypes { if regionalInstanceType.Region == region && regionalInstanceType.CpuNum == cpu && regionalInstanceType.MemoryGi == memory { return regionalInstanceType } } return nil } type SQLModeConfig struct { SQLMode string `mapstructure:"sqlMode"` TenantID []string `mapstructure:"tenantID"` ProjectID []string `mapstructure:"projectID"` ClusterID []string `mapstructure:"clusterID"` } type MultiTiCDCClustersFeatureGate struct { EnabledCloudProviders []string `mapstructure:"enabledCloudProviders"` EnabledDownstreamTypes []int32 `mapstructure:"enabledDownstreamTypes"` } type CDCMachineType struct { CPUNum uint32 `mapstructure:"cpuNum"` MemoryGi uint32 `mapstructure:"memoryGi"` } type CDCRegionalInstanceType struct { Region string `mapstructure:"region"` InstanceType string `mapstructure:"instanceType"` } func (f *MultiTiCDCClustersFeatureGate) IsEnabled(cloudProvider string, downstreamType dataflow.ConnectorType) bool { return pie.Contains(f.EnabledCloudProviders, cloudProvider) && pie.Contains(f.EnabledDownstreamTypes, int32(downstreamType)) } type DMConfig struct { PreCheckTimeout time.Duration `mapstructure:"preCheckTimeout" default:"30m"` // PrecheckSkipConfigs defines per-tenant/project/cluster precheck item blacklists. // Each entry specifies a matcher and a list of precheck items to skip for matched clusters. // Order matters if multiple entries match, the first match will be applied. PrecheckSkipConfigs PrecheckSkipConfigs `mapstructure:"precheckSkipConfigs"` // RCU price info for cloud provider & region. // https://github.com/tidbcloud/cluster-tier-config/blob/main/nightly/sku-dm.yaml (dev) // see how it's calculated https://pingcap.feishu.cn/docx/EdTLd1nJooeaqsxVshScWJdBnUb RCUPriceMap map[string]map[string]float64 `mapstructure:"rcuPriceMap"` EnableLightningPrecheck bool `mapstructure:"enableLightningPrecheck" default:"true"` // DM task configs for different matchers(tenant/project/cluster) DMTaskConfigs DMTaskConfigs `mapstructure:"dmTaskConfigs"` } // PrecheckSkipConfig skips specific precheck items for clusters matched by Matcher. type PrecheckSkipConfig struct { Matcher AllowListMatcher `mapstructure:"matcher"` SkipItems []string `mapstructure:"skipItems"` } // PrecheckSkipConfigs is a list of PrecheckSkipConfig entries. type PrecheckSkipConfigs []PrecheckSkipConfig // GetSkippedItems returns all precheck items that should be skipped for the given tenant/project/cluster. // The first match will be returned. func (c PrecheckSkipConfigs) GetSkippedItems(tenantID, projectID, clusterID uint64) map[string]bool { skipped := make(map[string]bool) for _, conf := range c { if !conf.Matcher.Match(tenantID, projectID, clusterID, "", "") { continue } for _, item := range conf.SkipItems { skipped[item] = true } break } return skipped } // ApplyBlacklist removes any items in the blacklist for the given tenant/project/cluster from items. func (c PrecheckSkipConfigs) ApplyBlacklist(items []string, tenantID, projectID, clusterID uint64) []string { skipped := c.GetSkippedItems(tenantID, projectID, clusterID) result := make([]string, 0, len(items)) for _, item := range items { if skipped[item] { continue } result = append(result, item) } return result } // DMTaskConfig // more config: https://docs.pingcap.com/tidb/stable/task-configuration-file-full/ type DMTaskConfig struct { Matcher AllowListMatcher `mapstructure:"matcher"` OnlineDDL *bool `mapstructure:"onlineDDL"` WorkerCount *int `mapstructure:"workerCount"` Batch *int `mapstructure:"batch"` } type DMTaskConfigs []DMTaskConfig func (c DMTaskConfigs) GetMatchedConfig(tenantID, projectID, clusterID uint64) *DMTaskConfig { for _, conf := range c { if conf.Matcher.Match(tenantID, projectID, clusterID, "", "") { copiedConf := conf return &copiedConf } } return nil } type ImportTaskConfig struct { Matcher AllowListMatcher `mapstructure:"matcher"` TiKVBackend *string `mapstructure:"tikvBackend"` } type ImportTaskConfigs []ImportTaskConfig func (c ImportTaskConfigs) GetMatchedConfig(tenantID, projectID, clusterID uint64) *ImportTaskConfig { for _, conf := range c { if conf.Matcher.Match(tenantID, projectID, clusterID, "", "") { copiedConf := conf return &copiedConf } } return nil } // BREnvs is k/v pairs of environment variables // because viper unmarshal map keys to lower case, we use slice to replace map // ref: https://github.com/spf13/viper/issues/373 type BREnvs []string func (c BREnvs) ToK8sEnvs() ([]corev1.EnvVar, error) { if len(c)%2 != 0 { return nil, errors.New("BREnvs must be [key1, val1, key2, val2, ...] format") } envVars := make([]corev1.EnvVar, 0, len(c)/2) for i := 0; i < len(c)-1; i += 2 { envVars = append(envVars, corev1.EnvVar{ Name: c[i], Value: c[i+1], }) } return envVars, nil } type BRResourceRequirements struct { Requests *BRResourceRequirementValues `mapstructure:"requests"` Limits *BRResourceRequirementValues `mapstructure:"limits"` InstanceType *string `mapstructure:"instanceType"` } func (r *BRResourceRequirements) ToK8sResourceRequirements() (*corev1.ResourceRequirements, error) { k8sResourceRequirements := &corev1.ResourceRequirements{} if r.Requests != nil { requests, err := r.Requests.ToK8sResourceList() if err != nil { return nil, errors.Annotate(err, "parse requests") } k8sResourceRequirements.Requests = requests } if r.Limits != nil { limits, err := r.Limits.ToK8sResourceList() if err != nil { return nil, errors.Annotate(err, "parse limits") } k8sResourceRequirements.Limits = limits } return k8sResourceRequirements, nil } type BRResourceRequirementValues struct { CPU string `mapstructure:"cpu"` Memory string `mapstructure:"memory"` } func (r *BRResourceRequirementValues) ToK8sResourceList() (corev1.ResourceList, error) { resourceList := make(corev1.ResourceList, 4) if len(r.CPU) > 0 { cpuQuantity, err := resource.ParseQuantity(r.CPU) if err != nil { return nil, errors.Annotatef(err, "parse cpu quantity %s", r.CPU) } resourceList[corev1.ResourceCPU] = cpuQuantity } if len(r.Memory) > 0 { memoryQuantity, err := resource.ParseQuantity(r.Memory) if err != nil { return nil, errors.Annotatef(err, "parse memory quantity %s", r.Memory) } resourceList[corev1.ResourceMemory] = memoryQuantity } return resourceList, nil } type CompactConfig struct { CompactBackupCycle string `mapstructure:"compactBackupCycle" default:"5h50m"` CompactBackupInterval string `mapstructure:"compactBackupInterval" default:"5m"` CompactBackupConcurrency int `mapstructure:"compactBackupConcurrency" default:"64"` CompactBackupCleanPolicy string `mapstructure:"compactBackupCleanPolicy" default:"Retain"` CompactBRInstanceTypeConfig *BRInstanceTypeConfig `mapstructure:"compactBrInstanceTypes"` CompactSyncTimeout string `mapstructure:"compactSyncTimeout" default:"5h"` CompactMinStartTS string `mapstructure:"compactMinStartTS" default:""` CompactOverrides *CompactOverrideConfig `mapstructure:"compactOverrides"` } // GetCompactInstanceType gets the instance type for compact backup based on cluster configuration func (cc *CompactConfig) GetCompactInstanceType(clusterID, tenantID, projectID uint64, cloudProvider, region string) *string { // If CompactBRInstanceTypeConfig is configured, use it to determine instance type if cc.CompactBRInstanceTypeConfig != nil { instanceType := cc.CompactBRInstanceTypeConfig.GetInstanceType( cloudProvider, region, fmt.Sprintf("%d", tenantID), fmt.Sprintf("%d", projectID), fmt.Sprintf("%d", clusterID), ) return &instanceType } // Default to m5.8xlarge if no configuration is available return ptr.To("m5.8xlarge") } // GenCompactK8SResourceRequirements gets CompactBackup K8s resource configuration func (cc *CompactConfig) GenCompactK8SResourceRequirements(clusterID, projectID, tenantID uint64) *corev1.ResourceRequirements { return nil } type BackupConfig struct { EBSBackupEnableMinVer string `mapstructure:"EBSBackupEnableMinVer" default:"v6.5.0"` AWSBackupExportPermissionPropagationDelay time.Duration `mapstructure:"awsBackupExportPermissionPropagationDelay" default:"1m" validate:"gt=0"` CleanupConcurrency int `mapstructure:"cleanupConcurrency" default:"8"` CleanupDeleteConcurrency int `mapstructure:"cleanupDeleteConcurrency" default:"4"` CalSnapshotSizeConcurrency int `mapstructure:"calSnapshotSizeConcurrency" default:"8"` LogBackupEnableMinVer string `mapstructure:"logBackupEnableMinVer" default:"v6.3.0"` LogBackupSizeConcurrency int `mapstructure:"logBackupSizeConcurrency" default:"16"` ServerlessBackupScheduleConcurrency int `mapstructure:"serverlessBackupScheduleConcurrency" default:"16"` DedicatedBackupScheduleConcurrency int `mapstructure:"dedicatedBackupScheduleConcurrency" default:"2"` EBSBackupEnableMinVerConstraint *semver.Constraints `mapstructure:"-"` LogBackupEnableMinVerConstraint *semver.Constraints `mapstructure:"-"` // CRRFEnableVersionConstraintStr checks with cluster version CRRFEnableVersionConstraintStr string `mapstructure:"crrfEnableVersionConstraint" default:">= v8.5.4, != v8.5.5, != v8.5.6, != v8.5.7, != v8.5.8"` CRRFEnableVersionConstraint *semver.Constraints `mapstructure:"-"` // PiTRAddIndexAsyncVersionConstraintStr checks with br version PiTRAddIndexAsyncVersionConstraintStr string `mapstructure:"piTRAddIndexAsyncVersionConstraint" default:">= v8.5.4, != v8.5.5, != v8.5.6, != v8.5.7, != v8.5.8"` PiTRAddIndexAsyncVersionConstraint *semver.Constraints `mapstructure:"-"` // ShardedCompactVersionConstraintStr checks with cluster version. It defines the valid cluster version required for sharded compaction. ShardedCompactVersionConstraintStr string `mapstructure:"shardedCompactVersionConstraint" default:">= v8.5.4, != v8.5.5, != v8.5.6, != v8.5.7, != v8.5.8"` ShardedCompactVersionConstraint *semver.Constraints `mapstructure:"-"` RestoreStoragePercentage float64 `mapstructure:"restoreStoragePercentage" default:"1.1"` ClusterBackupRegionalURLs map[string]string `mapstructure:"clusterBackupRegionalURLs"` ClusterBackupEnvs map[uint64]BREnvs `mapstructure:"clusterBackupEnvs"` ClusterBackupK8sEnvs map[uint64][]corev1.EnvVar `mapstructure:"-"` CommonBackupResourceRequirements *BRResourceRequirements `mapstructure:"commonBackupResourceRequirements"` CommonBackupK8sResourceRequirements *corev1.ResourceRequirements `mapstructure:"_"` // Deprecated: use BackupInstanceTypeConfig or RestoreInstanceTypeConfig instead BRInstanceTypeConfig *BRInstanceTypeConfig `mapstructure:"brInstanceTypes"` BackupInstanceTypeConfig *BRInstanceTypeConfig `mapstructure:"backupInstanceTypes"` RestoreInstanceTypeConfig *BRInstanceTypeConfig `mapstructure:"restoreInstanceTypes"` ClusterBackupResourceRequirements map[uint64]BRResourceRequirements `mapstructure:"clusterBackupResourceRequirements"` ClusterBackupK8sResourceRequirements map[uint64]*corev1.ResourceRequirements `mapstructure:"-"` ProjectBackupResourceRequirements map[uint64]BRResourceRequirements `mapstructure:"projectBackupResourceRequirements"` ProjectBackupK8sResourceRequirements map[uint64]*corev1.ResourceRequirements `mapstructure:"_"` TenantBackupResourceRequirements map[uint64]BRResourceRequirements `mapstructure:"tenantBackupResourceRequirements"` TenantBackupK8sResourceRequirements map[uint64]*corev1.ResourceRequirements `mapstructure:"-"` ClusterLogBackupEnvs map[uint64]BREnvs `mapstructure:"clusterLogBackupEnvs"` ClusterLogBackupK8sEnvs map[uint64][]corev1.EnvVar `mapstructure:"-"` CommonLogBackupResourceRequirements *BRResourceRequirements `mapstructure:"commonLogBackupResourceRequirements"` CommonLogBackupK8sResourceRequirements *corev1.ResourceRequirements `mapstructure:"_"` ClusterLogBackupResourceRequirements map[uint64]BRResourceRequirements `mapstructure:"clusterLogBackupResourceRequirements"` ClusterLogBackupK8sResourceRequirements map[uint64]*corev1.ResourceRequirements `mapstructure:"-"` ProjectLogBackupResourceRequirements map[uint64]BRResourceRequirements `mapstructure:"projectLogBackupResourceRequirements"` ProjectLogBackupK8sResourceRequirements map[uint64]*corev1.ResourceRequirements `mapstructure:"_"` TenantLogBackupResourceRequirements map[uint64]BRResourceRequirements `mapstructure:"tenantLogBackupResourceRequirements"` TenantLogBackupK8sResourceRequirements map[uint64]*corev1.ResourceRequirements `mapstructure:"-"` TenantRestoreEnvs map[uint64]BREnvs `mapstructure:"tenantRestoreEnvs"` TenantRestoreK8sEnvs map[uint64][]corev1.EnvVar `mapstructure:"-"` CommonRestoreResourceRequirements *BRResourceRequirements `mapstructure:"commonRestoreResourceRequirements"` CommonRestoreK8sResourceRequirements *corev1.ResourceRequirements `mapstructure:"_"` ProjectRestoreResourceRequirements map[uint64]BRResourceRequirements `mapstructure:"projectRestoreResourceRequirements"` ProjectRestoreK8sResourceRequirements map[uint64]*corev1.ResourceRequirements `mapstructure:"_"` TenantRestoreResourceRequirements map[uint64]BRResourceRequirements `mapstructure:"tenantRestoreResourceRequirements"` TenantRestoreK8sResourceRequirements map[uint64]*corev1.ResourceRequirements `mapstructure:"-"` CRRFOverrides CRRFOverrides `mapstructure:"crrfOverrides"` RestoreOverrides RestoreOverrides `mapstructure:"restoreOverrides"` MaxBucketDeleteBatchCount int `mapstructure:"maxBucketDeleteBatchCount" default:"20"` AzureStorageAccountResourceGroup string `mapstructure:"azureStorageAccountResourceGroup"` } // CRRFOverride set CRRF override for clusters matched by FeatureGate. type CRRFOverride struct { Gate FeatureGate `mapstructure:"gate"` // Image overrides the CRRF daemon image for matched clusters. Image string `mapstructure:"image"` // CalculatePiTRTimeRangeV2 controls whether to use the V2 PiTR time range grouping algorithm // for secondary regions when computing cross-region PiTR time ranges. CalculatePiTRTimeRangeV2 bool `mapstructure:"calculatePiTRTimeRangeV2" default:"false"` // AutoCRRF enables automatic CRRF enable/disable triggered by PiTR and dual-region backup changes. // When false (default), auto-CRRF is skipped even if other conditions are met. AutoCRRF bool `mapstructure:"autoCRRF" default:"false"` } // CRRFOverrides is a list of CRRFOverride entries. type CRRFOverrides []CRRFOverride // GetMatched returns the first matching CRRFOverride and true when a match is found for the given cluster, // or a zero CRRFOverride and false otherwise. func (o CRRFOverrides) GetMatched( tenantID, projectID, clusterID uint64, cloudProvider, region string, ) (_ CRRFOverride, ok bool) { for _, override := range o { if _, ok := override.Gate.Enabled(FeatureGateQueryParams{ Provider: cloudProvider, Region: region, TenantID: tenantID, ProjectID: projectID, ClusterID: clusterID, }); ok { return override, true } } return CRRFOverride{}, false } // RestoreOverride set Restore override for clusters matched by FeatureGate. type RestoreOverride struct { Gate FeatureGate `mapstructure:"gate"` // CompactionAcceleratedPiTR controls whether a backup compaction(compactbackup) is performed before PiTR. CompactionAcceleratedPiTR bool `mapstructure:"compactionAcceleratedPiTR" default:"false"` AddIndexAsyncForPiTR bool `mapstructure:"addIndexAsyncForPiTR" default:"false"` // AllowConcurrentPiTR bypasses the PiTR serialization constraint. // When false (default), at most one PiTR restore per log backup may be in PENDING/RUNNING state. // The serialization is enforced via two best-effort guards: // - Layer 1: a pre-check at entity creation time (makeCreateRestoreEntity) // - Layer 2: a re-check inside the compaction workflow action (RunCompactBackupForRestore.execute) // Neither layer is truly atomic; they work together to reduce — not eliminate — concurrent compaction risk. AllowConcurrentPiTR bool `mapstructure:"allowConcurrentPiTR" default:"false"` BRImageTag string `mapstructure:"brImageTag"` // Override BR image tag, e.g., "v8.5.3" // ComponentImageTags overrides per-component image tags for the restored cluster. // Valid keys: "tidb", "tikv", "pd", "tiflash", "tiproxy" // Example: {"tidb": "v8.5.0-custom", "tikv": "v8.5.0-custom"} ComponentImageTags map[string]string `mapstructure:"componentImageTags"` // TiKVDataAdditionalIops adds additional IOPS to TiKV DATA disk. // Final IOPS = SKU base IOPS + tikvDataAdditionalIops TiKVDataAdditionalIops int32 `mapstructure:"tikvDataAdditionalIops"` // CloneImageTags copies image tags from the source cluster for the specified components. // Valid items: "tidb", "tikv", "pd", "tiflash", "tiproxy" // Priority: ComponentImageTags > CloneImageTags // Example: ["tidb", "tikv"] - use the same image tags as the source cluster for TiDB and TiKV CloneImageTags []string `mapstructure:"cloneImageTags"` // CloneImageTagsIgnorePatchVersionDiff copies image tags (if CloneImageTags is enabled) from the source cluster, // ignoring patch version differences. // // For example, if the backup is from TiDB v8.5.0 and the source cluster is running with TiDB v8.5.3 now, // the image tag for TiDB will be copied from the source cluster even though the patch versions differ. // However, if the source cluster is running with TiDB v8.6.0, the image tag will not be copied because the minor version differs. CloneImageTagsIgnorePatchVersionDiff bool `mapstructure:"cloneImageTagsIgnorePatchVersionDiff" default:"true"` // CloneAdditionalIops copies additional IOPS from the source cluster for the specified components. // Valid items: "tikv", "tiflash" (only storage components have additional IOPS) // Priority: TiKVDataAdditionalIops > CloneAdditionalIops for TiKV // Example: ["tikv"] - use the same additional IOPS as the source cluster for TiKV CloneAdditionalIops []string `mapstructure:"cloneAdditionalIops"` } // RestoreOverrides is an ordered list of RestoreOverride entries. // The first entry whose Gate matches wins. type RestoreOverrides []RestoreOverride // GetMatched returns the first matching RestoreOverride and true when a match is found for the given cluster, // or a zero RestoreOverride and false otherwise. func (ro RestoreOverrides) GetMatched( tenantID, projectID, clusterID uint64, cloudProvider, region string, ) (_ RestoreOverride, ok bool) { for _, override := range ro { if _, ok := override.Gate.Enabled(FeatureGateQueryParams{ Provider: cloudProvider, Region: region, TenantID: tenantID, ProjectID: projectID, ClusterID: clusterID, }); ok { return override, true } } return RestoreOverride{}, false } func (bc *BackupConfig) Init() error { constraint, err := semver.NewConstraint(fmt.Sprintf(">=%s", bc.LogBackupEnableMinVer)) if err != nil { return errors.Annotatef(err, "new log backup enable version constraint: %s", bc.LogBackupEnableMinVer) } bc.LogBackupEnableMinVerConstraint = constraint constraint, err = semver.NewConstraint(fmt.Sprintf(">=%s", bc.EBSBackupEnableMinVer)) if err != nil { return errors.Annotatef(err, "new EBS backup enable version constraint: %s", bc.EBSBackupEnableMinVer) } bc.EBSBackupEnableMinVerConstraint = constraint constraint, err = semver.NewConstraint(bc.CRRFEnableVersionConstraintStr) if err != nil { return errors.Annotatef(err, "new CRRF enable version constraint: %s", bc.CRRFEnableVersionConstraintStr) } bc.CRRFEnableVersionConstraint = constraint constraint, err = semver.NewConstraint(bc.PiTRAddIndexAsyncVersionConstraintStr) if err != nil { return errors.Annotatef(err, "new PiTR add index async version constraint: %s", bc.PiTRAddIndexAsyncVersionConstraintStr) } bc.PiTRAddIndexAsyncVersionConstraint = constraint constraint, err = semver.NewConstraint(bc.ShardedCompactVersionConstraintStr) if err != nil { return errors.Annotatef(err, "new sharded compact version constraint: %s", bc.ShardedCompactVersionConstraintStr) } bc.ShardedCompactVersionConstraint = constraint if err := bc.InitBackupResourceConfig(); err != nil { return err } if err := bc.InitLogBackupResourceConfig(); err != nil { return err } if err := bc.InitRestoreResourceConfig(); err != nil { return err } return nil } func (bc *BackupConfig) InitRestoreResourceConfig() error { if bc.TenantRestoreEnvs != nil { envMap, err := bc.ParseBREnvMap(bc.TenantRestoreEnvs) if err != nil { return errors.Annotate(err, "parse TenantRestoreEnvs") } bc.TenantRestoreK8sEnvs = envMap } if bc.CommonRestoreResourceRequirements != nil { requirement, err := bc.ParseBRResourceRequirement(bc.CommonRestoreResourceRequirements) if err != nil { return errors.Annotate(err, "parse CommonRestoreResourceRequirements") } bc.CommonRestoreK8sResourceRequirements = requirement } if bc.TenantRestoreResourceRequirements != nil { requirements, err := bc.ParseBRResourceRequirementsMap(bc.TenantRestoreResourceRequirements) if err != nil { return errors.Annotate(err, "parse TenantRestoreResourceRequirements") } bc.TenantRestoreK8sResourceRequirements = requirements } if bc.ProjectRestoreResourceRequirements != nil { requirements, err := bc.ParseBRResourceRequirementsMap(bc.ProjectRestoreResourceRequirements) if err != nil { return errors.Annotate(err, "parse ProjectRestoreResourceRequirements") } bc.ProjectRestoreK8sResourceRequirements = requirements } return nil } func (bc *BackupConfig) InitBackupResourceConfig() error { if bc.ClusterBackupEnvs != nil { envMap, err := bc.ParseBREnvMap(bc.ClusterBackupEnvs) if err != nil { return errors.Annotate(err, "parse ClusterBackupEnvs") } bc.ClusterBackupK8sEnvs = envMap } if bc.CommonBackupResourceRequirements != nil { requirement, err := bc.ParseBRResourceRequirement(bc.CommonBackupResourceRequirements) if err != nil { return errors.Annotate(err, "parse commonBackupResourceRequirements") } bc.CommonBackupK8sResourceRequirements = requirement } if bc.ClusterBackupResourceRequirements != nil { requirements, err := bc.ParseBRResourceRequirementsMap(bc.ClusterBackupResourceRequirements) if err != nil { return errors.Annotate(err, "parse ClusterBackupResourceRequirements") } bc.ClusterBackupK8sResourceRequirements = requirements } if bc.ProjectBackupResourceRequirements != nil { requirements, err := bc.ParseBRResourceRequirementsMap(bc.ProjectBackupResourceRequirements) if err != nil { return errors.Annotatef(err, "parse ProjectBackupResourceRequirements") } bc.ProjectBackupK8sResourceRequirements = requirements } if bc.TenantBackupResourceRequirements != nil { requirements, err := bc.ParseBRResourceRequirementsMap(bc.TenantBackupResourceRequirements) if err != nil { return errors.Annotatef(err, "parse tenantBackupResourceRequirements") } bc.TenantBackupK8sResourceRequirements = requirements } return nil } func (bc *BackupConfig) InitLogBackupResourceConfig() error { if bc.ClusterLogBackupEnvs != nil { envMap, err := bc.ParseBREnvMap(bc.ClusterLogBackupEnvs) if err != nil { return errors.Annotate(err, "parse ClusterLogBackupEnvs") } bc.ClusterLogBackupK8sEnvs = envMap } if bc.CommonLogBackupResourceRequirements != nil { requirement, err := bc.ParseBRResourceRequirement(bc.CommonLogBackupResourceRequirements) if err != nil { return errors.Annotate(err, "parse commonLogBackupResourceRequirements") } bc.CommonLogBackupK8sResourceRequirements = requirement } if bc.ClusterLogBackupResourceRequirements != nil { requirements, err := bc.ParseBRResourceRequirementsMap(bc.ClusterLogBackupResourceRequirements) if err != nil { return errors.Annotate(err, "parse ClusterLogBackupResourceRequirements") } bc.ClusterLogBackupK8sResourceRequirements = requirements } if bc.ProjectLogBackupResourceRequirements != nil { requirements, err := bc.ParseBRResourceRequirementsMap(bc.ProjectLogBackupResourceRequirements) if err != nil { return errors.Annotatef(err, "parse ProjectLogBackupResourceRequirements") } bc.ProjectLogBackupK8sResourceRequirements = requirements } if bc.TenantLogBackupResourceRequirements != nil { requirements, err := bc.ParseBRResourceRequirementsMap(bc.TenantLogBackupResourceRequirements) if err != nil { return errors.Annotatef(err, "parse tenantLogBackupResourceRequirements") } bc.TenantLogBackupK8sResourceRequirements = requirements } return nil } func (bc *BackupConfig) ParseBREnvMap(envMap map[uint64]BREnvs) (map[uint64][]corev1.EnvVar, error) { if len(envMap) == 0 { return map[uint64][]corev1.EnvVar{}, nil } backupK8sEnvs := make(map[uint64][]corev1.EnvVar, len(envMap)) for id, envs := range envMap { k8sEnvs, err := envs.ToK8sEnvs() if err != nil { return nil, errors.Annotatef(err, "parse backup envs of id %d", id) } backupK8sEnvs[id] = k8sEnvs } return backupK8sEnvs, nil } func (bc *BackupConfig) ParseBRResourceRequirementsMap(input map[uint64]BRResourceRequirements) (map[uint64]*corev1.ResourceRequirements, error) { if len(input) == 0 { return map[uint64]*corev1.ResourceRequirements{}, nil } requirements := make(map[uint64]*corev1.ResourceRequirements, len(input)) for id, backupRequirements := range input { k8sBackupRequirements, err := backupRequirements.ToK8sResourceRequirements() if err != nil { return nil, errors.Annotatef(err, "parse backup resource requirements of id %d", id) } requirements[id] = k8sBackupRequirements } return requirements, nil } func (bc *BackupConfig) ParseBRResourceRequirement(input *BRResourceRequirements) (*corev1.ResourceRequirements, error) { if input == nil { return nil, nil } requirement, err := input.ToK8sResourceRequirements() if err != nil { return nil, errors.Annotatef(err, "parse backup resource requirements") } return requirement, nil } func (bc *BackupConfig) ParseCommonBackupResourceRequirements() error { k8sBackupRequirements, err := bc.CommonBackupResourceRequirements.ToK8sResourceRequirements() if err != nil { return errors.Annotatef(err, "parse common backup resource requirements") } bc.CommonBackupK8sResourceRequirements = k8sBackupRequirements return nil } func (bc *BackupConfig) ParseCommonRestoreResourceRequirements() error { k8sRestoreRequirements, err := bc.CommonRestoreResourceRequirements.ToK8sResourceRequirements() if err != nil { return errors.Annotatef(err, "parse common restore resource requirements") } bc.CommonRestoreK8sResourceRequirements = k8sRestoreRequirements return nil } func (bc *BackupConfig) GenLogBackupK8SResourceRequirements(b *appentity.LogBackup) *corev1.ResourceRequirements { if requirements, ok := bc.ClusterLogBackupK8sResourceRequirements[uint64(b.ClusterID)]; ok { return requirements } if requirements, ok := bc.ProjectLogBackupK8sResourceRequirements[uint64(b.ProjectID)]; ok { return requirements } if requirements, ok := bc.TenantLogBackupK8sResourceRequirements[uint64(b.TenantID)]; ok { return requirements } if bc.CommonLogBackupK8sResourceRequirements != nil { return bc.CommonLogBackupK8sResourceRequirements } return nil } func (bc *BackupConfig) GenBackupK8SResourceRequirements(b *appentity.Backup) *corev1.ResourceRequirements { if requirements, ok := bc.ClusterBackupK8sResourceRequirements[uint64(b.ClusterID)]; ok { return requirements } if requirements, ok := bc.ProjectBackupK8sResourceRequirements[uint64(b.ProjectID)]; ok { return requirements } if requirements, ok := bc.TenantBackupK8sResourceRequirements[uint64(b.TenantID)]; ok { return requirements } if bc.CommonBackupK8sResourceRequirements != nil { return bc.CommonBackupK8sResourceRequirements } return nil } func (bc *BackupConfig) GetBackupInstanceType(b *appentity.Backup, cloudProvider string, region string) *string { req, ok := bc.ClusterBackupResourceRequirements[uint64(b.ClusterID)] if ok && req.InstanceType != nil { return req.InstanceType } req, ok = bc.ProjectBackupResourceRequirements[uint64(b.ProjectID)] if ok && req.InstanceType != nil { return req.InstanceType } req, ok = bc.TenantBackupResourceRequirements[uint64(b.TenantID)] if ok && req.InstanceType != nil { return req.InstanceType } return ptr.To(bc.BackupInstanceTypeConfig.GetInstanceType( cloudProvider, region, fmt.Sprintf("%d", b.TenantID), fmt.Sprintf("%d", b.ProjectID), fmt.Sprintf("%d", b.ClusterID), )) } func (bc *BackupConfig) GetRestoreInstanceType( tenantID uint64, projectID uint64, clusterID uint64, cloudProvider string, region string, ) *string { req, ok := bc.ProjectRestoreResourceRequirements[projectID] if ok && req.InstanceType != nil { return req.InstanceType } req, ok = bc.TenantRestoreResourceRequirements[tenantID] if ok && req.InstanceType != nil { return req.InstanceType } return ptr.To(bc.RestoreInstanceTypeConfig.GetInstanceType( cloudProvider, region, fmt.Sprintf("%d", tenantID), fmt.Sprintf("%d", projectID), fmt.Sprintf("%d", clusterID), )) } func (bc *BackupConfig) GetRestoreK8SResourceRequirements( tenantID uint64, projectID uint64, ) *corev1.ResourceRequirements { if requirements, ok := bc.ProjectRestoreK8sResourceRequirements[projectID]; ok { return requirements } if requirements, ok := bc.TenantRestoreK8sResourceRequirements[tenantID]; ok { return requirements } if bc.CommonRestoreK8sResourceRequirements != nil { return bc.CommonRestoreK8sResourceRequirements } return nil } // GetCompactBackupIntervalDuration returns the compact backup interval as duration // This is a global setting and does not support cluster-level overrides func (cc *CompactConfig) GetCompactBackupIntervalDuration() (time.Duration, error) { return time.ParseDuration(cc.CompactBackupInterval) } // InitCompactResourceConfig initializes CompactConfig ResourceRequirements configuration // DISABLED FOR TESTING: Skip all ResourceRequirements initialization to ensure clean testing func (cc *CompactConfig) InitCompactResourceConfig() error { // DISABLED: Skip all initialization to ensure ResourceRequirements don't interfere with testing return nil } type SecretConfig struct { SecretKey string `mapstructure:"secretKey"` Region string `mapstructure:"region"` } type GrafanaConfig struct { Server string `mapstructure:"server" default:"https://grafana.oauth.dev.shared.aws.tidbcloud.com/"` APIKey string `mapstructure:"apiKey"` TidbDashboardTag string `mapstructure:"tidbDashboardTag" default:"tidb-mixin"` BackupGaugeMetrics []string `mapstructure:"backupGaugeMetrics"` CDCGaugeMetrics []string `mapstructure:"cdcGaugeMetrics"` } // infra configurations type InfraConfig struct { // dbaas central service region Region string `mapstructure:"region" required:"true" default:"us-west-2"` CMKArn string `mapstructure:"cmkArn" required:"true"` GCPCrossPlaneProvider string `mapstructure:"gcpCrossPlaneProvider" default:"provider-gcp"` AWSCrossPlaneProvider string `mapstructure:"gcpCrossPlaneProvider" default:"provider-aws"` } type Auth0Config struct { Auth0ManagementClientID string `mapstructure:"auth0ManagementClientID" required:"true"` Auth0ManagementClientSecret string `mapstructure:"auth0ManagementClientSecret" required:"true"` Auth0DatabaseConnectionName string `mapstructure:"auth0DatabaseConnectionName" required:"true" default:"custom-email"` Auth0Domain string `mapstructure:"auth0Domain" required:"true"` Auth0Audience string `mapstructure:"auth0Audience" required:"true"` Auth0Secret string `mapstructure:"auth0Secret" required:"true"` ManagementPortalAuth0Domain string `mapstructure:"managementPortalAuth0Domain" required:"true"` ManagementPortalAuth0Audience string `mapstructure:"managementPortalAuth0Audience" required:"true"` ManagementPortalAuth0Secret string `mapstructure:"managementPortalAuth0Secret" required:"true"` ManagementPortalAuth0ClientAudience string `mapstructure:"managementPortalAuth0ClientAudience" required:"true"` } type BillingConfig struct { // The billing system base url Billing2BaseURL string `mapstructure:"billing2BaseURL" default:"http://billing2:18888" required:"true"` Billing2URLPrefix string `mapstructure:"billing2URLPrefix" default:"/api/v1/billing" required:"true"` BillingPackageIDs BillingPackageIDs `mapstructure:"billingPackageIDs"` EnablePOCEndedCleanup bool `mapstructure:"enablePOCEndedCleanup" default:"true"` EnableCDCTCUMetering bool `mapstructure:"enableCDCTCUMetering" default:"false"` CDCTCUMappingTable [][]int `mapstructure:"cdcTCUMappingTable" default:"[[0,16], [48,24], [120,32], [169, 40]]"` DiscountDescription string `mapstructure:"discountDescription" default:""` } type BillingPackageIDs struct { OnDemand uint64 `mapstructure:"onDemand"` Poc uint64 `mapstructure:"poc"` DevTier uint64 `mapstructure:"devTier"` } type ProxyConfig struct { Auth0Domain string `mapstructure:"auth0Domain" required:"true"` Auth0ClientID string `mapstructure:"auth0ClientID" required:"true"` Auth0ClientSecret string `mapstructure:"auth0ClientSecret" required:"true"` Auth0Audience string `mapstructure:"auth0Audience" required:"true"` } type AwsConfig struct { PCAArn string `mapstructure:"pcaArn" required:"true"` PCARegion string `mapstructure:"pcaRegion" required:"true"` Region string `mapstructure:"region" required:"true"` AccountID string `mapstructure:"accountID" required:"true"` ProjectID string `mapstructure:"projectID" required:"true"` ReporterAccessKeyID string `mapstructure:"reporterAccessKeyID" required:"true"` ReporterSecretAccessKey string `mapstructure:"reporterSecretAccessKey" required:"true"` // s3 sdk request timeout in seconds S3SDKRequestTimeout int `mapstructure:"s3SDKRequestTimeout" default:"5"` } type GardenerConfig struct { GardenerProjectName string `mapstructure:"gardenerProjectName" default:"core" required:"true"` } type MysqlConfig struct { Host string `mapstructure:"host" default:"host"` User string `mapstructure:"user" default:"root"` Port uint16 `mapstructure:"port" default:"4000"` Password string `mapstructure:"password" required:"true"` Database string `mapstructure:"database" default:"dbaas"` } type TLSConfig struct { TlsCA string `json:"ca.crt"` TlsCert string `json:"tls.crt"` TlsKey string `json:"tls.key"` } type RBACConfig struct { Switch bool `mapstructure:"switch" default:"false"` RBACPathPattern2ResourceMap map[string]string `mapstructure:"rbacPathPattern2ResourceMap"` PathPattern2ResourceMap map[*regexp.Regexp]string `mapstructure:"-"` RBACPathWhitelist []string `mapstructure:"rbacPathWhitelist"` } func (r *RBACConfig) Init() error { if err := r.initRbacPathWhiteList(); err != nil { return err } return r.initPathPattern2ResourceMap() } func (r *RBACConfig) initRbacPathWhiteList() error { if len(r.RBACPathWhitelist) == 0 { return nil } for _, p := range r.RBACPathWhitelist { _, err := regexp.Compile(p) if err != nil { return errors.Annotatef(err, "whitelist path:%s invalid", p) } } return nil } func (r *RBACConfig) GetRBACResource(path string, method string) string { method = strings.ToLower(method) path = strings.ToLower(path) for p, uri := range r.PathPattern2ResourceMap { if p.MatchString(strings.ToLower(fmt.Sprintf("%s:%s", method, path))) { return uri } } return "" } func (r *RBACConfig) initPathPattern2ResourceMap() error { if r.RBACPathPattern2ResourceMap == nil || len(RBACPathPattern2ResourceMap) == 0 { return errors.New("resourceMap is empty") } var formatResourceMap = map[*regexp.Regexp]string{} if err := r.covertExprToRegexp(formatResourceMap, r.RBACPathPattern2ResourceMap); err != nil { return err } if err := r.covertExprToRegexp(formatResourceMap, RBACPathPattern2ResourceMap); err != nil { return err } r.PathPattern2ResourceMap = formatResourceMap return nil } func (r *RBACConfig) covertExprToRegexp( formatResourceMap map[*regexp.Regexp]string, exprMap map[string]string, ) error { for pathExpr, resourceExpr := range exprMap { pathExpr = strings.ToLower(pathExpr) pathPattern, method := r.parsePathPattern(pathExpr) if len(method) == 0 || len(pathPattern) == 0 { return errors.New("pathPattern or method is empty") } // pathPattern: /api/v1/orgs/(\d+)/poc-applications // resourceExpr: /orgs/{org_id}/owner pathExprMatchExpr := fmt.Sprintf(common.EXACT_MATCH_FORMATTER, pathExpr) // check the pathPattern valid reg, err := regexp.Compile(pathExprMatchExpr) if err != nil { return err } formatResourceMap[reg] = resourceExpr } return nil } func (r *RBACConfig) parsePathPattern(pathPattern string) (path, method string) { split := strings.Split(pathPattern, common.PATH_METHOD_SEPERATOR) if len(split) == 1 { path = split[0] return path, method } if len(split) == 2 { method = split[0] path = split[1] return path, method } method = split[0] path = strings.Join(split[1:], common.PATH_METHOD_SEPERATOR) return method, path } func (r *RBACConfig) IsIgnoreRBACPath(path string) bool { if len(r.RBACPathWhitelist) == 0 { return false } for _, p := range r.RBACPathWhitelist { reg := regexp.MustCompile(p) if reg.MatchString(path) { return true } } return false } type EnvType string const ( EnvTypeDev EnvType = "nightly" EnvTypeStaging EnvType = "staging" EnvTypeProd EnvType = "prod" )