Merge pull request #1929 from benluddy/installer-liveget-sync-receiver

OCPBUGS-45924: Fix "live get after write" in static pod installer controller.

Author: openshift-merge-bot[bot]

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)
+	}
+}