This repository has been archived by the owner on Mar 30, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #42 from RSE-Cambridge/fix-watching
Fix dacd watching
- Loading branch information
Showing
22 changed files
with
609 additions
and
119 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
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,52 @@ | ||
| package etcdregistry | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/RSE-Cambridge/data-acc/internal/pkg/keystoreregistry" | ||
| "github.com/coreos/etcd/clientv3" | ||
| ) | ||
|
|
||
| func (client *etcKeystore) Watch(ctxt context.Context, key string, withPrefix bool) keystoreregistry.KeyValueUpdateChan { | ||
| options := []clientv3.OpOption{clientv3.WithPrevKV()} | ||
| if withPrefix { | ||
| options = append(options, clientv3.WithPrefix()) | ||
| } | ||
| rch := client.Watcher.Watch(ctxt, key, options...) | ||
|
|
||
| c := make(chan keystoreregistry.KeyValueUpdate) | ||
|
|
||
| go processWatchEvents(rch, c) | ||
|
|
||
| return c | ||
| } | ||
|
|
||
| func processWatchEvents(watchChan clientv3.WatchChan, c chan keystoreregistry.KeyValueUpdate) { | ||
| for watchResponse := range watchChan { | ||
| // if error, send empty update with an error | ||
| err := watchResponse.Err() | ||
| if err != nil { | ||
| c <- keystoreregistry.KeyValueUpdate{Err: err} | ||
| } | ||
|
|
||
| // send all events in this watch response | ||
| for _, ev := range watchResponse.Events { | ||
| update := keystoreregistry.KeyValueUpdate{ | ||
| IsCreate: ev.IsCreate(), | ||
| IsModify: ev.IsModify(), | ||
| IsDelete: ev.Type == clientv3.EventTypeDelete, | ||
| } | ||
| if update.IsCreate || update.IsModify { | ||
| update.New = getKeyValueVersion(ev.Kv) | ||
| } | ||
| if update.IsDelete || update.IsModify { | ||
| update.Old = getKeyValueVersion(ev.PrevKv) | ||
| } | ||
|
|
||
| c <- update | ||
| } | ||
| } | ||
|
|
||
| // Assuming we get here when the context is cancelled or hits its timeout | ||
| // i.e. there are no more events, so we close the channel | ||
| close(c) | ||
| } |
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,119 @@ | ||
| package etcdregistry | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/coreos/etcd/clientv3" | ||
| "github.com/coreos/etcd/mvcc/mvccpb" | ||
| "github.com/stretchr/testify/assert" | ||
| "testing" | ||
| ) | ||
|
|
||
| type fakeWatcher struct { | ||
| t *testing.T | ||
| ch clientv3.WatchChan | ||
| opts []clientv3.OpOption | ||
| } | ||
|
|
||
| func (fw fakeWatcher) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan { | ||
| assert.Equal(fw.t, "key", key) | ||
| assert.EqualValues(fw.t, len(fw.opts), len(opts)) // TODO: how to assert this properly? | ||
| return fw.ch | ||
| } | ||
| func (fakeWatcher) Close() error { | ||
| panic("implement me") | ||
| } | ||
|
|
||
| func TestEtcKeystore_Watch_Nil(t *testing.T) { | ||
| keystore := etcKeystore{ | ||
| Watcher: fakeWatcher{ | ||
| t: t, ch: nil, | ||
| opts: []clientv3.OpOption{clientv3.WithPrevKV()}, | ||
| }, | ||
| } | ||
|
|
||
| response := keystore.Watch(context.TODO(), "key", false) | ||
|
|
||
| assert.Empty(t, response) | ||
| } | ||
|
|
||
| func TestEtcKeystore_Watch(t *testing.T) { | ||
| ch := make(chan clientv3.WatchResponse) | ||
|
|
||
| keystore := etcKeystore{ | ||
| Watcher: fakeWatcher{ | ||
| t: t, ch: ch, | ||
| opts: []clientv3.OpOption{clientv3.WithPrefix(), clientv3.WithPrevKV()}, | ||
| }, | ||
| } | ||
|
|
||
| go func() { | ||
| ch <- clientv3.WatchResponse{ | ||
| Events: []*clientv3.Event{ | ||
| {Type: clientv3.EventTypePut, Kv: &mvccpb.KeyValue{Key: []byte("key1")}}, | ||
| {Type: clientv3.EventTypePut, Kv: &mvccpb.KeyValue{Key: []byte("key2")}}, | ||
| }} | ||
| ch <- clientv3.WatchResponse{ | ||
| Events: []*clientv3.Event{ | ||
| { | ||
| Type: clientv3.EventTypePut, | ||
| Kv: &mvccpb.KeyValue{ModRevision: 1, Key: []byte("key2")}, | ||
| PrevKv: &mvccpb.KeyValue{ModRevision: 1, Key: []byte("key2")}, | ||
| }, | ||
| }} | ||
| ch <- clientv3.WatchResponse{ | ||
| Events: []*clientv3.Event{ | ||
| {Type: clientv3.EventTypeDelete, PrevKv: &mvccpb.KeyValue{Key: []byte("key2")}}, | ||
| {Type: clientv3.EventTypeDelete, PrevKv: &mvccpb.KeyValue{Key: []byte("key1")}}, | ||
| }} | ||
| ch <- clientv3.WatchResponse{Canceled: true} | ||
| close(ch) | ||
| }() | ||
|
|
||
| response := keystore.Watch(context.TODO(), "key", true) | ||
|
|
||
| ev1 := <-response | ||
| assert.True(t, ev1.IsCreate) | ||
| assert.False(t, ev1.IsModify) | ||
| assert.False(t, ev1.IsDelete) | ||
| assert.Nil(t, ev1.Old) | ||
| assert.EqualValues(t, "key1", ev1.New.Key) | ||
|
|
||
| ev2 := <-response | ||
| assert.True(t, ev2.IsCreate) | ||
| assert.False(t, ev2.IsModify) | ||
| assert.False(t, ev2.IsDelete) | ||
| assert.Nil(t, ev2.Old) | ||
| assert.EqualValues(t, "key2", ev2.New.Key) | ||
|
|
||
| ev3 := <-response | ||
| assert.False(t, ev3.IsCreate) | ||
| assert.True(t, ev3.IsModify) | ||
| assert.False(t, ev3.IsDelete) | ||
| assert.EqualValues(t, "key2", ev3.New.Key) | ||
| assert.EqualValues(t, "key2", ev3.Old.Key) | ||
|
|
||
| ev4 := <-response | ||
| assert.False(t, ev4.IsCreate) | ||
| assert.False(t, ev4.IsModify) | ||
| assert.True(t, ev4.IsDelete) | ||
| assert.Nil(t, ev4.New) | ||
| assert.EqualValues(t, "key2", ev4.Old.Key) | ||
|
|
||
| ev5 := <-response | ||
| assert.False(t, ev5.IsCreate) | ||
| assert.False(t, ev5.IsModify) | ||
| assert.True(t, ev5.IsDelete) | ||
| assert.Nil(t, ev5.New) | ||
| assert.EqualValues(t, "key1", ev5.Old.Key) | ||
|
|
||
| ev6 := <-response | ||
| assert.Equal(t, | ||
| "etcdserver: mvcc: required revision is a future revision", | ||
| ev6.Err.Error()) | ||
|
|
||
| // Check channels are closed | ||
| _, ok := <-response | ||
| assert.False(t, ok) | ||
| _, ok = <-ch | ||
| assert.False(t, ok) | ||
| } |
Oops, something went wrong.