-
Notifications
You must be signed in to change notification settings - Fork 27
Add move checks #560
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Add move checks #560
Changes from all commits
Commits
Show all changes
4 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| // Package check provides various configuration and health checks | ||
| // that can be run for move operations. | ||
| package check | ||
|
|
||
| import ( | ||
| "context" | ||
| "database/sql" | ||
| "log/slog" | ||
| "sync" | ||
|
|
||
| "github.com/block/spirit/pkg/applier" | ||
| "github.com/block/spirit/pkg/table" | ||
| "github.com/go-sql-driver/mysql" | ||
| ) | ||
|
|
||
| // ScopeFlag scopes a check | ||
| type ScopeFlag uint8 | ||
|
|
||
| const ( | ||
| ScopeNone ScopeFlag = iota | ||
| ScopePreRun | ||
| ScopePreflight | ||
| ScopePostSetup | ||
| ScopeResume | ||
| ) | ||
|
|
||
| // Resources contains the resources needed for move checks | ||
| type Resources struct { | ||
| SourceDB *sql.DB | ||
| SourceConfig *mysql.Config | ||
| Targets []applier.Target | ||
| SourceTables []*table.TableInfo | ||
| CreateSentinel bool | ||
| // For PreRun checks (before DB connections established) | ||
| SourceDSN string | ||
| TargetDSN string | ||
| } | ||
|
|
||
| type check struct { | ||
| callback func(context.Context, Resources, *slog.Logger) error | ||
| scope ScopeFlag | ||
| } | ||
|
|
||
| var ( | ||
| checks map[string]check | ||
| lock sync.Mutex | ||
| ) | ||
|
|
||
| // registerCheck registers a check (callback func) and a scope (aka time) that it is expected to be run | ||
| func registerCheck(name string, callback func(context.Context, Resources, *slog.Logger) error, scope ScopeFlag) { | ||
| lock.Lock() | ||
| defer lock.Unlock() | ||
| if checks == nil { | ||
| checks = make(map[string]check) | ||
| } | ||
| checks[name] = check{callback: callback, scope: scope} | ||
| } | ||
|
|
||
| // RunChecks runs all checks that are registered for the given scope | ||
| func RunChecks(ctx context.Context, r Resources, logger *slog.Logger, scope ScopeFlag) error { | ||
| for _, check := range checks { | ||
| if check.scope&scope == 0 { | ||
| continue | ||
| } | ||
| err := check.callback(ctx, r, logger) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } | ||
| return nil | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| package check | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "log/slog" | ||
| ) | ||
|
|
||
| func init() { | ||
| registerCheck("configuration", configurationCheck, ScopePreflight) | ||
| } | ||
|
|
||
| // configurationCheck verifies the MySQL configuration on the source database | ||
| // is suitable for move operations. Move operations require: | ||
| // - ROW binlog format for reading changes | ||
| // - Binary logging enabled | ||
| // - log_slave_updates enabled (unless we verify no replication) | ||
| // - FULL binlog_row_image (for buffered copy which is always used in move) | ||
| func configurationCheck(ctx context.Context, r Resources, logger *slog.Logger) error { | ||
| var binlogFormat, binlogRowImage, logBin, logSlaveUpdates, binlogRowValueOptions string | ||
| err := r.SourceDB.QueryRowContext(ctx, | ||
| `SELECT @@global.binlog_format, | ||
| @@global.binlog_row_image, | ||
| @@global.log_bin, | ||
| @@global.log_slave_updates, | ||
| @@global.binlog_row_value_options`).Scan( | ||
| &binlogFormat, | ||
| &binlogRowImage, | ||
| &logBin, | ||
| &logSlaveUpdates, | ||
| &binlogRowValueOptions, | ||
| ) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if binlogFormat != "ROW" { | ||
| return errors.New("binlog_format must be ROW") | ||
| } | ||
|
|
||
| // Move always uses buffered copy, which requires FULL binlog_row_image | ||
| if binlogRowImage != "FULL" { | ||
| return errors.New("binlog_row_image must be FULL for move operations (buffered copy requires reading all columns from binlog)") | ||
| } | ||
|
|
||
| if binlogRowValueOptions != "" { | ||
| return errors.New("binlog_row_value_options must be empty for move operations (buffered copy requires reading all columns from binlog)") | ||
| } | ||
|
|
||
| if logBin != "1" { | ||
| return errors.New("log_bin must be enabled") | ||
| } | ||
|
|
||
| if logSlaveUpdates != "1" { | ||
| // This is a hard requirement unless we enhance this to confirm | ||
| // it's not receiving any updates via the replication stream. | ||
| return errors.New("log_slave_updates must be enabled") | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| package check | ||
|
|
||
| import ( | ||
| "context" | ||
| "database/sql" | ||
| "log/slog" | ||
| "testing" | ||
|
|
||
| "github.com/block/spirit/pkg/testutils" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestConfigurationCheck(t *testing.T) { | ||
| if testutils.IsMinimalRBRTestRunner(t) { | ||
| t.Skip("Skipping test for minimal RBR test runner") | ||
| } | ||
|
|
||
| db, err := sql.Open("mysql", testutils.DSN()) | ||
| assert.NoError(t, err) | ||
| defer db.Close() | ||
|
|
||
| // Test with valid configuration | ||
| r := Resources{ | ||
| SourceDB: db, | ||
| } | ||
| err = configurationCheck(context.Background(), r, slog.Default()) | ||
| assert.NoError(t, err) | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| package check | ||
|
|
||
| import ( | ||
| "context" | ||
| "database/sql" | ||
| "errors" | ||
| "fmt" | ||
| "log/slog" | ||
| "slices" | ||
|
|
||
| "github.com/block/spirit/pkg/table" | ||
| ) | ||
|
|
||
| func init() { | ||
| registerCheck("resume_state", resumeStateCheck, ScopeResume) | ||
| } | ||
|
|
||
| // resumeStateCheck validates that the state is compatible with resuming from a checkpoint. | ||
| // This check runs at ScopeResume and verifies: | ||
| // 1. Checkpoint table exists on source | ||
| // 2. Target tables exist with matching schema to source | ||
| // 3. All source tables have corresponding target tables | ||
| // | ||
| // This check is the complement to target_state check: | ||
| // - target_state validates we can start a NEW copy (empty or matching empty tables) | ||
| // - resume_state validates we can RESUME from checkpoint (tables exist with data) | ||
| func resumeStateCheck(ctx context.Context, r Resources, logger *slog.Logger) error { | ||
| if len(r.SourceTables) == 0 { | ||
| return errors.New("no source tables available for resume validation") | ||
| } | ||
|
|
||
| // Check 1: Verify checkpoint table exists on source | ||
| checkpointTableName := "_spirit_checkpoint" | ||
| var checkpointExists int | ||
| err := r.SourceDB.QueryRowContext(ctx, | ||
| "SELECT 1 FROM information_schema.TABLES WHERE table_schema = ? AND table_name = ?", | ||
| r.SourceConfig.DBName, checkpointTableName).Scan(&checkpointExists) | ||
| if err == sql.ErrNoRows { | ||
| return fmt.Errorf("checkpoint table '%s.%s' does not exist; cannot resume", r.SourceConfig.DBName, checkpointTableName) | ||
| } | ||
| if err != nil { | ||
| return fmt.Errorf("failed to check for checkpoint table: %w", err) | ||
| } | ||
|
|
||
| logger.Info("checkpoint table exists, validating target tables for resume", | ||
| "checkpoint_table", fmt.Sprintf("%s.%s", r.SourceConfig.DBName, checkpointTableName)) | ||
|
|
||
| // Check 2: Verify all source tables have corresponding target tables with matching schema | ||
| for _, sourceTable := range r.SourceTables { | ||
| for i, target := range r.Targets { | ||
| // Check if table exists on target | ||
| var tableExists int | ||
| err := target.DB.QueryRowContext(ctx, | ||
| "SELECT 1 FROM information_schema.TABLES WHERE table_schema = ? AND table_name = ?", | ||
| target.Config.DBName, sourceTable.TableName).Scan(&tableExists) | ||
| if err == sql.ErrNoRows { | ||
| return fmt.Errorf("table '%s' does not exist on target %d (%s); cannot resume. Target state is incomplete", | ||
| sourceTable.TableName, i, target.Config.DBName) | ||
| } | ||
| if err != nil { | ||
| return fmt.Errorf("failed to check for table '%s' on target %d: %w", sourceTable.TableName, i, err) | ||
| } | ||
|
|
||
| // Verify schema matches | ||
| targetTable := table.NewTableInfo(target.DB, target.Config.DBName, sourceTable.TableName) | ||
| if err := targetTable.SetInfo(ctx); err != nil { | ||
| return fmt.Errorf("failed to get table info for target %d table '%s': %w", i, sourceTable.TableName, err) | ||
| } | ||
|
|
||
| if !slices.Equal(sourceTable.Columns, targetTable.Columns) { | ||
| return fmt.Errorf("table '%s' schema mismatch between source and target %d (%s); cannot resume safely. Schema may have changed since checkpoint was created", | ||
| sourceTable.TableName, i, target.Config.DBName) | ||
| } | ||
|
|
||
| logger.Debug("validated target table for resume", | ||
| "table", sourceTable.TableName, | ||
| "target", i, | ||
| "database", target.Config.DBName, | ||
| "columns", len(targetTable.Columns)) | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does anything in the checkpoint table itself need to be checked at this stage?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I could, but restoring the checkpoint is also able to return an error in
pkg/move/runner.go:resumeFromCheckpoint()My preference is not to because if I change the checkpoint format (it happens) the code is non-dry. This is because the
checkshould only check and is not expected to load or mutate structures.If I move checkpoints to a separate package, that would be a different argument though; they could call the same helper.