Skip to content

logcoordinator: improve log #4206

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

Merged
ti-chi-bot merged 4 commits into from
Feb 24, 2026
Merged

logcoordinator: improve log #4206

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f3489ea
logcoordinator: fix resign
lidezhu Feb 13, 2026
3b00729
fix
lidezhu Feb 13, 2026
7e5b543
Merge branch 'master' into ldz/fix-resign-log-coordinator
lidezhu Feb 24, 2026
02841db
fix format
lidezhu Feb 24, 2026
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
20 changes: 20 additions & 0 deletions pkg/etcd/etcd.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,8 @@ type OwnerCaptureInfoClient interface {

GetOwnerRevision(context.Context, config.CaptureID) (int64, error)

GetLogCoordinatorRevision(context.Context, config.CaptureID) (int64, error)

GetCaptures(context.Context) (int64, []*config.CaptureInfo, error)
}

Expand Down Expand Up @@ -599,6 +601,24 @@ func (c *CDCEtcdClientImpl) GetOwnerRevision(
return resp.Kvs[0].ModRevision, nil
}

// GetLogCoordinatorRevision gets the Etcd revision for the elected log coordinator.
func (c *CDCEtcdClientImpl) GetLogCoordinatorRevision(
ctx context.Context, captureID string,
) (rev int64, err error) {
resp, err := c.Client.Get(ctx, LogCoordinatorKey(c.ClusterID), clientv3.WithFirstCreate()...)
if err != nil {
return 0, errors.WrapError(errors.ErrPDEtcdAPIError, err)
}
if len(resp.Kvs) == 0 {
return 0, errors.ErrOwnerNotFound.GenWithStackByArgs()
}
// Checks that the given capture is indeed the log coordinator.
if string(resp.Kvs[0].Value) != captureID {
return 0, errors.ErrNotOwner.GenWithStackByArgs()
}
return resp.Kvs[0].ModRevision, nil
}

// GetGCServiceID returns the cdc gc service ID
func (c *CDCEtcdClientImpl) GetGCServiceID() string {
return fmt.Sprintf("ticdc-%s-%d", c.ClusterID, c.etcdClusterID)
Expand Down
30 changes: 30 additions & 0 deletions pkg/etcd/etcd_mock.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

68 changes: 68 additions & 0 deletions pkg/etcd/etcd_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"github.com/golang/mock/gomock"
"github.com/pingcap/errors"
"github.com/pingcap/ticdc/pkg/common"
cerrors "github.com/pingcap/ticdc/pkg/errors"
"github.com/stretchr/testify/require"
"go.etcd.io/etcd/api/v3/etcdserverpb"
"go.etcd.io/etcd/api/v3/mvccpb"
Expand Down Expand Up @@ -270,3 +271,70 @@ func TestCDCEtcdClientImpl_GetChangefeedInfoAndStatus(t *testing.T) {
})
}
}

func TestCDCEtcdClientImpl_GetLogCoordinatorRevision(t *testing.T) {
ctx := context.Background()
clusterID := "cluster-id"
captureID := "capture-1"

t.Run("get leader failed", func(t *testing.T) {
ctrl := gomock.NewController(t)
client := NewMockClient(ctrl)
client.EXPECT().
Get(gomock.Eq(ctx), gomock.Eq(LogCoordinatorKey(clusterID)), gomock.Any()).
Return(nil, errors.New("etcd failed")).
Times(1)

etcdClient := &CDCEtcdClientImpl{Client: client, ClusterID: clusterID}
_, err := etcdClient.GetLogCoordinatorRevision(ctx, captureID)
require.ErrorContains(t, err, "etcd failed")
})

t.Run("leader not found", func(t *testing.T) {
ctrl := gomock.NewController(t)
client := NewMockClient(ctrl)
client.EXPECT().
Get(gomock.Eq(ctx), gomock.Eq(LogCoordinatorKey(clusterID)), gomock.Any()).
Return(&clientv3.GetResponse{Kvs: nil}, nil).
Times(1)

etcdClient := &CDCEtcdClientImpl{Client: client, ClusterID: clusterID}
_, err := etcdClient.GetLogCoordinatorRevision(ctx, captureID)
require.True(t, cerrors.ErrOwnerNotFound.Equal(err))
})

t.Run("not the log coordinator", func(t *testing.T) {
ctrl := gomock.NewController(t)
client := NewMockClient(ctrl)
client.EXPECT().
Get(gomock.Eq(ctx), gomock.Eq(LogCoordinatorKey(clusterID)), gomock.Any()).
Return(&clientv3.GetResponse{
Kvs: []*mvccpb.KeyValue{
{Value: []byte("other-capture"), ModRevision: 99},
},
}, nil).
Times(1)

etcdClient := &CDCEtcdClientImpl{Client: client, ClusterID: clusterID}
_, err := etcdClient.GetLogCoordinatorRevision(ctx, captureID)
require.True(t, cerrors.ErrNotOwner.Equal(err))
})

t.Run("success", func(t *testing.T) {
ctrl := gomock.NewController(t)
client := NewMockClient(ctrl)
client.EXPECT().
Get(gomock.Eq(ctx), gomock.Eq(LogCoordinatorKey(clusterID)), gomock.Any()).
Return(&clientv3.GetResponse{
Kvs: []*mvccpb.KeyValue{
{Value: []byte(captureID), ModRevision: 123},
},
}, nil).
Times(1)

etcdClient := &CDCEtcdClientImpl{Client: client, ClusterID: clusterID}
rev, err := etcdClient.GetLogCoordinatorRevision(ctx, captureID)
require.NoError(t, err)
require.Equal(t, int64(123), rev)
})
}
32 changes: 21 additions & 11 deletions server/module_election.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -227,16 +227,22 @@ func (e *elector) campaignLogCoordinator(ctx context.Context) error {
if e.svr.liveness.Load() == api.LivenessCaptureStopping {
// If the server is stopping, resign actively.
log.Info("resign log coordinator actively, liveness is stopping")
if resignErr := e.resign(ctx); resignErr != nil {
if resignErr := e.resignLogCoordinator(); resignErr != nil {
log.Warn("resign log coordinator actively failed",
zap.String("nodeID", nodeID), zap.Error(resignErr))
return errors.Trace(err)
return nil
}
return nil
}

// FIXME: get log coordinator version from etcd and add it to log
log.Info("campaign log coordinator successfully", zap.String("nodeID", nodeID))
logCoordinatorVersion, err := e.svr.EtcdClient.GetLogCoordinatorRevision(ctx, config.CaptureID(nodeID))
if err != nil {
return errors.Trace(err)
}

log.Info("campaign log coordinator successfully",
zap.String("nodeID", nodeID),
zap.Int64("logCoordinatorVersion", logCoordinatorVersion))

co := logcoordinator.New()
err = co.Run(ctx)
Expand All @@ -247,12 +253,16 @@ func (e *elector) campaignLogCoordinator(ctx context.Context) error {
return errors.Trace(resignErr)
}
}
log.Warn("log coordinator exited with error", zap.String("nodeID", nodeID), zap.Error(err))
log.Warn("log coordinator exited with error",
zap.String("nodeID", nodeID), zap.Int64("logCoordinatorVersion", logCoordinatorVersion),
zap.Error(err))
return errors.Trace(err)
}

// If coordinator exits normally, continue the campaign loop and try to election coordinator again
log.Info("log coordinator exited normally", zap.String("nodeID", nodeID), zap.Error(err))
log.Info("log coordinator exited normally",
zap.String("nodeID", nodeID), zap.Int64("logCoordinatorVersion", logCoordinatorVersion),
zap.Error(err))
}
}

Expand Down Expand Up @@ -280,16 +290,16 @@ func (e *elector) resignLogCoordinator() error {
nodeID := string(e.svr.info.ID)
// use a new context to prevent the context from being cancelled.
resignCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if resignErr := e.logElection.Resign(resignCtx); resignErr != nil {
if errors.Is(errors.Cause(resignErr), context.DeadlineExceeded) {
log.Info("log coordinator resign failed",
log.Warn("log coordinator resign timeout",
zap.String("nodeID", nodeID), zap.Error(resignErr))
cancel()
return errors.Trace(resignErr)
return nil
}
log.Warn("log coordinator resign timeout",
log.Info("log coordinator resign failed",
zap.String("nodeID", nodeID), zap.Error(resignErr))
return errors.Trace(resignErr)
Comment on lines +300 to +302
Copy link

@gemini-code-assist gemini-code-assist bot Feb 13, 2026

Choose a reason for hiding this comment

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

medium

This log message is redundant. The primary caller of resignLogCoordinator in campaignLogCoordinator (line 231) already logs a WARN message when this function returns an error. To avoid duplicate logs and to keep the function's responsibility focused on the resign action, it's better to remove this log.Info call and just return the error. This change would also align resignLogCoordinator with the pattern used in the resign function, which does not perform logging itself.

return errors.Trace(resignErr)

}
cancel()
return nil
}