Fix "live get after write" in static pod installer controller.

The static pod installer controller builds an apply configuration for the status of the nodes it
manages based on potentially-stale state from an informer cache. The controller is written to assume
that it has observed the effect of its own previous writes. If the cache is stale, this assumption
can be violated which results in unpredictable installation decisions.

A mechanism was recently introduced requiring the installer controller to wait for its lister to
catch up to the latest version after performing a write. This mechanism did not work because it
depended on keeping state across calls to the Sync method, and because Sync has a value receiver,
field writes were not visible on subsequent calls.

Author: benluddy

pkg/operator/staticpod/controller/installer/installer_controller.go

@@ -53,6 +53,14 @@ const (
 //go:embed manifests/installer-pod.yaml
 var podTemplate []byte
 
+type OperatorClient interface {
+	GetStaticPodOperatorState() (spec *operatorv1.StaticPodOperatorSpec, status *operatorv1.StaticPodOperatorStatus, resourceVersion string, err error)
+	GetStaticPodOperatorStateWithQuorum(ctx context.Context) (spec *operatorv1.StaticPodOperatorSpec, status *operatorv1.StaticPodOperatorStatus, resourceVersion string, err error)
+	ApplyStaticPodOperatorStatus(ctx context.Context, fieldManager string, applyConfiguration *applyoperatorv1.StaticPodOperatorStatusApplyConfiguration) (err error)
+}
+
+var _ OperatorClient = v1helpers.StaticPodOperatorClient(nil)
+
 // InstallerController is a controller that watches the currentRevision and targetRevision fields for each node and spawn
 // installer pods to update the static pods on the master nodes.
 type InstallerController struct {
@@ -81,7 +89,7 @@ type InstallerController struct {
 	certSecrets    []UnrevisionedResource
 	certDir        string
 
-	operatorClient v1helpers.StaticPodOperatorClient
+	operatorClient OperatorClient
 
 	configMapsGetter corev1client.ConfigMapsGetter
 	secretsGetter    corev1client.SecretsGetter
@@ -92,7 +100,9 @@ type InstallerController struct {
 	// installerPodImageFn returns the image name for the installer pod
 	installerPodImageFn func() string
 	// ownerRefsFn sets the ownerrefs on the pruner pod
-	ownerRefsFn func(ctx context.Context, revision int32) ([]metav1.OwnerReference, error)
+	ownerRefsFn                    func(ctx context.Context, revision int32) ([]metav1.OwnerReference, error)
+	ensureRequiredResourcesExistFn func(ctx context.Context, revisionNumber int32) error
+	manageInstallationPodsFn       func(ctx context.Context, operatorSpec *operatorv1.StaticPodOperatorSpec, originalOperatorStatus *operatorv1.StaticPodOperatorStatus) (bool, time.Duration, *operatorv1.NodeStatus, func(), error)
 
 	installerPodMutationFns []InstallerPodMutationFunc
 
@@ -200,6 +210,8 @@ func NewInstallerController(
 	}
 
 	c.ownerRefsFn = c.setOwnerRefs
+	c.ensureRequiredResourcesExistFn = c.ensureRequiredResourcesExist
+	c.manageInstallationPodsFn = c.manageInstallationPods
 	c.factory = factory.New().
 		WithInformers(
 			operatorClient.Informer(),
@@ -1111,7 +1123,7 @@ func (c InstallerController) ensureRequiredResourcesExist(ctx context.Context, r
 	return fmt.Errorf("missing required resources: %v", aggregatedErr)
 }
 
-func (c InstallerController) Sync(ctx context.Context, syncCtx factory.SyncContext) error {
+func (c *InstallerController) Sync(ctx context.Context, syncCtx factory.SyncContext) error {
 	operatorSpec, originalOperatorStatus, operatorResourceVersion, err := c.operatorClient.GetStaticPodOperatorState()
 	if err != nil {
 		return err
@@ -1123,7 +1135,7 @@ func (c InstallerController) Sync(ctx context.Context, syncCtx factory.SyncConte
 	// Apply response.
 	if c.podOperatorStatusApplied {
 		if c.lastPodOperatorAppliedRV == 0 {
-			_, _, resourceVersion, err := c.operatorClient.GetOperatorStateWithQuorum(ctx)
+			_, _, resourceVersion, err := c.operatorClient.GetStaticPodOperatorStateWithQuorum(ctx)
 			if err != nil {
 				return err
 			}
@@ -1152,7 +1164,7 @@ func (c InstallerController) Sync(ctx context.Context, syncCtx factory.SyncConte
 		return nil
 	}
 
-	err = c.ensureRequiredResourcesExist(ctx, originalOperatorStatus.LatestAvailableRevision)
+	err = c.ensureRequiredResourcesExistFn(ctx, originalOperatorStatus.LatestAvailableRevision)
 
 	// Only manage installation pods when all required certs are present.
 	var updatedNode *operatorv1.NodeStatus
@@ -1161,7 +1173,7 @@ func (c InstallerController) Sync(ctx context.Context, syncCtx factory.SyncConte
 		var requeue bool
 		var after time.Duration
 		var syncErr error
-		requeue, after, updatedNode, updatedNodeReportOnSuccessfulUpdateFn, syncErr = c.manageInstallationPods(ctx, operatorSpec, operatorStatus)
+		requeue, after, updatedNode, updatedNodeReportOnSuccessfulUpdateFn, syncErr = c.manageInstallationPodsFn(ctx, operatorSpec, operatorStatus)
 		if requeue && syncErr == nil {
 			syncCtx.Queue().AddAfter(syncCtx.QueueKey(), after)
 			return nil

pkg/operator/staticpod/controller/installer/installer_controller_test.go

@@ -2651,3 +2651,121 @@ func metav1TimestampPtr(s string) *metav1.Time {
 	t := metav1Timestamp(s)
 	return &t
 }
+
+type GenericStaticPodOperator struct {
+	metav1.ObjectMeta `json:"metadata"`
+	Spec              operatorv1.StaticPodOperatorSpec   `json:"spec"`
+	Status            operatorv1.StaticPodOperatorStatus `json:"status"`
+}
+
+type StubOperatorClient struct {
+	live    *GenericStaticPodOperator
+	cached  *GenericStaticPodOperator
+	onApply func()
+}
+
+func (stub *StubOperatorClient) GetStaticPodOperatorState() (*operatorv1.StaticPodOperatorSpec, *operatorv1.StaticPodOperatorStatus, string, error) {
+	return &stub.cached.Spec, &stub.cached.Status, stub.cached.ResourceVersion, nil
+}
+
+func (stub *StubOperatorClient) GetStaticPodOperatorStateWithQuorum(context.Context) (*operatorv1.StaticPodOperatorSpec, *operatorv1.StaticPodOperatorStatus, string, error) {
+	return &stub.live.Spec, &stub.live.Status, stub.live.ResourceVersion, nil
+}
+
+func (stub *StubOperatorClient) ApplyStaticPodOperatorStatus(context.Context, string, *applyoperatorv1.StaticPodOperatorStatusApplyConfiguration) error {
+	stub.onApply()
+	return nil
+}
+
+func TestWaitToObserveWrites(t *testing.T) {
+	live := GenericStaticPodOperator{
+		ObjectMeta: metav1.ObjectMeta{
+			ResourceVersion: "1",
+		},
+		Status: operatorv1.StaticPodOperatorStatus{
+			NodeStatuses: []operatorv1.NodeStatus{
+				{NodeName: "test-node"},
+			},
+		},
+	}
+	cached := GenericStaticPodOperator{
+		ObjectMeta: metav1.ObjectMeta{
+			ResourceVersion: "1",
+		},
+		Status: operatorv1.StaticPodOperatorStatus{
+			NodeStatuses: []operatorv1.NodeStatus{
+				{NodeName: "test-node"},
+			},
+		},
+	}
+
+	nApplies := 0
+	operatorClient := StubOperatorClient{
+		live:   &live,
+		cached: &cached,
+		onApply: func() {
+			nApplies++
+			live = GenericStaticPodOperator{
+				ObjectMeta: metav1.ObjectMeta{
+					ResourceVersion: "2",
+				},
+				Status: operatorv1.StaticPodOperatorStatus{
+					NodeStatuses: []operatorv1.NodeStatus{
+						{NodeName: "test-node", TargetRevision: 1},
+					},
+				},
+			}
+		},
+	}
+
+	ctrl := InstallerController{
+		operatorClient: &operatorClient,
+		ensureRequiredResourcesExistFn: func(context.Context, int32) error {
+			return nil
+		},
+		manageInstallationPodsFn: func(context.Context, *operatorv1.StaticPodOperatorSpec, *operatorv1.StaticPodOperatorStatus) (bool, time.Duration, *operatorv1.NodeStatus, func(), error) {
+			return false, 0, &operatorv1.NodeStatus{NodeName: "test-node", TargetRevision: 1}, func() {}, nil
+		},
+	}
+
+	recorder := events.NewInMemoryRecorder("test", clocktesting.NewFakeClock(time.Date(1, 2, 3, 4, 5, 6, 7, time.UTC)))
+
+	// Write TargetRevision change.
+	if err := ctrl.Sync(context.TODO(), factory.NewSyncContext("test", recorder)); err != nil {
+		t.Fatal(err)
+	}
+
+	if want := 1; nApplies != want {
+		t.Fatalf("expected %d status apply, got %d", want, nApplies)
+	}
+
+	// Cache has not observed the previous write!
+	if err := ctrl.Sync(context.TODO(), factory.NewSyncContext("test", recorder)); err != nil {
+		t.Fatal(err)
+	}
+
+	if want := 1; nApplies != want {
+		t.Fatalf("expected %d status apply, got %d", want, nApplies)
+	}
+
+	// Cache observes the write.
+	cached = GenericStaticPodOperator{
+		ObjectMeta: metav1.ObjectMeta{
+			ResourceVersion: "2",
+		},
+		Status: operatorv1.StaticPodOperatorStatus{
+			NodeStatuses: []operatorv1.NodeStatus{
+				{NodeName: "test-node", TargetRevision: 1},
+			},
+		},
+	}
+
+	// Sync can continue with the next state transition now.
+	if err := ctrl.Sync(context.TODO(), factory.NewSyncContext("test", recorder)); err != nil {
+		t.Fatal(err)
+	}
+
+	if want := 2; nApplies != want {
+		t.Fatalf("expected %d status apply, got %d", want, nApplies)
+	}
+}