v1.14.6/physical/oci/oci_ha.go - hashicorp/vault - Git at Google

 // Copyright © 2019, Oracle and/or its affiliates.
 package oci

 import (
 	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"sync"
 	"sync/atomic"
 	"time"

 	"github.com/armon/go-metrics"
 	"github.com/hashicorp/go-uuid"
 	"github.com/hashicorp/vault/sdk/physical"
 	"github.com/oracle/oci-go-sdk/objectstorage"
 )

 // The lock implementation below prioritizes ensuring that there are not 2 primary at any given point in time
 // over high availability of the primary instance

 // Verify Backend satisfies the correct interfaces
 var (
 	_ physical.HABackend = (*Backend)(nil)
 	_ physical.Lock      = (*Lock)(nil)
 )

 const (
 	// LockRenewInterval is the time to wait between lock renewals.
 	LockRenewInterval = 3 * time.Second

 	// LockRetryInterval is the amount of time to wait if the lock fails before trying again.
 	LockRetryInterval = 5 * time.Second

 	// LockWatchRetryInterval is the amount of time to wait if a watch fails before trying again.
 	LockWatchRetryInterval = 2 * time.Second

 	// LockTTL is the default lock TTL.
 	LockTTL = 15 * time.Second

 	// LockWatchRetryMax is the number of times to retry a failed watch before signaling that leadership is lost.
 	LockWatchRetryMax = 4

 	// LockCacheMinAcceptableAge is minimum cache age in seconds to determine that its safe for a secondary instance
 	// to acquire lock.
 	LockCacheMinAcceptableAge = 45 * time.Second

 	// LockWriteRetriesOnFailures is the number of retries that are made on write 5xx failures.
 	LockWriteRetriesOnFailures = 4

 	ObjectStorageCallsReadTimeout = 3 * time.Second

 	ObjectStorageCallsWriteTimeout = 3 * time.Second
 )

 type LockCache struct {
 	// ETag values are unique identifiers generated by the OCI service and changed every time the object is modified.
 	etag       string
 	lastUpdate time.Time
 	lockRecord *LockRecord
 }

 type Lock struct {
 	// backend is the underlying physical backend.
 	backend *Backend

 	// Key is the name of the Key. Value is the Value of the Key.
 	key, value string

 	// held is a boolean indicating if the lock is currently held.
 	held bool

 	// Identity is the internal Identity of this Key (unique to this server instance).
 	identity string

 	internalLock sync.Mutex

 	// stopCh is the channel that stops all operations. It may be closed in the
 	// event of a leader loss or graceful shutdown. stopped is a boolean
 	// indicating if we are stopped - it exists to prevent double closing the
 	// channel. stopLock is a mutex around the locks.
 	stopCh   chan struct{}
 	stopped  bool
 	stopLock sync.Mutex

 	lockRecordCache atomic.Value

 	// Allow modifying the Lock durations for ease of unit testing.
 	renewInterval      time.Duration
 	retryInterval      time.Duration
 	ttl                time.Duration
 	watchRetryInterval time.Duration
 	watchRetryMax      int
 }

 type LockRecord struct {
 	Key      string
 	Value    string
 	Identity string
 }

 var (
 	metricLockUnlock  = []string{"oci", "lock", "unlock"}
 	metricLockLock    = []string{"oci", "lock", "lock"}
 	metricLockValue   = []string{"oci", "lock", "Value"}
 	metricLeaderValue = []string{"oci", "leader", "Value"}
 )

 func (b *Backend) HAEnabled() bool {
 	return b.haEnabled
 }

 // LockWith acquires a mutual exclusion based on the given Key.
 func (b *Backend) LockWith(key, value string) (physical.Lock, error) {
 	identity, err := uuid.GenerateUUID()
 	if err != nil {
 		return nil, fmt.Errorf("Lock with: %w", err)
 	}
 	return &Lock{
 		backend:  b,
 		key:      key,
 		value:    value,
 		identity: identity,
 		stopped:  true,

 		renewInterval:      LockRenewInterval,
 		retryInterval:      LockRetryInterval,
 		ttl:                LockTTL,
 		watchRetryInterval: LockWatchRetryInterval,
 		watchRetryMax:      LockWatchRetryMax,
 	}, nil
 }

 func (l *Lock) Lock(stopCh <-chan struct{}) (<-chan struct{}, error) {
 	l.backend.logger.Debug("Lock() called")
 	defer metrics.MeasureSince(metricLockLock, time.Now().UTC())
 	l.internalLock.Lock()
 	defer l.internalLock.Unlock()
 	if l.held {
 		return nil, errors.New("lock already held")
 	}

 	// Attempt to lock - this function blocks until a lock is acquired or an error
 	// occurs.
 	acquired, err := l.attemptLock(stopCh)
 	if err != nil {
 		return nil, fmt.Errorf("lock: %w", err)
 	}
 	if !acquired {
 		return nil, nil
 	}

 	// We have the lock now
 	l.held = true

 	// Build the locks
 	l.stopLock.Lock()
 	l.stopCh = make(chan struct{})
 	l.stopped = false
 	l.stopLock.Unlock()

 	// Periodically renew and watch the lock
 	go l.renewLock()
 	go l.watchLock()

 	return l.stopCh, nil
 }

 // attemptLock attempts to acquire a lock. If the given channel is closed, the
 // acquisition attempt stops. This function returns when a lock is acquired or
 // an error occurs.
 func (l *Lock) attemptLock(stopCh <-chan struct{}) (bool, error) {
 	l.backend.logger.Debug("AttemptLock() called")
 	ticker := time.NewTicker(l.retryInterval)
 	defer ticker.Stop()

 	for {
 		select {
 		case <-ticker.C:
 			acquired, err := l.writeLock()
 			if err != nil {
 				return false, fmt.Errorf("attempt lock: %w", err)
 			}
 			if !acquired {
 				continue
 			}

 			return true, nil
 		case <-stopCh:
 			return false, nil
 		}
 	}
 }

 // renewLock renews the given lock until the channel is closed.
 func (l *Lock) renewLock() {
 	l.backend.logger.Debug("RenewLock() called")
 	ticker := time.NewTicker(l.renewInterval)
 	defer ticker.Stop()

 	for {
 		select {
 		case <-ticker.C:
 			l.writeLock()
 		case <-l.stopCh:
 			return
 		}
 	}
 }

 func loadLockRecordCache(l *Lock) *LockCache {
 	lockRecordCache := l.lockRecordCache.Load()
 	if lockRecordCache == nil {
 		return nil
 	}
 	return lockRecordCache.(*LockCache)
 }

 // watchLock checks whether the lock has changed in the table and closes the
 // leader channel accordingly. If an error occurs during the check, watchLock
 // will retry the operation and then close the leader channel if it can't
 // succeed after retries.
 func (l *Lock) watchLock() {
 	l.backend.logger.Debug("WatchLock() called")
 	retries := 0
 	ticker := time.NewTicker(l.watchRetryInterval)
 	defer ticker.Stop()

 OUTER:
 	for {
 		// Check if the channel is already closed
 		select {
 		case <-l.stopCh:
 			l.backend.logger.Debug("WatchLock():Stop lock signaled/closed.")
 			break OUTER
 		default:
 		}

 		// Check if we've exceeded retries
 		if retries >= l.watchRetryMax-1 {
 			l.backend.logger.Debug("WatchLock: Failed to get lock data from object storage. Giving up the lease after max retries")
 			break OUTER
 		}

 		// Wait for the timer
 		select {
 		case <-ticker.C:
 		case <-l.stopCh:
 			break OUTER
 		}

 		lockRecordCache := loadLockRecordCache(l)
 		if (lockRecordCache == nil) ||
 			(lockRecordCache.lockRecord == nil) ||
 			(lockRecordCache.lockRecord.Identity != l.identity) ||
 			(time.Now().Sub(lockRecordCache.lastUpdate) > l.ttl) {
 			l.backend.logger.Debug("WatchLock: Lock record cache is nil, stale or does not belong to self.")
 			break OUTER
 		}

 		lockRecord, _, err := l.get(context.Background())
 		if err != nil {
 			retries++
 			l.backend.logger.Debug("WatchLock: Failed to get lock data from object storage. Retrying..")
 			metrics.SetGauge(metricHaWatchLockRetriable, 1)
 			continue
 		}

 		if (lockRecord == nil) || (lockRecord.Identity != l.identity) {
 			l.backend.logger.Debug("WatchLock: Lock record cache is nil or does not belong to self.")
 			break OUTER
 		}

 		// reset retries counter on success
 		retries = 0
 		l.backend.logger.Debug("WatchLock() successful")
 		metrics.SetGauge(metricHaWatchLockRetriable, 0)
 	}

 	l.stopLock.Lock()
 	defer l.stopLock.Unlock()
 	if !l.stopped {
 		l.stopped = true
 		l.backend.logger.Debug("Closing the stop channel to give up leadership.")
 		close(l.stopCh)
 	}
 }

 func (l *Lock) Unlock() error {
 	l.backend.logger.Debug("Unlock() called")
 	defer metrics.MeasureSince(metricLockUnlock, time.Now().UTC())

 	l.internalLock.Lock()
 	defer l.internalLock.Unlock()
 	if !l.held {
 		return nil
 	}

 	// Stop any existing locking or renewal attempts
 	l.stopLock.Lock()
 	if !l.stopped {
 		l.stopped = true
 		close(l.stopCh)
 	}
 	l.stopLock.Unlock()

 	// We are no longer holding the lock
 	l.held = false

 	// Get current lock record
 	currentLockRecord, etag, err := l.get(context.Background())
 	if err != nil {
 		return fmt.Errorf("error reading lock record: %w", err)
 	}

 	if currentLockRecord != nil && currentLockRecord.Identity == l.identity {

 		defer metrics.MeasureSince(metricDeleteHa, time.Now())
 		opcClientRequestId, err := uuid.GenerateUUID()
 		if err != nil {
 			l.backend.logger.Debug("Unlock: error generating UUID")
 			return fmt.Errorf("failed to generate UUID: %w", err)
 		}
 		l.backend.logger.Debug("Unlock", "opc-client-request-id", opcClientRequestId)
 		request := objectstorage.DeleteObjectRequest{
 			NamespaceName:      &l.backend.namespaceName,
 			BucketName:         &l.backend.lockBucketName,
 			ObjectName:         &l.key,
 			IfMatch:            &etag,
 			OpcClientRequestId: &opcClientRequestId,
 		}

 		response, err := l.backend.client.DeleteObject(context.Background(), request)
 		l.backend.logRequest("deleteHA", response.RawResponse, response.OpcClientRequestId, response.OpcRequestId, err)

 		if err != nil {
 			metrics.IncrCounter(metricDeleteFailed, 1)
 			return fmt.Errorf("write lock: %w", err)
 		}
 	}

 	return nil
 }

 func (l *Lock) Value() (bool, string, error) {
 	l.backend.logger.Debug("Value() called")
 	defer metrics.MeasureSince(metricLockValue, time.Now().UTC())

 	lockRecord, _, err := l.get(context.Background())
 	if err != nil {
 		return false, "", err
 	}
 	if lockRecord == nil {
 		return false, "", err
 	}
 	return true, lockRecord.Value, nil
 }

 // get retrieves the Value for the lock.
 func (l *Lock) get(ctx context.Context) (*LockRecord, string, error) {
 	l.backend.logger.Debug("Called getLockRecord()")

 	// Read lock Key

 	defer metrics.MeasureSince(metricGetHa, time.Now())
 	opcClientRequestId, err := uuid.GenerateUUID()
 	if err != nil {
 		l.backend.logger.Error("getHa: error generating UUID")
 		return nil, "", fmt.Errorf("failed to generate UUID: %w", err)
 	}
 	l.backend.logger.Debug("getHa", "opc-client-request-id", opcClientRequestId)

 	request := objectstorage.GetObjectRequest{
 		NamespaceName:      &l.backend.namespaceName,
 		BucketName:         &l.backend.lockBucketName,
 		ObjectName:         &l.key,
 		OpcClientRequestId: &opcClientRequestId,
 	}

 	ctx, cancel := context.WithTimeout(ctx, ObjectStorageCallsReadTimeout)
 	defer cancel()

 	response, err := l.backend.client.GetObject(ctx, request)
 	l.backend.logRequest("getHA", response.RawResponse, response.OpcClientRequestId, response.OpcRequestId, err)

 	if err != nil {
 		if response.RawResponse != nil && response.RawResponse.StatusCode == http.StatusNotFound {
 			return nil, "", nil
 		}

 		metrics.IncrCounter(metricGetFailed, 1)
 		l.backend.logger.Error("Error calling GET", "err", err)
 		return nil, "", fmt.Errorf("failed to read Value for %q: %w", l.key, err)
 	}

 	defer response.RawResponse.Body.Close()

 	body, err := ioutil.ReadAll(response.Content)
 	if err != nil {
 		metrics.IncrCounter(metricGetFailed, 1)
 		l.backend.logger.Error("Error reading content", "err", err)
 		return nil, "", fmt.Errorf("failed to decode Value into bytes: %w", err)
 	}

 	var lockRecord LockRecord
 	err = json.Unmarshal(body, &lockRecord)
 	if err != nil {
 		metrics.IncrCounter(metricGetFailed, 1)
 		l.backend.logger.Error("Error un-marshalling content", "err", err)
 		return nil, "", fmt.Errorf("failed to read Value for %q: %w", l.key, err)
 	}

 	return &lockRecord, *response.ETag, nil
 }

 func (l *Lock) writeLock() (bool, error) {
 	l.backend.logger.Debug("WriteLock() called")

 	// Create a transaction to read and the update (maybe)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

 	// The transaction will be retried, and it could sit in a queue behind, say,
 	// the delete operation. To stop the transaction, we close the context when
 	// the associated stopCh is received.
 	go func() {
 		select {
 		case <-l.stopCh:
 			cancel()
 		case <-ctx.Done():
 		}
 	}()

 	lockRecordCache := loadLockRecordCache(l)
 	if (lockRecordCache == nil) || lockRecordCache.lockRecord == nil ||
 		lockRecordCache.lockRecord.Identity != l.identity ||
 		time.Now().Sub(lockRecordCache.lastUpdate) > l.ttl {
 		// case secondary
 		currentLockRecord, currentEtag, err := l.get(ctx)
 		if err != nil {
 			return false, fmt.Errorf("error reading lock record: %w", err)
 		}

 		if (lockRecordCache == nil) || lockRecordCache.etag != currentEtag {
 			// update cached lock record
 			l.lockRecordCache.Store(&LockCache{
 				etag:       currentEtag,
 				lastUpdate: time.Now().UTC(),
 				lockRecord: currentLockRecord,
 			})

 			lockRecordCache = loadLockRecordCache(l)
 		}

 		// Current lock record being null implies that there is no leader. In this case we want to try acquiring lock.
 		if currentLockRecord != nil && time.Now().Sub(lockRecordCache.lastUpdate) < LockCacheMinAcceptableAge {
 			return false, nil
 		}
 		// cache is old enough and current, try acquiring lock as secondary
 	}

 	newLockRecord := &LockRecord{
 		Key:      l.key,
 		Value:    l.value,
 		Identity: l.identity,
 	}

 	newLockRecordJson, err := json.Marshal(newLockRecord)
 	if err != nil {
 		return false, fmt.Errorf("error reading lock record: %w", err)
 	}

 	defer metrics.MeasureSince(metricPutHa, time.Now())

 	opcClientRequestId, err := uuid.GenerateUUID()
 	if err != nil {
 		l.backend.logger.Error("putHa: error generating UUID")
 		return false, fmt.Errorf("failed to generate UUID: %w", err)
 	}
 	l.backend.logger.Debug("putHa", "opc-client-request-id", opcClientRequestId)
 	size := int64(len(newLockRecordJson))
 	putRequest := objectstorage.PutObjectRequest{
 		NamespaceName:      &l.backend.namespaceName,
 		BucketName:         &l.backend.lockBucketName,
 		ObjectName:         &l.key,
 		ContentLength:      &size,
 		PutObjectBody:      ioutil.NopCloser(bytes.NewReader(newLockRecordJson)),
 		OpcMeta:            nil,
 		OpcClientRequestId: &opcClientRequestId,
 	}

 	if lockRecordCache.etag == "" {
 		noneMatch := "*"
 		putRequest.IfNoneMatch = &noneMatch
 	} else {
 		putRequest.IfMatch = &lockRecordCache.etag
 	}

 	newtEtag := ""
 	for i := 1; i <= LockWriteRetriesOnFailures; i++ {
 		writeCtx, writeCancel := context.WithTimeout(ctx, ObjectStorageCallsWriteTimeout)
 		defer writeCancel()

 		putObjectResponse, putObjectError := l.backend.client.PutObject(writeCtx, putRequest)
 		l.backend.logRequest("putHA", putObjectResponse.RawResponse, putObjectResponse.OpcClientRequestId, putObjectResponse.OpcRequestId, putObjectError)

 		if putObjectError == nil {
 			newtEtag = *putObjectResponse.ETag
 			putObjectResponse.RawResponse.Body.Close()
 			break
 		}

 		err = putObjectError

 		if putObjectResponse.RawResponse == nil {
 			metrics.IncrCounter(metricPutFailed, 1)
 			l.backend.logger.Error("PUT", "err", err)
 			break
 		}

 		putObjectResponse.RawResponse.Body.Close()

 		// Retry if the return code is 5xx
 		if (putObjectResponse.RawResponse.StatusCode / 100) == 5 {
 			metrics.IncrCounter(metricPutFailed, 1)
 			l.backend.logger.Warn("PUT. Retrying..", "err", err)
 			time.Sleep(time.Duration(100*i) * time.Millisecond)
 		} else {
 			l.backend.logger.Error("PUT", "err", err)
 			break
 		}
 	}

 	if err != nil {
 		return false, fmt.Errorf("write lock: %w", err)
 	}

 	l.backend.logger.Debug("Lock written", string(newLockRecordJson))

 	l.lockRecordCache.Store(&LockCache{
 		etag:       newtEtag,
 		lastUpdate: time.Now().UTC(),
 		lockRecord: newLockRecord,
 	})

 	metrics.SetGauge(metricLeaderValue, 1)
 	return true, nil
 }
	// Copyright © 2019, Oracle and/or its affiliates.
	package oci

	import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"net/http"
	"sync"
	"sync/atomic"
	"time"

	"github.com/armon/go-metrics"
	"github.com/hashicorp/go-uuid"
	"github.com/hashicorp/vault/sdk/physical"
	"github.com/oracle/oci-go-sdk/objectstorage"
	)

	// The lock implementation below prioritizes ensuring that there are not 2 primary at any given point in time
	// over high availability of the primary instance

	// Verify Backend satisfies the correct interfaces
	var (
	_ physical.HABackend = (*Backend)(nil)
	_ physical.Lock = (*Lock)(nil)
	)

	const (
	// LockRenewInterval is the time to wait between lock renewals.
	LockRenewInterval = 3 * time.Second

	// LockRetryInterval is the amount of time to wait if the lock fails before trying again.
	LockRetryInterval = 5 * time.Second

	// LockWatchRetryInterval is the amount of time to wait if a watch fails before trying again.
	LockWatchRetryInterval = 2 * time.Second

	// LockTTL is the default lock TTL.
	LockTTL = 15 * time.Second

	// LockWatchRetryMax is the number of times to retry a failed watch before signaling that leadership is lost.
	LockWatchRetryMax = 4

	// LockCacheMinAcceptableAge is minimum cache age in seconds to determine that its safe for a secondary instance
	// to acquire lock.
	LockCacheMinAcceptableAge = 45 * time.Second

	// LockWriteRetriesOnFailures is the number of retries that are made on write 5xx failures.
	LockWriteRetriesOnFailures = 4

	ObjectStorageCallsReadTimeout = 3 * time.Second

	ObjectStorageCallsWriteTimeout = 3 * time.Second
	)

	type LockCache struct {
	// ETag values are unique identifiers generated by the OCI service and changed every time the object is modified.
	etag string
	lastUpdate time.Time
	lockRecord *LockRecord
	}

	type Lock struct {
	// backend is the underlying physical backend.
	backend *Backend

	// Key is the name of the Key. Value is the Value of the Key.
	key, value string

	// held is a boolean indicating if the lock is currently held.
	held bool

	// Identity is the internal Identity of this Key (unique to this server instance).
	identity string

	internalLock sync.Mutex

	// stopCh is the channel that stops all operations. It may be closed in the
	// event of a leader loss or graceful shutdown. stopped is a boolean
	// indicating if we are stopped - it exists to prevent double closing the
	// channel. stopLock is a mutex around the locks.
	stopCh chan struct{}
	stopped bool
	stopLock sync.Mutex

	lockRecordCache atomic.Value

	// Allow modifying the Lock durations for ease of unit testing.
	renewInterval time.Duration
	retryInterval time.Duration
	ttl time.Duration
	watchRetryInterval time.Duration
	watchRetryMax int
	}

	type LockRecord struct {
	Key string
	Value string
	Identity string
	}

	var (
	metricLockUnlock = []string{"oci", "lock", "unlock"}
	metricLockLock = []string{"oci", "lock", "lock"}
	metricLockValue = []string{"oci", "lock", "Value"}
	metricLeaderValue = []string{"oci", "leader", "Value"}
	)

	func (b *Backend) HAEnabled() bool {
	return b.haEnabled
	}

	// LockWith acquires a mutual exclusion based on the given Key.
	func (b *Backend) LockWith(key, value string) (physical.Lock, error) {
	identity, err := uuid.GenerateUUID()
	if err != nil {
	return nil, fmt.Errorf("Lock with: %w", err)
	}
	return &Lock{
	backend: b,
	key: key,
	value: value,
	identity: identity,
	stopped: true,

	renewInterval: LockRenewInterval,
	retryInterval: LockRetryInterval,
	ttl: LockTTL,
	watchRetryInterval: LockWatchRetryInterval,
	watchRetryMax: LockWatchRetryMax,
	}, nil
	}

	func (l *Lock) Lock(stopCh <-chan struct{}) (<-chan struct{}, error) {
	l.backend.logger.Debug("Lock() called")
	defer metrics.MeasureSince(metricLockLock, time.Now().UTC())
	l.internalLock.Lock()
	defer l.internalLock.Unlock()
	if l.held {
	return nil, errors.New("lock already held")
	}

	// Attempt to lock - this function blocks until a lock is acquired or an error
	// occurs.
	acquired, err := l.attemptLock(stopCh)
	if err != nil {
	return nil, fmt.Errorf("lock: %w", err)
	}
	if !acquired {
	return nil, nil
	}

	// We have the lock now
	l.held = true

	// Build the locks
	l.stopLock.Lock()
	l.stopCh = make(chan struct{})
	l.stopped = false
	l.stopLock.Unlock()

	// Periodically renew and watch the lock
	go l.renewLock()
	go l.watchLock()

	return l.stopCh, nil
	}

	// attemptLock attempts to acquire a lock. If the given channel is closed, the
	// acquisition attempt stops. This function returns when a lock is acquired or
	// an error occurs.
	func (l *Lock) attemptLock(stopCh <-chan struct{}) (bool, error) {
	l.backend.logger.Debug("AttemptLock() called")
	ticker := time.NewTicker(l.retryInterval)
	defer ticker.Stop()

	for {
	select {
	case <-ticker.C:
	acquired, err := l.writeLock()
	if err != nil {
	return false, fmt.Errorf("attempt lock: %w", err)
	}
	if !acquired {
	continue
	}

	return true, nil
	case <-stopCh:
	return false, nil
	}
	}
	}

	// renewLock renews the given lock until the channel is closed.
	func (l *Lock) renewLock() {
	l.backend.logger.Debug("RenewLock() called")
	ticker := time.NewTicker(l.renewInterval)
	defer ticker.Stop()

	for {
	select {
	case <-ticker.C:
	l.writeLock()
	case <-l.stopCh:
	return
	}
	}
	}

	func loadLockRecordCache(l Lock) LockCache {
	lockRecordCache := l.lockRecordCache.Load()
	if lockRecordCache == nil {
	return nil
	}
	return lockRecordCache.(*LockCache)
	}

	// watchLock checks whether the lock has changed in the table and closes the
	// leader channel accordingly. If an error occurs during the check, watchLock
	// will retry the operation and then close the leader channel if it can't
	// succeed after retries.
	func (l *Lock) watchLock() {
	l.backend.logger.Debug("WatchLock() called")
	retries := 0
	ticker := time.NewTicker(l.watchRetryInterval)
	defer ticker.Stop()

	OUTER:
	for {
	// Check if the channel is already closed
	select {
	case <-l.stopCh:
	l.backend.logger.Debug("WatchLock():Stop lock signaled/closed.")
	break OUTER
	default:
	}

	// Check if we've exceeded retries
	if retries >= l.watchRetryMax-1 {
	l.backend.logger.Debug("WatchLock: Failed to get lock data from object storage. Giving up the lease after max retries")
	break OUTER
	}

	// Wait for the timer
	select {
	case <-ticker.C:
	case <-l.stopCh:
	break OUTER
	}

	lockRecordCache := loadLockRecordCache(l)
	if (lockRecordCache == nil) \|\|
	(lockRecordCache.lockRecord == nil) \|\|
	(lockRecordCache.lockRecord.Identity != l.identity) \|\|
	(time.Now().Sub(lockRecordCache.lastUpdate) > l.ttl) {
	l.backend.logger.Debug("WatchLock: Lock record cache is nil, stale or does not belong to self.")
	break OUTER
	}

	lockRecord, _, err := l.get(context.Background())
	if err != nil {
	retries++
	l.backend.logger.Debug("WatchLock: Failed to get lock data from object storage. Retrying..")
	metrics.SetGauge(metricHaWatchLockRetriable, 1)
	continue
	}

	if (lockRecord == nil) \|\| (lockRecord.Identity != l.identity) {
	l.backend.logger.Debug("WatchLock: Lock record cache is nil or does not belong to self.")
	break OUTER
	}

	// reset retries counter on success
	retries = 0
	l.backend.logger.Debug("WatchLock() successful")
	metrics.SetGauge(metricHaWatchLockRetriable, 0)
	}

	l.stopLock.Lock()
	defer l.stopLock.Unlock()
	if !l.stopped {
	l.stopped = true
	l.backend.logger.Debug("Closing the stop channel to give up leadership.")
	close(l.stopCh)
	}
	}

	func (l *Lock) Unlock() error {
	l.backend.logger.Debug("Unlock() called")
	defer metrics.MeasureSince(metricLockUnlock, time.Now().UTC())

	l.internalLock.Lock()
	defer l.internalLock.Unlock()
	if !l.held {
	return nil
	}

	// Stop any existing locking or renewal attempts
	l.stopLock.Lock()
	if !l.stopped {
	l.stopped = true
	close(l.stopCh)
	}
	l.stopLock.Unlock()

	// We are no longer holding the lock
	l.held = false

	// Get current lock record
	currentLockRecord, etag, err := l.get(context.Background())
	if err != nil {
	return fmt.Errorf("error reading lock record: %w", err)
	}

	if currentLockRecord != nil && currentLockRecord.Identity == l.identity {

	defer metrics.MeasureSince(metricDeleteHa, time.Now())
	opcClientRequestId, err := uuid.GenerateUUID()
	if err != nil {
	l.backend.logger.Debug("Unlock: error generating UUID")
	return fmt.Errorf("failed to generate UUID: %w", err)
	}
	l.backend.logger.Debug("Unlock", "opc-client-request-id", opcClientRequestId)
	request := objectstorage.DeleteObjectRequest{
	NamespaceName: &l.backend.namespaceName,
	BucketName: &l.backend.lockBucketName,
	ObjectName: &l.key,
	IfMatch: &etag,
	OpcClientRequestId: &opcClientRequestId,
	}

	response, err := l.backend.client.DeleteObject(context.Background(), request)
	l.backend.logRequest("deleteHA", response.RawResponse, response.OpcClientRequestId, response.OpcRequestId, err)

	if err != nil {
	metrics.IncrCounter(metricDeleteFailed, 1)
	return fmt.Errorf("write lock: %w", err)
	}
	}

	return nil
	}

	func (l *Lock) Value() (bool, string, error) {
	l.backend.logger.Debug("Value() called")
	defer metrics.MeasureSince(metricLockValue, time.Now().UTC())

	lockRecord, _, err := l.get(context.Background())
	if err != nil {
	return false, "", err
	}
	if lockRecord == nil {
	return false, "", err
	}
	return true, lockRecord.Value, nil
	}

	// get retrieves the Value for the lock.
	func (l Lock) get(ctx context.Context) (LockRecord, string, error) {
	l.backend.logger.Debug("Called getLockRecord()")

	// Read lock Key

	defer metrics.MeasureSince(metricGetHa, time.Now())
	opcClientRequestId, err := uuid.GenerateUUID()
	if err != nil {
	l.backend.logger.Error("getHa: error generating UUID")
	return nil, "", fmt.Errorf("failed to generate UUID: %w", err)
	}
	l.backend.logger.Debug("getHa", "opc-client-request-id", opcClientRequestId)

	request := objectstorage.GetObjectRequest{
	NamespaceName: &l.backend.namespaceName,
	BucketName: &l.backend.lockBucketName,
	ObjectName: &l.key,
	OpcClientRequestId: &opcClientRequestId,
	}

	ctx, cancel := context.WithTimeout(ctx, ObjectStorageCallsReadTimeout)
	defer cancel()

	response, err := l.backend.client.GetObject(ctx, request)
	l.backend.logRequest("getHA", response.RawResponse, response.OpcClientRequestId, response.OpcRequestId, err)

	if err != nil {
	if response.RawResponse != nil && response.RawResponse.StatusCode == http.StatusNotFound {
	return nil, "", nil
	}

	metrics.IncrCounter(metricGetFailed, 1)
	l.backend.logger.Error("Error calling GET", "err", err)
	return nil, "", fmt.Errorf("failed to read Value for %q: %w", l.key, err)
	}

	defer response.RawResponse.Body.Close()

	body, err := ioutil.ReadAll(response.Content)
	if err != nil {
	metrics.IncrCounter(metricGetFailed, 1)
	l.backend.logger.Error("Error reading content", "err", err)
	return nil, "", fmt.Errorf("failed to decode Value into bytes: %w", err)
	}

	var lockRecord LockRecord
	err = json.Unmarshal(body, &lockRecord)
	if err != nil {
	metrics.IncrCounter(metricGetFailed, 1)
	l.backend.logger.Error("Error un-marshalling content", "err", err)
	return nil, "", fmt.Errorf("failed to read Value for %q: %w", l.key, err)
	}

	return &lockRecord, *response.ETag, nil
	}

	func (l *Lock) writeLock() (bool, error) {
	l.backend.logger.Debug("WriteLock() called")

	// Create a transaction to read and the update (maybe)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// The transaction will be retried, and it could sit in a queue behind, say,
	// the delete operation. To stop the transaction, we close the context when
	// the associated stopCh is received.
	go func() {
	select {
	case <-l.stopCh:
	cancel()
	case <-ctx.Done():
	}
	}()

	lockRecordCache := loadLockRecordCache(l)
	if (lockRecordCache == nil) \|\| lockRecordCache.lockRecord == nil \|\|
	lockRecordCache.lockRecord.Identity != l.identity \|\|
	time.Now().Sub(lockRecordCache.lastUpdate) > l.ttl {
	// case secondary
	currentLockRecord, currentEtag, err := l.get(ctx)
	if err != nil {
	return false, fmt.Errorf("error reading lock record: %w", err)
	}

	if (lockRecordCache == nil) \|\| lockRecordCache.etag != currentEtag {
	// update cached lock record
	l.lockRecordCache.Store(&LockCache{
	etag: currentEtag,
	lastUpdate: time.Now().UTC(),
	lockRecord: currentLockRecord,
	})

	lockRecordCache = loadLockRecordCache(l)
	}

	// Current lock record being null implies that there is no leader. In this case we want to try acquiring lock.
	if currentLockRecord != nil && time.Now().Sub(lockRecordCache.lastUpdate) < LockCacheMinAcceptableAge {
	return false, nil
	}
	// cache is old enough and current, try acquiring lock as secondary
	}

	newLockRecord := &LockRecord{
	Key: l.key,
	Value: l.value,
	Identity: l.identity,
	}

	newLockRecordJson, err := json.Marshal(newLockRecord)
	if err != nil {
	return false, fmt.Errorf("error reading lock record: %w", err)
	}

	defer metrics.MeasureSince(metricPutHa, time.Now())

	opcClientRequestId, err := uuid.GenerateUUID()
	if err != nil {
	l.backend.logger.Error("putHa: error generating UUID")
	return false, fmt.Errorf("failed to generate UUID: %w", err)
	}
	l.backend.logger.Debug("putHa", "opc-client-request-id", opcClientRequestId)
	size := int64(len(newLockRecordJson))
	putRequest := objectstorage.PutObjectRequest{
	NamespaceName: &l.backend.namespaceName,
	BucketName: &l.backend.lockBucketName,
	ObjectName: &l.key,
	ContentLength: &size,
	PutObjectBody: ioutil.NopCloser(bytes.NewReader(newLockRecordJson)),
	OpcMeta: nil,
	OpcClientRequestId: &opcClientRequestId,
	}

	if lockRecordCache.etag == "" {
	noneMatch := "*"
	putRequest.IfNoneMatch = &noneMatch
	} else {
	putRequest.IfMatch = &lockRecordCache.etag
	}

	newtEtag := ""
	for i := 1; i <= LockWriteRetriesOnFailures; i++ {
	writeCtx, writeCancel := context.WithTimeout(ctx, ObjectStorageCallsWriteTimeout)
	defer writeCancel()

	putObjectResponse, putObjectError := l.backend.client.PutObject(writeCtx, putRequest)
	l.backend.logRequest("putHA", putObjectResponse.RawResponse, putObjectResponse.OpcClientRequestId, putObjectResponse.OpcRequestId, putObjectError)

	if putObjectError == nil {
	newtEtag = *putObjectResponse.ETag
	putObjectResponse.RawResponse.Body.Close()
	break
	}

	err = putObjectError

	if putObjectResponse.RawResponse == nil {
	metrics.IncrCounter(metricPutFailed, 1)
	l.backend.logger.Error("PUT", "err", err)
	break
	}

	putObjectResponse.RawResponse.Body.Close()

	// Retry if the return code is 5xx
	if (putObjectResponse.RawResponse.StatusCode / 100) == 5 {
	metrics.IncrCounter(metricPutFailed, 1)
	l.backend.logger.Warn("PUT. Retrying..", "err", err)
	time.Sleep(time.Duration(100i) time.Millisecond)
	} else {
	l.backend.logger.Error("PUT", "err", err)
	break
	}
	}

	if err != nil {
	return false, fmt.Errorf("write lock: %w", err)
	}

	l.backend.logger.Debug("Lock written", string(newLockRecordJson))

	l.lockRecordCache.Store(&LockCache{
	etag: newtEtag,
	lastUpdate: time.Now().UTC(),
	lockRecord: newLockRecord,
	})

	metrics.SetGauge(metricLeaderValue, 1)
	return true, nil
	}