| // 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 |
| } |