Skip to content
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.

Sign up for GitHub

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

Add more logs for clickhouse migrate #1178

Closed
wants to merge 1 commit into from
Closed

Add more logs for clickhouse migrate #1178

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions database/clickhouse/clickhouse.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,42 +134,53 @@
}

func (ch *ClickHouse) Run(r io.Reader) error {
fmt.Println(fmt.Sprintf("Try to run clickhouse. MultiStatementEnabled:%+v", ch.config.MultiStatementEnabled))

Check failure on line 137 in database/clickhouse/clickhouse.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
if ch.config.MultiStatementEnabled {
var err error
if e := multistmt.Parse(r, multiStmtDelimiter, ch.config.MultiStatementMaxSize, func(m []byte) bool {
fmt.Println(fmt.Sprintf("Try to exec query clickhouse. query:%+v", string(m)))

Check failure on line 141 in database/clickhouse/clickhouse.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
tq := strings.TrimSpace(string(m))
if tq == "" {
return true
}
if _, e := ch.conn.Exec(string(m)); e != nil {
fmt.Println(fmt.Errorf("exec query clickhouse error:%+v", err))
err = database.Error{OrigErr: e, Err: "migration failed", Query: m}
return false
}
return true
}); e != nil {
fmt.Println(fmt.Errorf("multistmt parse error:%+v", err))
return e
}
return err
}

fmt.Println(fmt.Sprintf("Try to get clickhouse migration."))

Check failure on line 159 in database/clickhouse/clickhouse.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
migration, err := io.ReadAll(r)
if err != nil {
fmt.Println(fmt.Errorf("get clickhouse migration error:%+v", err))
return err
}

fmt.Println(fmt.Sprintf("Try to exec migration clickhouse. migration:%+v", string(migration)))

Check failure on line 166 in database/clickhouse/clickhouse.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
if _, err := ch.conn.Exec(string(migration)); err != nil {
fmt.Println(fmt.Errorf("exec migration clickhouse error:%+v", err))
return database.Error{OrigErr: err, Err: "migration failed", Query: migration}
}

return nil
}
func (ch *ClickHouse) Version() (int, bool, error) {

var (
version int
dirty uint8
query = "SELECT version, dirty FROM `" + ch.config.MigrationsTable + "` ORDER BY sequence DESC LIMIT 1"
)
fmt.Println(fmt.Sprintf("Try to get version clickhouse. query:%+v", query))

Check failure on line 181 in database/clickhouse/clickhouse.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
if err := ch.conn.QueryRow(query).Scan(&version, &dirty); err != nil {
fmt.Println(fmt.Errorf("get version clickhouse err:%+v", err))
if err == sql.ErrNoRows {
return database.NilVersion, false, nil
}
Expand All @@ -179,6 +190,7 @@
}

func (ch *ClickHouse) SetVersion(version int, dirty bool) error {
fmt.Println(fmt.Sprintf("Try to set version clickhouse. version:%+v dirty:%+v", version, dirty))

Check failure on line 193 in database/clickhouse/clickhouse.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
var (
bool = func(v bool) uint8 {
if v {
Expand All @@ -189,11 +201,14 @@
tx, err = ch.conn.Begin()
)
if err != nil {
fmt.Println(fmt.Errorf("set version clickhouse begin error:%+v", err))
return err
}

query := "INSERT INTO " + ch.config.MigrationsTable + " (version, dirty, sequence) VALUES (?, ?, ?)"
fmt.Println(fmt.Sprintf("Try to set version clickhouse exec. query:%+v", query))

Check failure on line 209 in database/clickhouse/clickhouse.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
if _, err := tx.Exec(query, version, bool(dirty), time.Now().UnixNano()); err != nil {
fmt.Println(fmt.Errorf("set version clickhouse exec error:%+v", err))
return &database.Error{OrigErr: err, Query: []byte(query)}
}

Expand Down
26 changes: 25 additions & 1 deletion migrate.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,6 +273,7 @@
if err != nil {
return m.unlockErr(err)
}
fmt.Println(fmt.Sprintf("Try to do up. curVersion:%+v dirty:%+v", curVersion, dirty))

Check failure on line 276 in migrate.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)

if dirty {
return m.unlockErr(ErrDirty{curVersion})
Expand Down Expand Up @@ -534,9 +535,11 @@
func (m *Migrate) readUp(from int, limit int, ret chan<- interface{}) {
defer close(ret)

fmt.Println(fmt.Sprintf("Try to read up. from:%+v limit:%+v", from, limit))

Check failure on line 538 in migrate.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
// check if from version exists
if from >= 0 {
if err := m.versionExists(suint(from)); err != nil {
fmt.Println(fmt.Errorf("version exists error:%+v", err))
ret <- err
return
}
Expand All @@ -549,27 +552,34 @@

count := 0
for count < limit || limit == -1 {
fmt.Println(fmt.Sprintf("Try to loop read up. count:%+v from:%+v", count, from))

Check failure on line 555 in migrate.go

View workflow job for this annotation

GitHub Actions / lint 

S1038: should use fmt.Printf instead of fmt.Println(fmt.Sprintf(...)) (but don't forget the newline) (gosimple)
if m.stop() {
fmt.Println(fmt.Sprintf("Exit loop read up."))
return
}

// apply first migration if from is nil version
if from == -1 {
fmt.Println(fmt.Sprintf("Try to get first version."))
firstVersion, err := m.sourceDrv.First()
if err != nil {
fmt.Println(fmt.Errorf("get first version error:%+v", err))
ret <- err
return
}

fmt.Println(fmt.Sprintf("Try to new first migration. firstVersion: %+v", firstVersion))
migr, err := m.newMigration(firstVersion, int(firstVersion))
if err != nil {
fmt.Println(fmt.Errorf("new first migration error:%+v", err))
ret <- err
return
}

ret <- migr
go func() {
if err := migr.Buffer(); err != nil {
fmt.Println(fmt.Errorf("buffer migr error:%+v", err))
m.logErr(err)
}
}()
Expand All @@ -578,6 +588,7 @@
continue
}

fmt.Println(fmt.Sprintf("Try to get next version."))
// apply next migration
next, err := m.sourceDrv.Next(suint(from))
if errors.Is(err, os.ErrNotExist) {
Expand Down Expand Up @@ -605,19 +616,23 @@
}
}
if err != nil {
fmt.Println(fmt.Errorf("get next version error:%+v", err))
ret <- err
return
}

fmt.Println(fmt.Sprintf("Try to new next migration. nextVersion: %+v", next))
migr, err := m.newMigration(next, int(next))
if err != nil {
fmt.Println(fmt.Errorf("new next migration error:%+v", err))
ret <- err
return
}

ret <- migr
go func() {
if err := migr.Buffer(); err != nil {
fmt.Println(fmt.Errorf("buffer migr error:%+v", err))
m.logErr(err)
}
}()
Expand Down Expand Up @@ -723,39 +738,48 @@
// to stop execution because it might have received a stop signal on the
// GracefulStop channel.
func (m *Migrate) runMigrations(ret <-chan interface{}) error {
fmt.Println(fmt.Sprintf("Try to run migrations."))
for r := range ret {

if m.stop() {
fmt.Println(fmt.Sprintf("Exit loop run migration."))
return nil
}

switch r := r.(type) {
case error:
fmt.Println(fmt.Errorf("run migration error:%+v", r))
return r

case *Migration:
migr := r

fmt.Println(fmt.Sprintf("Try to set dirty version. TargetVersion:%+v", migr.TargetVersion))
// set version with dirty state
if err := m.databaseDrv.SetVersion(migr.TargetVersion, true); err != nil {
fmt.Println(fmt.Errorf("set dirty version error:%+v", err))
return err
}

if migr.Body != nil {
fmt.Println(fmt.Sprintf("Try to loop run migrations. LogString:%+v", migr.LogString()))
m.logVerbosePrintf("Read and execute %v\n", migr.LogString())
if err := m.databaseDrv.Run(migr.BufferedBody); err != nil {
fmt.Println(fmt.Errorf("run error:%+v", err))
return err
}
}

fmt.Println(fmt.Sprintf("Try to set clenn version. TargetVersion:%+v", migr.TargetVersion))
// set clean state
if err := m.databaseDrv.SetVersion(migr.TargetVersion, false); err != nil {
fmt.Println(fmt.Errorf("set clean version error:%+v", err))
return err
}

endTime := time.Now()
readTime := migr.FinishedReading.Sub(migr.StartedBuffering)
runTime := endTime.Sub(migr.FinishedReading)
fmt.Println(fmt.Sprintf("Finished %+v (read %+v, ran %+v)", migr.LogString(), readTime, runTime))

// log either verbose or normal
if m.Log != nil {
Expand Down
Loading