This repository has been archived by the owner on Feb 6, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 15
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: implement failover based distributed etcd lock (#146)
- Loading branch information
1 parent
19f90f5
commit f8d24fd
Showing
4 changed files
with
257 additions
and
1 deletion.
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,174 @@ | ||
| // Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0. | ||
|
|
||
| package coordinator | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "strconv" | ||
| "strings" | ||
| "sync" | ||
|
|
||
| "github.com/CeresDB/ceresdbproto/golang/pkg/metaeventpb" | ||
| "github.com/CeresDB/ceresmeta/pkg/log" | ||
| "github.com/CeresDB/ceresmeta/server/storage" | ||
| "github.com/pkg/errors" | ||
| "go.etcd.io/etcd/api/v3/mvccpb" | ||
| clientv3 "go.etcd.io/etcd/client/v3" | ||
| "go.uber.org/zap" | ||
| "google.golang.org/protobuf/proto" | ||
| ) | ||
|
|
||
| const ( | ||
| shardPath = "shards" | ||
| keySep = "/" | ||
| ) | ||
|
|
||
| type ShardRegisterEvent struct { | ||
| ShardID storage.ShardID | ||
| NewLeaderNode string | ||
| } | ||
|
|
||
| type ShardExpireEvent struct { | ||
| ShardID storage.ShardID | ||
| OldLeaderNode string | ||
| } | ||
|
|
||
| type ShardEventCallback interface { | ||
| OnShardRegistered(event ShardRegisterEvent) error | ||
| OnShardExpired(event ShardExpireEvent) error | ||
| } | ||
|
|
||
| // ShardWatch used to watch the distributed lock of shard, and provide the corresponding callback function. | ||
| type ShardWatch struct { | ||
| rootPath string | ||
| etcdClient *clientv3.Client | ||
| eventCallbacks []ShardEventCallback | ||
|
|
||
| lock sync.RWMutex | ||
| isRunning bool | ||
| cancel context.CancelFunc | ||
| } | ||
|
|
||
| func NewWatch(rootPath string, client *clientv3.Client) *ShardWatch { | ||
| return &ShardWatch{ | ||
| rootPath: rootPath, | ||
| etcdClient: client, | ||
| eventCallbacks: []ShardEventCallback{}, | ||
| } | ||
| } | ||
|
|
||
| func (w *ShardWatch) Start(ctx context.Context) error { | ||
| w.lock.Lock() | ||
| defer w.lock.Unlock() | ||
|
|
||
| if w.isRunning { | ||
| return nil | ||
| } | ||
|
|
||
| shardKeyPrefix := encodeShardKeyPrefix(w.rootPath, shardPath) | ||
| if err := w.startWatch(ctx, shardKeyPrefix); err != nil { | ||
| return errors.WithMessage(err, "etcd register watch failed") | ||
| } | ||
|
|
||
| w.isRunning = true | ||
| return nil | ||
| } | ||
|
|
||
| func (w *ShardWatch) Stop(_ context.Context) error { | ||
| w.lock.Lock() | ||
| defer w.lock.Unlock() | ||
|
|
||
| w.isRunning = false | ||
| w.cancel() | ||
| return nil | ||
| } | ||
|
|
||
| func (w *ShardWatch) RegisteringEventCallback(eventCallback ShardEventCallback) { | ||
| w.eventCallbacks = append(w.eventCallbacks, eventCallback) | ||
| } | ||
|
|
||
| func (w *ShardWatch) startWatch(ctx context.Context, path string) error { | ||
| log.Info("register shard watch", zap.String("watchPath", path)) | ||
| go func() { | ||
| ctxWithCancel, cancel := context.WithCancel(ctx) | ||
| w.cancel = cancel | ||
| respChan := w.etcdClient.Watch(ctxWithCancel, path, clientv3.WithPrefix(), clientv3.WithPrevKV()) | ||
| for resp := range respChan { | ||
| for _, event := range resp.Events { | ||
| if err := w.processEvent(event); err != nil { | ||
| log.Error("process event", zap.Error(err)) | ||
| } | ||
| } | ||
| } | ||
| }() | ||
| return nil | ||
| } | ||
|
|
||
| func (w *ShardWatch) processEvent(event *clientv3.Event) error { | ||
| switch event.Type { | ||
| case mvccpb.DELETE: | ||
| shardID, err := decodeShardKey(string(event.Kv.Key)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| shardLockValue, err := convertShardLockValueToPB(event.PrevKv.Value) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| log.Info("receive delete event", zap.String("preKV", fmt.Sprintf("%v", event.PrevKv)), zap.String("event", fmt.Sprintf("%v", event)), zap.Uint64("shardID", shardID), zap.String("oldLeader", shardLockValue.NodeName)) | ||
| for _, callback := range w.eventCallbacks { | ||
| if err := callback.OnShardExpired(ShardExpireEvent{ | ||
| ShardID: storage.ShardID(shardID), | ||
| OldLeaderNode: shardLockValue.NodeName, | ||
| }); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| case mvccpb.PUT: | ||
| shardID, err := decodeShardKey(string(event.Kv.Key)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| shardLockValue, err := convertShardLockValueToPB(event.Kv.Value) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| log.Info("receive put event", zap.String("event", fmt.Sprintf("%v", event)), zap.Uint64("shardID", shardID), zap.String("oldLeader", shardLockValue.NodeName)) | ||
| for _, callback := range w.eventCallbacks { | ||
| if err := callback.OnShardRegistered(ShardRegisterEvent{ | ||
| ShardID: storage.ShardID(shardID), | ||
| NewLeaderNode: shardLockValue.NodeName, | ||
| }); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func decodeShardKey(keyPath string) (uint64, error) { | ||
| pathList := strings.Split(keyPath, keySep) | ||
| shardID, err := strconv.ParseUint(pathList[len(pathList)-1], 10, 64) | ||
| if err != nil { | ||
| return 0, errors.WithMessage(err, "decode etcd event key failed") | ||
| } | ||
| return shardID, nil | ||
| } | ||
|
|
||
| func encodeShardKeyPrefix(rootPath, shardPath string) string { | ||
| return strings.Join([]string{rootPath, shardPath}, keySep) | ||
| } | ||
|
|
||
| func encodeShardKey(rootPath string, shardPath string, shardID uint64) string { | ||
| shardKeyPrefix := encodeShardKeyPrefix(rootPath, shardPath) | ||
| return strings.Join([]string{shardKeyPrefix, strconv.FormatUint(shardID, 10)}, keySep) | ||
| } | ||
|
|
||
| func convertShardLockValueToPB(value []byte) (*metaeventpb.ShardLockValue, error) { | ||
| shardLockValue := &metaeventpb.ShardLockValue{} | ||
| if err := proto.Unmarshal(value, shardLockValue); err != nil { | ||
| return shardLockValue, errors.WithMessage(err, "unmarshal shardLockValue failed") | ||
| } | ||
| return shardLockValue, 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,74 @@ | ||
| // Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0. | ||
|
|
||
| package coordinator | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/CeresDB/ceresdbproto/golang/pkg/metaeventpb" | ||
| "github.com/CeresDB/ceresmeta/server/etcdutil" | ||
| "github.com/CeresDB/ceresmeta/server/storage" | ||
| "github.com/stretchr/testify/require" | ||
| clientv3 "go.etcd.io/etcd/client/v3" | ||
| "google.golang.org/protobuf/proto" | ||
| ) | ||
|
|
||
| const ( | ||
| TestRootPath = "/rootPath" | ||
| TestShardPath = "shards" | ||
| TestShardID = 1 | ||
| TestNodeName = "testNode" | ||
| ) | ||
|
|
||
| func TestWatch(t *testing.T) { | ||
| re := require.New(t) | ||
| ctx := context.Background() | ||
|
|
||
| _, client, _ := etcdutil.PrepareEtcdServerAndClient(t) | ||
| watch := NewWatch(TestRootPath, client) | ||
| err := watch.Start(ctx) | ||
| re.NoError(err) | ||
|
|
||
| testCallback := testShardEventCallback{ | ||
| result: 0, | ||
| re: re, | ||
| } | ||
|
|
||
| watch.RegisteringEventCallback(&testCallback) | ||
|
|
||
| // Valid that callback function is executed and the params are as expected. | ||
| b, err := proto.Marshal(&metaeventpb.ShardLockValue{NodeName: TestNodeName}) | ||
| re.NoError(err) | ||
|
|
||
| keyPath := encodeShardKey(TestRootPath, TestShardPath, TestShardID) | ||
| _, err = client.Put(ctx, keyPath, string(b)) | ||
| re.NoError(err) | ||
| time.Sleep(time.Millisecond * 10) | ||
| re.Equal(2, testCallback.result) | ||
|
|
||
| _, err = client.Delete(ctx, keyPath, clientv3.WithPrevKV()) | ||
| re.NoError(err) | ||
| time.Sleep(time.Millisecond * 10) | ||
| re.Equal(1, testCallback.result) | ||
| } | ||
|
|
||
| type testShardEventCallback struct { | ||
| result int | ||
| re *require.Assertions | ||
| } | ||
|
|
||
| func (c *testShardEventCallback) OnShardRegistered(event ShardRegisterEvent) error { | ||
| c.result = 2 | ||
| c.re.Equal(storage.ShardID(TestShardID), event.ShardID) | ||
| c.re.Equal(TestNodeName, event.NewLeaderNode) | ||
| return nil | ||
| } | ||
|
|
||
| func (c *testShardEventCallback) OnShardExpired(event ShardExpireEvent) error { | ||
| c.result = 1 | ||
| c.re.Equal(storage.ShardID(TestShardID), event.ShardID) | ||
| c.re.Equal(TestNodeName, event.OldLeaderNode) | ||
| return nil | ||
| } |