feat: add dry run and individual db config parameters
Add a dry-run mode that validates connections and counts source rows without writing, and accept per-connection host/port/user/password/options alongside URLs.
This commit is contained in:
17
.env.example
17
.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
|
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
|
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
|
LOG_LEVEL=INFO
|
||||||
|
|
||||||
AZ_STORAGE_ENABLED=false
|
AZ_STORAGE_ENABLED=false
|
||||||
|
|||||||
112
cmd/go_migrate/dryrun.go
Normal file
112
cmd/go_migrate/dryrun.go
Normal file
@@ -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)
|
||||||
|
}
|
||||||
@@ -23,6 +23,7 @@ func main() {
|
|||||||
|
|
||||||
configPath := flag.String("config", "", "path to migration config file")
|
configPath := flag.String("config", "", "path to migration config file")
|
||||||
validate := flag.Bool("validate", false, "count rows in source and target per job and compare")
|
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()
|
flag.Parse()
|
||||||
|
|
||||||
if flag.NArg() > 1 {
|
if flag.NArg() > 1 {
|
||||||
@@ -42,17 +43,24 @@ func main() {
|
|||||||
|
|
||||||
startTime := time.Now()
|
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())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
log.Info("=== Starting migration ===")
|
|
||||||
|
|
||||||
var wgConnect errgroup.Group
|
var wgConnect errgroup.Group
|
||||||
var sourceDb, targetDb dbwrapper.DbWrapper
|
var sourceDb, targetDb dbwrapper.DbWrapper
|
||||||
|
|
||||||
wgConnect.Go(func() error {
|
wgConnect.Go(func() error {
|
||||||
var err 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 {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -63,7 +71,7 @@ func main() {
|
|||||||
|
|
||||||
wgConnect.Go(func() error {
|
wgConnect.Go(func() error {
|
||||||
var err 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 {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -79,13 +87,29 @@ func main() {
|
|||||||
defer sourceDb.Close()
|
defer sourceDb.Close()
|
||||||
defer targetDb.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 {
|
if *validate {
|
||||||
validationResults := validateJobs(ctx, sourceDb, targetDb, migrationConfig.Jobs, migrationConfig.MaxParallelWorkers)
|
validationResults := validateJobs(ctx, sourceDb, targetDb, migrationConfig.Jobs, migrationConfig.MaxParallelWorkers)
|
||||||
printValidationReport(validationResults)
|
printValidationReport(validationResults)
|
||||||
return
|
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 ===")
|
log.Info("=== RESUMEN DE MIGRACIÓN ===")
|
||||||
var totalProcessed, totalErrors int64
|
var totalProcessed, totalErrors int64
|
||||||
@@ -116,6 +140,7 @@ func processMigrationJobs(
|
|||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
sourceDb dbwrapper.DbWrapper,
|
sourceDb dbwrapper.DbWrapper,
|
||||||
targetDb dbwrapper.DbWrapper,
|
targetDb dbwrapper.DbWrapper,
|
||||||
|
azureClient *azure.Client,
|
||||||
jobs []config.Job,
|
jobs []config.Job,
|
||||||
maxParallelWorkers int,
|
maxParallelWorkers int,
|
||||||
) []models.JobResult {
|
) []models.JobResult {
|
||||||
@@ -143,15 +168,6 @@ func processMigrationJobs(
|
|||||||
extractor := extractors.NewExtractor(sourceDb)
|
extractor := extractors.NewExtractor(sourceDb)
|
||||||
loader := loaders.NewGenericLoader(targetDb)
|
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 {
|
for i := range maxParallelWorkers {
|
||||||
wgJobs.Go(func() {
|
wgJobs.Go(func() {
|
||||||
for job := range chJobs {
|
for job := range chJobs {
|
||||||
|
|||||||
@@ -70,6 +70,15 @@ func (c *Client) UploadBuffer(ctx context.Context, containerName, blobPath strin
|
|||||||
return nil
|
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) {
|
func (c *Client) UploadAndGetURL(ctx context.Context, blobPath string, buffer []byte) (string, error) {
|
||||||
if blobPath == "" || buffer == nil {
|
if blobPath == "" || buffer == nil {
|
||||||
return "", ErrInvalidInput
|
return "", ErrInvalidInput
|
||||||
|
|||||||
@@ -1,6 +1,10 @@
|
|||||||
package config
|
package config
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"fmt"
|
||||||
|
"maps"
|
||||||
|
"net/url"
|
||||||
|
|
||||||
"github.com/ilyakaznacheev/cleanenv"
|
"github.com/ilyakaznacheev/cleanenv"
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
)
|
)
|
||||||
@@ -16,10 +20,95 @@ type AzureStorageConfig struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type appConfig struct {
|
type appConfig struct {
|
||||||
SourceDbUrl string `env:"SOURCE_DB_URL" env-required:"true"`
|
SourceDbUrl string `env:"SOURCE_DB_URL"`
|
||||||
TargetDbUrl string `env:"TARGET_DB_URL" env-required:"true"`
|
SourceDbHost string `env:"SOURCE_DB_HOST"`
|
||||||
LogLevel string `env:"LOG_LEVEL" env-default:"INFO"`
|
SourceDbPort string `env:"SOURCE_DB_PORT"`
|
||||||
AzureStorage AzureStorageConfig
|
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 {
|
func getAppConfig() appConfig {
|
||||||
|
|||||||
Reference in New Issue
Block a user