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

Jump to bottom

binlog: fix show pump/drainer status #44764

Merged
merged 4 commits into from
Jun 27, 2023
Merged

binlog: fix show pump/drainer status #44764

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
14924ad
fix
lichunzhu Jun 19, 2023
e8e75c6
fix close problem
lichunzhu Jun 20, 2023
43cf74f
fix bazel
lichunzhu Jun 20, 2023
fccba30
address comment
lichunzhu Jun 27, 2023
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
1 change: 1 addition & 0 deletions executor/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,7 @@ go_library(
"//privilege/privileges",
"//session/txninfo",
"//sessionctx",
"//sessionctx/binloginfo",
"//sessionctx/sessionstates",
"//sessionctx/stmtctx",
"//sessionctx/variable",
Expand Down
7 changes: 6 additions & 1 deletion executor/change.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,15 @@ type ChangeExec struct {
func (e *ChangeExec) Next(ctx context.Context, req *chunk.Chunk) error {
kind := strings.ToLower(e.NodeType)
urls := config.GetGlobalConfig().Path
registry, err := createRegistry(urls)
registry, needToClose, err := createRegistry(urls)
if err != nil {
return err
}
if needToClose {
defer func() {
_ = registry.Close()
}()
}
nodes, _, err := registry.Nodes(ctx, node.NodePrefix[kind])
if err != nil {
return err
Expand Down
26 changes: 16 additions & 10 deletions executor/show.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ import (
"github.com/pingcap/tidb/privilege"
"github.com/pingcap/tidb/privilege/privileges"
"github.com/pingcap/tidb/sessionctx"
"github.com/pingcap/tidb/sessionctx/binloginfo"
"github.com/pingcap/tidb/sessionctx/sessionstates"
"github.com/pingcap/tidb/sessionctx/stmtctx"
"github.com/pingcap/tidb/sessionctx/variable"
Expand Down Expand Up @@ -1876,16 +1877,18 @@ func (e *ShowExec) fetchShowWarnings(errOnly bool) error {

// fetchShowPumpOrDrainerStatus gets status of all pumps or drainers and fill them into e.rows.
func (e *ShowExec) fetchShowPumpOrDrainerStatus(kind string) error {
registry, err := createRegistry(config.GetGlobalConfig().Path)
registry, needToClose, err := createRegistry(config.GetGlobalConfig().Path)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it better to use a new function? This client, I think, is used for Pump/Drainer specifically.

Copy link
Contributor Author

@lichunzhu lichunzhu Jun 27, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about changing function's name to getOrCreateBinlogRegistry to avoid abuse? @okJiang

if err != nil {
return errors.Trace(err)
}

nodes, _, err := registry.Nodes(context.Background(), node.NodePrefix[kind])
if err != nil {
return errors.Trace(err)
if needToClose {
defer func() {
_ = registry.Close()
}()
}
err = registry.Close()

nodes, _, err := registry.Nodes(context.Background(), node.NodePrefix[kind])
if err != nil {
return errors.Trace(err)
}
Expand All @@ -1900,18 +1903,21 @@ func (e *ShowExec) fetchShowPumpOrDrainerStatus(kind string) error {
return nil
}

// createRegistry returns an ectd registry
func createRegistry(urls string) (*node.EtcdRegistry, error) {
// createRegistry returns an ectd registry, need to close, and error
func createRegistry(urls string) (*node.EtcdRegistry, bool, error) {
if pumpClient := binloginfo.GetPumpsClient(); pumpClient != nil && pumpClient.EtcdRegistry != nil {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need tolerate pumpClient is nil?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

createRegistry is used in excutor/change.go too. Should it use the PumpsClient?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need tolerate pumpClient is nil?

Yes. When TiDB doesn't enable binlog, pumpClient is nil. If we use show drainer status now TiDB will panic which is not acceptable.

createRegistry is used in excutor/change.go too. Should it use the PumpsClient?

Yes. It's also binlog's related sql. It also have the same problem.

return pumpClient.EtcdRegistry, false, nil
}
ectdEndpoints, err := util.ParseHostPortAddr(urls)
if err != nil {
return nil, errors.Trace(err)
return nil, false, errors.Trace(err)
}
cli, err := etcd.NewClientFromCfg(ectdEndpoints, etcdDialTimeout, node.DefaultRootPath, nil)
if err != nil {
return nil, errors.Trace(err)
return nil, false, errors.Trace(err)
}

return node.NewEtcdRegistry(cli, etcdDialTimeout), nil
return node.NewEtcdRegistry(cli, etcdDialTimeout), true, nil
}

func (e *ShowExec) getTable() (table.Table, error) {
Expand Down