snap_restore: resend recover_region while there are TiKV restarts
#45361
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
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,29 @@ | ||
| load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test") | ||
|
|
||
| go_library( | ||
| name = "storewatch", | ||
| srcs = ["watching.go"], | ||
| importpath = "github.com/pingcap/tidb/br/pkg/utils/storewatch", | ||
| visibility = ["//visibility:public"], | ||
| deps = [ | ||
| "//br/pkg/conn", | ||
| "//br/pkg/conn/util", | ||
| "@com_github_pingcap_errors//:errors", | ||
| "@com_github_pingcap_kvproto//pkg/metapb", | ||
| ], | ||
| ) | ||
|
|
||
| go_test( | ||
| name = "storewatch_test", | ||
| timeout = "short", | ||
| srcs = ["watching_test.go"], | ||
| flaky = True, | ||
| shard_count = 3, | ||
| deps = [ | ||
| ":storewatch", | ||
| "//br/pkg/conn/util", | ||
| "@com_github_pingcap_kvproto//pkg/metapb", | ||
| "@com_github_stretchr_testify//require", | ||
| "@com_github_tikv_pd_client//:client", | ||
| ], | ||
| ) |
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,136 @@ | ||
| // package storewatch provides a `Watcher` type which allows | ||
| // the user to listen the events of lifetime of stores. | ||
| package storewatch | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/pingcap/errors" | ||
| "github.com/pingcap/kvproto/pkg/metapb" | ||
| "github.com/pingcap/tidb/br/pkg/conn" | ||
| "github.com/pingcap/tidb/br/pkg/conn/util" | ||
| ) | ||
|
|
||
| // Callback will be called the supported event triggered. | ||
| type Callback interface { | ||
| OnNewStoreRegistered(store *metapb.Store) | ||
| OnDisconnect(store *metapb.Store) | ||
| OnReboot(store *metapb.Store) | ||
| } | ||
|
|
||
| // DynCallback is a function based callback set. | ||
| type DynCallback struct { | ||
| onNewStoreRegistered func(*metapb.Store) | ||
| onDisconnect func(*metapb.Store) | ||
| onReboot func(*metapb.Store) | ||
| } | ||
|
|
||
| // OnNewStoreRegistered will be called once new region added to be watched. | ||
| func (cb *DynCallback) OnNewStoreRegistered(store *metapb.Store) { | ||
| if cb.onNewStoreRegistered != nil { | ||
| cb.onNewStoreRegistered(store) | ||
| } | ||
| } | ||
|
|
||
| // OnDisconnect will be called once the store is disconnected. | ||
| func (cb *DynCallback) OnDisconnect(store *metapb.Store) { | ||
| if cb.onDisconnect != nil { | ||
| cb.onDisconnect(store) | ||
| } | ||
| } | ||
|
|
||
| // OnReboot will be called once the store is rebooted. | ||
| func (cb *DynCallback) OnReboot(store *metapb.Store) { | ||
| if cb.onReboot != nil { | ||
| cb.onReboot(store) | ||
| } | ||
| } | ||
|
|
||
| // DynCallbackOpt is the option for DynCallback. | ||
| type DynCallbackOpt func(*DynCallback) | ||
|
|
||
| // WithOnNewStoreRegistered adds a hook to the callback. | ||
| func WithOnNewStoreRegistered(f func(*metapb.Store)) DynCallbackOpt { | ||
| return func(cb *DynCallback) { | ||
| cb.onNewStoreRegistered = f | ||
| } | ||
| } | ||
|
|
||
| // WithOnDisconnect adds a hook to the callback. | ||
| func WithOnDisconnect(f func(*metapb.Store)) DynCallbackOpt { | ||
| return func(cb *DynCallback) { | ||
| cb.onDisconnect = f | ||
| } | ||
| } | ||
|
|
||
| // WithOnReboot adds a hook to the callback. | ||
| func WithOnReboot(f func(*metapb.Store)) DynCallbackOpt { | ||
| return func(cb *DynCallback) { | ||
| cb.onReboot = f | ||
| } | ||
| } | ||
|
|
||
| // MakeCallback creates a callback with the given options. | ||
| // Allowed options: WithOnNewStoreRegistered, WithOnDisconnect, WithOnReboot. | ||
| func MakeCallback(opts ...DynCallbackOpt) Callback { | ||
| cb := &DynCallback{} | ||
| for _, opt := range opts { | ||
| opt(cb) | ||
| } | ||
| return cb | ||
| } | ||
|
|
||
| // Watcher watches the lifetime of stores. | ||
| // generally it should be advanced by calling the `Step` call. | ||
| type Watcher struct { | ||
| cli util.StoreMeta | ||
| cb Callback | ||
|
|
||
| lastStores map[uint64]*metapb.Store | ||
| } | ||
|
|
||
| func New(cli util.StoreMeta, cb Callback) *Watcher { | ||
| return &Watcher{ | ||
| cli: cli, | ||
| cb: cb, | ||
| lastStores: make(map[uint64]*metapb.Store), | ||
| } | ||
| } | ||
|
|
||
| func (w *Watcher) Step(ctx context.Context) error { | ||
| liveStores, err := conn.GetAllTiKVStoresWithRetry(ctx, w.cli, util.SkipTiFlash) | ||
| if err != nil { | ||
| return errors.Annotate(err, "failed to update store list") | ||
| } | ||
| recorded := map[uint64]struct{}{} | ||
| for _, store := range liveStores { | ||
| w.updateStore(store) | ||
| recorded[store.GetId()] = struct{}{} | ||
| } | ||
| w.retain(recorded) | ||
| return nil | ||
| } | ||
|
|
||
| // updateStore updates the current store. and call the hooks needed. | ||
| func (w *Watcher) updateStore(newStore *metapb.Store) { | ||
| lastStore, ok := w.lastStores[newStore.GetId()] | ||
| w.lastStores[newStore.GetId()] = newStore | ||
| if !ok { | ||
| w.cb.OnNewStoreRegistered(newStore) | ||
| return | ||
| } | ||
| if lastStore.GetState() == metapb.StoreState_Up && newStore.GetState() == metapb.StoreState_Offline { | ||
| w.cb.OnDisconnect(newStore) | ||
| } | ||
| if lastStore.StartTimestamp != newStore.StartTimestamp { | ||
| w.cb.OnReboot(newStore) | ||
| } | ||
| } | ||
|
|
||
| func (w *Watcher) retain(storeSet map[uint64]struct{}) { | ||
| for id := range w.lastStores { | ||
| if _, ok := storeSet[id]; !ok { | ||
| delete(w.lastStores, id) | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.