[VOL-3187]Pass Context down the execution call hierarchy across ofagent codebase
Change-Id: Ia5f2fa1509beefe0ddc427b83e39d2702782db8f
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go
index 1e23a0f..f595dc1 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go
@@ -47,7 +47,7 @@
}
// NewBackend creates a new instance of a Backend structure
-func NewBackend(storeType string, address string, timeout time.Duration, pathPrefix string) *Backend {
+func NewBackend(ctx context.Context, storeType string, address string, timeout time.Duration, pathPrefix string) *Backend {
var err error
b := &Backend{
@@ -59,8 +59,8 @@
alive: false, // connection considered down at start
}
- if b.Client, err = b.newClient(address, timeout); err != nil {
- logger.Errorw("failed-to-create-kv-client",
+ if b.Client, err = b.newClient(ctx, address, timeout); err != nil {
+ logger.Errorw(ctx, "failed-to-create-kv-client",
log.Fields{
"type": storeType, "address": address,
"timeout": timeout, "prefix": pathPrefix,
@@ -71,22 +71,22 @@
return b
}
-func (b *Backend) newClient(address string, timeout time.Duration) (kvstore.Client, error) {
+func (b *Backend) newClient(ctx context.Context, address string, timeout time.Duration) (kvstore.Client, error) {
switch b.StoreType {
case "consul":
- return kvstore.NewConsulClient(address, timeout)
+ return kvstore.NewConsulClient(ctx, address, timeout)
case "etcd":
- return kvstore.NewEtcdClient(address, timeout, log.WarnLevel)
+ return kvstore.NewEtcdClient(ctx, address, timeout, log.WarnLevel)
}
return nil, errors.New("unsupported-kv-store")
}
-func (b *Backend) makePath(key string) string {
+func (b *Backend) makePath(ctx context.Context, key string) string {
path := fmt.Sprintf("%s/%s", b.PathPrefix, key)
return path
}
-func (b *Backend) updateLiveness(alive bool) {
+func (b *Backend) updateLiveness(ctx context.Context, alive bool) {
// Periodically push stream of liveness data to the channel,
// so that in a live state, the core does not timeout and
// send a forced liveness message. Push alive state if the
@@ -94,11 +94,11 @@
if b.liveness != nil {
if b.alive != alive {
- logger.Debug("update-liveness-channel-reason-change")
+ logger.Debug(ctx, "update-liveness-channel-reason-change")
b.liveness <- alive
b.lastLivenessTime = time.Now()
} else if time.Since(b.lastLivenessTime) > b.LivenessChannelInterval {
- logger.Debug("update-liveness-channel-reason-interval")
+ logger.Debug(ctx, "update-liveness-channel-reason-interval")
b.liveness <- alive
b.lastLivenessTime = time.Now()
}
@@ -106,7 +106,7 @@
// Emit log message only for alive state change
if b.alive != alive {
- logger.Debugw("change-kvstore-alive-status", log.Fields{"alive": alive})
+ logger.Debugw(ctx, "change-kvstore-alive-status", log.Fields{"alive": alive})
b.alive = alive
}
}
@@ -115,9 +115,9 @@
// post on Liveness channel
func (b *Backend) PerformLivenessCheck(ctx context.Context) bool {
alive := b.Client.IsConnectionUp(ctx)
- logger.Debugw("kvstore-liveness-check-result", log.Fields{"alive": alive})
+ logger.Debugw(ctx, "kvstore-liveness-check-result", log.Fields{"alive": alive})
- b.updateLiveness(alive)
+ b.updateLiveness(ctx, alive)
return alive
}
@@ -126,11 +126,11 @@
// or not the connection is still Live. This channel is then picked up
// by the service (i.e. rw_core / ro_core) to update readiness status
// and/or take other actions.
-func (b *Backend) EnableLivenessChannel() chan bool {
- logger.Debug("enable-kvstore-liveness-channel")
+func (b *Backend) EnableLivenessChannel(ctx context.Context) chan bool {
+ logger.Debug(ctx, "enable-kvstore-liveness-channel")
if b.liveness == nil {
- logger.Debug("create-kvstore-liveness-channel")
+ logger.Debug(ctx, "create-kvstore-liveness-channel")
// Channel size of 10 to avoid any possibility of blocking in Load conditions
b.liveness = make(chan bool, 10)
@@ -144,7 +144,7 @@
}
// Extract Alive status of Kvstore based on type of error
-func (b *Backend) isErrorIndicatingAliveKvstore(err error) bool {
+func (b *Backend) isErrorIndicatingAliveKvstore(ctx context.Context, err error) bool {
// Alive unless observed an error indicating so
alive := true
@@ -182,64 +182,64 @@
// List retrieves one or more items that match the specified key
func (b *Backend) List(ctx context.Context, key string) (map[string]*kvstore.KVPair, error) {
- formattedPath := b.makePath(key)
- logger.Debugw("listing-key", log.Fields{"key": key, "path": formattedPath})
+ formattedPath := b.makePath(ctx, key)
+ logger.Debugw(ctx, "listing-key", log.Fields{"key": key, "path": formattedPath})
pair, err := b.Client.List(ctx, formattedPath)
- b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
+ b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(ctx, err))
return pair, err
}
// Get retrieves an item that matches the specified key
func (b *Backend) Get(ctx context.Context, key string) (*kvstore.KVPair, error) {
- formattedPath := b.makePath(key)
- logger.Debugw("getting-key", log.Fields{"key": key, "path": formattedPath})
+ formattedPath := b.makePath(ctx, key)
+ logger.Debugw(ctx, "getting-key", log.Fields{"key": key, "path": formattedPath})
pair, err := b.Client.Get(ctx, formattedPath)
- b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
+ b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(ctx, err))
return pair, err
}
// Put stores an item value under the specifed key
func (b *Backend) Put(ctx context.Context, key string, value interface{}) error {
- formattedPath := b.makePath(key)
- logger.Debugw("putting-key", log.Fields{"key": key, "path": formattedPath})
+ formattedPath := b.makePath(ctx, key)
+ logger.Debugw(ctx, "putting-key", log.Fields{"key": key, "path": formattedPath})
err := b.Client.Put(ctx, formattedPath, value)
- b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
+ b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(ctx, err))
return err
}
// Delete removes an item under the specified key
func (b *Backend) Delete(ctx context.Context, key string) error {
- formattedPath := b.makePath(key)
- logger.Debugw("deleting-key", log.Fields{"key": key, "path": formattedPath})
+ formattedPath := b.makePath(ctx, key)
+ logger.Debugw(ctx, "deleting-key", log.Fields{"key": key, "path": formattedPath})
err := b.Client.Delete(ctx, formattedPath)
- b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
+ b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(ctx, err))
return err
}
// CreateWatch starts watching events for the specified key
func (b *Backend) CreateWatch(ctx context.Context, key string, withPrefix bool) chan *kvstore.Event {
- formattedPath := b.makePath(key)
- logger.Debugw("creating-key-watch", log.Fields{"key": key, "path": formattedPath})
+ formattedPath := b.makePath(ctx, key)
+ logger.Debugw(ctx, "creating-key-watch", log.Fields{"key": key, "path": formattedPath})
return b.Client.Watch(ctx, formattedPath, withPrefix)
}
// DeleteWatch stops watching events for the specified key
-func (b *Backend) DeleteWatch(key string, ch chan *kvstore.Event) {
- formattedPath := b.makePath(key)
- logger.Debugw("deleting-key-watch", log.Fields{"key": key, "path": formattedPath})
+func (b *Backend) DeleteWatch(ctx context.Context, key string, ch chan *kvstore.Event) {
+ formattedPath := b.makePath(ctx, key)
+ logger.Debugw(ctx, "deleting-key-watch", log.Fields{"key": key, "path": formattedPath})
- b.Client.CloseWatch(formattedPath, ch)
+ b.Client.CloseWatch(ctx, formattedPath, ch)
}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/common.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/common.go
index 1cf2e1c..fe84b46 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/common.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/common.go
@@ -19,12 +19,12 @@
"github.com/opencord/voltha-lib-go/v3/pkg/log"
)
-var logger log.Logger
+var logger log.CLogger
func init() {
// Setup this package so that it's log level can be modified at run time
var err error
- logger, err = log.AddPackage(log.JSON, log.ErrorLevel, log.Fields{"pkg": "db"})
+ logger, err = log.RegisterPackage(log.JSON, log.ErrorLevel, log.Fields{"pkg": "db"})
if err != nil {
panic(err)
}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/client.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/client.go
index 158e626..480d476 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/client.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/client.go
@@ -88,6 +88,6 @@
AcquireLock(ctx context.Context, lockName string, timeout time.Duration) error
ReleaseLock(lockName string) error
IsConnectionUp(ctx context.Context) bool // timeout in second
- CloseWatch(key string, ch chan *Event)
- Close()
+ CloseWatch(ctx context.Context, key string, ch chan *Event)
+ Close(ctx context.Context)
}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/common.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/common.go
index aa7aeb0..0de395f 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/common.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/common.go
@@ -19,12 +19,12 @@
"github.com/opencord/voltha-lib-go/v3/pkg/log"
)
-var logger log.Logger
+var logger log.CLogger
func init() {
// Setup this package so that it's log level can be modified at run time
var err error
- logger, err = log.AddPackage(log.JSON, log.ErrorLevel, log.Fields{"pkg": "kvstore"})
+ logger, err = log.RegisterPackage(log.JSON, log.ErrorLevel, log.Fields{"pkg": "kvstore"})
if err != nil {
panic(err)
}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/consulclient.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/consulclient.go
index d2544dd..c2cd841 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/consulclient.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/consulclient.go
@@ -44,14 +44,13 @@
}
// NewConsulClient returns a new client for the Consul KV store
-func NewConsulClient(addr string, timeout time.Duration) (*ConsulClient, error) {
-
+func NewConsulClient(ctx context.Context, addr string, timeout time.Duration) (*ConsulClient, error) {
config := consulapi.DefaultConfig()
config.Address = addr
config.WaitTime = timeout
consul, err := consulapi.NewClient(config)
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return nil, err
}
@@ -63,7 +62,7 @@
// IsConnectionUp returns whether the connection to the Consul KV store is up
func (c *ConsulClient) IsConnectionUp(ctx context.Context) bool {
- logger.Error("Unimplemented function")
+ logger.Error(ctx, "Unimplemented function")
return false
}
@@ -80,7 +79,7 @@
// For now we ignore meta data
kvps, _, err := kv.List(key, &queryOptions)
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return nil, err
}
m := make(map[string]*KVPair)
@@ -103,7 +102,7 @@
// For now we ignore meta data
kvp, _, err := kv.Get(key, &queryOptions)
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return nil, err
}
if kvp != nil {
@@ -122,7 +121,7 @@
var val []byte
var er error
if val, er = ToByte(value); er != nil {
- logger.Error(er)
+ logger.Error(ctx, er)
return er
}
@@ -134,7 +133,7 @@
defer c.writeLock.Unlock()
_, err := kv.Put(&kvp, &writeOptions)
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return err
}
return nil
@@ -149,26 +148,26 @@
defer c.writeLock.Unlock()
_, err := kv.Delete(key, &writeOptions)
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return err
}
return nil
}
-func (c *ConsulClient) deleteSession() {
+func (c *ConsulClient) deleteSession(ctx context.Context) {
if c.sessionID != "" {
- logger.Debug("cleaning-up-session")
+ logger.Debug(ctx, "cleaning-up-session")
session := c.consul.Session()
_, err := session.Destroy(c.sessionID, nil)
if err != nil {
- logger.Errorw("error-cleaning-session", log.Fields{"session": c.sessionID, "error": err})
+ logger.Errorw(ctx, "error-cleaning-session", log.Fields{"session": c.sessionID, "error": err})
}
}
c.sessionID = ""
c.session = nil
}
-func (c *ConsulClient) createSession(ttl time.Duration, retries int) (*consulapi.Session, string, error) {
+func (c *ConsulClient) createSession(ctx context.Context, ttl time.Duration, retries int) (*consulapi.Session, string, error) {
session := c.consul.Session()
entry := &consulapi.SessionEntry{
Behavior: consulapi.SessionBehaviorDelete,
@@ -178,17 +177,17 @@
for {
id, meta, err := session.Create(entry, nil)
if err != nil {
- logger.Errorw("create-session-error", log.Fields{"error": err})
+ logger.Errorw(ctx, "create-session-error", log.Fields{"error": err})
if retries == 0 {
return nil, "", err
}
} else if meta.RequestTime == 0 {
- logger.Errorw("create-session-bad-meta-data", log.Fields{"meta-data": meta})
+ logger.Errorw(ctx, "create-session-bad-meta-data", log.Fields{"meta-data": meta})
if retries == 0 {
return nil, "", errors.New("bad-meta-data")
}
} else if id == "" {
- logger.Error("create-session-nil-id")
+ logger.Error(ctx, "create-session-nil-id")
if retries == 0 {
return nil, "", errors.New("ID-nil")
}
@@ -199,7 +198,7 @@
if retries > 0 {
retries--
}
- logger.Debug("retrying-session-create-after-a-second-delay")
+ logger.Debug(ctx, "retrying-session-create-after-a-second-delay")
time.Sleep(time.Duration(1) * time.Second)
}
}
@@ -226,30 +225,30 @@
var val []byte
var er error
if val, er = ToByte(value); er != nil {
- logger.Error(er)
+ logger.Error(ctx, er)
return nil, er
}
// Cleanup any existing session and recreate new ones. A key is reserved against a session
if c.sessionID != "" {
- c.deleteSession()
+ c.deleteSession(ctx)
}
// Clear session if reservation is not successful
reservationSuccessful := false
defer func() {
if !reservationSuccessful {
- logger.Debug("deleting-session")
- c.deleteSession()
+ logger.Debug(ctx, "deleting-session")
+ c.deleteSession(ctx)
}
}()
- session, sessionID, err := c.createSession(ttl, -1)
+ session, sessionID, err := c.createSession(ctx, ttl, -1)
if err != nil {
- logger.Errorw("no-session-created", log.Fields{"error": err})
+ logger.Errorw(ctx, "no-session-created", log.Fields{"error": err})
return "", errors.New("no-session-created")
}
- logger.Debugw("session-created", log.Fields{"session-id": sessionID})
+ logger.Debugw(ctx, "session-created", log.Fields{"session-id": sessionID})
c.sessionID = sessionID
c.session = session
@@ -258,11 +257,11 @@
kvp := consulapi.KVPair{Key: key, Value: val, Session: c.sessionID}
result, _, err := kv.Acquire(&kvp, nil)
if err != nil {
- logger.Errorw("error-acquiring-keys", log.Fields{"error": err})
+ logger.Errorw(ctx, "error-acquiring-keys", log.Fields{"error": err})
return nil, err
}
- logger.Debugw("key-acquired", log.Fields{"key": key, "status": result})
+ logger.Debugw(ctx, "key-acquired", log.Fields{"key": key, "status": result})
// Irrespective whether we were successful in acquiring the key, let's read it back and see if it's us.
m, err := c.Get(ctx, key)
@@ -270,7 +269,7 @@
return nil, err
}
if m != nil {
- logger.Debugw("response-received", log.Fields{"key": m.Key, "m.value": string(m.Value.([]byte)), "value": value})
+ logger.Debugw(ctx, "response-received", log.Fields{"key": m.Key, "m.value": string(m.Value.([]byte)), "value": value})
if m.Key == key && isEqual(m.Value, value) {
// My reservation is successful - register it. For now, support is only for 1 reservation per key
// per session.
@@ -300,11 +299,11 @@
kvp = consulapi.KVPair{Key: key, Value: value.([]byte), Session: c.sessionID}
result, _, err = kv.Release(&kvp, nil)
if err != nil {
- logger.Errorw("cannot-release-reservation", log.Fields{"key": key, "error": err})
+ logger.Errorw(ctx, "cannot-release-reservation", log.Fields{"key": key, "error": err})
return err
}
if !result {
- logger.Errorw("cannot-release-reservation", log.Fields{"key": key})
+ logger.Errorw(ctx, "cannot-release-reservation", log.Fields{"key": key})
}
delete(c.keyReservations, key)
}
@@ -384,21 +383,21 @@
// CloseWatch closes a specific watch. Both the key and the channel are required when closing a watch as there
// may be multiple listeners on the same key. The previously created channel serves as a key
-func (c *ConsulClient) CloseWatch(key string, ch chan *Event) {
+func (c *ConsulClient) CloseWatch(ctx context.Context, key string, ch chan *Event) {
// First close the context
var ok bool
var watchedChannelsContexts []*channelContextMap
c.writeLock.Lock()
defer c.writeLock.Unlock()
if watchedChannelsContexts, ok = c.watchedChannelsContext[key]; !ok {
- logger.Errorw("key-has-no-watched-context-or-channel", log.Fields{"key": key})
+ logger.Errorw(ctx, "key-has-no-watched-context-or-channel", log.Fields{"key": key})
return
}
// Look for the channels
var pos = -1
for i, chCtxMap := range watchedChannelsContexts {
if chCtxMap.channel == ch {
- logger.Debug("channel-found")
+ logger.Debug(ctx, "channel-found")
chCtxMap.cancel()
//close the channel
close(ch)
@@ -410,7 +409,7 @@
if pos >= 0 {
c.watchedChannelsContext[key] = append(c.watchedChannelsContext[key][:pos], c.watchedChannelsContext[key][pos+1:]...)
}
- logger.Debugw("watched-channel-exiting", log.Fields{"key": key, "channel": c.watchedChannelsContext[key]})
+ logger.Debugw(ctx, "watched-channel-exiting", log.Fields{"key": key, "channel": c.watchedChannelsContext[key]})
}
func (c *ConsulClient) isKVEqual(kv1 *consulapi.KVPair, kv2 *consulapi.KVPair) bool {
@@ -430,10 +429,10 @@
return true
}
-func (c *ConsulClient) listenForKeyChange(watchContext context.Context, key string, ch chan *Event) {
- logger.Debugw("start-watching-channel", log.Fields{"key": key, "channel": ch})
+func (c *ConsulClient) listenForKeyChange(ctx context.Context, key string, ch chan *Event) {
+ logger.Debugw(ctx, "start-watching-channel", log.Fields{"key": key, "channel": ch})
- defer c.CloseWatch(key, ch)
+ defer c.CloseWatch(ctx, key, ch)
kv := c.consul.KV()
var queryOptions consulapi.QueryOptions
queryOptions.WaitTime = defaultKVGetTimeout
@@ -441,7 +440,7 @@
// Get the existing value, if any
previousKVPair, meta, err := kv.Get(key, &queryOptions)
if err != nil {
- logger.Debug(err)
+ logger.Debug(ctx, err)
}
lastIndex := meta.LastIndex
@@ -449,37 +448,37 @@
//var waitOptions consulapi.QueryOptions
var pair *consulapi.KVPair
//watchContext, _ := context.WithCancel(context.Background())
- waitOptions := queryOptions.WithContext(watchContext)
+ waitOptions := queryOptions.WithContext(ctx)
for {
//waitOptions = consulapi.QueryOptions{WaitIndex: lastIndex}
waitOptions.WaitIndex = lastIndex
pair, meta, err = kv.Get(key, waitOptions)
select {
- case <-watchContext.Done():
- logger.Debug("done-event-received-exiting")
+ case <-ctx.Done():
+ logger.Debug(ctx, "done-event-received-exiting")
return
default:
if err != nil {
- logger.Warnw("error-from-watch", log.Fields{"error": err})
+ logger.Warnw(ctx, "error-from-watch", log.Fields{"error": err})
ch <- NewEvent(CONNECTIONDOWN, key, []byte(""), -1)
} else {
- logger.Debugw("index-state", log.Fields{"lastindex": lastIndex, "newindex": meta.LastIndex, "key": key})
+ logger.Debugw(ctx, "index-state", log.Fields{"lastindex": lastIndex, "newindex": meta.LastIndex, "key": key})
}
}
if err != nil {
- logger.Debug(err)
+ logger.Debug(ctx, err)
// On error, block for 10 milliseconds to prevent endless loop
time.Sleep(10 * time.Millisecond)
} else if meta.LastIndex <= lastIndex {
- logger.Info("no-index-change-or-negative")
+ logger.Info(ctx, "no-index-change-or-negative")
} else {
- logger.Debugw("update-received", log.Fields{"pair": pair})
+ logger.Debugw(ctx, "update-received", log.Fields{"pair": pair})
if pair == nil {
ch <- NewEvent(DELETE, key, []byte(""), -1)
} else if !c.isKVEqual(pair, previousKVPair) {
// Push the change onto the channel if the data has changed
// For now just assume it's a PUT change
- logger.Debugw("pair-details", log.Fields{"session": pair.Session, "key": pair.Key, "value": pair.Value})
+ logger.Debugw(ctx, "pair-details", log.Fields{"session": pair.Session, "key": pair.Key, "value": pair.Value})
ch <- NewEvent(PUT, pair.Key, pair.Value, -1)
}
previousKVPair = pair
@@ -489,7 +488,7 @@
}
// Close closes the KV store client
-func (c *ConsulClient) Close() {
+func (c *ConsulClient) Close(ctx context.Context) {
var writeOptions consulapi.WriteOptions
// Inform any goroutine it's time to say goodbye.
c.writeLock.Lock()
@@ -500,7 +499,7 @@
// Clear the sessionID
if _, err := c.consul.Session().Destroy(c.sessionID, &writeOptions); err != nil {
- logger.Errorw("error-closing-client", log.Fields{"error": err})
+ logger.Errorw(ctx, "error-closing-client", log.Fields{"error": err})
}
}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go
index 8d4a462..0165e18 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go
@@ -40,7 +40,7 @@
}
// NewEtcdClient returns a new client for the Etcd KV store
-func NewEtcdClient(addr string, timeout time.Duration, level log.LogLevel) (*EtcdClient, error) {
+func NewEtcdClient(ctx context.Context, addr string, timeout time.Duration, level log.LogLevel) (*EtcdClient, error) {
logconfig := log.ConstructZapConfig(log.JSON, level, log.Fields{})
c, err := v3Client.New(v3Client.Config{
@@ -49,7 +49,7 @@
LogConfig: &logconfig,
})
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return nil, err
}
@@ -77,7 +77,7 @@
func (c *EtcdClient) List(ctx context.Context, key string) (map[string]*KVPair, error) {
resp, err := c.ectdAPI.Get(ctx, key, v3Client.WithPrefix())
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return nil, err
}
m := make(map[string]*KVPair)
@@ -94,7 +94,7 @@
resp, err := c.ectdAPI.Get(ctx, key)
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return nil, err
}
for _, ev := range resp.Kvs {
@@ -131,13 +131,13 @@
if err != nil {
switch err {
case context.Canceled:
- logger.Warnw("context-cancelled", log.Fields{"error": err})
+ logger.Warnw(ctx, "context-cancelled", log.Fields{"error": err})
case context.DeadlineExceeded:
- logger.Warnw("context-deadline-exceeded", log.Fields{"error": err})
+ logger.Warnw(ctx, "context-deadline-exceeded", log.Fields{"error": err})
case v3rpcTypes.ErrEmptyKey:
- logger.Warnw("etcd-client-error", log.Fields{"error": err})
+ logger.Warnw(ctx, "etcd-client-error", log.Fields{"error": err})
default:
- logger.Warnw("bad-endpoints", log.Fields{"error": err})
+ logger.Warnw(ctx, "bad-endpoints", log.Fields{"error": err})
}
return err
}
@@ -150,10 +150,10 @@
// delete the key
if _, err := c.ectdAPI.Delete(ctx, key); err != nil {
- logger.Errorw("failed-to-delete-key", log.Fields{"key": key, "error": err})
+ logger.Errorw(ctx, "failed-to-delete-key", log.Fields{"key": key, "error": err})
return err
}
- logger.Debugw("key(s)-deleted", log.Fields{"key": key})
+ logger.Debugw(ctx, "key(s)-deleted", log.Fields{"key": key})
return nil
}
@@ -172,7 +172,7 @@
resp, err := c.ectdAPI.Grant(ctx, int64(ttl.Seconds()))
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return nil, err
}
// Register the lease id
@@ -185,7 +185,7 @@
defer func() {
if !reservationSuccessful {
if err = c.ReleaseReservation(context.Background(), key); err != nil {
- logger.Error("cannot-release-lease")
+ logger.Error(ctx, "cannot-release-lease")
}
}
}()
@@ -240,7 +240,7 @@
for key, leaseID := range c.keyReservations {
_, err := c.ectdAPI.Revoke(ctx, *leaseID)
if err != nil {
- logger.Errorw("cannot-release-reservation", log.Fields{"key": key, "error": err})
+ logger.Errorw(ctx, "cannot-release-reservation", log.Fields{"key": key, "error": err})
return err
}
delete(c.keyReservations, key)
@@ -251,7 +251,7 @@
// ReleaseReservation releases reservation for a specific key.
func (c *EtcdClient) ReleaseReservation(ctx context.Context, key string) error {
// Get the leaseid using the key
- logger.Debugw("Release-reservation", log.Fields{"key": key})
+ logger.Debugw(ctx, "Release-reservation", log.Fields{"key": key})
var ok bool
var leaseID *v3Client.LeaseID
c.keyReservationsLock.Lock()
@@ -263,7 +263,7 @@
if leaseID != nil {
_, err := c.ectdAPI.Revoke(ctx, *leaseID)
if err != nil {
- logger.Error(err)
+ logger.Error(ctx, err)
return err
}
delete(c.keyReservations, key)
@@ -288,7 +288,7 @@
if leaseID != nil {
_, err := c.ectdAPI.KeepAliveOnce(ctx, *leaseID)
if err != nil {
- logger.Errorw("lease-may-have-expired", log.Fields{"error": err})
+ logger.Errorw(ctx, "lease-may-have-expired", log.Fields{"error": err})
return err
}
} else {
@@ -320,9 +320,9 @@
// Changing the log field (from channelMaps) as the underlying logger cannot format the map of channels into a
// json format.
- logger.Debugw("watched-channels", log.Fields{"len": len(channelMaps)})
+ logger.Debugw(ctx, "watched-channels", log.Fields{"len": len(channelMaps)})
// Launch a go routine to listen for updates
- go c.listenForKeyChange(channel, ch, cancel)
+ go c.listenForKeyChange(ctx, channel, ch, cancel)
return ch
@@ -369,23 +369,23 @@
// CloseWatch closes a specific watch. Both the key and the channel are required when closing a watch as there
// may be multiple listeners on the same key. The previously created channel serves as a key
-func (c *EtcdClient) CloseWatch(key string, ch chan *Event) {
+func (c *EtcdClient) CloseWatch(ctx context.Context, key string, ch chan *Event) {
// Get the array of channels mapping
var watchedChannels []map[chan *Event]v3Client.Watcher
var ok bool
if watchedChannels, ok = c.getChannelMaps(key); !ok {
- logger.Warnw("key-has-no-watched-channels", log.Fields{"key": key})
+ logger.Warnw(ctx, "key-has-no-watched-channels", log.Fields{"key": key})
return
}
// Look for the channels
var pos = -1
for i, chMap := range watchedChannels {
if t, ok := chMap[ch]; ok {
- logger.Debug("channel-found")
+ logger.Debug(ctx, "channel-found")
// Close the etcd watcher before the client channel. This should close the etcd channel as well
if err := t.Close(); err != nil {
- logger.Errorw("watcher-cannot-be-closed", log.Fields{"key": key, "error": err})
+ logger.Errorw(ctx, "watcher-cannot-be-closed", log.Fields{"key": key, "error": err})
}
pos = i
break
@@ -397,11 +397,11 @@
if pos >= 0 {
channelMaps = c.removeChannelMap(key, pos)
}
- logger.Infow("watcher-channel-exiting", log.Fields{"key": key, "channel": channelMaps})
+ logger.Infow(ctx, "watcher-channel-exiting", log.Fields{"key": key, "channel": channelMaps})
}
-func (c *EtcdClient) listenForKeyChange(channel v3Client.WatchChan, ch chan<- *Event, cancel context.CancelFunc) {
- logger.Debug("start-listening-on-channel ...")
+func (c *EtcdClient) listenForKeyChange(ctx context.Context, channel v3Client.WatchChan, ch chan<- *Event, cancel context.CancelFunc) {
+ logger.Debug(ctx, "start-listening-on-channel ...")
defer cancel()
defer close(ch)
for resp := range channel {
@@ -409,7 +409,7 @@
ch <- NewEvent(getEventType(ev), ev.Kv.Key, ev.Kv.Value, ev.Kv.Version)
}
}
- logger.Debug("stop-listening-on-channel ...")
+ logger.Debug(ctx, "stop-listening-on-channel ...")
}
func getEventType(event *v3Client.Event) int {
@@ -423,9 +423,9 @@
}
// Close closes the KV store client
-func (c *EtcdClient) Close() {
+func (c *EtcdClient) Close(ctx context.Context) {
if err := c.ectdAPI.Close(); err != nil {
- logger.Errorw("error-closing-client", log.Fields{"error": err})
+ logger.Errorw(ctx, "error-closing-client", log.Fields{"error": err})
}
}