fix: gitops-engine fixes, plus enable Numaflow Controller deleted and then recreated

cmd/main.go

@@ -218,7 +218,7 @@ func main() {
 		numaLogger.Fatal(err, "Unable to create NumaflowController controller")
 	}
 
-	if err = numaflowControllerReconciler.SetupWithManager(mgr); err != nil {
+	if err = numaflowControllerReconciler.SetupWithManager(ctx, mgr); err != nil {
 		numaLogger.Fatal(err, "Unable to set up NumaflowController controller")
 	}
 

internal/controller/numaflowcontrollerrollout/numaflowcontrollerrollout_controller.go

@@ -321,7 +321,7 @@ func (r *NumaflowControllerRolloutReconciler) processExistingNumaflowController(
 	}
 
 	numaLogger.
-		WithValues("numaflowControllerNeedsToUpdate", numaflowControllerNeedsToUpdate, "upgradeStrategyType", upgradeStrategyType).
+		WithValues("numaflowControllerNeedsToUpdate", numaflowControllerNeedsToUpdate, "upgradeStrategyType", upgradeStrategyType, "numaflowControllerIsUpdating", numaflowControllerIsUpdating).
 		Debug("Upgrade decision result")
 
 	// set the Status appropriately to "Pending" or "Deployed"
@@ -503,16 +503,18 @@ func (r *NumaflowControllerRolloutReconciler) processNumaflowControllerStatus(
 
 		healthyChildCond := existingNumaflowControllerStatus.GetCondition(apiv1.ConditionChildResourceHealthy)
 
-		if existingNumaflowControllerStatus.IsHealthy() &&
-			healthyChildCond != nil && existingNumaflowControllerDef.GetGeneration() <= healthyChildCond.ObservedGeneration &&
-			healthyChildCond.Status == metav1.ConditionTrue {
-
+		if existingNumaflowControllerDef.GetGeneration() > existingNumaflowControllerStatus.ObservedGeneration {
+			nfcRollout.Status.MarkChildResourcesUnhealthy("Progressing",
+				fmt.Sprintf("observedGeneration %d < generation %d", existingNumaflowControllerStatus.ObservedGeneration, existingNumaflowControllerDef.GetGeneration()),
+				nfcRollout.Generation)
+		} else if existingNumaflowControllerStatus.IsHealthy() &&
+			healthyChildCond != nil && healthyChildCond.Status == metav1.ConditionTrue {
 			nfcRollout.Status.MarkChildResourcesHealthy(nfcRollout.Generation)
 		} else {
 			if healthyChildCond != nil {
 				nfcRollout.Status.MarkChildResourcesUnhealthy(healthyChildCond.Reason, healthyChildCond.Message, nfcRollout.Generation)
 			} else {
-				nfcRollout.Status.MarkChildResourcesUnhealthy(apiv1.ProgressingReasonString, "Progressing", nfcRollout.Generation)
+				nfcRollout.Status.MarkChildResourcesUnhealthy("Unhealthy", "Unhealthy", nfcRollout.Generation)
 			}
 		}
 	}

internal/controller/pipelinerollout/pipelinerollout_controller.go

@@ -563,10 +563,16 @@ func (r *PipelineRolloutReconciler) processPipelineStatus(ctx context.Context, p
 
 	// Only fetch the latest pipeline object while deleting the pipeline object, i.e. when pipelineRollout.DeletionTimestamp.IsZero() is false
 	if existingPipelineDef == nil {
-		pipelineDef, err := r.makePromotedPipelineDefinition(ctx, pipelineRollout)
+		// determine name of the promoted Pipeline
+		pipelineName, err := progressive.GetChildName(ctx, pipelineRollout, r, common.LabelValueUpgradePromoted, r.client, true)
 		if err != nil {
-			return err
+			return fmt.Errorf("Unable to process pipeline status: err=%s", err)
 		}
+		pipelineDef := &unstructured.Unstructured{}
+		pipelineDef.SetGroupVersionKind(numaflowv1.PipelineGroupVersionKind)
+		pipelineDef.SetNamespace(pipelineRollout.Namespace)
+		pipelineDef.SetName(pipelineName)
+
 		livePipelineDef, err := kubernetes.GetLiveResource(ctx, pipelineDef, "pipelines")
 		if err != nil {
 			if apierrors.IsNotFound(err) {

internal/sync/cache.go

@@ -65,7 +65,7 @@ var (
 )
 
 type ResourceInfo struct {
-	Name string
+	OwnerName string
 
 	Health       *health.HealthStatus
 	manifestHash string
@@ -77,9 +77,11 @@ type ResourceInfo struct {
 type LiveStateCache interface {
 	// GetClusterCache returns synced cluster cache
 	GetClusterCache() (clustercache.ClusterCache, error)
-	// GetManagedLiveObjs returns state of live objects which correspond to target
-	// objects with the specified ResourceInfo name and matching namespace.
-	GetManagedLiveObjs(name, namespace string, targetObjs []*unstructured.Unstructured) (map[kube.ResourceKey]*unstructured.Unstructured, error)
+	// GetManagedLiveObjects returns the live objects associated with the owner name
+	GetManagedLiveObjects(ownerName, namespace string) (map[kube.ResourceKey]*clustercache.Resource, error)
+	// GetManagedLiveObjsFromResourceList returns state of live objects which correspond to target
+	// objects with the specified ResourceInfo name and matching namespace, associated with the owner name
+	GetManagedLiveObjsFromResourceList(ownerName, namespace string, targetObjs []*unstructured.Unstructured) (map[kube.ResourceKey]*unstructured.Unstructured, error)
 	// Init must be executed before cache can be used
 	Init(numaLogger *logger.NumaLogger) error
 	// PopulateResourceInfo is called by the cache to update ResourceInfo struct for a managed resource
@@ -219,8 +221,8 @@ func (c *liveStateCache) PopulateResourceInfo(un *unstructured.Unstructured, isR
 	res.Health, _ = health.GetResourceHealth(un, settings.clusterSettings.ResourceHealthOverride)
 
 	numaplaneInstanceName := getNumaplaneInstanceName(un)
-	if isRoot && numaplaneInstanceName != "" {
-		res.Name = numaplaneInstanceName
+	if /*isRoot &&*/ numaplaneInstanceName != "" { // TODO: can probably put this back now that we don't have OwnerRef - need to test
+		res.OwnerName = numaplaneInstanceName
 	}
 
 	gvk := un.GroupVersionKind()
@@ -236,7 +238,7 @@ func (c *liveStateCache) PopulateResourceInfo(un *unstructured.Unstructured, isR
 
 	// edge case. we do not label CRDs, so they miss the tracking label we inject. But we still
 	// want the full resource to be available in our cache (to diff), so we store all CRDs
-	return res, res.Name != "" || gvk.Kind == kube.CustomResourceDefinitionKind
+	return res, res.OwnerName != "" || gvk.Kind == kube.CustomResourceDefinitionKind
 }
 
 // getNumaplaneInstanceName gets the Numaplane Object that owns the resource from a label in the resource
@@ -432,17 +434,39 @@ func (c *liveStateCache) GetClusterCache() (clustercache.ClusterCache, error) {
 	return clusterCache, nil
 }
 
-func (c *liveStateCache) GetManagedLiveObjs(
-	name, namespace string,
+func (c *liveStateCache) GetManagedLiveObjects(
+	ownerName, namespace string,
+) (map[kube.ResourceKey]*clustercache.Resource, error) {
+	clusterInfo, err := c.getSyncedCluster()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get cluster info: %w", err)
+	}
+
+	liveObjs := clusterInfo.FindResources(namespace, func(r *clustercache.Resource) bool {
+		matches := resInfo(r).OwnerName == ownerName && r.Ref.Namespace == namespace
+		fmt.Printf("deletethis: viewing live objects, I see: r.Ref.Kind=%q, r.Ref.Name=%q, r.Ref.Namespace=%q, resInfo(r)=%+v, checking against namespace=%q, ownerName=%q, matches=%t\n",
+			r.Ref.Kind, r.Ref.Name, r.Ref.Namespace, resInfo(r), namespace, ownerName, matches)
+		// distinguish it by numaplane resource's name and namespace
+		return matches
+	})
+	return liveObjs, err
+}
+
+func (c *liveStateCache) GetManagedLiveObjsFromResourceList(
+	ownerName, namespace string,
 	targetObjs []*unstructured.Unstructured,
 ) (map[kube.ResourceKey]*unstructured.Unstructured, error) {
 	clusterInfo, err := c.getSyncedCluster()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get cluster info: %w", err)
 	}
+
 	liveObjs, err := clusterInfo.GetManagedLiveObjs(targetObjs, func(r *clustercache.Resource) bool {
+		matches := resInfo(r).OwnerName == ownerName && r.Ref.Namespace == namespace
+		fmt.Printf("deletethis: viewing live objects, I see: r.Ref.Kind=%q, r.Ref.Name=%q, r.Ref.Namespace=%q, resInfo(r)=%+v, checking against namespace=%q, ownerName=%q, matches=%t\n",
+			r.Ref.Kind, r.Ref.Name, r.Ref.Namespace, resInfo(r), namespace, ownerName, matches)
 		// distinguish it by numaplane resource's name and namespace
-		return resInfo(r).Name == name && r.Ref.Namespace == namespace
+		return matches
 	})
 	c.customMetrics.KubeResourceMonitored.WithLabelValues().Set(float64(len(liveObjs)))
 	return liveObjs, err

internal/sync/cache_test.go

@@ -173,7 +173,7 @@ metadata:
   labels:
     numaplane.numaproj.io/tracking-id: my-example`)
 
-	managedObjs, err := clusterCache.GetManagedLiveObjs(testName, testNamespace, []*unstructured.Unstructured{targetDeploy})
+	managedObjs, err := clusterCache.GetManagedLiveObjsFromResourceList(testName, testNamespace, []*unstructured.Unstructured{targetDeploy})
 	require.NoError(t, err)
 	assert.Equal(t, map[kube.ResourceKey]*unstructured.Unstructured{
 		kube.NewResourceKey("apps", "Deployment", "default", "my-app"): mustToUnstructured(testDeploy()),