[eno] Prune input revisions for refs removed from the synthesizer

e2e/input_removal_test.go

@@ -0,0 +1,160 @@
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	flow "github.com/Azure/go-workflow"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	apiv1 "github.com/Azure/eno/api/v1"
+	fw "github.com/Azure/eno/e2e/framework"
+)
+
+// TestSynthesizerInputRemoval verifies that removing a ref from a synthesizer doesn't block resynthesis.
+//
+// The composition keeps its binding for the removed ref (as when bindings are inherited from a Symphony),
+// and the removed input carries a synthesizer-generation annotation. Unless its input revision is pruned,
+// that stale revision holds the composition out of lockstep and the new synthesis never happens.
+func TestSynthesizerInputRemoval(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	cli := fw.NewClient(t)
+
+	synthName := fw.UniqueName("input-removal-synth")
+	compName := fw.UniqueName("input-removal-comp")
+	outputName := fw.UniqueName("input-removal-output")
+	compKey := types.NamespacedName{Name: compName, Namespace: "default"}
+
+	configA := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{Name: fw.UniqueName("input-removal-configa"), Namespace: "default"},
+		Data:       map[string]string{"valueA": "alpha"},
+	}
+	configB := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fw.UniqueName("input-removal-configb"),
+			Namespace: "default",
+			// Ties configb's revision to the synthesizer's first generation, so it goes stale once the ref is removed
+			Annotations: map[string]string{"eno.azure.io/synthesizer-generation": "1"},
+		},
+		Data: map[string]string{"valueB": "bravo"},
+	}
+
+	// Copies every "valueA"/"valueB" entry found in the inputs into the output ConfigMap
+	script := fmt.Sprintf(`data=$(grep -o '"value[AB]":"[^"]*"' | sort -u | paste -sd, - || true)
+printf '{"apiVersion":"config.kubernetes.io/v1","kind":"ResourceList","items":[{"apiVersion":"v1","kind":"ConfigMap","metadata":{"name":"%s","namespace":"default"},"data":{%%s}}]}' "$data"`, outputName)
+
+	configMapRef := apiv1.ResourceRef{Version: "v1", Kind: "ConfigMap"}
+	synth := fw.NewMinimalSynthesizer(synthName, fw.WithCommand([]string{"/bin/bash", "-c", script}))
+	synth.Spec.Refs = []apiv1.Ref{
+		{Key: "configa", Resource: configMapRef},
+		{Key: "configb", Resource: configMapRef},
+	}
+
+	comp := fw.NewComposition(compName, "default", fw.WithSynthesizerRefs(apiv1.SynthesizerRef{Name: synthName}))
+	comp.Spec.Bindings = []apiv1.Binding{
+		{Key: "configa", Resource: apiv1.ResourceBinding{Name: configA.Name, Namespace: configA.Namespace}},
+		{Key: "configb", Resource: apiv1.ResourceBinding{Name: configB.Name, Namespace: configB.Namespace}},
+	}
+
+	var initialSynthGen int64
+
+	// -- Define workflow steps --
+
+	createConfigA := fw.CreateStep(t, "createConfigA", cli, configA)
+	createConfigB := fw.CreateStep(t, "createConfigB", cli, configB)
+	createSynthesizer := fw.CreateStep(t, "createSynthesizer", cli, synth)
+	createComposition := fw.CreateStep(t, "createComposition", cli, comp)
+
+	waitReady := flow.Func("waitReady", func(ctx context.Context) error {
+		fw.WaitForCompositionReady(t, ctx, cli, compKey, 3*time.Minute)
+		current := &apiv1.Composition{}
+		require.NoError(t, cli.Get(ctx, compKey, current))
+		require.NotNil(t, current.Status.CurrentSynthesis)
+		initialSynthGen = current.Status.CurrentSynthesis.ObservedSynthesizerGeneration
+		t.Logf("initial ObservedSynthesizerGeneration: %d", initialSynthGen)
+		return nil
+	})
+
+	verifyInitialOutput := flow.Func("verifyInitialOutput", func(ctx context.Context) error {
+		waitForOutputData(t, ctx, cli, outputName, map[string]string{"valueA": "alpha", "valueB": "bravo"})
+		t.Log("verified output contains both inputs")
+		return nil
+	})
+
+	removeConfigBRef := flow.Func("removeConfigBRef", func(ctx context.Context) error {
+		require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: synthName}, synth))
+		synth.Spec.Refs = []apiv1.Ref{{Key: "configa", Resource: configMapRef}}
+		t.Log("removing configb ref from synthesizer (composition binding is left in place)")
+		return cli.Update(ctx, synth)
+	})
+
+	waitResynthesized := flow.Func("waitResynthesized", func(ctx context.Context) error {
+		fw.WaitForCompositionResynthesized(t, ctx, cli, compKey, initialSynthGen, 3*time.Minute)
+		t.Log("composition re-synthesized and ready")
+		return nil
+	})
+
+	verifyUpdatedOutput := flow.Func("verifyUpdatedOutput", func(ctx context.Context) error {
+		waitForOutputData(t, ctx, cli, outputName, map[string]string{"valueA": "alpha"})
+
+		current := &apiv1.Composition{}
+		require.NoError(t, cli.Get(ctx, compKey, current))
+		for _, rev := range current.Status.InputRevisions {
+			require.NotEqual(t, "configb", rev.Key, "input revision for removed ref should be pruned")
+		}
+		t.Log("verified output contains only the remaining input")
+		return nil
+	})
+
+	deleteComposition := fw.DeleteStep(t, "deleteComposition", cli, comp)
+
+	verifyOutputDeleted := flow.Func("verifyOutputDeleted", func(ctx context.Context) error {
+		output := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: outputName, Namespace: "default"}}
+		fw.WaitForResourceDeleted(t, ctx, cli, output, 60*time.Second)
+		return nil
+	})
+
+	cleanup := fw.CleanupStep(t, "cleanup", cli, synth, configA, configB)
+
+	// -- Wire the DAG --
+
+	w := new(flow.Workflow)
+	w.Add(
+		flow.Step(createComposition).DependsOn(createConfigA, createConfigB, createSynthesizer),
+		flow.Step(waitReady).DependsOn(createComposition),
+		flow.Step(verifyInitialOutput).DependsOn(waitReady),
+		flow.Step(removeConfigBRef).DependsOn(verifyInitialOutput),
+		flow.Step(waitResynthesized).DependsOn(removeConfigBRef),
+		flow.Step(verifyUpdatedOutput).DependsOn(waitResynthesized),
+		flow.Step(deleteComposition).DependsOn(verifyUpdatedOutput),
+		flow.Step(verifyOutputDeleted).DependsOn(deleteComposition),
+		flow.Step(cleanup).DependsOn(verifyOutputDeleted),
+	)
+
+	require.NoError(t, w.Do(ctx))
+}
+
+// waitForOutputData polls until the output ConfigMap's data exactly matches expected.
+func waitForOutputData(t *testing.T, ctx context.Context, cli client.Client, name string, expected map[string]string) {
+	t.Helper()
+	err := wait.PollUntilContextTimeout(ctx, 2*time.Second, time.Minute, true, func(ctx context.Context) (bool, error) {
+		output := &corev1.ConfigMap{}
+		if err := cli.Get(ctx, types.NamespacedName{Name: name, Namespace: "default"}, output); err != nil {
+			return false, nil
+		}
+		t.Logf("output configmap %s data: %v", name, output.Data)
+		return reflect.DeepEqual(expected, output.Data), nil
+	})
+	require.NoError(t, err, "timed out waiting for output configmap %s to have data %v", name, expected)
+}

internal/controllers/watch/integration_test.go

@@ -519,6 +519,77 @@ func TestRemoveInput(t *testing.T) {
 	})
 }
 
+// TestRemoveSynthesizerRefWithLingeringBinding covers a synthesizer dropping a ref while the
+// composition still carries a binding for it (e.g. inherited from a Symphony).
+// The stale revision must be pruned, otherwise it holds the composition out of lockstep forever.
+func TestRemoveSynthesizerRefWithLingeringBinding(t *testing.T) {
+	mgr := testutil.NewManager(t)
+	require.NoError(t, NewController(mgr.Manager))
+	mgr.Start(t)
+
+	ctx := testutil.NewContext(t)
+	cli := mgr.GetClient()
+
+	for _, name := range []string{"input-a", "input-b"} {
+		cm := &corev1.ConfigMap{}
+		cm.Name = name
+		cm.Namespace = "default"
+		cm.Annotations = map[string]string{"eno.azure.io/synthesizer-generation": "1"}
+		require.NoError(t, cli.Create(ctx, cm))
+	}
+
+	refB := apiv1.Ref{Key: "b", Resource: apiv1.ResourceRef{Version: "v1", Kind: "ConfigMap"}}
+	synth := &apiv1.Synthesizer{}
+	synth.Name = "test-synth"
+	synth.Spec.Refs = []apiv1.Ref{
+		{Key: "a", Resource: apiv1.ResourceRef{Version: "v1", Kind: "ConfigMap"}},
+		refB,
+	}
+	require.NoError(t, cli.Create(ctx, synth))
+
+	comp := &apiv1.Composition{}
+	comp.Name = "test-comp"
+	comp.Namespace = "default"
+	comp.Spec.Synthesizer.Name = synth.Name
+	comp.Spec.Bindings = []apiv1.Binding{
+		{Key: "a", Resource: apiv1.ResourceBinding{Name: "input-a", Namespace: "default"}},
+		{Key: "b", Resource: apiv1.ResourceBinding{Name: "input-b", Namespace: "default"}},
+	}
+	require.NoError(t, cli.Create(ctx, comp))
+
+	testutil.Eventually(t, func() bool {
+		cli.Get(ctx, client.ObjectKeyFromObject(comp), comp)
+		return len(comp.Status.InputRevisions) == 2
+	})
+
+	// Remove ref "b" from the synthesizer but leave the binding on the composition
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		cli.Get(ctx, client.ObjectKeyFromObject(synth), synth)
+		synth.Spec.Refs = synth.Spec.Refs[:1]
+		return cli.Update(ctx, synth)
+	})
+	require.NoError(t, err)
+
+	testutil.Eventually(t, func() bool {
+		cli.Get(ctx, client.ObjectKeyFromObject(comp), comp)
+		return len(comp.Status.InputRevisions) == 1 && comp.Status.InputRevisions[0].Key == "a"
+	})
+	assert.Len(t, comp.Spec.Bindings, 2, "binding should be left untouched")
+
+	// Adding the ref back restores the revision
+	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		cli.Get(ctx, client.ObjectKeyFromObject(synth), synth)
+		synth.Spec.Refs = append(synth.Spec.Refs, refB)
+		return cli.Update(ctx, synth)
+	})
+	require.NoError(t, err)
+
+	testutil.Eventually(t, func() bool {
+		cli.Get(ctx, client.ObjectKeyFromObject(comp), comp)
+		return len(comp.Status.InputRevisions) == 2
+	})
+}
+
 // TestConcurrentStatusUpdates verifies that the retry mechanism handles
 // concurrent updates to composition status without causing a retry storm.
 // This tests the fix for the issue where conflict errors would cause immediate

internal/controllers/watch/pruning.go

@@ -4,10 +4,13 @@ import (
 	"context"
 
 	apiv1 "github.com/Azure/eno/api/v1"
+	"github.com/Azure/eno/internal/manager"
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/api/errors"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
 type pruningController struct {
@@ -33,40 +36,81 @@ func (c *pruningController) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	synth := &apiv1.Synthesizer{}
 	synth.Name = comp.Spec.Synthesizer.Name
 	err = c.client.Get(ctx, client.ObjectKeyFromObject(synth), synth)
-	if client.IgnoreNotFound(err) != nil {
+	if errors.IsNotFound(err) {
+		// Without the synthesizer we can't tell which refs are still declared, so don't prune anything
+		return ctrl.Result{}, nil
+	}
+	if err != nil {
 		logger.Error(err, "failed to get synthesizer")
 		return ctrl.Result{}, err
 	}
 
-	for i, ir := range comp.Status.InputRevisions {
-		if hasBindingKey(comp, synth, ir.Key) {
+	kept := make([]apiv1.InputRevisions, 0, len(comp.Status.InputRevisions))
+	var pruned []string
+	for _, ir := range comp.Status.InputRevisions {
+		if isInputBound(comp, synth, ir.Key) {
+			kept = append(kept, ir)
 			continue
 		}
-		logger.Info("pruning input revision - no longer has binding", "key", ir.Key, "revision", ir.Revision, "index", i)
-		comp.Status.InputRevisions = append(comp.Status.InputRevisions[:i], comp.Status.InputRevisions[i+1:]...)
-		err = c.client.Status().Update(ctx, comp)
-		if err != nil {
-			logger.Error(err, "failed to update composition status")
-			return ctrl.Result{}, err
-		}
-
-		logger.Info("pruned old input revision from composition status", "compositionName", comp.Name, "compositionNamespace", comp.Namespace, "ref", ir.Key)
+		pruned = append(pruned, ir.Key)
+	}
+	if len(pruned) == 0 {
 		return ctrl.Result{}, nil
 	}
 
+	logger.Info("pruning input revisions - no longer declared by synthesizer or bound by composition", "keys", pruned, "synthesizerGeneration", synth.Generation)
+	comp.Status.InputRevisions = kept
+	err = c.client.Status().Update(ctx, comp)
+	if err != nil {
+		logger.Error(err, "failed to update composition status")
+		return ctrl.Result{}, err
+	}
+
+	logger.Info("pruned old input revisions from composition status", "keys", pruned)
 	return ctrl.Result{}, nil
 }
 
-func hasBindingKey(comp *apiv1.Composition, synth *apiv1.Synthesizer, key string) bool {
-	for _, b := range comp.Spec.Bindings {
-		if b.Key == key {
-			return true
-		}
-	}
+// isInputBound returns true when the synthesizer still declares a ref for the key and that ref
+// is bound either explicitly by the composition or implicitly by the ref itself.
+// A composition binding alone is not enough: bindings can outlive the ref (e.g. bindings inherited from a Symphony).
+func isInputBound(comp *apiv1.Composition, synth *apiv1.Synthesizer, key string) bool {
 	for _, ref := range synth.Spec.Refs {
-		if ref.Key == key && ref.Resource.Name != "" {
+		if ref.Key != key {
+			continue
+		}
+		if ref.Resource.Name != "" {
 			return true // implicit binding
 		}
+		for _, b := range comp.Spec.Bindings {
+			if b.Key == key {
+				return true
+			}
+		}
+		return false
 	}
 	return false
 }
+
+// newSynthEventHandler enqueues every composition that uses the synthesizer, so revisions for refs
+// removed from the synthesizer are pruned without waiting for an unrelated composition event.
+func (c *pruningController) newSynthEventHandler() handler.TypedEventHandler[*apiv1.Synthesizer, reconcile.Request] {
+	fn := func(ctx context.Context, synth *apiv1.Synthesizer) []reconcile.Request {
+		logger := logr.FromContextOrDiscard(ctx)
+
+		list := &apiv1.CompositionList{}
+		err := c.client.List(ctx, list, client.MatchingFields{
+			manager.IdxCompositionsBySynthesizer: synth.Name,
+		})
+		if err != nil {
+			logger.Error(err, "failed to list compositions for synthesizer", "synthesizerName", synth.Name)
+			return nil
+		}
+
+		reqs := make([]reconcile.Request, 0, len(list.Items))
+		for i := range list.Items {
+			reqs = append(reqs, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(&list.Items[i])})
+		}
+		return reqs
+	}
+	return handler.TypedEnqueueRequestsFromMapFunc(fn)
+}