-
Notifications
You must be signed in to change notification settings - Fork 5.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
log-backup: Refactor daemon (#36763)
close #36762
- Loading branch information
Showing
8 changed files
with
287 additions
and
79 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
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,17 @@ | ||
| // Copyright 2022 PingCAP, Inc. Licensed under Apache-2.0. | ||
|
|
||
| package daemon | ||
|
|
||
| import "context" | ||
|
|
||
| // Interface describes the lifetime hook of a daemon application. | ||
| type Interface interface { | ||
| // OnStart would be called once become the owner. | ||
| // The context passed in would be canceled once it is no more the owner. | ||
| OnStart(ctx context.Context) | ||
| // OnTick would be called periodically. | ||
| // The error can be recorded. | ||
| OnTick(ctx context.Context) error | ||
| // Name returns the name which is used for tracing the daemon. | ||
| Name() string | ||
| } |
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,94 @@ | ||
| // Copyright 2022 PingCAP, Inc. Licensed under Apache-2.0. | ||
|
|
||
| package daemon | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/pingcap/log" | ||
| "github.com/pingcap/tidb/br/pkg/logutil" | ||
| "github.com/pingcap/tidb/owner" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| // OwnerDaemon is a wrapper for running "daemon" in the TiDB cluster. | ||
| // Generally, it uses the etcd election API (wrapped in the `owner.Manager` interface), | ||
| // and shares nothing between nodes. | ||
| // Please make sure the daemon is "stateless" (i.e. it doesn't depend on the local storage or memory state.) | ||
| // This struct is "synchronous" (which means there are no race accessing of these variables.). | ||
| type OwnerDaemon struct { | ||
| daemon Interface | ||
| manager owner.Manager | ||
| tickInterval time.Duration | ||
|
|
||
| // When not `nil`, implies the daemon is running. | ||
| cancel context.CancelFunc | ||
| } | ||
|
|
||
| // New creates a new owner daemon. | ||
| func New(daemon Interface, manager owner.Manager, tickInterval time.Duration) *OwnerDaemon { | ||
| return &OwnerDaemon{ | ||
| daemon: daemon, | ||
| manager: manager, | ||
| tickInterval: tickInterval, | ||
| } | ||
| } | ||
|
|
||
| // Running tests whether the daemon is running (i.e. is it the owner?) | ||
| func (od *OwnerDaemon) Running() bool { | ||
| return od.cancel != nil | ||
| } | ||
|
|
||
| func (od *OwnerDaemon) cancelRun() { | ||
| if od.Running() { | ||
| log.Info("cancel running daemon", zap.String("daemon", od.daemon.Name())) | ||
| od.cancel() | ||
| od.cancel = nil | ||
| } | ||
| } | ||
|
|
||
| func (od *OwnerDaemon) ownerTick(ctx context.Context) { | ||
| // If not running, switching to running. | ||
| if !od.Running() { | ||
| cx, cancel := context.WithCancel(ctx) | ||
| od.cancel = cancel | ||
| log.Info("daemon became owner", zap.String("id", od.manager.ID()), zap.String("daemon-id", od.daemon.Name())) | ||
| // Note: maybe save the context so we can cancel the tick when we are not owner? | ||
| od.daemon.OnStart(cx) | ||
| } | ||
|
|
||
| // Tick anyway. | ||
| if err := od.daemon.OnTick(ctx); err != nil { | ||
| log.Warn("failed on tick", logutil.ShortError(err)) | ||
| } | ||
| } | ||
|
|
||
| // Begin starts the daemon. | ||
| // It would do some bootstrap task, and return a closure that would begin the main loop. | ||
| func (od *OwnerDaemon) Begin(ctx context.Context) (func(), error) { | ||
| log.Info("begin advancer daemon", zap.String("daemon-id", od.daemon.Name())) | ||
| if err := od.manager.CampaignOwner(); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| tick := time.NewTicker(od.tickInterval) | ||
| loop := func() { | ||
| log.Info("begin running daemon", zap.String("id", od.manager.ID()), zap.String("daemon-id", od.daemon.Name())) | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| log.Info("daemon loop exits", zap.String("id", od.manager.ID()), zap.String("daemon-id", od.daemon.Name())) | ||
| return | ||
| case <-tick.C: | ||
| log.Debug("daemon tick start", zap.Bool("is-owner", od.manager.IsOwner()), zap.String("daemon-id", od.daemon.Name())) | ||
| if od.manager.IsOwner() { | ||
| od.ownerTick(ctx) | ||
| } else { | ||
| od.cancelRun() | ||
| } | ||
| } | ||
| } | ||
| } | ||
| return loop, nil | ||
| } |
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,144 @@ | ||
| // Copyright 2022 PingCAP, Inc. Licensed under Apache-2.0. | ||
|
|
||
| package daemon_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/pingcap/log" | ||
| "github.com/pingcap/tidb/br/pkg/streamhelper/daemon" | ||
| "github.com/pingcap/tidb/owner" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| type anApp struct { | ||
| sync.Mutex | ||
| begun bool | ||
|
|
||
| tickingMessenger chan struct{} | ||
| tickingMessengerOnce *sync.Once | ||
| stopMessenger chan struct{} | ||
| startMessenger chan struct{} | ||
|
|
||
| tCtx *testing.T | ||
| } | ||
|
|
||
| func newTestApp(t *testing.T) *anApp { | ||
| return &anApp{ | ||
| tCtx: t, | ||
| startMessenger: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| // OnStart would be called once become the owner. | ||
| // The context passed in would be canceled once it is no more the owner. | ||
| func (a *anApp) OnStart(ctx context.Context) { | ||
| a.Lock() | ||
| defer a.Unlock() | ||
| if a.begun { | ||
| a.tCtx.Fatalf("failed: an app is started twice") | ||
| } | ||
| a.begun = true | ||
| a.tickingMessenger = make(chan struct{}) | ||
| a.tickingMessengerOnce = new(sync.Once) | ||
| a.stopMessenger = make(chan struct{}) | ||
| go func() { | ||
| <-ctx.Done() | ||
|
|
||
| a.Lock() | ||
| defer a.Unlock() | ||
|
|
||
| a.begun = false | ||
| a.tickingMessenger = nil | ||
| a.startMessenger = make(chan struct{}) | ||
| close(a.stopMessenger) | ||
| }() | ||
| close(a.startMessenger) | ||
| } | ||
|
|
||
| // OnTick would be called periodically. | ||
| // The error can be recorded. | ||
| func (a *anApp) OnTick(ctx context.Context) error { | ||
| log.Info("tick") | ||
| a.Lock() | ||
| defer a.Unlock() | ||
| if !a.begun { | ||
| a.tCtx.Fatal("failed: an app is ticking before start") | ||
| } | ||
| a.tickingMessengerOnce.Do(func() { | ||
| log.Info("close") | ||
| close(a.tickingMessenger) | ||
| }) | ||
| return nil | ||
| } | ||
|
|
||
| // Name returns the name which is used for tracing the daemon. | ||
| func (a *anApp) Name() string { | ||
| return "testing" | ||
| } | ||
|
|
||
| func (a *anApp) Running() bool { | ||
| a.Lock() | ||
| defer a.Unlock() | ||
|
|
||
| return a.begun | ||
| } | ||
|
|
||
| func (a *anApp) AssertTick(timeout time.Duration) { | ||
| a.Lock() | ||
| messenger := a.tickingMessenger | ||
| a.Unlock() | ||
| log.Info("waiting") | ||
| select { | ||
| case <-messenger: | ||
| case <-time.After(timeout): | ||
| a.tCtx.Fatalf("tick not triggered after %s", timeout) | ||
| } | ||
| } | ||
|
|
||
| func (a *anApp) AssertNotRunning(timeout time.Duration) { | ||
| a.Lock() | ||
| messenger := a.stopMessenger | ||
| a.Unlock() | ||
| select { | ||
| case <-messenger: | ||
| case <-time.After(timeout): | ||
| a.tCtx.Fatalf("stop not triggered after %s", timeout) | ||
| } | ||
| } | ||
|
|
||
| func (a *anApp) AssertStart(timeout time.Duration) { | ||
| a.Lock() | ||
| messenger := a.startMessenger | ||
| a.Unlock() | ||
| select { | ||
| case <-messenger: | ||
| case <-time.After(timeout): | ||
| a.tCtx.Fatalf("start not triggered after %s", timeout) | ||
| } | ||
| } | ||
|
|
||
| func TestDaemon(t *testing.T) { | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
| req := require.New(t) | ||
| app := newTestApp(t) | ||
| ow := owner.NewMockManager(ctx, "owner_daemon_test") | ||
| d := daemon.New(app, ow, 100*time.Millisecond) | ||
|
|
||
| f, err := d.Begin(ctx) | ||
| req.NoError(err) | ||
| go f() | ||
| app.AssertStart(1 * time.Second) | ||
| app.AssertTick(1 * time.Second) | ||
| ow.RetireOwner() | ||
| req.False(ow.IsOwner()) | ||
| app.AssertNotRunning(1 * time.Second) | ||
| ow.CampaignOwner() | ||
| req.True(ow.IsOwner()) | ||
| app.AssertStart(1 * time.Second) | ||
| app.AssertTick(1 * time.Second) | ||
| } |
Oops, something went wrong.