package config import ( "fmt" "os" "reflect" "strings" "github.com/elliotchance/pie/v2" "github.com/tidbcloud/dataflow-service-e2e/internal/login" "github.com/tidbcloud/dataflow-service-e2e/internal/misc" . "github.com/tidbcloud/dataflow-service-e2e/internal/model" "gopkg.in/yaml.v2" ) const ( LocalENV = "local" DevENV = "dev" StagingENV = "staging" ProdENV = "prod" ) const ( DomainAll = "all" DomainDM = "dm" DomainCDC = "cdc" DomainBR = "br" DomainCRRF = "crrf" DomainImport = "import" DomainExport = "export" ) const ( ApiModeConsoleAPI = "consoleapi" ApiModeOpenAPI = "openapi" ) // Canonical import data formats. The values match the backend import file-type // enums (essential ImportFileTypeEnum, dedicated DataflowImportDataFormat) so // the importData config maps to API requests without translation. const ( DataFormatSQL = "SQL" DataFormatCSV = "CSV" DataFormatParquet = "PARQUET" DataFormatAuroraSnapshot = "AURORA_SNAPSHOT" ) // SuitePlan identifies the E2E suite whose region matrix is selected. // It is deliberately distinct from TiDB Serverless cluster- and service-plan values. type SuitePlan string const ( SuitePlanDedicated SuitePlan = "dedicated" SuitePlanPremium SuitePlan = "premium" SuitePlanEssential SuitePlan = "essential" ) var ( LegalENVs = []string{LocalENV, DevENV, StagingENV, ProdENV} AllDomains = []string{DomainAll, DomainDM, DomainCDC, DomainBR, DomainCRRF, DomainImport, DomainExport} AllApiModes = []string{ApiModeConsoleAPI, ApiModeOpenAPI} AllSuitePlans = []SuitePlan{SuitePlanDedicated, SuitePlanPremium, SuitePlanEssential} // legalDataFormats is the supported importData dataFormat vocabulary; it is // the same set the backend import file-type enums accept. legalDataFormats = map[string]struct{}{ DataFormatSQL: {}, DataFormatCSV: {}, DataFormatParquet: {}, DataFormatAuroraSnapshot: {}, } ) type loginFuncType func(loginInCI *bool, url, user, password string) (string, error) var GlobalConfig *config var loginToTidbCloud loginFuncType = login.Login func InitConfig(env, domain, apiMode, envConfigFile, executor, casesFlag, caseGroup string, suitePlan SuitePlan) { commonConfig := config{} data, err := os.ReadFile("./config.common.yaml") if err != nil { panic("fail to load common config: " + err.Error()) } err = yaml.Unmarshal(data, &commonConfig) if err != nil { panic("fail to unmarshal common config: " + err.Error()) } envConfig := config{} envConfigPath := resolveEnvConfigPath(env, envConfigFile) if envConfigFile != "" { fmt.Printf("Environment config file overridden from flag: %s\n", envConfigPath) } // override the common config data, err = os.ReadFile(envConfigPath) if err != nil { panic(fmt.Sprintf("fail to load env specified config(%s): %s", envConfigPath, err.Error())) } err = yaml.Unmarshal(data, &envConfig) if err != nil { panic(fmt.Sprintf("fail to unmarshal env specified config(%s): %s", envConfigPath, err.Error())) } combinedConfig := combineCommonAndEnvConfigs(commonConfig, envConfig) combinedConfig = initConfigWithDependencies(combinedConfig, env, domain, apiMode, executor, casesFlag, caseGroup, suitePlan, loginToTidbCloud) GlobalConfig = &combinedConfig fmt.Printf("Config loaded:\n--------\n%s--------\n", misc.PrettyYAML(configSummary(GlobalConfig))) } func resolveEnvConfigPath(env, envConfigFile string) string { if envConfigFile != "" { fmt.Printf("use env config file overridden from flag: %s\n", envConfigFile) return envConfigFile } return fmt.Sprintf("./config.%s.yaml", env) } func initConfigWithDependencies(combinedConfig config, env, domain, apiMode, executor, casesFlag, caseGroup string, suitePlan SuitePlan, loginFn loginFuncType) config { combinedConfig.env = env combinedConfig.SelectRegionsForSuitePlan(suitePlan) if domain != "" { combinedConfig.Domain = domain fmt.Printf("Domain overridden from flag: %s\n", domain) } if apiMode != "" { combinedConfig.ApiMode = apiMode fmt.Printf("API mode overridden from flag: %s\n", apiMode) } if executor != "" { combinedConfig.Executor = executor fmt.Printf("Executor overridden from flag: %s\n", executor) } if casesFlag != "" { combinedConfig.Cases2Run = strings.Split(casesFlag, ",") fmt.Printf("Cases2Run overridden from flag: %v\n", combinedConfig.Cases2Run) } if caseGroup != "" { combinedConfig.CaseGroup2Run = caseGroup fmt.Printf("CaseGroup2Run overridden from flag: %s\n", caseGroup) } combinedConfig.Validate() // Allow AliCloud OSS import credentials to be supplied at runtime via env, // overriding whatever was merged from config. CI injects these from secrets. applyImportSampleDataEnvOverrides(&combinedConfig.ImportSampleData) // login tidbcloud console to get the token if combinedConfig.TidbCloudToken == "" { fmt.Printf("Login tidb cloud console with url(%s), user(%s), password(xxx) to get the token...\n", combinedConfig.TiDBCloudURL, combinedConfig.TiDBCloudUser) token, err := loginFn(combinedConfig.RunAsCI, combinedConfig.TiDBCloudURL, combinedConfig.TiDBCloudUser, combinedConfig.TiDBCloudPassword) if err != nil { panic(fmt.Sprintf("failed to login to tidb cloud console with url(%s), user(%s), password(xxx), error: %s", combinedConfig.TiDBCloudURL, combinedConfig.TiDBCloudUser, err.Error())) } combinedConfig.TidbCloudToken = token fmt.Println("Login to tidb cloud console succeeded") } else { fmt.Println("Token already configured, no need to login to tidb cloud console") } return combinedConfig } func mask(cf *config) config { data, _ := yaml.Marshal(cf) var tmp config yaml.Unmarshal(data, &tmp) tmp.TiDBCloudPassword = "xxx" tmp.TidbCloudToken = "xxx" tmp.OpenApi.PublicKey = "xxx" tmp.OpenApi.PrivateKey = "xxx" for _, kafka := range tmp.Kafka { if kafka != nil { kafka.Password = "xxx" kafka.User = "xxx" } } for _, mysql := range tmp.MySQL { if mysql != nil { mysql.Password = "xxx" mysql.User = "xxx" } } for _, s3 := range tmp.S3 { if s3 != nil { s3.SecretKey = "xxx" s3.AccessKey = "xxx" } } for _, azure := range tmp.AzureBlobStorage { if azure != nil { azure.SASToken = "xxx" } } for _, pulsar := range tmp.Pulsar { if pulsar != nil { pulsar.Password = "xxx" pulsar.Token = "xxx" pulsar.User = "xxx" } } if tmp.SchemaRegistry != nil { tmp.SchemaRegistry.Password = "xxx" tmp.SchemaRegistry.Username = "xxx" } tmp.BackupExport.AWS.TargetURI = "xxx" tmp.BackupExport.AWS.RoleARN = "xxx" tmp.BackupExport.GCP.TargetURI = "xxx" tmp.BackupExport.AliCloud.TargetURI = "xxx" tmp.BackupExport.AliCloud.RoleARN = "xxx" tmp.EssentialExport.AWS.TargetURI = "xxx" tmp.EssentialExport.AWS.RoleARN = "xxx" tmp.EssentialExport.GCP.TargetURI = "xxx" tmp.EssentialExport.AliCloud.TargetURI = "xxx" tmp.EssentialExport.AliCloud.RoleARN = "xxx" tmp.ImportSampleData.AzureBlob.SASToken = "xxx" tmp.ImportSampleData.OSS.AccessKeyID = "xxx" tmp.ImportSampleData.OSS.AccessKeySecret = "xxx" for _, items := range [][]ImportDataItem{tmp.ImportData.S3, tmp.ImportData.GCS, tmp.ImportData.AzureBlob, tmp.ImportData.OSS} { for i := range items { items[i].AccessKeyID = "xxx" items[i].AccessKeySecret = "xxx" items[i].SASToken = "xxx" } } return tmp } type safeConfigSummary struct { ENV string `yaml:"env"` RunAsCI *bool `yaml:"runAsCI"` MustCleanup *bool `yaml:"mustCleanup"` Executor string `yaml:"executor"` Domain string `yaml:"domain"` APIMode string `yaml:"apiMode"` PremiumClusterCount int `yaml:"premiumClusterCount"` DedicatedClusterCount int `yaml:"dedicatedClusterCount"` ServerlessClusterCount int `yaml:"serverlessClusterCount"` CasesToRun []string `yaml:"casesToRun"` CaseGroupToRun string `yaml:"caseGroupToRun"` } func configSummary(cf *config) safeConfigSummary { return safeConfigSummary{ ENV: cf.env, RunAsCI: cf.RunAsCI, MustCleanup: cf.MustCleanup, Executor: cf.Executor, Domain: cf.Domain, APIMode: cf.ApiMode, PremiumClusterCount: len(cf.TiDBPremiumClusterIds), DedicatedClusterCount: len(cf.TiDBDedicatedClusterIds), ServerlessClusterCount: len(cf.TiDBServerlessClusterIds), CasesToRun: cf.Cases2Run, CaseGroupToRun: cf.CaseGroup2Run, } } type config struct { env string RunAsCI *bool `yaml:"runAsCI"` MustCleanup *bool `yaml:"mustCleanup"` Executor string `yaml:"executor"` Domain string `yaml:"domain"` ApiMode string `yaml:"apiMode"` TiDBCloudURL string `yaml:"tidbCloudURL"` TiDBCloudConsoleAPIURL string `yaml:"tidbCloudConsoleAPIURL"` TiDBCloudUser string `yaml:"tidbCloudUser"` TiDBCloudPassword string `yaml:"tidbCloudPassword"` TidbCloudToken string `yaml:"tidbCloudToken"` OrgID string `yaml:"orgID"` ProjectID string `yaml:"projectID"` // Regions is the selected suite's runtime-only projection. YAML must use // SuitePlanRegions so different E2E suites cannot share a flat region set. Regions cloudRegion `yaml:"-"` SuitePlanRegions map[SuitePlan]cloudRegion `yaml:"suitePlanRegions"` TiDBDedicatedClusterIds []string `yaml:"tidbDedicatedClusterIds"` TiDBPremiumClusterIds []string `yaml:"tiDBPremiumClusterIds"` TiDBServerlessClusterIds []string `yaml:"tiDBServerlessClusterIds"` Kafka []*KafkaEntity `yaml:"kafka"` MySQL []*MySQLEntity `yaml:"mysql"` S3 []*S3Entity `yaml:"s3"` GCS []*GcsEntity `yaml:"gcs"` AzureBlobStorage []*AzureBlobStorageEntity `yaml:"azureBlobStorage"` Pulsar []*PulsarEntity `yaml:"pulsar"` SchemaRegistry *SchemaRegistry `yaml:"schemaRegistry"` Cases2Run []string `yaml:"cases2Run"` CaseGroup2Run string `yaml:"caseGroup2Run"` OpenApi OpenApiConfig `yaml:"openApi"` ServerlessRegion string `yaml:"serverlessRegion"` ImportSampleData ImportSampleDataConfig `yaml:"importSampleData"` ImportData ImportDataConfig `yaml:"importData"` BackupExport BackupExportConfig `yaml:"backupExport"` EssentialExport BackupExportConfig `yaml:"essentialExport"` } type OpenApiConfig struct { ApiURL string `yaml:"apiURL"` PremiumApiURL string `yaml:"premiumApiURL"` PublicKey string `yaml:"publicKey"` PrivateKey string `yaml:"privateKey"` } type ImportSampleDataCSVFormatConfig struct { Separator string `yaml:"separator"` Delimiter string `yaml:"delimiter"` LineTerminator string `yaml:"lineTerminator"` EscapeChar string `yaml:"escapeChar"` SkipRows int64 `yaml:"skipRows"` NullValue string `yaml:"nullValue"` SplitFile bool `yaml:"splitFile"` } type ImportDBTableName struct { Database string `yaml:"database"` Table string `yaml:"table"` } type ImportSampleDataOSSConfig struct { SourceURL string `yaml:"sourceURL"` Endpoint string `yaml:"endpoint"` AccessKeyID string `yaml:"accessKeyID"` AccessKeySecret string `yaml:"accessKeySecret"` ImportedDBTable *ImportDBTableName `yaml:"importedDBTable"` CustomCSVFormat *ImportSampleDataCSVFormatConfig `yaml:"customCSVFormat"` } type ImportSampleDataConfig struct { S3 struct { SourceURL string `yaml:"sourceURL"` RoleARN string `yaml:"roleARN"` } `yaml:"s3"` GCS struct { SourceURL string `yaml:"sourceURL"` } `yaml:"gcs"` AzureBlob struct { SourceURL string `yaml:"sourceURL"` SASToken string `yaml:"sasToken"` } `yaml:"azureBlob"` OSS ImportSampleDataOSSConfig `yaml:"oss"` } // ImportDataTableName declares one target table of an importData fixture and // the row count it must yield. ExpectedRows 0 means only assert at least one // row; a positive value asserts the exact count. type ImportDataTableName struct { Table string `yaml:"table"` ExpectedRows int64 `yaml:"expectedRows"` } // ImportDataDB declares the target database of an importData fixture and every // table it populates, so cases can assert the real imported contents. type ImportDataDB struct { Database string `yaml:"database"` Tables []ImportDataTableName `yaml:"tables"` } // ImportDataItem is one configured import source with an explicit dataFormat // (see DataFormatSQL/CSV/Parquet/AuroraSnapshot). Cases pick items by provider // and data format; importedDBTable declares the database and the tables the // fixture populates, with expected row counts, so cases can assert the real // imported rows. type ImportDataItem struct { DataFormat string `yaml:"dataFormat"` SourceURL string `yaml:"sourceURL"` RoleARN string `yaml:"roleARN"` Endpoint string `yaml:"endpoint"` AccessKeyID string `yaml:"accessKeyID"` AccessKeySecret string `yaml:"accessKeySecret"` SASToken string `yaml:"sasToken"` ImportedDBTable *ImportDataDB `yaml:"importedDBTable"` CustomCSVFormat *ImportSampleDataCSVFormatConfig `yaml:"customCSVFormat"` } // ImportDataConfig is the importData section: one list of items per provider // (same storage-type keys as importSampleData), so several data formats can // share one provider. New formats are added as extra list items. type ImportDataConfig struct { S3 []ImportDataItem `yaml:"s3"` GCS []ImportDataItem `yaml:"gcs"` AzureBlob []ImportDataItem `yaml:"azureBlob"` OSS []ImportDataItem `yaml:"oss"` } // Items returns the configured import data items for the given cloud provider. func (c ImportDataConfig) Items(provider CloudProviderType) []ImportDataItem { switch provider { case AWS: return c.S3 case GCP: return c.GCS case AZURE: return c.AzureBlob case AliCloud: return c.OSS default: return nil } } // Item returns the first import data item of the given data format for the // provider, or an error when it is not configured. func (c ImportDataConfig) Item(provider CloudProviderType, dataFormat string) (ImportDataItem, error) { for _, item := range c.Items(provider) { if item.DataFormat == dataFormat { return item, nil } } return ImportDataItem{}, fmt.Errorf("importData has no %s item for provider %s", dataFormat, provider) } // RequireItem resolves the configured import data item and validates that it is // complete enough to run an import: the source must be configured and declare // the target database and at least one table with a non-negative expected row // count for row validation. func (c ImportDataConfig) RequireItem(provider CloudProviderType, dataFormat string) (ImportDataItem, error) { item, err := c.Item(provider, dataFormat) if err != nil { return ImportDataItem{}, err } if strings.TrimSpace(item.SourceURL) == "" { return ImportDataItem{}, fmt.Errorf("importData %s item for provider %s has an empty sourceURL", dataFormat, provider) } if provider == AWS && strings.TrimSpace(item.RoleARN) == "" { return ImportDataItem{}, fmt.Errorf("importData %s item for provider %s has an empty roleARN", dataFormat, provider) } if item.ImportedDBTable == nil || strings.TrimSpace(item.ImportedDBTable.Database) == "" || len(item.ImportedDBTable.Tables) == 0 { return ImportDataItem{}, fmt.Errorf("importData %s item for provider %s must declare importedDBTable.database and at least one importedDBTable.tables entry for row validation", dataFormat, provider) } for _, table := range item.ImportedDBTable.Tables { if strings.TrimSpace(table.Table) == "" { return ImportDataItem{}, fmt.Errorf("importData %s item for provider %s has an importedDBTable.tables entry without a table name", dataFormat, provider) } if table.ExpectedRows < 0 { return ImportDataItem{}, fmt.Errorf("importData %s item for provider %s table %s has a negative expectedRows", dataFormat, provider, table.Table) } } return item, nil } // DataFormatLabel returns a PascalCase label for an import data format, used // to compose registry case names (e.g. AURORA_SNAPSHOT -> AuroraSnapshot). func DataFormatLabel(dataFormat string) string { switch dataFormat { case DataFormatAuroraSnapshot: return "AuroraSnapshot" case DataFormatParquet: return "Parquet" case DataFormatCSV: return "CSV" case DataFormatSQL: return "SQL" default: return dataFormat } } type BackupExportTargetConfig struct { TargetURI string `yaml:"targetURI"` Region string `yaml:"region"` RoleARN string `yaml:"roleARN"` Endpoint string `yaml:"endpoint"` } type BackupExportConfig struct { AWS BackupExportTargetConfig `yaml:"aws"` GCP BackupExportTargetConfig `yaml:"gcp"` AliCloud BackupExportTargetConfig `yaml:"alicloud"` } type cloudRegion []struct { CloudProvider CloudProviderType `yaml:"cloudProvider"` Regions []string `yaml:"regions"` } type SchemaRegistry struct { Username string `yaml:"username"` Password string `yaml:"password"` Endpoints string `yaml:"endpoints"` } func (t config) ENV() string { return t.env } // SelectRegionsForSuitePlan projects the selected plan's region matrix into // the runtime-only Regions field consumed by cases and dependency drivers. func (t *config) SelectRegionsForSuitePlan(suitePlan SuitePlan) { if !pie.Contains(AllSuitePlans, suitePlan) { panic(fmt.Sprintf("invalid suite plan %q, must be one of: %v", suitePlan, AllSuitePlans)) } regions, ok := t.SuitePlanRegions[suitePlan] if !ok { panic(fmt.Sprintf("regions not configured for suite plan %q", suitePlan)) } t.Regions = regions } func (t config) RegionsOf(cp CloudProviderType) []string { for _, v := range t.Regions { if v.CloudProvider == cp { return v.Regions } } return nil } func (t config) GetSchemaRegistry() *SchemaRegistry { return t.SchemaRegistry } func (t config) DefaultRegionOf(cp CloudProviderType) string { regions := t.RegionsOf(cp) if len(regions) == 0 { panic("can not find default region for provider " + cp) } return regions[0] } func (t config) SecondRegionOf(cp CloudProviderType) string { regions := t.RegionsOf(cp) if len(regions) < 2 { panic("can not find second region for provider " + cp) } return regions[1] } func (t config) AllRegions() []string { var regions []string for _, regs := range t.Regions { for _, it := range regs.Regions { regions = append(regions, it) } } return regions } func (t *config) Validate() { // set default domain if not specified if t.Domain == "" { t.Domain = DomainCDC } // validate domain if !pie.Contains(AllDomains, t.Domain) { panic(fmt.Sprintf("invalid domain '%s', must be one of: %v", t.Domain, AllDomains)) } if t.ApiMode == "" { t.ApiMode = ApiModeConsoleAPI } if !pie.Contains(AllApiModes, t.ApiMode) { panic(fmt.Sprintf("invalid api mode '%s', must be one of: %v", t.ApiMode, AllApiModes)) } validateCloudRegions(t.Regions) for suitePlan, regions := range t.SuitePlanRegions { if !pie.Contains(AllSuitePlans, suitePlan) { panic(fmt.Sprintf("invalid suite plan %q in suitePlanRegions, must be one of: %v", suitePlan, AllSuitePlans)) } validateCloudRegions(regions) } t.validateImportData() } func validateCloudRegions(regions cloudRegion) { providerMap := map[CloudProviderType]bool{} regionMap := map[string]bool{} for _, it := range regions { if _, ok := providerMap[it.CloudProvider]; ok { panic(fmt.Sprintf("duplicate CloudProvider(%s) in cloudRegions", it.CloudProvider)) } providerMap[it.CloudProvider] = true for _, v := range it.Regions { if _, ok := regionMap[v]; ok { panic(fmt.Sprintf("duplicate region (%s) in cloudRegions, may be in a same provider or different", v)) } regionMap[v] = true } } } // validateImportData panics on malformed importData items so misconfigured // fixtures fail at startup instead of deep inside a case run. func (t config) validateImportData() { for provider, items := range map[string][]ImportDataItem{ "s3": t.ImportData.S3, "gcs": t.ImportData.GCS, "azureBlob": t.ImportData.AzureBlob, "oss": t.ImportData.OSS, } { for _, item := range items { if strings.TrimSpace(item.DataFormat) == "" { panic(fmt.Sprintf("importData.%s contains an item without dataFormat", provider)) } if _, ok := legalDataFormats[item.DataFormat]; !ok { panic(fmt.Sprintf("importData.%s has unsupported dataFormat %q, must be one of SQL, CSV, PARQUET, AURORA_SNAPSHOT", provider, item.DataFormat)) } } } } // default override the fields except Kafka and TiDBClusters func combineCommonAndEnvConfigs(commonConfig, envConfig config) config { src := toGenericMapStruct(commonConfig) dst := toGenericMapStruct(envConfig) mergeFields := []string{"tidbDedicatedClusterIds", "kafka", "mysql", "schemaRegistry"} // handle merge fields for key, dstValue := range dst { if !pie.Contains(mergeFields, key) { continue } if isValueEmpty(dstValue) { dst[key] = src[key] continue } srcValue := src[key] switch v := dstValue.(type) { case []interface{}: if srcSlice, ok := srcValue.([]interface{}); ok { dst[key] = append(v, srcSlice...) } case map[string]interface{}: if srcMap, ok := srcValue.(map[string]interface{}); ok { for k, vv := range srcMap { dst[key].(map[string]interface{})[k] = vv } } } } // handle zero values in dst for key, dstValue := range dst { if isValueEmpty(dstValue) { dst[key] = src[key] } } // handle dst missing fields for key, srcValue := range src { if _, ok := dst[key]; !ok { dst[key] = srcValue } } combined := genericMapStruct2Config(dst) combined.ImportSampleData = mergeImportSampleDataWithFallback(commonConfig.ImportSampleData, combined.ImportSampleData) combined.ImportData = mergeImportDataWithFallback(commonConfig.ImportData, combined.ImportData) combined.BackupExport = mergeBackupExportWithFallback(commonConfig.BackupExport, combined.BackupExport) essentialExportCommon := commonConfig.EssentialExport if isEmptyBackupExport(essentialExportCommon) { essentialExportCommon = commonConfig.BackupExport } combined.EssentialExport = mergeBackupExportWithFallback(essentialExportCommon, combined.EssentialExport) if isEmptyBackupExport(combined.EssentialExport) { combined.EssentialExport = combined.BackupExport } return combined } func isEmptyBackupExport(value BackupExportConfig) bool { return value == (BackupExportConfig{}) } func mergeBackupExportWithFallback(common, merged BackupExportConfig) BackupExportConfig { if merged.AWS.TargetURI == "" { merged.AWS.TargetURI = common.AWS.TargetURI } if merged.AWS.Region == "" { merged.AWS.Region = common.AWS.Region } if merged.AWS.RoleARN == "" { merged.AWS.RoleARN = common.AWS.RoleARN } if merged.AWS.Endpoint == "" { merged.AWS.Endpoint = common.AWS.Endpoint } if merged.GCP.TargetURI == "" { merged.GCP.TargetURI = common.GCP.TargetURI } if merged.AliCloud.TargetURI == "" { merged.AliCloud.TargetURI = common.AliCloud.TargetURI } if merged.AliCloud.Region == "" { merged.AliCloud.Region = common.AliCloud.Region } if merged.AliCloud.RoleARN == "" { merged.AliCloud.RoleARN = common.AliCloud.RoleARN } if merged.AliCloud.Endpoint == "" { merged.AliCloud.Endpoint = common.AliCloud.Endpoint } return merged } // mergeImportDataWithFallback fills providers the environment config did not // override with the common importData items. An environment list, when present, // fully replaces the common list for that provider. func mergeImportDataWithFallback(common, merged ImportDataConfig) ImportDataConfig { if len(merged.S3) == 0 { merged.S3 = common.S3 } if len(merged.GCS) == 0 { merged.GCS = common.GCS } if len(merged.AzureBlob) == 0 { merged.AzureBlob = common.AzureBlob } if len(merged.OSS) == 0 { merged.OSS = common.OSS } return merged } func mergeImportSampleDataWithFallback(common, merged ImportSampleDataConfig) ImportSampleDataConfig { if merged.S3.SourceURL == "" { merged.S3.SourceURL = common.S3.SourceURL } if merged.S3.RoleARN == "" { merged.S3.RoleARN = common.S3.RoleARN } if merged.GCS.SourceURL == "" { merged.GCS.SourceURL = common.GCS.SourceURL } if merged.AzureBlob.SourceURL == "" { merged.AzureBlob.SourceURL = common.AzureBlob.SourceURL } if merged.AzureBlob.SASToken == "" { merged.AzureBlob.SASToken = common.AzureBlob.SASToken } if merged.OSS.SourceURL == "" { merged.OSS.SourceURL = common.OSS.SourceURL } if merged.OSS.Endpoint == "" { merged.OSS.Endpoint = common.OSS.Endpoint } if merged.OSS.AccessKeyID == "" { merged.OSS.AccessKeyID = common.OSS.AccessKeyID } if merged.OSS.AccessKeySecret == "" { merged.OSS.AccessKeySecret = common.OSS.AccessKeySecret } if merged.OSS.ImportedDBTable == nil { merged.OSS.ImportedDBTable = common.OSS.ImportedDBTable } if merged.OSS.ImportedDBTable != nil { if merged.OSS.ImportedDBTable.Database == "" { merged.OSS.ImportedDBTable.Database = common.OSS.ImportedDBTable.Database } if merged.OSS.ImportedDBTable.Table == "" { merged.OSS.ImportedDBTable.Table = common.OSS.ImportedDBTable.Table } } if merged.OSS.CustomCSVFormat == nil { merged.OSS.CustomCSVFormat = common.OSS.CustomCSVFormat } return merged } // applyImportSampleDataEnvOverrides overrides the import sample-data // credentials from the environment when set. This mirrors how the S3 and // Azure Blob dependency drivers accept credentials at runtime: the workflow // injects secret-backed env vars into the test process, and any non-empty // value takes precedence over the config. func applyImportSampleDataEnvOverrides(sampleData *ImportSampleDataConfig) { if token := os.Getenv("AZURE_SAS_TOKEN"); token != "" { sampleData.AzureBlob.SASToken = token } if ak := os.Getenv("OSS_ACCESS_KEY"); ak != "" { sampleData.OSS.AccessKeyID = ak } if sk := os.Getenv("OSS_SECRET_KEY"); sk != "" { sampleData.OSS.AccessKeySecret = sk } } func isValueEmpty(obj interface{}) bool { if obj == nil { return true } switch v := obj.(type) { case []interface{}: return len(v) == 0 case map[string]interface{}: if len(v) == 0 { return true } for _, vv := range v { if !isValueEmpty(vv) { return false } } return true case map[interface{}]interface{}: // yaml.v2 unmarshals nested maps as map[interface{}]interface{}, not map[string]interface{}. // Treat such a map as empty only when every leaf value is also empty. if len(v) == 0 { return true } for _, vv := range v { if !isValueEmpty(vv) { return false } } return true case bool: return false default: return reflect.DeepEqual(v, reflect.Zero(reflect.TypeOf(v)).Interface()) } } func genericMapStruct2Config(obj map[string]interface{}) config { data, err := yaml.Marshal(obj) if err != nil { panic(err) } var cfg config err = yaml.Unmarshal(data, &cfg) if err != nil { panic(err) } return cfg } func toGenericMapStruct(obj interface{}) map[string]interface{} { data, err := yaml.Marshal(obj) if err != nil { panic(err) } var m map[string]interface{} err = yaml.Unmarshal(data, &m) if err != nil { panic(err) } return m }