React to Program objects changing

At present the controller will notice changes to a Program object when
it reruns a Stack that refers to that object, either because it failed
the previous time, or because it requeued it on a schedule.

This adds an index keeping track of which Stacks reference which
Programs, and a watch that will requeue all the Stacks referring to a
Program when that program changes.

Signed-off-by: Michael Bridgen <[email protected]>

pkg/controller/stack/stack_controller.go

@@ -44,6 +44,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	ctrlhandler "sigs.k8s.io/controller-runtime/pkg/handler"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
@@ -60,6 +61,7 @@ var (
 const (
 	pulumiFinalizer                = "finalizer.stack.pulumi.com"
 	defaultMaxConcurrentReconciles = 10
+	programRefIndexFieldName       = ".spec.programRef.name" // this is an arbitrary string, named for the field it indexes
 )
 
 const (
@@ -148,7 +150,38 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		return err
 	}
 
-	return nil
+	// Watch Programs, and look up which (if any) Stack refers to them when they change
+	indexer := mgr.GetFieldIndexer()
+	if err = indexer.IndexField(context.Background(), &pulumiv1.Stack{}, programRefIndexFieldName, func(o client.Object) []string {
+		stack := o.(*pulumiv1.Stack)
+		if stack.Spec.ProgramRef != nil {
+			return []string{stack.Spec.ProgramRef.Name}
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	enqueueStacksForProgram := func(program client.Object) []reconcile.Request {
+		var stacks pulumiv1.StackList
+		err := mgr.GetClient().List(context.TODO(), &stacks,
+			client.InNamespace(program.GetNamespace()),
+			client.MatchingFields{programRefIndexFieldName: program.GetName()})
+		if err == nil {
+			reqs := make([]reconcile.Request, len(stacks.Items), len(stacks.Items))
+			for i := range stacks.Items {
+				reqs[i].NamespacedName = client.ObjectKeyFromObject(&stacks.Items[i])
+			}
+			return reqs
+		}
+		// we don't get to return an error; only to fail quietly
+		mgr.GetLogger().Error(err, "failed to fetch stacks referring to program", "name", program.GetName(), "namespace", program.GetNamespace())
+		return nil
+	}
+
+	err = c.Watch(&source.Kind{Type: &pulumiv1.Program{}}, ctrlhandler.EnqueueRequestsFromMapFunc(enqueueStacksForProgram))
+
+	return err
 }
 
 // blank assignment to verify that ReconcileStack implements reconcile.Reconciler

test/program_test.go

@@ -90,6 +90,7 @@ var _ = Describe("Creating a YAML program", func() {
 						"PULUMI_CONFIG_PASSPHRASE": shared.NewLiteralResourceRef("password"),
 						"KUBECONFIG":               shared.NewLiteralResourceRef(kubeconfig),
 					},
+					ResyncFrequencySeconds: 3600, // make sure it doesn't run again unless there's another reason to
 				},
 			}
 			// stack name left to test cases
@@ -157,8 +158,41 @@ var _ = Describe("Creating a YAML program", func() {
 			Expect(stack.Status.LastUpdate.State).To(Equal(shared.SucceededStackStateMessage))
 			Expect(apimeta.IsStatusConditionTrue(stack.Status.Conditions, pulumiv1.ReadyCondition)).To(BeTrue())
 		})
-	})
 
+		When("the program is changed", func() {
+			var revisionOnFirstSuccess string
+
+			BeforeEach(func() {
+				prog := programFromFile("./testdata/test-program.yaml")
+				Expect(k8sClient.Create(context.TODO(), &prog)).To(Succeed())
+
+				stack.Spec.ProgramRef = &shared.ProgramReference{
+					Name: prog.Name,
+				}
+
+				// Set DestroyOnFinalize to clean up the configmap for repeat runs.
+				stack.Spec.DestroyOnFinalize = true
+				stack.Name = "changing-program-" + randString()
+				Expect(k8sClient.Create(context.TODO(), &stack)).To(Succeed())
+				waitForStackSuccess(&stack)
+
+				Expect(stack.Status.LastUpdate).ToNot(BeNil())
+				revisionOnFirstSuccess = stack.Status.LastUpdate.LastSuccessfulCommit
+				fmt.Fprintf(GinkgoWriter, ".status.lastUpdate.LastSuccessfulCommit before changing program: %s", revisionOnFirstSuccess)
+				prog2 := programFromFile("testdata/test-program-changed.yaml")
+				prog.Program = prog2.Program
+				resetWaitForStack()
+				Expect(k8sClient.Update(context.TODO(), &prog)).To(Succeed())
+			})
+
+			It("reruns the stack", func() {
+				waitForStackSuccess(&stack)
+				Expect(stack.Status.LastUpdate).ToNot(BeNil())
+				fmt.Fprintf(GinkgoWriter, ".status.lastUpdate.LastSuccessfulCommit after changing program: %s", stack.Status.LastUpdate.LastSuccessfulCommit)
+				Expect(stack.Status.LastUpdate.LastSuccessfulCommit).NotTo(Equal(revisionOnFirstSuccess))
+			})
+		})
+	})
 })
 
 func programFromFile(path string) pulumiv1.Program {

test/testdata/test-program-changed.yaml

@@ -0,0 +1,19 @@
+configuration:
+    foo:
+        type: String
+        default: "different"
+resources:
+    provider:
+        type: pulumi:providers:kubernetes
+    example:
+        type: kubernetes:core/v1:ConfigMap
+        properties:
+            metadata:
+                annotations:
+                    pulumi.com/patchForce: "true"
+                name:
+                    test-configmap
+            data:
+                foo: ${foo}
+        options:
+            provider: ${provider}