Repository: keikoproj/upgrade-manager (go)

Issue:
Check node object is gone before proceeding
**Is this a BUG REPORT or FEATURE REQUEST?**:
Feature Request

**What happened**:
Rolling upgrade waits a predefined amount of time after terminating an instance, but even if the node object is still around, it moves on to terminate the next instance

**What you expected to happen**:
it would be nice if outside of waiting the post-terminate defined time, rollingupgrade also verified that the node object is gone before moving to the next instance.
This is very useful with components such as `lifecycle-manager` for cases where there is a delay with the termination of a node.



Pull request:
wait for node unjoined
Fixes #33 

This PR waits for node object to be unjoined before proceeding with rolling upgrade.

There is a hard limit on how long to wait which is now capped at an hour, if timeout occurs, upgrade will disregard and proceed anyway - we can make this configurable in CR if desired.

Added some unit test for new method.

## Testing

- Unit tests passing
- Manual testing - TBD in progress

Patch:
diff --git a/controllers/rollingupgrade_controller.go b/controllers/rollingupgrade_controller.go
index c49eaf9..73620ce 100644
--- a/controllers/rollingupgrade_controller.go
+++ b/controllers/rollingupgrade_controller.go
@@ -41,6 +41,7 @@ import (
 	"github.com/go-logr/logr"
 	log "github.com/keikoproj/upgrade-manager/pkg/log"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	v1errors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -76,6 +77,13 @@ const (
 	ShellBinary = "/bin/sh"
 )
 
+var (
+	// TerminationTimeoutSeconds is the timeout threshold for waiting for a node object unjoin
+	TerminationTimeoutSeconds = 3600
+	// TerminationSleepIntervalSeconds is the polling interval for checking if a node object is unjoined
+	TerminationSleepIntervalSeconds = 30
+)
+
 var DefaultRetryer = awsclient.DefaultRetryer{
 	NumMaxRetries:    250,
 	MinThrottleDelay: time.Second * 5,
@@ -222,6 +230,28 @@ func (r *RollingUpgradeReconciler) CallKubectlDrain(ctx context.Context, nodeNam
 	errChan <- nil
 }
 
+func (r *RollingUpgradeReconciler) WaitForTermination(nodeName string, nodeInterface v1.NodeInterface) (bool, error) {
+	var (
+		started = time.Now()
+	)
+	for {
+		if time.Since(started) >= (time.Second * time.Duration(TerminationTimeoutSeconds)) {
+			log.Println("WaitForTermination timed out while waiting for node to unjoin")
+			return false, nil
+		}
+
+		_, err := nodeInterface.Get(nodeName, metav1.GetOptions{})
+		if v1errors.IsNotFound(err) {
+			log.Printf("node %s is unjoined from cluster, upgrade will proceed", nodeName)
+			break
+		}
+
+		log.Printf("node %s is still joined to clutster, will wait %vs and retry", nodeName, TerminationSleepIntervalSeconds)
+		time.Sleep(time.Duration(TerminationSleepIntervalSeconds) * time.Second)
+	}
+	return true, nil
+}
+
 // TerminateNode actually terminates the given node.
 func (r *RollingUpgradeReconciler) TerminateNode(ruObj *upgrademgrv1alpha1.RollingUpgrade, instanceID string, svc autoscalingiface.AutoScalingAPI) error {
 
@@ -761,6 +791,16 @@ func (r *RollingUpgradeReconciler) UpdateInstance(ctx *context.Context,
 		return
 	}
 
+	unjoined, err := r.WaitForTermination(nodeName, r.generatedClient.CoreV1().Nodes())
+	if err != nil {
+		ch <- err
+		return
+	}
+
+	if !unjoined {
+		log.Warnf("termination waiter completed but %s is still joined, will proceed with upgrade", nodeName)
+	}
+
 	ruObj.Status.NodesProcessed = ruObj.Status.NodesProcessed + 1
 	r.Update(*ctx, ruObj)
 


Test patch:
diff --git a/controllers/rollingupgrade_controller_test.go b/controllers/rollingupgrade_controller_test.go
index 08bb31a..275efae 100644
--- a/controllers/rollingupgrade_controller_test.go
+++ b/controllers/rollingupgrade_controller_test.go
@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
 	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
@@ -1928,6 +1929,44 @@ func TestRunRestackNoNodeInAsg(t *testing.T) {
 	g.Expect(err).To(gomega.BeNil())
 }
 
+func TestWaitForTermination(t *testing.T) {
+	g := gomega.NewGomegaWithT(t)
+
+	TerminationTimeoutSeconds = 1
+	TerminationSleepIntervalSeconds = 1
+
+	mockNodeName := "node-123"
+	mockNode := &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: mockNodeName,
+		},
+	}
+	kuberenetesClient := fake.NewSimpleClientset()
+	nodeInterface := kuberenetesClient.CoreV1().Nodes()
+
+	mgr, err := manager.New(cfg, manager.Options{})
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+
+	rcRollingUpgrade := &RollingUpgradeReconciler{
+		Client:          mgr.GetClient(),
+		generatedClient: kubernetes.NewForConfigOrDie(mgr.GetConfig()),
+		admissionMap:    sync.Map{},
+		ruObjNameToASG:  sync.Map{},
+		ClusterState:    NewClusterState(),
+	}
+	nodeInterface.Create(mockNode)
+
+	unjoined, err := rcRollingUpgrade.WaitForTermination(mockNodeName, nodeInterface)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	g.Expect(unjoined).To(gomega.BeFalse())
+
+	nodeInterface.Delete(mockNodeName, &metav1.DeleteOptions{})
+
+	unjoined, err = rcRollingUpgrade.WaitForTermination(mockNodeName, nodeInterface)
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	g.Expect(unjoined).To(gomega.BeTrue())
+}
+
 func TestRunRestackWithNodesLessThanMaxUnavailable(t *testing.T) {
 	g := gomega.NewGomegaWithT(t)