github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-27 05:27:21 +00:00
Code Issues Projects Releases Wiki Activity

Refactor operation keys for NestedPendingOperations

Browse Source
This commit is contained in:
Cheng Xing 2020-01-14 10:56:58 -08:00
parent 3fca0a6e41
commit 37957e2a0d
1 changed files with 47 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
pkg/volume/util/nestedpendingoperations/nestedpendingoperations.go
View File

@ -88,8 +88,7 @@ type nestedPendingOperations struct {
} }
type operation struct { type operation struct {
volumeName v1.UniqueVolumeName key operationKey
podName types.UniquePodName
operationName string operationName string
operationPending bool operationPending bool
expBackoff exponentialbackoff.ExponentialBackoff expBackoff exponentialbackoff.ExponentialBackoff
@ -101,18 +100,19 @@ func (grm *nestedPendingOperations) Run(
generatedOperations types.GeneratedOperations) error { generatedOperations types.GeneratedOperations) error {
grm.lock.Lock() grm.lock.Lock()
defer grm.lock.Unlock() defer grm.lock.Unlock()
opExists, previousOpIndex := grm.isOperationExists(volumeName, podName)
opKey := operationKey{volumeName, podName}
opExists, previousOpIndex := grm.isOperationExists(opKey)
if opExists { if opExists {
previousOp := grm.operations[previousOpIndex] previousOp := grm.operations[previousOpIndex]
// Operation already exists // Operation already exists
if previousOp.operationPending { if previousOp.operationPending {
// Operation is pending // Operation is pending
operationKey := getOperationKey(volumeName, podName) return NewAlreadyExistsError(opKey)
return NewAlreadyExistsError(operationKey)
} }
operationKey := getOperationKey(volumeName, podName) backOffErr := previousOp.expBackoff.SafeToRetry(opKey.String())
backOffErr := previousOp.expBackoff.SafeToRetry(operationKey)
if backOffErr != nil { if backOffErr != nil {
if previousOp.operationName == generatedOperations.OperationName { if previousOp.operationName == generatedOperations.OperationName {
return backOffErr return backOffErr
@ -124,15 +124,13 @@ func (grm *nestedPendingOperations) Run(
// Update existing operation to mark as pending. // Update existing operation to mark as pending.
grm.operations[previousOpIndex].operationPending = true grm.operations[previousOpIndex].operationPending = true
grm.operations[previousOpIndex].volumeName = volumeName grm.operations[previousOpIndex].key = opKey
grm.operations[previousOpIndex].podName = podName
} else { } else {
// Create a new operation // Create a new operation
grm.operations = append(grm.operations, grm.operations = append(grm.operations,
operation{ operation{
key: opKey,
operationPending: true, operationPending: true,
volumeName: volumeName,
podName: podName,
operationName: generatedOperations.OperationName, operationName: generatedOperations.OperationName,
expBackoff: exponentialbackoff.ExponentialBackoff{}, expBackoff: exponentialbackoff.ExponentialBackoff{},
}) })
@ -142,7 +140,7 @@ func (grm *nestedPendingOperations) Run(
// Handle unhandled panics (very unlikely) // Handle unhandled panics (very unlikely)
defer k8sRuntime.HandleCrash() defer k8sRuntime.HandleCrash()
// Handle completion of and error, if any, from operationFunc() // Handle completion of and error, if any, from operationFunc()
defer grm.operationComplete(volumeName, podName, &detailedErr) defer grm.operationComplete(opKey, &detailedErr)
return generatedOperations.Run() return generatedOperations.Run()
}() }()
@ -156,7 +154,8 @@ func (grm *nestedPendingOperations) IsOperationPending(
grm.lock.RLock() grm.lock.RLock()
defer grm.lock.RUnlock() defer grm.lock.RUnlock()
exist, previousOpIndex := grm.isOperationExists(volumeName, podName) opKey := operationKey{volumeName, podName}
exist, previousOpIndex := grm.isOperationExists(opKey)
if exist && grm.operations[previousOpIndex].operationPending { if exist && grm.operations[previousOpIndex].operationPending {
return true return true
} }
@ -164,59 +163,49 @@ func (grm *nestedPendingOperations) IsOperationPending(
} }
// This is an internal function and caller should acquire and release the lock // This is an internal function and caller should acquire and release the lock
func (grm *nestedPendingOperations) isOperationExists( func (grm *nestedPendingOperations) isOperationExists(key operationKey) (bool, int) {
volumeName v1.UniqueVolumeName,
podName types.UniquePodName) (bool, int) {
// If volumeName is empty, operation can be executed concurrently // If volumeName is empty, operation can be executed concurrently
if volumeName == EmptyUniqueVolumeName { if key.volumeName == EmptyUniqueVolumeName {
return false, -1 return false, -1
} }
for previousOpIndex, previousOp := range grm.operations { for previousOpIndex, previousOp := range grm.operations {
if previousOp.volumeName != volumeName { volumeNameMatch := previousOp.key.volumeName == key.volumeName
// No match, keep searching
continue
}
if previousOp.podName != EmptyUniquePodName && podNameMatch := previousOp.key.podName == EmptyUniquePodName ||
podName != EmptyUniquePodName && key.podName == EmptyUniquePodName ||
previousOp.podName != podName { previousOp.key.podName == key.podName
// No match, keep searching
continue
}
// Match
return true, previousOpIndex if volumeNameMatch && podNameMatch {
return true, previousOpIndex
}
} }
return false, -1 return false, -1
} }
func (grm *nestedPendingOperations) getOperation( func (grm *nestedPendingOperations) getOperation(key operationKey) (uint, error) {
volumeName v1.UniqueVolumeName,
podName types.UniquePodName) (uint, error) {
// Assumes lock has been acquired by caller. // Assumes lock has been acquired by caller.
for i, op := range grm.operations { for i, op := range grm.operations {
if op.volumeName == volumeName && if op.key.volumeName == key.volumeName &&
op.podName == podName { op.key.podName == key.podName {
return uint(i), nil return uint(i), nil
} }
} }
logOperationKey := getOperationKey(volumeName, podName) return 0, fmt.Errorf("Operation %q not found", key)
return 0, fmt.Errorf("Operation %q not found", logOperationKey)
} }
func (grm *nestedPendingOperations) deleteOperation( func (grm *nestedPendingOperations) deleteOperation(key operationKey) {
// Assumes lock has been acquired by caller. // Assumes lock has been acquired by caller.
volumeName v1.UniqueVolumeName,
podName types.UniquePodName) {
opIndex := -1 opIndex := -1
for i, op := range grm.operations { for i, op := range grm.operations {
if op.volumeName == volumeName && if op.key.volumeName == key.volumeName &&
op.podName == podName { op.key.podName == key.podName {
opIndex = i opIndex = i
break break
} }
@ -227,8 +216,7 @@ func (grm *nestedPendingOperations) deleteOperation(
grm.operations = grm.operations[:len(grm.operations)-1] grm.operations = grm.operations[:len(grm.operations)-1]
} }
func (grm *nestedPendingOperations) operationComplete( func (grm *nestedPendingOperations) operationComplete(key operationKey, err *error) {
volumeName v1.UniqueVolumeName, podName types.UniquePodName, err *error) {
// Defer operations are executed in Last-In is First-Out order. In this case // Defer operations are executed in Last-In is First-Out order. In this case
// the lock is acquired first when operationCompletes begins, and is // the lock is acquired first when operationCompletes begins, and is
// released when the method finishes, after the lock is released cond is // released when the method finishes, after the lock is released cond is
@ -239,24 +227,20 @@ func (grm *nestedPendingOperations) operationComplete(
if *err == nil || !grm.exponentialBackOffOnError { if *err == nil || !grm.exponentialBackOffOnError {
// Operation completed without error, or exponentialBackOffOnError disabled // Operation completed without error, or exponentialBackOffOnError disabled
grm.deleteOperation(volumeName, podName) grm.deleteOperation(key)
if *err != nil { if *err != nil {
// Log error // Log error
logOperationKey := getOperationKey(volumeName, podName) klog.Errorf("operation %s failed with: %v", key, *err)
klog.Errorf("operation %s failed with: %v",
logOperationKey,
*err)
} }
return return
} }
// Operation completed with error and exponentialBackOffOnError Enabled // Operation completed with error and exponentialBackOffOnError Enabled
existingOpIndex, getOpErr := grm.getOperation(volumeName, podName) existingOpIndex, getOpErr := grm.getOperation(key)
if getOpErr != nil { if getOpErr != nil {
// Failed to find existing operation // Failed to find existing operation
logOperationKey := getOperationKey(volumeName, podName)
klog.Errorf("Operation %s completed. error: %v. exponentialBackOffOnError is enabled, but failed to get operation to update.", klog.Errorf("Operation %s completed. error: %v. exponentialBackOffOnError is enabled, but failed to get operation to update.",
logOperationKey, key,
*err) *err)
return return
} }
@ -265,10 +249,8 @@ func (grm *nestedPendingOperations) operationComplete(
grm.operations[existingOpIndex].operationPending = false grm.operations[existingOpIndex].operationPending = false
// Log error // Log error
operationKey :=
getOperationKey(volumeName, podName)
klog.Errorf("%v", grm.operations[existingOpIndex].expBackoff. klog.Errorf("%v", grm.operations[existingOpIndex].expBackoff.
GenerateNoRetriesPermittedMsg(operationKey)) GenerateNoRetriesPermittedMsg(key.String()))
} }
func (grm *nestedPendingOperations) Wait() { func (grm *nestedPendingOperations) Wait() {
@ -280,21 +262,22 @@ func (grm *nestedPendingOperations) Wait() {
} }
} }
func getOperationKey( type operationKey struct {
volumeName v1.UniqueVolumeName, podName types.UniquePodName) string { volumeName v1.UniqueVolumeName
podNameStr := "" podName types.UniquePodName
if podName != EmptyUniquePodName { }
podNameStr = fmt.Sprintf(" (%q)", podName)
} func (key operationKey) String() string {
podNameStr := fmt.Sprintf(" (%q)", key.podName)
return fmt.Sprintf("%q%s", return fmt.Sprintf("%q%s",
volumeName, key.volumeName,
podNameStr) podNameStr)
} }
// NewAlreadyExistsError returns a new instance of AlreadyExists error. // NewAlreadyExistsError returns a new instance of AlreadyExists error.
func NewAlreadyExistsError(operationKey string) error { func NewAlreadyExistsError(key operationKey) error {
return alreadyExistsError{operationKey} return alreadyExistsError{key}
} }
// IsAlreadyExists returns true if an error returned from // IsAlreadyExists returns true if an error returned from
@ -313,7 +296,7 @@ func IsAlreadyExists(err error) bool {
// new operation can not be started because an operation with the same operation // new operation can not be started because an operation with the same operation
// name is already executing. // name is already executing.
type alreadyExistsError struct { type alreadyExistsError struct {
operationKey string operationKey operationKey
} }
var _ error = alreadyExistsError{} var _ error = alreadyExistsError{}