-
Notifications
You must be signed in to change notification settings - Fork 12
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: <[email protected]>
- Loading branch information
Showing
5 changed files
with
148 additions
and
7 deletions.
There are no files selected for viewing
This file contains 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 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,141 @@ | ||
| /* | ||
| Copyright ApeCloud, Inc. | ||
| Licensed under the Apache v2(found in the LICENSE file in the root directory). | ||
| */ | ||
|
|
||
| package tabletserver | ||
|
|
||
| import ( | ||
| "context" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "vitess.io/vitess/go/sqltypes" | ||
| "vitess.io/vitess/go/vt/dbconfigs" | ||
| "vitess.io/vitess/go/vt/sqlparser" | ||
| "vitess.io/vitess/go/vt/vttablet/tabletserver/connpool" | ||
| "vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv" | ||
| ) | ||
|
|
||
| const SeleteWorkflowNameFromBranchJobs = "SELECT * from mysql.branch_jobs;" | ||
|
|
||
| const SeleteVReplicationByWorkflow = "SELECT * from mysql.vreplication where workflow=%a;" | ||
|
|
||
| const UpdateBranchJobStatusByWorkflow = "update mysql.branch_jobs set status=%a,message=%a where workflow_name=%a" | ||
|
|
||
| const ( | ||
| BranchStateOfPrepare = "prepare" | ||
| BranchStateOfRunning = "running" | ||
| BranchStateOfStop = "Stop" | ||
| BranchStateOfCompleted = "completed" | ||
| BranchStateOfError = "error" | ||
| ) | ||
|
|
||
| const UpdateInterval = 5 * time.Second | ||
|
|
||
| type BranchWatcher struct { | ||
| dbConfig dbconfigs.Connector | ||
| conns *connpool.Pool | ||
|
|
||
| ticker *time.Ticker | ||
| updateInterval time.Duration | ||
| running bool | ||
| } | ||
|
|
||
| func NewBranchWatcher(env tabletenv.Env, dbConfig dbconfigs.Connector) *BranchWatcher { | ||
| branchWatcher := BranchWatcher{ | ||
| running: false, | ||
| ticker: time.NewTicker(UpdateInterval), | ||
| updateInterval: UpdateInterval, | ||
| dbConfig: dbConfig, | ||
| } | ||
| branchWatcher.conns = connpool.NewPool(env, "", tabletenv.ConnPoolConfig{ | ||
| Size: 2, | ||
| IdleTimeoutSeconds: env.Config().OltpReadPool.IdleTimeoutSeconds, | ||
| }) | ||
| return &branchWatcher | ||
| } | ||
|
|
||
| func (b *BranchWatcher) updateBranchState(ctx context.Context, conn *connpool.DBConn, state string, message, workflow string) error { | ||
| query, err := sqlparser.ParseAndBind(UpdateBranchJobStatusByWorkflow, | ||
| sqltypes.StringBindVariable(state), | ||
| sqltypes.StringBindVariable(message), | ||
| sqltypes.StringBindVariable(workflow)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| _, err = conn.Exec(ctx, query, -1, false) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (b *BranchWatcher) watch() error { | ||
| ctx := context.Background() | ||
| conn, err := b.conns.Get(ctx, nil) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| defer conn.Recycle() | ||
| qr, err := conn.Exec(ctx, SeleteWorkflowNameFromBranchJobs, -1, true) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| for _, row := range qr.Named().Rows { | ||
| workflow := row["workflow_name"].ToString() | ||
| query, err := sqlparser.ParseAndBind(SeleteVReplicationByWorkflow, sqltypes.StringBindVariable(workflow)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| vreplication, err := conn.Exec(ctx, query, -1, true) | ||
| rows := vreplication.Named().Row() | ||
| vState := rows["state"].ToString() | ||
| message := rows["message"].ToString() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| switch vState { | ||
| case "Stopped": | ||
| if strings.Contains(message, "Stopped after copy") { | ||
| err = b.updateBranchState(ctx, conn, BranchStateOfCompleted, message, workflow) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } else { | ||
| err = b.updateBranchState(ctx, conn, BranchStateOfError, message, workflow) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } | ||
| case "Error": | ||
| err = b.updateBranchState(ctx, conn, BranchStateOfError, message, workflow) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| case "Copying": | ||
| err = b.updateBranchState(ctx, conn, BranchStateOfRunning, message, workflow) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (b *BranchWatcher) Open() { | ||
| b.conns.Open(b.dbConfig, b.dbConfig, b.dbConfig) | ||
| b.ticker.Reset(b.updateInterval) | ||
| if !b.running { | ||
| go func() { | ||
| for range b.ticker.C { | ||
| b.watch() | ||
| } | ||
| }() | ||
| b.running = true | ||
| } | ||
| } | ||
| func (b *BranchWatcher) Close() { | ||
| b.conns.Close() | ||
| b.ticker.Stop() | ||
| } |
This file contains 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 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 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