Repository: ariga/atlas
Pull request #1110 (2022-08-31 07:27:09)

sql/migrate: set operator version from CLI on migrate apply
Integration tests will follow once the status command has been updated


Patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 740ea65e..d296349b 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -456,7 +456,11 @@ func (tx *tx) commit() error {
 }
 
 func executorOptions(l migrate.Logger) []migrate.ExecutorOption {
-	opts := []migrate.ExecutorOption{migrate.WithLogger(l)}
+	v, _ := parse(version)
+	opts := []migrate.ExecutorOption{
+		migrate.WithLogger(l),
+		migrate.WithOperatorVersion("Atlas CLI - " + v),
+	}
 	if MigrateFlags.Apply.AllowDirty {
 		opts = append(opts, migrate.WithAllowDirty(true))
 	}
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index 9569c5e7..3d279490 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -215,6 +215,7 @@ type (
 		fromVer     string             // Calculate pending files from the given version (including it).
 		baselineVer string             // Start the first migration after the given baseline version.
 		allowDirty  bool               // Allow start working on a non-clean database.
+		operator    string             // Revision.OperatorVersion
 	}
 
 	// ExecutorOption allows configuring an Executor using functional arguments.
@@ -390,7 +391,7 @@ var (
 	ErrRevisionNotExist = errors.New("sql/migrate: revision not found")
 )
 
-// NewExecutor creates a new Executor with default values. // TODO(masseelch): Operator Version and other Meta
+// NewExecutor creates a new Executor with default values.
 func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOption) (*Executor, error) {
 	if drv == nil {
 		return nil, errors.New("sql/migrate: execute: no driver given")
@@ -422,6 +423,24 @@ func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOp
 	return ex, nil
 }
 
+// WithAllowDirty defines if we can start working on a non-clean database
+// in the first migration execution.
+func WithAllowDirty(b bool) ExecutorOption {
+	return func(ex *Executor) error {
+		ex.allowDirty = b
+		return nil
+	}
+}
+
+// WithBaselineVersion allows setting the baseline version of the database on the
+// first migration. Hence, all versions up to and including this version are skipped.
+func WithBaselineVersion(v string) ExecutorOption {
+	return func(ex *Executor) error {
+		ex.baselineVer = v
+		return nil
+	}
+}
+
 // WithLogger sets the Logger of an Executor.
 func WithLogger(log Logger) ExecutorOption {
 	return func(ex *Executor) error {
@@ -439,20 +458,11 @@ func WithFromVersion(v string) ExecutorOption {
 	}
 }
 
-// WithBaselineVersion allows setting the baseline version of the database on the
-// first migration. Hence, all versions up to and including this version are skipped.
-func WithBaselineVersion(v string) ExecutorOption {
+// WithOperatorVersion sets the operator version to save on the revisions
+// when executing migration files.
+func WithOperatorVersion(v string) ExecutorOption {
 	return func(ex *Executor) error {
-		ex.baselineVer = v
-		return nil
-	}
-}
-
-// WithAllowDirty defines if we can start working on a non-clean database
-// in the first migration execution.
-func WithAllowDirty(b bool) ExecutorOption {
-	return func(ex *Executor) error {
-		ex.allowDirty = b
+		ex.operator = v
 		return nil
 	}
 }
@@ -498,7 +508,7 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 			}
 			f := migrations[baseline]
 			// Mark the revision in the database as baseline revision.
-			if err := writeRevision(ctx, e.rrw, &Revision{Version: f.Version(), Description: f.Desc(), Type: RevisionTypeBaseline}); err != nil {
+			if err := e.writeRevision(ctx, &Revision{Version: f.Version(), Description: f.Desc(), Type: RevisionTypeBaseline}); err != nil {
 				return nil, err
 			}
 			pending = migrations[baseline+1:]
@@ -576,15 +586,15 @@ func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 		}
 	}
 	// Save once to mark as started in the database.
-	if err = writeRevision(ctx, e.rrw, r); err != nil {
+	if err = e.writeRevision(ctx, r); err != nil {
 		return err
 	}
 	// Make sure to store the Revision information.
-	defer func(ctx context.Context, rrw RevisionReadWriter, r *Revision) {
-		if err2 := writeRevision(ctx, rrw, r); err2 != nil {
+	defer func(ctx context.Context, e *Executor, r *Revision) {
+		if err2 := e.writeRevision(ctx, r); err2 != nil {
 			err = wrap(err2, err)
 		}
-	}(ctx, e.rrw, r)
+	}(ctx, e, r)
 	if r.Applied > 0 {
 		// If the file has been applied partially before, check if the
 		// applied statements have not changed.
@@ -606,7 +616,7 @@ func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 		}
 		r.PartialHashes = append(r.PartialHashes, "h1:"+sums[r.Applied])
 		r.Applied++
-		if err = writeRevision(ctx, e.rrw, r); err != nil {
+		if err = e.writeRevision(ctx, r); err != nil {
 			return err
 		}
 	}
@@ -614,9 +624,10 @@ func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 	return
 }
 
-func writeRevision(ctx context.Context, w RevisionReadWriter, r *Revision) error {
+func (e *Executor) writeRevision(ctx context.Context, r *Revision) error {
 	r.ExecutedAt = time.Now()
-	if err := w.WriteRevision(ctx, r); err != nil {
+	r.OperatorVersion = e.operator
+	if err := e.rrw.WriteRevision(ctx, r); err != nil {
 		return fmt.Errorf("sql/migrate: execute: write revision: %w", err)
 	}
 	return nil
@@ -741,25 +752,11 @@ func (NopRevisionReadWriter) WriteRevision(context.Context, *Revision) error {
 
 var _ RevisionReadWriter = (*NopRevisionReadWriter)(nil)
 
-// Complete returns if a revision has been applied without errors and in full.
-func (r *Revision) Complete() error {
-	if r.Error != "" {
-		return errors.New(r.Error)
-	}
-	return nil
-}
-
 // done computes and sets the ExecutionTime.
 func (r *Revision) done() {
 	r.ExecutionTime = time.Now().Sub(r.ExecutedAt)
 }
 
-func (r *Revision) setGoErr(err error) error {
-	r.done()
-	r.Error = fmt.Sprintf("Go:\n%s", err)
-	return err
-}
-
 func (r *Revision) setSQLErr(stmt string, err error) {
 	r.done()
 	r.Error = fmt.Sprintf("Statement:\n%s\n\nError:\n%s", stmt, err)


Test patch:
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index 9cfef6b2..83ada4bd 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -217,7 +217,7 @@ func TestExecutor(t *testing.T) {
 	dir, err = migrate.NewLocalDir(t.TempDir())
 	require.NoError(t, err)
 	require.NoError(t, dir.WriteFile("atlas.sum", hash))
-	ex, err = migrate.NewExecutor(&mockDriver{}, dir, &mockRevisionReadWriter{})
+	ex, err = migrate.NewExecutor(&mockDriver{}, dir, &mockRevisionReadWriter{}, migrate.WithOperatorVersion("op"))
 	require.NoError(t, err)
 	require.NotNil(t, ex)
 	require.ErrorIs(t, ex.ExecuteN(context.Background(), 0), migrate.ErrChecksumMismatch)
@@ -228,25 +228,27 @@ func TestExecutor(t *testing.T) {
 		rrw  = &mockRevisionReadWriter{}
 		log  = &mockLogger{}
 		rev1 = &migrate.Revision{
-			Version:     "1.a",
-			Description: "sub.up",
-			Type:        migrate.RevisionTypeExecute,
-			Applied:     2,
-			Total:       2,
-			Hash:        "nXyZR020M/mH7LxkoTkJr7BcQkipVg90imQ9I4595dw=",
+			Version:         "1.a",
+			Description:     "sub.up",
+			Type:            migrate.RevisionTypeExecute,
+			Applied:         2,
+			Total:           2,
+			Hash:            "nXyZR020M/mH7LxkoTkJr7BcQkipVg90imQ9I4595dw=",
+			OperatorVersion: "op",
 		}
 		rev2 = &migrate.Revision{
-			Version:     "2.10.x-20",
-			Description: "description",
-			Type:        migrate.RevisionTypeExecute,
-			Applied:     1,
-			Total:       1,
-			Hash:        "wQB3Vh3PHVXQg9OD3Gn7TBxbZN3r1Qb7TtAE1g3q9mQ=",
+			Version:         "2.10.x-20",
+			Description:     "description",
+			Type:            migrate.RevisionTypeExecute,
+			Applied:         1,
+			Total:           1,
+			Hash:            "wQB3Vh3PHVXQg9OD3Gn7TBxbZN3r1Qb7TtAE1g3q9mQ=",
+			OperatorVersion: "op",
 		}
 	)
 	dir, err = migrate.NewLocalDir(filepath.Join("testdata/migrate", "sub"))
 	require.NoError(t, err)
-	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log))
+	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log), migrate.WithOperatorVersion("op"))
 	require.NoError(t, err)
 
 	// Applies two of them.
@@ -313,12 +315,13 @@ func TestExecutor(t *testing.T) {
 	revs, err := rrw.ReadRevisions(context.Background())
 	require.NoError(t, err)
 	requireEqualRevision(t, &migrate.Revision{
-		Version:     "3",
-		Description: "partly",
-		Type:        migrate.RevisionTypeExecute,
-		Applied:     1,
-		Total:       2,
-		Error:       "Statement:\nALTER TABLE t_sub ADD c4 int;\n\nError:\nthis is an error",
+		Version:         "3",
+		Description:     "partly",
+		Type:            migrate.RevisionTypeExecute,
+		Applied:         1,
+		Total:           2,
+		Error:           "Statement:\nALTER TABLE t_sub ADD c4 int;\n\nError:\nthis is an error",
+		OperatorVersion: "op",
 	}, revs[len(revs)-1])
 
 	// Will fail if applied contents hash has changed (like when editing a partially applied file to fix an error).