diff --git a/.env.example b/.env.example index d139d2d..50afc22 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,23 @@ SOURCE_DB_URL=sqlserver://sa:password@localhost:1433?database=master&packet+size=32767&loc=UTC&dial+timeout=120&connection+timeout=120&KeepAlive=30 + +# used only when SOURCE_DB_URL is not set +# SOURCE_DB_HOST=localhost +# SOURCE_DB_PORT=1433 +# SOURCE_DB_NAME=master +# SOURCE_DB_USER=sa +# SOURCE_DB_PWD=secure_password!123 +# SOURCE_DB_OPTIONS="packet+size=32767&loc=UTC&dial+timeout=120&connection+timeout=120&KeepAlive=30" + TARGET_DB_URL=postgresql://postgres:password@localhost:5432/db +# used only when TARGET_DB_URL is not set +# TARGET_DB_HOST=localhost +# TARGET_DB_PORT=5432 +# TARGET_DB_NAME=db +# TARGET_DB_USER=postgres +# TARGET_DB_PWD=secure_password!123 +# TARGET_DB_OPTIONS="" + LOG_LEVEL=INFO AZ_STORAGE_ENABLED=false diff --git a/cmd/go_migrate/dryrun.go b/cmd/go_migrate/dryrun.go new file mode 100644 index 0000000..e496318 --- /dev/null +++ b/cmd/go_migrate/dryrun.go @@ -0,0 +1,112 @@ +package main + +import ( + "context" + "fmt" + "sync" + + "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/azure" + "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config" + dbwrapper "git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db-wrapper" + log "github.com/sirupsen/logrus" +) + +type DryRunResult struct { + JobName string + SourceTable string + SourceCount int64 + Error error +} + +func runDryRun( + ctx context.Context, + azureClient *azure.Client, + sourceDb dbwrapper.DbWrapper, + jobs []config.Job, + maxParallelWorkers int, +) { + log.Info("=== DRY RUN ===") + log.Info("[DB] Source connection: OK") + log.Info("[DB] Target connection: OK") + + if azureClient != nil { + if err := azureClient.Ping(ctx); err != nil { + log.Errorf("[STORAGE] Azure: FAIL — %v", err) + } else { + log.Info("[STORAGE] Azure: OK") + } + } else { + log.Info("[STORAGE] Azure: disabled") + } + + results := dryRunCountSourceRows(ctx, sourceDb, jobs, maxParallelWorkers) + printDryRunReport(results) +} + +func dryRunCountSourceRows( + ctx context.Context, + sourceDb dbwrapper.DbWrapper, + jobs []config.Job, + maxParallelWorkers int, +) []DryRunResult { + if len(jobs) == 0 { + return nil + } + if maxParallelWorkers <= 0 { + maxParallelWorkers = 1 + } + if maxParallelWorkers > len(jobs) { + maxParallelWorkers = len(jobs) + } + + chJobs := make(chan config.Job, len(jobs)) + var mu sync.Mutex + var results []DryRunResult + var wg sync.WaitGroup + + for range maxParallelWorkers { + wg.Go(func() { + for job := range chJobs { + res := DryRunResult{ + JobName: job.Name, + SourceTable: fmt.Sprintf("[%s].[%s]", job.SourceTable.Schema, job.SourceTable.Table), + } + count, err := countSourceRows(ctx, sourceDb, job) + if err != nil { + res.Error = err + } else { + res.SourceCount = count + } + mu.Lock() + results = append(results, res) + mu.Unlock() + } + }) + } + + for _, job := range jobs { + chJobs <- job + } + close(chJobs) + wg.Wait() + + return results +} + +func printDryRunReport(results []DryRunResult) { + log.Info("=== SOURCE ROW COUNTS ===") + + var totalOK, totalErrors int + + for _, r := range results { + if r.Error != nil { + log.Errorf("[%s] %s — ERROR: %v", r.JobName, r.SourceTable, r.Error) + totalErrors++ + } else { + log.Infof("[%s] %s — rows: %d", r.JobName, r.SourceTable, r.SourceCount) + totalOK++ + } + } + + log.Infof("=== Dry run complete: %d OK, %d errors ===", totalOK, totalErrors) +} diff --git a/cmd/go_migrate/main.go b/cmd/go_migrate/main.go index fcd7679..c17e5f1 100644 --- a/cmd/go_migrate/main.go +++ b/cmd/go_migrate/main.go @@ -23,6 +23,7 @@ func main() { configPath := flag.String("config", "", "path to migration config file") validate := flag.Bool("validate", false, "count rows in source and target per job and compare") + dryRun := flag.Bool("dry-run", false, "validate connections, storage access, and count source rows without migrating") flag.Parse() if flag.NArg() > 1 { @@ -42,17 +43,24 @@ func main() { startTime := time.Now() + sourceDbUrl, err := config.App.ResolveSourceDbUrl(migrationConfig.SourceDbType) + if err != nil { + log.Fatalf("source DB config error: %v", err) + } + targetDbUrl, err := config.App.ResolveTargetDbUrl(migrationConfig.TargetDbType) + if err != nil { + log.Fatalf("target DB config error: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) defer cancel() - log.Info("=== Starting migration ===") - var wgConnect errgroup.Group var sourceDb, targetDb dbwrapper.DbWrapper wgConnect.Go(func() error { var err error - sourceDb, err = connectWithTimeout(ctx, migrationConfig.SourceDbType, config.App.SourceDbUrl, 20*time.Second) + sourceDb, err = connectWithTimeout(ctx, migrationConfig.SourceDbType, sourceDbUrl, 20*time.Second) if err != nil { return err } @@ -63,7 +71,7 @@ func main() { wgConnect.Go(func() error { var err error - targetDb, err = connectWithTimeout(ctx, migrationConfig.TargetDbType, config.App.TargetDbUrl, 20*time.Second) + targetDb, err = connectWithTimeout(ctx, migrationConfig.TargetDbType, targetDbUrl, 20*time.Second) if err != nil { return err } @@ -79,13 +87,29 @@ func main() { defer sourceDb.Close() defer targetDb.Close() + var azureClient *azure.Client + if config.App.AzureStorage.Enabled { + var err error + azureClient, err = azure.NewClient(config.App.AzureStorage) + if err != nil { + log.Fatalf("Failed to create Azure storage client: %v", err) + } + } + + if *dryRun { + runDryRun(ctx, azureClient, sourceDb, migrationConfig.Jobs, migrationConfig.MaxParallelWorkers) + return + } + if *validate { validationResults := validateJobs(ctx, sourceDb, targetDb, migrationConfig.Jobs, migrationConfig.MaxParallelWorkers) printValidationReport(validationResults) return } - results := processMigrationJobs(ctx, sourceDb, targetDb, migrationConfig.Jobs, migrationConfig.MaxParallelWorkers) + log.Info("=== Starting migration ===") + + results := processMigrationJobs(ctx, sourceDb, targetDb, azureClient, migrationConfig.Jobs, migrationConfig.MaxParallelWorkers) log.Info("=== RESUMEN DE MIGRACIÓN ===") var totalProcessed, totalErrors int64 @@ -116,6 +140,7 @@ func processMigrationJobs( ctx context.Context, sourceDb dbwrapper.DbWrapper, targetDb dbwrapper.DbWrapper, + azureClient *azure.Client, jobs []config.Job, maxParallelWorkers int, ) []models.JobResult { @@ -143,15 +168,6 @@ func processMigrationJobs( extractor := extractors.NewExtractor(sourceDb) loader := loaders.NewGenericLoader(targetDb) - var azureClient *azure.Client - if config.App.AzureStorage.Enabled { - var err error - azureClient, err = azure.NewClient(config.App.AzureStorage) - if err != nil { - log.Fatalf("Failed to create Azure storage client: %v", err) - } - } - for i := range maxParallelWorkers { wgJobs.Go(func() { for job := range chJobs { diff --git a/internal/app/azure/main.go b/internal/app/azure/main.go index 6b32a84..96f7041 100644 --- a/internal/app/azure/main.go +++ b/internal/app/azure/main.go @@ -70,6 +70,15 @@ func (c *Client) UploadBuffer(ctx context.Context, containerName, blobPath strin return nil } +func (c *Client) Ping(ctx context.Context) error { + pager := c.client.NewListBlobsFlatPager(c.azureStorageConfig.Container, nil) + _, err := pager.NextPage(ctx) + if err != nil { + return fmt.Errorf("storage access check failed: %w", err) + } + return nil +} + func (c *Client) UploadAndGetURL(ctx context.Context, blobPath string, buffer []byte) (string, error) { if blobPath == "" || buffer == nil { return "", ErrInvalidInput diff --git a/internal/app/config/main.go b/internal/app/config/main.go index 5e5e6db..2270a5e 100644 --- a/internal/app/config/main.go +++ b/internal/app/config/main.go @@ -1,6 +1,10 @@ package config import ( + "fmt" + "maps" + "net/url" + "github.com/ilyakaznacheev/cleanenv" log "github.com/sirupsen/logrus" ) @@ -16,10 +20,95 @@ type AzureStorageConfig struct { } type appConfig struct { - SourceDbUrl string `env:"SOURCE_DB_URL" env-required:"true"` - TargetDbUrl string `env:"TARGET_DB_URL" env-required:"true"` - LogLevel string `env:"LOG_LEVEL" env-default:"INFO"` - AzureStorage AzureStorageConfig + SourceDbUrl string `env:"SOURCE_DB_URL"` + SourceDbHost string `env:"SOURCE_DB_HOST"` + SourceDbPort string `env:"SOURCE_DB_PORT"` + SourceDbName string `env:"SOURCE_DB_NAME"` + SourceDbUser string `env:"SOURCE_DB_USER"` + SourceDbPwd string `env:"SOURCE_DB_PWD"` + SourceDbOptions string `env:"SOURCE_DB_OPTIONS"` + TargetDbUrl string `env:"TARGET_DB_URL"` + TargetDbHost string `env:"TARGET_DB_HOST"` + TargetDbPort string `env:"TARGET_DB_PORT"` + TargetDbName string `env:"TARGET_DB_NAME"` + TargetDbUser string `env:"TARGET_DB_USER"` + TargetDbPwd string `env:"TARGET_DB_PWD"` + TargetDbOptions string `env:"TARGET_DB_OPTIONS"` + LogLevel string `env:"LOG_LEVEL" env-default:"INFO"` + AzureStorage AzureStorageConfig +} + +func (c *appConfig) ResolveSourceDbUrl(dbType string) (string, error) { + if c.SourceDbUrl != "" { + return c.SourceDbUrl, nil + } + u, err := buildDbUrl(dbType, c.SourceDbHost, c.SourceDbPort, c.SourceDbName, c.SourceDbUser, c.SourceDbPwd, c.SourceDbOptions) + if err != nil { + return "", fmt.Errorf("source DB: %w", err) + } + return u, nil +} + +func (c *appConfig) ResolveTargetDbUrl(dbType string) (string, error) { + if c.TargetDbUrl != "" { + return c.TargetDbUrl, nil + } + u, err := buildDbUrl(dbType, c.TargetDbHost, c.TargetDbPort, c.TargetDbName, c.TargetDbUser, c.TargetDbPwd, c.TargetDbOptions) + if err != nil { + return "", fmt.Errorf("target DB: %w", err) + } + return u, nil +} + +func buildDbUrl(dbType, host, port, name, user, pwd, options string) (string, error) { + if host == "" { + return "", fmt.Errorf("DB_HOST is required when DB_URL is not set") + } + if name == "" { + return "", fmt.Errorf("DB_NAME is required when DB_URL is not set") + } + if user == "" { + return "", fmt.Errorf("DB_USER is required when DB_URL is not set") + } + + switch dbType { + case "sqlserver": + if port == "" { + port = "1433" + } + q := url.Values{} + if options != "" { + extra, err := url.ParseQuery(options) + if err != nil { + return "", fmt.Errorf("invalid DB_OPTIONS: %w", err) + } + maps.Copy(q, extra) + } + q.Set("database", name) + u := &url.URL{ + Scheme: "sqlserver", + Host: host + ":" + port, + User: url.UserPassword(user, pwd), + RawQuery: q.Encode(), + } + return u.String(), nil + + case "postgres": + if port == "" { + port = "5432" + } + u := &url.URL{ + Scheme: "postgres", + Host: host + ":" + port, + User: url.UserPassword(user, pwd), + Path: "/" + name, + RawQuery: options, + } + return u.String(), nil + + default: + return "", fmt.Errorf("unknown db type %q — cannot build URL from individual components", dbType) + } } func getAppConfig() appConfig {