prep codebase for addition of integration tests (#144)

Signed-off-by: Alex Price <aprice@atlassian.com>

Author: awprice

cmd/manager/main.go

@@ -49,6 +49,18 @@ var (
 	defaultCNScyclingExpiry          = app.Flag("default-cns-cycling-expiry", "Fail the CNS if it has been cycling for this long").Default("3h").Duration()
 	unhealthyPodTerminationThreshold = app.Flag("unhealthy-pod-termination-after", "How long to tolerate an un-evictable yet unhealthy pod before forcefully removing it").Default("5m").Duration()
 
+	cnrScaleUpWait              = app.Flag("cnr-scale-up-wait", "Minimum time to wait after scaling up before checking if replacement nodes are Ready").Default("1m").Duration()
+	cnrScaleUpLimit             = app.Flag("cnr-scale-up-limit", "Maximum total time to wait for replacement nodes to come up before failing the CNR").Default("20m").Duration()
+	cnrNodeEquilibriumWaitLimit = app.Flag("cnr-node-equilibrium-wait-limit", "Maximum time to wait for the kube-node-set and cloud-provider-instance-set to converge during the Initialised phase").Default("5m").Duration()
+	cnrTransitionDuration       = app.Flag("cnr-transition-duration", "RequeueAfter used when moving the CNR between phases").Default("10s").Duration()
+	cnrRequeueDuration          = app.Flag("cnr-requeue-duration", "RequeueAfter used while the CNR is waiting on an external condition within a phase").Default("30s").Duration()
+
+	cnsTransitionDuration        = app.Flag("cns-transition-duration", "RequeueAfter used when moving the CNS between phases").Default("10s").Duration()
+	cnsWaitingPodsRequeue        = app.Flag("cns-waiting-pods-requeue", "RequeueAfter used while waiting for pods on the cycling node to finish naturally (Method=Wait)").Default("60s").Duration()
+	cnsRemovingLabelsPodsRequeue = app.Flag("cns-removing-labels-pods-requeue", "RequeueAfter used while removing labels from pods on the cycling node").Default("1s").Duration()
+	cnsDrainingRetryRequeue      = app.Flag("cns-draining-retry-requeue", "RequeueAfter used when the apiserver returns 429 TooManyRequests (PDB-blocked) during drain").Default("15s").Duration()
+	cnsDrainingPodsRequeue       = app.Flag("cns-draining-pods-requeue", "RequeueAfter used while waiting for the in-flight drain to finish").Default("30s").Duration()
+
 	nodeControllerReconcileConcurrency = app.Flag("node-controller-reconcile-concurrency", "Maximum number of concurrent node controller reconciles").Default("1").Int()
 	nodeControllerRequeueAfter         = app.Flag("node-controller-requeue-after", "How often the node controller rechecks annotated nodes that are still covered by an active CNR").Default("5m").Duration()
 )
@@ -120,16 +132,26 @@ func main() {
 
 	// Configure the CNR transitioner options
 	cnrOptions := cnrTransitioner.Options{
-		DeleteCNR:          *deleteCNR,
-		DeleteCNRExpiry:    *deleteCNRExpiry,
-		DeleteCNRRequeue:   *deleteCNRRequeue,
-		HealthCheckTimeout: *healthCheckTimeout,
+		DeleteCNR:                *deleteCNR,
+		DeleteCNRExpiry:          *deleteCNRExpiry,
+		DeleteCNRRequeue:         *deleteCNRRequeue,
+		HealthCheckTimeout:       *healthCheckTimeout,
+		ScaleUpWait:              *cnrScaleUpWait,
+		ScaleUpLimit:             *cnrScaleUpLimit,
+		NodeEquilibriumWaitLimit: *cnrNodeEquilibriumWaitLimit,
+		TransitionDuration:       *cnrTransitionDuration,
+		RequeueDuration:          *cnrRequeueDuration,
 	}
 
 	// Configure the CNS transitioner options
 	cnsOptions := cnsTransitioner.Options{
 		DefaultCNScyclingExpiry:          *defaultCNScyclingExpiry,
 		UnhealthyPodTerminationThreshold: *unhealthyPodTerminationThreshold,
+		TransitionDuration:               *cnsTransitionDuration,
+		WaitingPodsRequeue:               *cnsWaitingPodsRequeue,
+		RemovingLabelsPodsRequeue:        *cnsRemovingLabelsPodsRequeue,
+		DrainingRetryRequeue:             *cnsDrainingRetryRequeue,
+		DrainingPodsRequeue:              *cnsDrainingPodsRequeue,
 	}
 
 	// Configure the node controller options

pkg/cloudprovider/aws/builder.go

@@ -4,11 +4,12 @@ import (
 	"fmt"
 
 	"github.com/atlassian-labs/cyclops/pkg/cloudprovider"
-	fakeaws "github.com/atlassian-labs/cyclops/pkg/cloudprovider/aws/fake"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/autoscaling"
+	"github.com/aws/aws-sdk-go/service/autoscaling/autoscalingiface"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/aws/aws-sdk-go/service/ec2/ec2iface"
 	"github.com/go-logr/logr"
 )
 
@@ -52,10 +53,13 @@ func NewCloudProvider(logger logr.Logger) (cloudprovider.CloudProvider, error) {
 	return p, nil
 }
 
-// NewGenericCloudProvider returns a new mock AWS cloud provider
-func NewGenericCloudProvider(autoscalingiface *fakeaws.Autoscaling, ec2iface *fakeaws.Ec2) cloudprovider.CloudProvider {
+// NewGenericCloudProvider returns a cloud provider built around the supplied
+// Autoscaling and EC2 clients. Production code uses NewCloudProvider; this
+// constructor lets tests inject fakes (or instrumented decorators around the
+// real clients).
+func NewGenericCloudProvider(asg autoscalingiface.AutoScalingAPI, ec2c ec2iface.EC2API) cloudprovider.CloudProvider {
 	return &provider{
-		autoScalingService: autoscalingiface,
-		ec2Service:         ec2iface,
+		autoScalingService: asg,
+		ec2Service:         ec2c,
 	}
 }

pkg/controller/cyclenoderequest/transitioner/integration_test.go

@@ -34,7 +34,7 @@ func TestOriginalIssue_IOTimeoutCausesHealing(t *testing.T) {
 		}
 
 		rm := createTestResourceManager(t, cnr, node, mockCP)
-		transitioner := NewCycleNodeRequestTransitioner(cnr, rm, Options{})
+		transitioner := NewCycleNodeRequestTransitioner(cnr, rm, defaultTestTransitionerOptions())
 
 		// Execute
 		_, err := transitioner.transitionPending()
@@ -58,7 +58,7 @@ func TestOriginalIssue_IOTimeoutCausesHealing(t *testing.T) {
 		}
 
 		rm := createTestResourceManager(t, cnr, node, mockCP)
-		transitioner := NewCycleNodeRequestTransitioner(cnr, rm, Options{})
+		transitioner := NewCycleNodeRequestTransitioner(cnr, rm, defaultTestTransitionerOptions())
 
 		// Execute
 		result, err := transitioner.transitionPending()
@@ -139,7 +139,7 @@ func TestEndToEnd_TransientErrorRecovery(t *testing.T) {
 			recovered := false
 
 			for attempt := 1; attempt <= maxAttempts; attempt++ {
-				transitioner := NewCycleNodeRequestTransitioner(cnr, rm, Options{})
+				transitioner := NewCycleNodeRequestTransitioner(cnr, rm, defaultTestTransitionerOptions())
 				result, err := transitioner.transitionPending()
 
 				t.Logf("Attempt %d: Phase=%s, Requeue=%v, Error=%v",
@@ -209,7 +209,7 @@ func TestEndToEnd_PermanentErrorsStillFail(t *testing.T) {
 			}
 
 			rm := createTestResourceManager(t, cnr, node, mockCP)
-			transitioner := NewCycleNodeRequestTransitioner(cnr, rm, Options{})
+			transitioner := NewCycleNodeRequestTransitioner(cnr, rm, defaultTestTransitionerOptions())
 
 			// Execute
 			result, err := transitioner.transitionPending()
@@ -246,7 +246,7 @@ func TestEndToEnd_EquilibriumTimeout(t *testing.T) {
 		}
 
 		rm := createTestResourceManager(t, cnr, node, mockCP)
-		transitioner := NewCycleNodeRequestTransitioner(cnr, rm, Options{})
+		transitioner := NewCycleNodeRequestTransitioner(cnr, rm, defaultTestTransitionerOptions())
 
 		// Execute
 		_, err := transitioner.transitionPending()
@@ -279,7 +279,7 @@ func TestEndToEnd_CompareBeforeAndAfter(t *testing.T) {
 		node := createTestNode("test-node-1", "aws:///us-west-2a/i-1234567890abcdef0")
 		mockCP := &mockCloudProviderAlwaysFails{error: testError}
 		rm := createTestResourceManager(t, cnr, node, mockCP)
-		transitioner := NewCycleNodeRequestTransitioner(cnr, rm, Options{})
+		transitioner := NewCycleNodeRequestTransitioner(cnr, rm, defaultTestTransitionerOptions())
 
 		result, err := transitioner.transitionPending()
 

pkg/controller/cyclenoderequest/transitioner/test_helpers.go

@@ -2,6 +2,7 @@ package transitioner
 
 import (
 	"net/http"
+	"time"
 
 	v1 "github.com/atlassian-labs/cyclops/pkg/apis/atlassian/v1"
 	"github.com/atlassian-labs/cyclops/pkg/controller"
@@ -10,6 +11,20 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+// defaultTestTransitionerOptions mirrors the production defaults declared
+// in cmd/manager/main.go so unit tests using NewFakeTransitioner behave
+// the same way as the running operator. Individual tests can replace any
+// field via WithTransitionerOptions.
+func defaultTestTransitionerOptions() Options {
+	return Options{
+		ScaleUpWait:              1 * time.Minute,
+		ScaleUpLimit:             20 * time.Minute,
+		NodeEquilibriumWaitLimit: 5 * time.Minute,
+		TransitionDuration:       10 * time.Second,
+		RequeueDuration:          30 * time.Second,
+	}
+}
+
 type Option func(t *Transitioner)
 
 func WithCloudProviderInstances(nodes []*mock.Node) Option {
@@ -57,7 +72,7 @@ func NewFakeTransitioner(cnr *v1.CycleNodeRequest, opts ...Option) *Transitioner
 		CloudProviderInstances: make([]*mock.Node, 0),
 		KubeNodes:              make([]*mock.Node, 0),
 		extraKubeObjects:       []client.Object{cnr},
-		transitionerOptions:    Options{},
+		transitionerOptions:    defaultTestTransitionerOptions(),
 	}
 
 	for _, opt := range opts {

pkg/controller/cyclenoderequest/transitioner/transitioner.go

@@ -33,19 +33,18 @@ const (
 // Value: "false" or missing/empty → default enabled (annotation management enabled)
 const nodeGroupAnnotationKey = "cyclops.atlassian.com/disable-annotation-management"
 
-var (
-	transitionDuration = 10 * time.Second
-	requeueDuration    = 30 * time.Second
-)
-
 // CycleNodeRequestTransitioner takes a cycleNodeRequest and attempts to transition it to the next phase
 type CycleNodeRequestTransitioner struct {
 	cycleNodeRequest *v1.CycleNodeRequest
 	rm               *controller.ResourceManager
 	options          Options
 }
 
-// Options stores configurable options for the CycleNodeRequestTransitioner
+// Options stores configurable options for the CycleNodeRequestTransitioner.
+//
+// All fields are required to be set by the caller. The cyclops manager
+// (cmd/manager/main.go) provides defaults via kingpin CLI flags; tests
+// construct Options directly with whatever values they need.
 type Options struct {
 	// DeleteCNR enables/disables deleting successful CycleNodeRequests after a certain amount of time
 	DeleteCNR bool
@@ -58,9 +57,32 @@ type Options struct {
 
 	// HealthCheckTimeout controls the duration of the timeout period for health checks performed on nodes
 	HealthCheckTimeout time.Duration
+
+	// ScaleUpWait is the minimum time the transitioner waits after detaching
+	// instances before checking whether replacement Kubernetes nodes have
+	// become Ready.
+	ScaleUpWait time.Duration
+
+	// ScaleUpLimit is the maximum total time spent waiting for replacement
+	// nodes to come up before the CNR transitions to Healing.
+	ScaleUpLimit time.Duration
+
+	// NodeEquilibriumWaitLimit caps how long the transitioner will wait for
+	// the kube-node-set and cloud-provider-instance-set to converge during
+	// the Initialised phase.
+	NodeEquilibriumWaitLimit time.Duration
+
+	// TransitionDuration is the RequeueAfter used when moving the CNR
+	// between phases.
+	TransitionDuration time.Duration
+
+	// RequeueDuration is the RequeueAfter used while the CNR is waiting on
+	// an external condition within a phase (e.g. ScalingUp readiness,
+	// WaitingTermination).
+	RequeueDuration time.Duration
 }
 
-// NewCycleNodeRequestTransitioner returns a new cycleNodeRequest transitioner
+// NewCycleNodeRequestTransitioner returns a new cycleNodeRequest transitioner.
 func NewCycleNodeRequestTransitioner(
 	cycleNodeRequest *v1.CycleNodeRequest,
 	rm *controller.ResourceManager,

pkg/controller/cyclenoderequest/transitioner/transitions.go

@@ -18,12 +18,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
-const (
-	scaleUpWait              = 1 * time.Minute
-	scaleUpLimit             = 20 * time.Minute
-	nodeEquilibriumWaitLimit = 5 * time.Minute
-)
-
 // transitionUndefined transitions any CycleNodeRequests in the undefined phase to the pending phase
 // It checks to ensure that a valid selector has been provided.
 func (t *CycleNodeRequestTransitioner) transitionUndefined() (reconcile.Result, error) {
@@ -81,7 +75,7 @@ func (t *CycleNodeRequestTransitioner) transitionPending() (reconcile.Result, er
 			// Requeue with backoff instead of transitioning to Healing
 			return reconcile.Result{
 				Requeue:      true,
-				RequeueAfter: requeueDuration,
+				RequeueAfter: t.options.RequeueDuration,
 			}, nil
 		}
 		// Non-retryable error, transition to Healing
@@ -123,7 +117,7 @@ func (t *CycleNodeRequestTransitioner) transitionPending() (reconcile.Result, er
 		// After working through these attempts, requeue to run through the Pending phase from the
 		// beginning to check the full state of nodes again. If there are any problem nodes we should
 		// not proceed and keep requeuing until the state is fixed or the timeout has been reached.
-		return reconcile.Result{Requeue: true, RequeueAfter: requeueDuration}, nil
+		return reconcile.Result{Requeue: true, RequeueAfter: t.options.RequeueDuration}, nil
 	}
 
 	valid, err := t.validateInstanceState(validNodeGroupInstances)
@@ -133,7 +127,7 @@ func (t *CycleNodeRequestTransitioner) transitionPending() (reconcile.Result, er
 
 	if !valid {
 		t.rm.Logger.Info("instance state not valid, requeuing")
-		return reconcile.Result{Requeue: true, RequeueAfter: requeueDuration}, nil
+		return reconcile.Result{Requeue: true, RequeueAfter: t.options.RequeueDuration}, nil
 	}
 
 	t.rm.Logger.Info("instance state valid, proceeding")
@@ -319,17 +313,17 @@ func (t *CycleNodeRequestTransitioner) transitionScalingUp() (reconcile.Result,
 	scaleUpStarted := t.cycleNodeRequest.Status.ScaleUpStarted
 
 	// Check we have waited long enough - give the node some time to start up
-	if time.Since(scaleUpStarted.Time) <= scaleUpWait {
+	if time.Since(scaleUpStarted.Time) <= t.options.ScaleUpWait {
 		t.rm.LogEvent(t.cycleNodeRequest, "ScalingUpWaiting", "Waiting for new nodes to be warmed up")
-		return reconcile.Result{Requeue: true, RequeueAfter: scaleUpWait}, nil
+		return reconcile.Result{Requeue: true, RequeueAfter: t.options.ScaleUpWait}, nil
 	}
 
 	nodeGroups, err := t.rm.CloudProvider.GetNodeGroups(t.cycleNodeRequest.GetNodeGroupNames())
 	if err != nil {
 		return t.transitionToHealing(err)
 	}
 	// If we have exceeded the max scale up time, then fail
-	if scaleUpStarted.Add(scaleUpLimit).Before(time.Now()) {
+	if scaleUpStarted.Add(t.options.ScaleUpLimit).Before(time.Now()) {
 		return t.transitionToHealing(
 			fmt.Errorf("all nodes failed to come up in time - instances not ready in cloud provider: %+v",
 				nodeGroups.NotReadyInstances()))
@@ -376,7 +370,7 @@ func (t *CycleNodeRequestTransitioner) transitionScalingUp() (reconcile.Result,
 	// also check if at least one Ready node was created after the scaleUpStarted time
 	if !allInstancesReady || !allKubernetesNodesReady || numNodesCreatedAfterScaleUpStarted <= 0 {
 		t.rm.LogEvent(t.cycleNodeRequest, "ScalingUpWaiting", "Waiting for new nodes to be ready")
-		return reconcile.Result{Requeue: true, RequeueAfter: requeueDuration}, nil
+		return reconcile.Result{Requeue: true, RequeueAfter: t.options.RequeueDuration}, nil
 	}
 
 	// Remove any nodes from the CNR object which are found to have been removed prematurely due to a race condition
@@ -404,7 +398,7 @@ func (t *CycleNodeRequestTransitioner) transitionScalingUp() (reconcile.Result,
 				return t.transitionToHealing(err)
 			}
 
-			return reconcile.Result{Requeue: true, RequeueAfter: requeueDuration}, nil
+			return reconcile.Result{Requeue: true, RequeueAfter: t.options.RequeueDuration}, nil
 		}
 	}
 
@@ -508,7 +502,7 @@ func (t *CycleNodeRequestTransitioner) transitionCordoning() (reconcile.Result,
 			return t.transitionToHealing(err)
 		}
 
-		return reconcile.Result{Requeue: true, RequeueAfter: requeueDuration}, nil
+		return reconcile.Result{Requeue: true, RequeueAfter: t.options.RequeueDuration}, nil
 	}
 
 	// The scale up + cordon is finished, we no longer need this list of nodes
@@ -617,7 +611,7 @@ func (t *CycleNodeRequestTransitioner) transitionFailed() (reconcile.Result, err
 		return t.transitionToFailed(err)
 	}
 	if shouldRequeue {
-		return reconcile.Result{Requeue: true, RequeueAfter: transitionDuration}, nil
+		return reconcile.Result{Requeue: true, RequeueAfter: t.options.TransitionDuration}, nil
 	}
 
 	return reconcile.Result{}, nil
@@ -631,7 +625,7 @@ func (t *CycleNodeRequestTransitioner) transitionSuccessful() (reconcile.Result,
 	}
 
 	if shouldRequeue {
-		return reconcile.Result{Requeue: true, RequeueAfter: transitionDuration}, nil
+		return reconcile.Result{Requeue: true, RequeueAfter: t.options.TransitionDuration}, nil
 	}
 
 	// Delete failed sibling CNRs regardless of whether the CNR for the

pkg/controller/cyclenoderequest/transitioner/transitions_pending_test.go

@@ -537,7 +537,7 @@ func TestPendingTimeoutReached(t *testing.T) {
 
 	// Simulate waiting for 1s more than the wait limit
 	cnr.Status.EquilibriumWaitStarted = &metav1.Time{
-		Time: time.Now().Add(-nodeEquilibriumWaitLimit - time.Second),
+		Time: time.Now().Add(-(5 * time.Minute) - time.Second),
 	}
 
 	// This time should transition to the healing phase
@@ -593,7 +593,7 @@ func TestPendingReattachedCloudProviderNode(t *testing.T) {
 
 	// Simulate waiting for 1s less than the wait limit
 	cnr.Status.EquilibriumWaitStarted = &metav1.Time{
-		Time: time.Now().Add(-nodeEquilibriumWaitLimit + time.Second),
+		Time: time.Now().Add(-(5 * time.Minute) + time.Second),
 	}
 
 	_, err = fakeTransitioner.Autoscaling.AttachInstances(&autoscaling.AttachInstancesInput{
@@ -661,7 +661,7 @@ func TestPendingReattachedCloudProviderNodeTooLate(t *testing.T) {
 
 	// Simulate waiting for 1s more than the wait limit
 	cnr.Status.EquilibriumWaitStarted = &metav1.Time{
-		Time: time.Now().Add(-nodeEquilibriumWaitLimit - time.Second),
+		Time: time.Now().Add(-(5 * time.Minute) - time.Second),
 	}
 
 	_, err = fakeTransitioner.Autoscaling.AttachInstances(&autoscaling.AttachInstancesInput{

pkg/controller/cyclenoderequest/transitioner/util.go

@@ -113,7 +113,7 @@ func (t *CycleNodeRequestTransitioner) transitionObject(desiredPhase v1.CycleNod
 
 	return reconcile.Result{
 		Requeue:      true,
-		RequeueAfter: transitionDuration,
+		RequeueAfter: t.options.TransitionDuration,
 	}, nil
 }
 
@@ -132,7 +132,7 @@ func (t *CycleNodeRequestTransitioner) equilibriumWaitTimedOut() (bool, error) {
 		}
 	}
 
-	return time.Now().After(t.cycleNodeRequest.Status.EquilibriumWaitStarted.Add(nodeEquilibriumWaitLimit)), nil
+	return time.Now().After(t.cycleNodeRequest.Status.EquilibriumWaitStarted.Add(t.options.NodeEquilibriumWaitLimit)), nil
 }
 
 // reapChildren reaps CycleNodeStatus children. It returns the state that should be
@@ -481,7 +481,7 @@ func (t *CycleNodeRequestTransitioner) errorIfEquilibriumTimeoutReached() error
 	if timedOut {
 		return fmt.Errorf(
 			"node count mismatch, number of kubernetes nodes does not match number of cloud provider instances after %v",
-			nodeEquilibriumWaitLimit,
+			t.options.NodeEquilibriumWaitLimit,
 		)
 	}