Repository: ariga/atlas
Pull request #793 (2022-05-13 09:57:26)

sql/migrate: introduce basic execution logging
Package consumers can then use the logs for whatever they need it to be (JSON / pretty TTY / Sentry / ...)

~Will be rebased against master once the base (#781) is merged.~


Patch:
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index 0b1e41d6..aaa295de 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -226,6 +226,7 @@ type (
 		drv Driver             // The Driver to access and manage the database.
 		dir Dir                // The Dir with migration files to use.
 		rrw RevisionReadWriter // The RevisionReadWriter to read and write database revisions to.
+		log Logger             // The Logger to use.
 	}
 
 	// ExecutorOption allows configuring an Executor using functional arguments.
@@ -354,6 +355,14 @@ func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOp
 	return p, nil
 }
 
+// WithLogger sets the Logger of an Executor.
+func WithLogger(log Logger) ExecutorOption {
+	return func(ex *Executor) error {
+		ex.log = log
+		return nil
+	}
+}
+
 // Execute executes n missing migration files on the database. Passing in n <= 0 means running all pending migrations.
 func (e *Executor) Execute(ctx context.Context, n int) (err error) {
 	unlock, err := e.drv.(schema.Locker).Lock(ctx, "atlas_migration_execute", 0)
@@ -401,7 +410,7 @@ func (e *Executor) Execute(ctx context.Context, n int) (err error) {
 			return fmt.Errorf("sql/migrate: execute: scan description from %q: %w", m.Name(), err)
 		}
 		r := revisions[i]
-		if v != r.Version || d != r.Description || hf[i].H != r.Hash { // TODO(masseelch): version / desc check necessary?
+		if v != r.Version || d != r.Description || hf[i].H != r.Hash {
 			return fmt.Errorf("sql/migrate: execute: revisions and migrations mismatch: rev %q <> file %q", v, r.Version)
 		}
 	}
@@ -430,10 +439,30 @@ func (e *Executor) Execute(ctx context.Context, n int) (err error) {
 		}
 		pending = migrations[len(revisions) : len(revisions)+n]
 	}
+	if e.log != nil {
+		names := make([]string, len(pending))
+		for i := range pending {
+			names[i] = pending[i].Name()
+		}
+		e.log.Log(LogExecution{names})
+	}
 	// TODO(masseelch): run in a transaction
 	for _, m := range pending {
 		r := &Revision{ExecutedAt: time.Now(), ExecutionState: stateOngoing}
 		revisions = append(revisions, r)
+		v, err := sc.Version(m)
+		if err != nil {
+			return r.setGoErr(fmt.Errorf("sql/migrate: execute: scan version from %q: %w", m.Name(), err))
+		}
+		r.Version = v
+		d, err := sc.Desc(m)
+		if err != nil {
+			return r.setGoErr(fmt.Errorf("sql/migrate: execute: scan description from %q: %w", m.Name(), err))
+		}
+		r.Description = d
+		if e.log != nil {
+			e.log.Log(LogFile{v, d})
+		}
 		h, err := hf.sumByName(m.Name())
 		if err != nil {
 			return r.setGoErr(fmt.Errorf("sql/migrate: execute: scanning checksum for file %q: %w", m.Name(), err))
@@ -444,16 +473,9 @@ func (e *Executor) Execute(ctx context.Context, n int) (err error) {
 			return r.setGoErr(fmt.Errorf("sql/migrate: execute: scanning statements from file %q: %w", m.Name(), err))
 		}
 		for _, stmt := range stmts {
-			v, err := sc.Version(m)
-			if err != nil {
-				return r.setGoErr(fmt.Errorf("sql/migrate: execute: scan version from %q: %w", m.Name(), err))
-			}
-			r.Version = v
-			d, err := sc.Desc(m)
-			if err != nil {
-				return r.setGoErr(fmt.Errorf("sql/migrate: execute: scan description from %q: %w", m.Name(), err))
+			if e.log != nil {
+				e.log.Log(LogStmt{stmt})
 			}
-			r.Description = d
 			if _, err := e.drv.ExecContext(ctx, stmt); err != nil {
 				return r.setSQLErr(
 					fmt.Errorf("sql/migrate: execute: executing statement %q from version %q: %w", stmt, v, err),
@@ -587,6 +609,41 @@ func (r *Revision) setSQLErr(err error, stmt string) error {
 	return err
 }
 
+type (
+	// A Logger logs migration execution.
+	Logger interface {
+		Log(LogEntry)
+	}
+
+	// LogEntry marks several types of logs to be passed to a Logger.
+	LogEntry interface {
+		logEntry()
+	}
+
+	// LogExecution is sent once when execution of the migration files has been started.
+	// It holds the filenames of the pending migration files.
+	LogExecution struct {
+		Files []string
+	}
+
+	// LogFile is sent if a new migration file is executed.
+	LogFile struct {
+		// Version executed.
+		Version string
+		// Desc of migration executed.
+		Desc string
+	}
+
+	// LogStmt is sent if a new SQL statement is executed.
+	LogStmt struct {
+		SQL string
+	}
+)
+
+func (LogExecution) logEntry() {}
+func (LogFile) logEntry()      {}
+func (LogStmt) logEntry()      {}
+
 // LocalDir implements Dir for a local path. It implements the Scanner interface compatible with
 // migration files generated by the DefaultFormatter.
 type LocalDir struct {


Test patch:
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index 1ea3f8ab..e237b634 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -321,6 +321,7 @@ func TestExecutor(t *testing.T) {
 	var (
 		drv  = &lockMockDriver{&mockDriver{}}
 		rrw  = &mockRevisionReadWriter{}
+		log  = &mockLogger{}
 		rev1 = &migrate.Revision{
 			Version:        "1.a",
 			Description:    "sub.up",
@@ -336,7 +337,7 @@ func TestExecutor(t *testing.T) {
 	)
 	dir, err = migrate.NewLocalDir(filepath.Join("testdata", "sub"))
 	require.NoError(t, err)
-	ex, err = migrate.NewExecutor(drv, dir, rrw)
+	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log))
 	require.NoError(t, err)
 
 	// Applies all of them.
@@ -345,6 +346,14 @@ func TestExecutor(t *testing.T) {
 		"CREATE TABLE t_sub(c int);", "ALTER TABLE t_sub ADD c1 int;", "ALTER TABLE t_sub ADD c2 int;",
 	})
 	requireEqualRevisions(t, migrate.Revisions{rev1, rev2}, migrate.Revisions(*rrw))
+	require.Equal(t, []migrate.LogEntry{
+		migrate.LogExecution{Files: []string{"1.a_sub.up.sql", "2.10.x-20_description.sql"}},
+		migrate.LogFile{Version: "1.a", Desc: "sub.up"},
+		migrate.LogStmt{SQL: "CREATE TABLE t_sub(c int);"},
+		migrate.LogStmt{SQL: "ALTER TABLE t_sub ADD c1 int;"},
+		migrate.LogFile{Version: "2.10.x-20", Desc: "description"},
+		migrate.LogStmt{SQL: "ALTER TABLE t_sub ADD c2 int;"},
+	}, []migrate.LogEntry(*log))
 
 	// No pending files.
 	require.ErrorIs(t, ex.Execute(context.Background(), 0), migrate.ErrNoPendingFiles)
@@ -451,6 +460,10 @@ func (r *mockRevisionReadWriter) clean() {
 	*r = mockRevisionReadWriter(migrate.Revisions{})
 }
 
+type mockLogger []migrate.LogEntry
+
+func (m *mockLogger) Log(e migrate.LogEntry) { *m = append(*m, e) }
+
 func requireEqualRevisions(t *testing.T, expected, actual migrate.Revisions) {
 	require.Equal(t, len(expected), len(actual))
 	for i := range expected {