[VOL-1997] Remove transaction timeout for a non-active rw_core
This commit cleans up the transaction processing between two
cores in a pair. It prevents the core not processing the request
to grab the request based on a timeout only.
Since this update heavily relies on the etcd mechanism then customized
local tests (not unit as could not find a full-featured etcd mock)
were run against it as well as some basic manual tests with
kind-voltha.
There is a TODO item in this commit to implement a peer-probe
mechanism to guarantee that a core in a pair has actually died
before a switch over is done.
Minor updates after first review.
Comments updates after second review
Change-Id: Ifc1442471595a979b39251535b8ee9210e1a52df
(cherry picked from commit cc40904e208892dea8e1a2a73b52e6465d3c6d59)
diff --git a/db/kvstore/etcdclient.go b/db/kvstore/etcdclient.go
index 88c13ae..3af1ef2 100644
--- a/db/kvstore/etcdclient.go
+++ b/db/kvstore/etcdclient.go
@@ -127,7 +127,15 @@
c.writeLock.Lock()
defer c.writeLock.Unlock()
- _, err := c.ectdAPI.Put(ctx, key, val)
+
+ var err error
+ // Check if there is already a lease for this key - if there is then use it, otherwise a PUT will make
+ // that KV key permanent instead of automatically removing it after a lease expiration
+ if leaseID, ok := c.keyReservations[key]; ok {
+ _, err = c.ectdAPI.Put(ctx, key, val, v3Client.WithLease(*leaseID))
+ } else {
+ _, err = c.ectdAPI.Put(ctx, key, val)
+ }
cancel()
if err != nil {
switch err {
@@ -158,8 +166,8 @@
c.writeLock.Lock()
defer c.writeLock.Unlock()
- // delete the keys
- if _, err := c.ectdAPI.Delete(ctx, key, v3Client.WithPrefix()); err != nil {
+ // delete the key
+ if _, err := c.ectdAPI.Delete(ctx, key); err != nil {
log.Errorw("failed-to-delete-key", log.Fields{"key": key, "error": err})
return err
}
@@ -309,7 +317,7 @@
func (c *EtcdClient) Watch(key string) chan *Event {
w := v3Client.NewWatcher(c.ectdAPI)
ctx, cancel := context.WithCancel(context.Background())
- channel := w.Watch(ctx, key, v3Client.WithPrefix())
+ channel := w.Watch(ctx, key)
// Create a new channel
ch := make(chan *Event, maxClientChannelBufferSize)
@@ -317,8 +325,6 @@
// Keep track of the created channels so they can be closed when required
channelMap := make(map[chan *Event]v3Client.Watcher)
channelMap[ch] = w
- //c.writeLock.Lock()
- //defer c.writeLock.Unlock()
channelMaps := c.addChannelMap(key, channelMap)
@@ -412,7 +418,6 @@
defer close(ch)
for resp := range channel {
for _, ev := range resp.Events {
- //log.Debugf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
ch <- NewEvent(getEventType(ev), ev.Kv.Key, ev.Kv.Value, ev.Kv.Version)
}
}