craigcondit commented on code in PR #565:
URL: https://github.com/apache/yunikorn-k8shim/pull/565#discussion_r1155347692
##########
pkg/appmgmt/appmgmt_recovery.go:
##########
@@ -71,33 +71,47 @@ func (svc *AppManagementService) recoverApps()
(map[string]interfaces.ManagedApp
return recoveringApps, nil
}
-func (svc *AppManagementService) waitForAppRecovery(
- recoveringApps map[string]interfaces.ManagedApp, maxTimeout
time.Duration) error {
- if len(recoveringApps) > 0 {
- log.Logger().Info("wait for app recovery",
- zap.Int("appToRecover", len(recoveringApps)))
- // check app states periodically, ensure all apps exit from
recovering state
- if err := utils.WaitForCondition(func() bool {
- for _, app := range recoveringApps {
- log.Logger().Debug("appInfo",
- zap.String("appId",
app.GetApplicationID()),
- zap.String("state",
app.GetApplicationState()))
- if app.GetApplicationState() ==
cache.ApplicationStates().Accepted {
- delete(recoveringApps,
app.GetApplicationID())
- }
- }
-
- if len(recoveringApps) == 0 {
- log.Logger().Info("app recovery is successful")
- return true
- }
-
+func (svc *AppManagementService) waitForAppRecovery(recoveringApps
map[string]interfaces.ManagedApp) bool {
+ svc.cancelRecovery.Store(false) // reset cancellation token
+ recoveryStartTime := time.Now()
+ counter := 0
+ for {
+ // check for cancellation token
+ if svc.cancelRecovery.Load() {
+ log.Logger().Info("Waiting for recovery canceled.")
+ svc.cancelRecovery.Store(false)
return false
- }, 1*time.Second, maxTimeout); err != nil {
- return fmt.Errorf("timeout waiting for app recovery in
%s",
- maxTimeout.String())
}
+
+ svc.removeRecoveredApps(recoveringApps)
+ if len(recoveringApps) == 0 {
+ log.Logger().Info("Application recovery complete.")
+ return true
+ }
+ counter++
+ if counter%10 == 0 {
+ log.Logger().Info("Waiting for application recovery",
+ zap.Duration("timeElapsed",
time.Since(recoveryStartTime).Round(time.Second)),
+ zap.Int("appsRemaining", len(recoveringApps)))
+ }
+ time.Sleep(1 * time.Second)
}
+}
- return nil
+// cancelWaitForAppRecovery is used by testing code to ensure that
waitForAppRecovery() does not block forever
+func (svc *AppManagementService) cancelWaitForAppRecovery() {
+ svc.cancelRecovery.Store(true)
+}
+
+// removeRecoveredApps() is used to walk the currently recovering apps list
and remove those that have finished recovering
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
