blob: 887a167a6066164180c44ea66483f1aa0f8c8c0a [file] [log] [blame]
khenaidoobf6e7bb2018-08-14 22:27:29 -04001/*
2 * Copyright 2018-present Open Networking Foundation
3
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7
8 * http://www.apache.org/licenses/LICENSE-2.0
9
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
khenaidoocfee5f42018-07-19 22:47:38 -040016package kvstore
17
18import (
khenaidoocfee5f42018-07-19 22:47:38 -040019 "context"
20 "errors"
khenaidoocfee5f42018-07-19 22:47:38 -040021 "fmt"
Stephane Barbarie260a5632019-02-26 16:12:49 -050022 "github.com/opencord/voltha-go/common/log"
khenaidoob9203542018-09-17 22:56:37 -040023 v3Client "go.etcd.io/etcd/clientv3"
Stephane Barbarie260a5632019-02-26 16:12:49 -050024 v3Concurrency "go.etcd.io/etcd/clientv3/concurrency"
khenaidoob9203542018-09-17 22:56:37 -040025 v3rpcTypes "go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
khenaidoo5c11af72018-07-20 17:21:05 -040026 "sync"
khenaidoocfee5f42018-07-19 22:47:38 -040027)
28
29// EtcdClient represents the Etcd KV store client
30type EtcdClient struct {
Stephane Barbariec53a2752019-03-08 17:50:10 -050031 ectdAPI *v3Client.Client
32 leaderRev v3Client.Client
33 keyReservations map[string]*v3Client.LeaseID
34 watchedChannels sync.Map
35 writeLock sync.Mutex
36 lockToMutexMap map[string]*v3Concurrency.Mutex
khenaidoobdcb8e02019-03-06 16:28:56 -050037 lockToSessionMap map[string]*v3Concurrency.Session
Stephane Barbariec53a2752019-03-08 17:50:10 -050038 lockToMutexLock sync.Mutex
khenaidoocfee5f42018-07-19 22:47:38 -040039}
40
41// NewEtcdClient returns a new client for the Etcd KV store
42func NewEtcdClient(addr string, timeout int) (*EtcdClient, error) {
khenaidoocfee5f42018-07-19 22:47:38 -040043 duration := GetDuration(timeout)
44
45 c, err := v3Client.New(v3Client.Config{
46 Endpoints: []string{addr},
47 DialTimeout: duration,
48 })
49 if err != nil {
50 log.Error(err)
51 return nil, err
52 }
Stephane Barbariec53a2752019-03-08 17:50:10 -050053
khenaidoocfee5f42018-07-19 22:47:38 -040054 reservations := make(map[string]*v3Client.LeaseID)
khenaidoobdcb8e02019-03-06 16:28:56 -050055 lockMutexMap := make(map[string]*v3Concurrency.Mutex)
56 lockSessionMap := make(map[string]*v3Concurrency.Session)
Stephane Barbarie260a5632019-02-26 16:12:49 -050057
Stephane Barbariec53a2752019-03-08 17:50:10 -050058 return &EtcdClient{ectdAPI: c, keyReservations: reservations, lockToMutexMap: lockMutexMap,
59 lockToSessionMap: lockSessionMap}, nil
khenaidoocfee5f42018-07-19 22:47:38 -040060}
61
62// List returns an array of key-value pairs with key as a prefix. Timeout defines how long the function will
63// wait for a response
Stephane Barbarie260a5632019-02-26 16:12:49 -050064func (c *EtcdClient) List(key string, timeout int, lock ...bool) (map[string]*KVPair, error) {
khenaidoocfee5f42018-07-19 22:47:38 -040065 duration := GetDuration(timeout)
66
67 ctx, cancel := context.WithTimeout(context.Background(), duration)
Stephane Barbarie260a5632019-02-26 16:12:49 -050068
69 // DO NOT lock by default; otherwise lock per instructed value
70 if len(lock) > 0 && lock[0] {
71 session, _ := v3Concurrency.NewSession(c.ectdAPI, v3Concurrency.WithContext(ctx))
Stephane Barbariec53a2752019-03-08 17:50:10 -050072 mu := v3Concurrency.NewMutex(session, "/lock"+key)
Stephane Barbarie260a5632019-02-26 16:12:49 -050073 mu.Lock(context.Background())
74 defer mu.Unlock(context.Background())
75 defer session.Close()
76 }
77
khenaidoocfee5f42018-07-19 22:47:38 -040078 resp, err := c.ectdAPI.Get(ctx, key, v3Client.WithPrefix())
79 cancel()
80 if err != nil {
81 log.Error(err)
82 return nil, err
83 }
84 m := make(map[string]*KVPair)
85 for _, ev := range resp.Kvs {
86 m[string(ev.Key)] = NewKVPair(string(ev.Key), ev.Value, "", ev.Lease)
87 }
88 return m, nil
89}
90
91// Get returns a key-value pair for a given key. Timeout defines how long the function will
92// wait for a response
Stephane Barbarie260a5632019-02-26 16:12:49 -050093func (c *EtcdClient) Get(key string, timeout int, lock ...bool) (*KVPair, error) {
khenaidoocfee5f42018-07-19 22:47:38 -040094 duration := GetDuration(timeout)
95
96 ctx, cancel := context.WithTimeout(context.Background(), duration)
Stephane Barbarie260a5632019-02-26 16:12:49 -050097
98 // Lock by default; otherwise lock per instructed value
Stephane Barbariec53a2752019-03-08 17:50:10 -050099 if len(lock) > 0 && lock[0] {
Stephane Barbarie260a5632019-02-26 16:12:49 -0500100 session, _ := v3Concurrency.NewSession(c.ectdAPI, v3Concurrency.WithContext(ctx))
Stephane Barbariec53a2752019-03-08 17:50:10 -0500101 mu := v3Concurrency.NewMutex(session, "/lock"+key)
Stephane Barbarie260a5632019-02-26 16:12:49 -0500102 mu.Lock(context.Background())
103 defer mu.Unlock(context.Background())
104 defer session.Close()
105 }
106
khenaidoocfee5f42018-07-19 22:47:38 -0400107 resp, err := c.ectdAPI.Get(ctx, key)
108 cancel()
109 if err != nil {
110 log.Error(err)
111 return nil, err
112 }
113 for _, ev := range resp.Kvs {
114 // Only one value is returned
115 return NewKVPair(string(ev.Key), ev.Value, "", ev.Lease), nil
116 }
117 return nil, nil
118}
119
120// Put writes a key-value pair to the KV store. Value can only be a string or []byte since the etcd API
121// accepts only a string as a value for a put operation. Timeout defines how long the function will
122// wait for a response
Stephane Barbarie260a5632019-02-26 16:12:49 -0500123func (c *EtcdClient) Put(key string, value interface{}, timeout int, lock ...bool) error {
khenaidoocfee5f42018-07-19 22:47:38 -0400124
125 // Validate that we can convert value to a string as etcd API expects a string
126 var val string
127 var er error
128 if val, er = ToString(value); er != nil {
129 return fmt.Errorf("unexpected-type-%T", value)
130 }
131
132 duration := GetDuration(timeout)
133
134 ctx, cancel := context.WithTimeout(context.Background(), duration)
Stephane Barbarie260a5632019-02-26 16:12:49 -0500135
136 // Lock by default; otherwise lock per instructed value
137 if len(lock) == 0 || lock[0] {
138 session, _ := v3Concurrency.NewSession(c.ectdAPI, v3Concurrency.WithContext(ctx))
Stephane Barbariec53a2752019-03-08 17:50:10 -0500139 mu := v3Concurrency.NewMutex(session, "/lock"+key)
Stephane Barbarie260a5632019-02-26 16:12:49 -0500140 mu.Lock(context.Background())
141 defer mu.Unlock(context.Background())
142 defer session.Close()
143 }
144
khenaidoocfee5f42018-07-19 22:47:38 -0400145 c.writeLock.Lock()
146 defer c.writeLock.Unlock()
147 _, err := c.ectdAPI.Put(ctx, key, val)
148 cancel()
149 if err != nil {
150 switch err {
151 case context.Canceled:
khenaidoo5c11af72018-07-20 17:21:05 -0400152 log.Warnw("context-cancelled", log.Fields{"error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400153 case context.DeadlineExceeded:
khenaidoo5c11af72018-07-20 17:21:05 -0400154 log.Warnw("context-deadline-exceeded", log.Fields{"error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400155 case v3rpcTypes.ErrEmptyKey:
khenaidoo5c11af72018-07-20 17:21:05 -0400156 log.Warnw("etcd-client-error", log.Fields{"error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400157 default:
khenaidoo5c11af72018-07-20 17:21:05 -0400158 log.Warnw("bad-endpoints", log.Fields{"error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400159 }
160 return err
161 }
162 return nil
163}
164
165// Delete removes a key from the KV store. Timeout defines how long the function will
166// wait for a response
Stephane Barbarie260a5632019-02-26 16:12:49 -0500167func (c *EtcdClient) Delete(key string, timeout int, lock ...bool) error {
khenaidoocfee5f42018-07-19 22:47:38 -0400168
169 duration := GetDuration(timeout)
170
171 ctx, cancel := context.WithTimeout(context.Background(), duration)
Stephane Barbarie260a5632019-02-26 16:12:49 -0500172
173 // Lock by default; otherwise lock per instructed value
174 if len(lock) == 0 || lock[0] {
175 session, _ := v3Concurrency.NewSession(c.ectdAPI, v3Concurrency.WithContext(ctx))
Stephane Barbariec53a2752019-03-08 17:50:10 -0500176 mu := v3Concurrency.NewMutex(session, "/lock"+key)
Stephane Barbarie260a5632019-02-26 16:12:49 -0500177 mu.Lock(context.Background())
178 defer mu.Unlock(context.Background())
179 defer session.Close()
180 }
181
khenaidoocfee5f42018-07-19 22:47:38 -0400182 defer cancel()
183
184 c.writeLock.Lock()
185 defer c.writeLock.Unlock()
186
khenaidoocfee5f42018-07-19 22:47:38 -0400187 // delete the keys
khenaidoo1ce37ad2019-03-24 22:07:24 -0400188 if _, err := c.ectdAPI.Delete(ctx, key, v3Client.WithPrefix()); err != nil {
189 log.Errorw("failed-to-delete-key", log.Fields{"key": key, "error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400190 return err
191 }
khenaidoo1ce37ad2019-03-24 22:07:24 -0400192 log.Debugw("key(s)-deleted", log.Fields{"key": key})
khenaidoocfee5f42018-07-19 22:47:38 -0400193 return nil
194}
195
196// Reserve is invoked to acquire a key and set it to a given value. Value can only be a string or []byte since
197// the etcd API accepts only a string. Timeout defines how long the function will wait for a response. TTL
198// defines how long that reservation is valid. When TTL expires the key is unreserved by the KV store itself.
199// If the key is acquired then the value returned will be the value passed in. If the key is already acquired
200// then the value assigned to that key will be returned.
201func (c *EtcdClient) Reserve(key string, value interface{}, ttl int64) (interface{}, error) {
202 // Validate that we can convert value to a string as etcd API expects a string
203 var val string
204 var er error
205 if val, er = ToString(value); er != nil {
206 return nil, fmt.Errorf("unexpected-type%T", value)
207 }
208
209 // Create a lease
210 resp, err := c.ectdAPI.Grant(context.Background(), ttl)
211 if err != nil {
212 log.Error(err)
213 return nil, err
214 }
215 // Register the lease id
216 c.writeLock.Lock()
217 c.keyReservations[key] = &resp.ID
218 c.writeLock.Unlock()
219
220 // Revoke lease if reservation is not successful
221 reservationSuccessful := false
222 defer func() {
223 if !reservationSuccessful {
224 if err = c.ReleaseReservation(key); err != nil {
khenaidoo1ce37ad2019-03-24 22:07:24 -0400225 log.Error("cannot-release-lease")
khenaidoocfee5f42018-07-19 22:47:38 -0400226 }
227 }
228 }()
229
230 // Try to grap the Key with the above lease
231 c.ectdAPI.Txn(context.Background())
232 txn := c.ectdAPI.Txn(context.Background())
233 txn = txn.If(v3Client.Compare(v3Client.Version(key), "=", 0))
234 txn = txn.Then(v3Client.OpPut(key, val, v3Client.WithLease(resp.ID)))
235 txn = txn.Else(v3Client.OpGet(key))
236 result, er := txn.Commit()
237 if er != nil {
238 return nil, er
239 }
240
241 if !result.Succeeded {
242 // Verify whether we are already the owner of that Key
243 if len(result.Responses) > 0 &&
244 len(result.Responses[0].GetResponseRange().Kvs) > 0 {
245 kv := result.Responses[0].GetResponseRange().Kvs[0]
246 if string(kv.Value) == val {
247 reservationSuccessful = true
248 return value, nil
249 }
250 return kv.Value, nil
251 }
252 } else {
253 // Read the Key to ensure this is our Key
Stephane Barbarie260a5632019-02-26 16:12:49 -0500254 m, err := c.Get(key, defaultKVGetTimeout, false)
khenaidoocfee5f42018-07-19 22:47:38 -0400255 if err != nil {
256 return nil, err
257 }
258 if m != nil {
259 if m.Key == key && isEqual(m.Value, value) {
260 // My reservation is successful - register it. For now, support is only for 1 reservation per key
261 // per session.
262 reservationSuccessful = true
263 return value, nil
264 }
265 // My reservation has failed. Return the owner of that key
266 return m.Value, nil
267 }
268 }
269 return nil, nil
270}
271
272// ReleaseAllReservations releases all key reservations previously made (using Reserve API)
273func (c *EtcdClient) ReleaseAllReservations() error {
274 c.writeLock.Lock()
275 defer c.writeLock.Unlock()
276 for key, leaseID := range c.keyReservations {
277 _, err := c.ectdAPI.Revoke(context.Background(), *leaseID)
278 if err != nil {
khenaidoo5c11af72018-07-20 17:21:05 -0400279 log.Errorw("cannot-release-reservation", log.Fields{"key": key, "error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400280 return err
281 }
282 delete(c.keyReservations, key)
283 }
284 return nil
285}
286
287// ReleaseReservation releases reservation for a specific key.
288func (c *EtcdClient) ReleaseReservation(key string) error {
289 // Get the leaseid using the key
khenaidoo1ce37ad2019-03-24 22:07:24 -0400290 log.Debugw("Release-reservation", log.Fields{"key":key})
khenaidoocfee5f42018-07-19 22:47:38 -0400291 var ok bool
292 var leaseID *v3Client.LeaseID
293 c.writeLock.Lock()
294 defer c.writeLock.Unlock()
295 if leaseID, ok = c.keyReservations[key]; !ok {
khenaidoofc1314d2019-03-14 09:34:21 -0400296 return nil
khenaidoocfee5f42018-07-19 22:47:38 -0400297 }
298 if leaseID != nil {
299 _, err := c.ectdAPI.Revoke(context.Background(), *leaseID)
300 if err != nil {
301 log.Error(err)
302 return err
303 }
304 delete(c.keyReservations, key)
305 }
306 return nil
307}
308
309// RenewReservation renews a reservation. A reservation will go stale after the specified TTL (Time To Live)
310// period specified when reserving the key
311func (c *EtcdClient) RenewReservation(key string) error {
312 // Get the leaseid using the key
313 var ok bool
314 var leaseID *v3Client.LeaseID
315 c.writeLock.Lock()
316 defer c.writeLock.Unlock()
317 if leaseID, ok = c.keyReservations[key]; !ok {
318 return errors.New("key-not-reserved")
319 }
320
321 if leaseID != nil {
322 _, err := c.ectdAPI.KeepAliveOnce(context.Background(), *leaseID)
323 if err != nil {
khenaidoo5c11af72018-07-20 17:21:05 -0400324 log.Errorw("lease-may-have-expired", log.Fields{"error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400325 return err
326 }
327 } else {
328 return errors.New("lease-expired")
329 }
330 return nil
331}
332
333// Watch provides the watch capability on a given key. It returns a channel onto which the callee needs to
334// listen to receive Events.
335func (c *EtcdClient) Watch(key string) chan *Event {
336 w := v3Client.NewWatcher(c.ectdAPI)
337 channel := w.Watch(context.Background(), key, v3Client.WithPrefix())
338
339 // Create a new channel
340 ch := make(chan *Event, maxClientChannelBufferSize)
341
342 // Keep track of the created channels so they can be closed when required
343 channelMap := make(map[chan *Event]v3Client.Watcher)
344 channelMap[ch] = w
345 //c.writeLock.Lock()
346 //defer c.writeLock.Unlock()
khenaidoocfee5f42018-07-19 22:47:38 -0400347
Stephane Barbariec53a2752019-03-08 17:50:10 -0500348 channelMaps := c.addChannelMap(key, channelMap)
349
350 log.Debugw("watched-channels", log.Fields{"channels": channelMaps})
khenaidoocfee5f42018-07-19 22:47:38 -0400351 // Launch a go routine to listen for updates
352 go c.listenForKeyChange(channel, ch)
353
354 return ch
355
356}
357
Stephane Barbariec53a2752019-03-08 17:50:10 -0500358func (c *EtcdClient) addChannelMap(key string, channelMap map[chan *Event]v3Client.Watcher) []map[chan *Event]v3Client.Watcher {
359 var channels interface{}
360 var exists bool
361
362 if channels, exists = c.watchedChannels.Load(key); exists {
363 channels = append(channels.([]map[chan *Event]v3Client.Watcher), channelMap)
364 } else {
365 channels = []map[chan *Event]v3Client.Watcher{channelMap}
366 }
367 c.watchedChannels.Store(key, channels)
368
369 return channels.([]map[chan *Event]v3Client.Watcher)
370}
371
372func (c *EtcdClient) removeChannelMap(key string, pos int) []map[chan *Event]v3Client.Watcher {
373 var channels interface{}
374 var exists bool
375
376 if channels, exists = c.watchedChannels.Load(key); exists {
377 channels = append(channels.([]map[chan *Event]v3Client.Watcher)[:pos], channels.([]map[chan *Event]v3Client.Watcher)[pos+1:]...)
378 c.watchedChannels.Store(key, channels)
379 }
380
381 return channels.([]map[chan *Event]v3Client.Watcher)
382}
383
384func (c *EtcdClient) getChannelMaps(key string) ([]map[chan *Event]v3Client.Watcher, bool) {
385 var channels interface{}
386 var exists bool
387
388 channels, exists = c.watchedChannels.Load(key)
389
khenaidoodaefa372019-03-15 14:04:25 -0400390 if channels == nil {
391 return nil, exists
392 }
393
Stephane Barbariec53a2752019-03-08 17:50:10 -0500394 return channels.([]map[chan *Event]v3Client.Watcher), exists
395}
396
khenaidoocfee5f42018-07-19 22:47:38 -0400397// CloseWatch closes a specific watch. Both the key and the channel are required when closing a watch as there
398// may be multiple listeners on the same key. The previously created channel serves as a key
399func (c *EtcdClient) CloseWatch(key string, ch chan *Event) {
400 // Get the array of channels mapping
401 var watchedChannels []map[chan *Event]v3Client.Watcher
402 var ok bool
403 c.writeLock.Lock()
404 defer c.writeLock.Unlock()
405
Stephane Barbariec53a2752019-03-08 17:50:10 -0500406 if watchedChannels, ok = c.getChannelMaps(key); !ok {
khenaidoo5c11af72018-07-20 17:21:05 -0400407 log.Warnw("key-has-no-watched-channels", log.Fields{"key": key})
khenaidoocfee5f42018-07-19 22:47:38 -0400408 return
409 }
410 // Look for the channels
411 var pos = -1
412 for i, chMap := range watchedChannels {
413 if t, ok := chMap[ch]; ok {
414 log.Debug("channel-found")
415 // Close the etcd watcher before the client channel. This should close the etcd channel as well
416 if err := t.Close(); err != nil {
khenaidoo5c11af72018-07-20 17:21:05 -0400417 log.Errorw("watcher-cannot-be-closed", log.Fields{"key": key, "error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400418 }
419 close(ch)
420 pos = i
421 break
422 }
423 }
Stephane Barbariec53a2752019-03-08 17:50:10 -0500424
425 channelMaps, _ := c.getChannelMaps(key)
khenaidoocfee5f42018-07-19 22:47:38 -0400426 // Remove that entry if present
427 if pos >= 0 {
Stephane Barbariec53a2752019-03-08 17:50:10 -0500428 channelMaps = c.removeChannelMap(key, pos)
khenaidoocfee5f42018-07-19 22:47:38 -0400429 }
Stephane Barbariec53a2752019-03-08 17:50:10 -0500430 log.Infow("watcher-channel-exiting", log.Fields{"key": key, "channel": channelMaps})
khenaidoocfee5f42018-07-19 22:47:38 -0400431}
432
433func (c *EtcdClient) listenForKeyChange(channel v3Client.WatchChan, ch chan<- *Event) {
khenaidoo5c11af72018-07-20 17:21:05 -0400434 log.Infow("start-listening-on-channel", log.Fields{"channel": ch})
khenaidoocfee5f42018-07-19 22:47:38 -0400435 for resp := range channel {
436 for _, ev := range resp.Events {
437 //log.Debugf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
438 ch <- NewEvent(getEventType(ev), ev.Kv.Key, ev.Kv.Value)
439 }
440 }
441 log.Info("stop-listening-on-channel")
442}
443
444func getEventType(event *v3Client.Event) int {
445 switch event.Type {
446 case v3Client.EventTypePut:
447 return PUT
448 case v3Client.EventTypeDelete:
449 return DELETE
450 }
451 return UNKNOWN
452}
453
454// Close closes the KV store client
455func (c *EtcdClient) Close() {
456 c.writeLock.Lock()
457 defer c.writeLock.Unlock()
458 if err := c.ectdAPI.Close(); err != nil {
khenaidoo5c11af72018-07-20 17:21:05 -0400459 log.Errorw("error-closing-client", log.Fields{"error": err})
khenaidoocfee5f42018-07-19 22:47:38 -0400460 }
461}
khenaidoobdcb8e02019-03-06 16:28:56 -0500462
463func (c *EtcdClient) addLockName(lockName string, lock *v3Concurrency.Mutex, session *v3Concurrency.Session) {
464 c.lockToMutexLock.Lock()
465 defer c.lockToMutexLock.Unlock()
466 c.lockToMutexMap[lockName] = lock
467 c.lockToSessionMap[lockName] = session
468}
469
470func (c *EtcdClient) deleteLockName(lockName string) {
471 c.lockToMutexLock.Lock()
472 defer c.lockToMutexLock.Unlock()
473 delete(c.lockToMutexMap, lockName)
474 delete(c.lockToSessionMap, lockName)
475}
476
477func (c *EtcdClient) getLock(lockName string) (*v3Concurrency.Mutex, *v3Concurrency.Session) {
478 c.lockToMutexLock.Lock()
479 defer c.lockToMutexLock.Unlock()
480 var lock *v3Concurrency.Mutex
481 var session *v3Concurrency.Session
482 if l, exist := c.lockToMutexMap[lockName]; exist {
483 lock = l
484 }
485 if s, exist := c.lockToSessionMap[lockName]; exist {
486 session = s
487 }
488 return lock, session
489}
490
Stephane Barbariec53a2752019-03-08 17:50:10 -0500491func (c *EtcdClient) AcquireLock(lockName string, timeout int) error {
khenaidoobdcb8e02019-03-06 16:28:56 -0500492 duration := GetDuration(timeout)
493 ctx, cancel := context.WithTimeout(context.Background(), duration)
494 session, _ := v3Concurrency.NewSession(c.ectdAPI, v3Concurrency.WithContext(ctx))
Stephane Barbariec53a2752019-03-08 17:50:10 -0500495 mu := v3Concurrency.NewMutex(session, "/devicelock_"+lockName)
khenaidoobdcb8e02019-03-06 16:28:56 -0500496 if err := mu.Lock(context.Background()); err != nil {
497 return err
498 }
499 c.addLockName(lockName, mu, session)
500 cancel()
501 return nil
502}
503
Stephane Barbariec53a2752019-03-08 17:50:10 -0500504func (c *EtcdClient) ReleaseLock(lockName string) error {
khenaidoobdcb8e02019-03-06 16:28:56 -0500505 lock, session := c.getLock(lockName)
506 var err error
507 if lock != nil {
508 if e := lock.Unlock(context.Background()); e != nil {
509 err = e
510 }
511 }
512 if session != nil {
513 if e := session.Close(); e != nil {
514 err = e
515 }
516 }
517 c.deleteLockName(lockName)
518
519 return err
Stephane Barbariec53a2752019-03-08 17:50:10 -0500520}