feat: tenant owned resources problem enhancements

Signed-off-by: Bence Csati <bence.csati@axoflow.com>

Author: csatib02

controllers/telemetry/route_controller.go

@@ -19,7 +19,7 @@ import (
 	"fmt"
 	"reflect"
 	"slices"
-	"sort"
+	"strings"
 
 	"emperror.dev/errors"
 	apiv1 "k8s.io/api/core/v1"
@@ -98,11 +98,12 @@ func (r *RouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl
 	}
 	for _, step := range steps {
 		if err := step.fn(); err != nil {
-			return r.handleReconcileError(ctx, baseManager, tenant, step.name, err)
+			return r.handleTenantReconcileError(ctx, &baseManager, tenant, step.name, err)
 		}
 	}
 
 	tenant.Status.State = state.StateReady
+	tenant.ClearProblems()
 	if !reflect.DeepEqual(originalTenantStatus, tenant.Status) {
 		baseManager.Info("tenant status changed")
 		if updateErr := r.Status().Update(ctx, tenant); updateErr != nil {
@@ -217,12 +218,11 @@ func (r *RouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return builder.Complete(r)
 }
 
-// handleReconcileError handles errors that occur during reconciliation steps
-func (r *RouteReconciler) handleReconcileError(ctx context.Context, baseManager manager.BaseManager, tenant *v1alpha1.Tenant, stepName string, err error) (ctrl.Result, error) {
+// handleTenantReconcileError handles errors that occur during reconciliation steps
+func (r *RouteReconciler) handleTenantReconcileError(ctx context.Context, baseManager *manager.BaseManager, tenant *v1alpha1.Tenant, stepName string, err error) (ctrl.Result, error) {
 	wrappedErr := errors.Wrapf(err, "failed to %s for tenant %s", stepName, tenant.Name)
 
-	tenant.Status.Problems = append(tenant.Status.Problems, wrappedErr.Error())
-	tenant.Status.ProblemsCount = len(tenant.Status.Problems)
+	tenant.AddProblem(wrappedErr.Error())
 	tenant.Status.State = state.StateFailed
 
 	baseManager.Error(errors.WithStack(err), fmt.Sprintf("failed to %s", stepName), "tenant", tenant.Name)
@@ -247,6 +247,10 @@ func handleLogSourceNamespaces(ctx context.Context, tenantResManager *manager.Te
 }
 
 func handleOwnedResources(ctx context.Context, tenantResManager *manager.TenantResourceManager, tenant *v1alpha1.Tenant) error {
+	// Caching all outputs for the tenant to validate subscriptions against
+	var allOutputsForTenant []model.ResourceOwnedByTenant
+
+	// Process outputs first to ensure they have their tenant set before validating subscriptions
 	tenantOwnedResources := []model.ResourceOwnedByTenant{
 		&v1alpha1.Output{},
 		&v1alpha1.Subscription{},
@@ -255,11 +259,7 @@ func handleOwnedResources(ctx context.Context, tenantResManager *manager.TenantR
 		resourcesForTenant, resourceUpdateList, err := tenantResManager.GetResourceOwnedByTenant(ctx, resource, tenant)
 		if err != nil {
 			tenantResManager.Error(errors.WithStack(err), fmt.Sprintf("failed to get %T for tenant", resource), "tenant", tenant.Name)
-			if updateErr := tenantResManager.Status().Update(ctx, tenant); updateErr != nil {
-				tenantResManager.Error(errors.WithStack(updateErr), "failed updating tenant status", "tenant", tenant.Name)
-				return errors.Append(err, updateErr)
-			}
-			return err
+			return fmt.Errorf("failed to get %T for tenant %s: %w", resource, tenant.Name, err)
 		}
 
 		// Add all newly updated resources here
@@ -271,38 +271,60 @@ func handleOwnedResources(ctx context.Context, tenantResManager *manager.TenantR
 		}
 		tenantResManager.DisownResources(ctx, resourcesToDisown)
 
+		if _, ok := resource.(*v1alpha1.Output); ok {
+			allOutputsForTenant = resourcesForTenant
+		}
+
 		if _, ok := resource.(*v1alpha1.Subscription); ok {
 			subscriptionNames := manager.GetResourceNamesFromResource(resourcesForTenant)
 			components.SortNamespacedNames(subscriptionNames)
 			tenant.Status.Subscriptions = subscriptionNames
 
-			if err := validateSubscriptionOutputs(ctx, tenantResManager, tenant, resourcesForTenant); err != nil {
+			if err := validateSubscriptionOutputs(ctx, tenantResManager, tenant, resourcesForTenant, allOutputsForTenant); err != nil {
 				return err
 			}
 		}
+
+		for _, res := range resourcesForTenant {
+			if res.GetState() == state.StateFailed {
+				tenantResManager.Error(errors.New("resource failed"), "failed resource", "resource", res.GetName())
+				return fmt.Errorf("resource %s is in a failed state", res.GetName())
+			}
+		}
 	}
 
 	return nil
 }
 
-func validateSubscriptionOutputs(ctx context.Context, tenantResManager *manager.TenantResourceManager, tenant *v1alpha1.Tenant, subscriptionsForTenant []model.ResourceOwnedByTenant) error {
+func validateSubscriptionOutputs(ctx context.Context, tenantResManager *manager.TenantResourceManager, tenant *v1alpha1.Tenant, subscriptionsForTenant []model.ResourceOwnedByTenant, outputsForTenant []model.ResourceOwnedByTenant) error {
 	realSubscriptionsForTenant, err := utils.GetConcreteTypeFromList[*v1alpha1.Subscription](utils.ToObject(subscriptionsForTenant))
 	if err != nil {
 		tenantResManager.Error(errors.WithStack(err), "failed to get concrete type from list", "tenant", tenant.Name)
 		return err
 	}
 
+	realOutputsForTenant, err := utils.GetConcreteTypeFromList[*v1alpha1.Output](utils.ToObject(outputsForTenant))
+	if err != nil {
+		tenantResManager.Error(errors.WithStack(err), "failed to get concrete type from list for outputs", "tenant", tenant.Name)
+		return err
+	}
+
+	outputMap := make(map[v1alpha1.NamespacedName]*v1alpha1.Output)
+	for _, output := range realOutputsForTenant {
+		outputMap[output.NamespacedName()] = output
+	}
+
 	for _, subscription := range realSubscriptionsForTenant {
 		originalSubscriptionStatus := subscription.Status.DeepCopy()
-		validOutputs, invalidOutputs := tenantResManager.ValidateSubscriptionOutputs(ctx, subscription)
-
-		if len(invalidOutputs) > 0 {
-			subscription.Status.Problems = append(subscription.Status.Problems, fmt.Sprintf("invalid outputs for subscription %s: %v", subscription.Name, invalidOutputs))
-			subscription.Status.ProblemsCount = len(subscription.Status.Problems)
+		validOutputs, invalidOutputs := tenantResManager.ValidateSubscriptionReferencedOutputsWithCache(ctx, subscription, outputMap)
+		switch len(invalidOutputs) {
+		case 0:
+			subscription.Status.State = state.StateReady
+			subscription.ClearProblems()
+		default:
 			subscription.Status.State = state.StateFailed
-			tenantResManager.UpdateOutputs(ctx, tenant, invalidOutputs)
+			subscription.AddProblem(fmt.Sprintf("invalid outputs referenced by subscription %s: %s", subscription.Name, strings.Join(invalidOutputs, ", ")))
 		}
-
 		components.SortNamespacedNames(validOutputs)
 		subscription.Status.Outputs = validOutputs
 
@@ -320,34 +342,40 @@ func validateSubscriptionOutputs(ctx context.Context, tenantResManager *manager.
 func handleBridgeResources(ctx context.Context, bridgeManager *manager.BridgeManager, tenant *v1alpha1.Tenant) error {
 	bridgesForTenant, err := bridgeManager.GetBridgesForTenant(ctx, tenant.Name)
 	if err != nil {
-		tenant.Status.State = state.StateFailed
 		bridgeManager.Error(errors.WithStack(err), "failed to get bridges for tenant", "tenant", tenant.Name)
-		if updateErr := bridgeManager.Status().Update(ctx, tenant); updateErr != nil {
-			bridgeManager.Error(errors.WithStack(updateErr), "failed updating tenant status", "tenant", tenant.Name)
-			return errors.Append(err, updateErr)
-		}
-
-		return err
+		return fmt.Errorf("failed to get bridges for tenant %s: %w", tenant.Name, err)
 	}
 
 	bridgesForTenantNames := manager.GetBridgeNamesFromBridges(bridgesForTenant)
-	sort.Strings(bridgesForTenantNames)
+	slices.Sort(bridgesForTenantNames)
 	tenant.Status.ConnectedBridges = bridgesForTenantNames
 
-	for _, bridge := range bridgesForTenant {
-		if err := bridgeManager.ValidateBridgeConnection(ctx, tenant.Name, &bridge); err != nil {
-			bridge.Status.Problems = append(bridge.Status.Problems, errors.Wrapf(err, "bridge %s validation failed", bridge.Name).Error())
-			bridge.Status.ProblemsCount = len(bridge.Status.Problems)
+	for i := range bridgesForTenant {
+		bridge := &bridgesForTenant[i]
+		originalBridgeStatus := bridge.Status.DeepCopy()
+
+		if err := bridgeManager.ValidateBridgeConnection(ctx, tenant.Name, bridge); err != nil {
+			bridge.AddProblem(errors.Wrapf(err, "bridge %s validation failed", bridge.Name).Error())
 			bridge.Status.State = state.StateFailed
 
 			bridgeManager.Error(errors.WithStack(err), "failed to check bridge connection", "bridge", bridge.Name)
-			if updateErr := bridgeManager.Status().Update(ctx, tenant); updateErr != nil {
-				bridgeManager.Error(errors.WithStack(updateErr), "failed updating tenant status", "tenant", tenant.Name)
+			if updateErr := bridgeManager.Status().Update(ctx, bridge); updateErr != nil {
+				bridgeManager.Error(errors.WithStack(updateErr), "failed updating bridge status", "bridge", bridge.Name)
 				return errors.Append(err, updateErr)
 			}
 
 			return err
 		}
+
+		bridge.Status.State = state.StateReady
+		bridge.ClearProblems()
+
+		if !reflect.DeepEqual(originalBridgeStatus, &bridge.Status) {
+			if updateErr := bridgeManager.Status().Update(ctx, bridge); updateErr != nil {
+				bridgeManager.Error(errors.WithStack(updateErr), "failed updating bridge status", "bridge", bridge.Name)
+				return updateErr
+			}
+		}
 	}
 
 	return nil

pkg/resources/manager/bridge_manager.go

@@ -75,7 +75,7 @@ func (b *BridgeManager) getTenants(ctx context.Context, listOpts *client.ListOpt
 	return tenants.Items, nil
 }
 
-func (b *BridgeManager) CheckBridgeConnection(ctx context.Context, tenantName string, bridge *v1alpha1.Bridge) error {
+func (b *BridgeManager) ValidateBridgeConnection(ctx context.Context, tenantName string, bridge *v1alpha1.Bridge) error {
 	for _, tenantReference := range []string{bridge.Spec.SourceTenant, bridge.Spec.TargetTenant} {
 		if tenantReference != tenantName {
 			listOpts := &client.ListOptions{

pkg/resources/manager/tenant_resource_manager.go

@@ -26,7 +26,6 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
-	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kube-logging/telemetry-controller/api/telemetry/v1alpha1"
@@ -99,12 +98,21 @@ func (t *TenantResourceManager) GetResourceOwnedByTenant(ctx context.Context, re
 		currentTenant := res.GetTenant()
 		if currentTenant != "" && currentTenant != tenant.Name {
 			t.Error(
-				fmt.Errorf("resource is owned by another tenant"),
+				fmt.Errorf("resource: %s of kind: %s is owned by another tenant", res.GetName(), res.GetObjectKind().GroupVersionKind().Kind),
 				"skipping reconciliation",
 				"current_tenant", currentTenant,
 				"desired_tenant", tenant.Name,
 				"action_required", "remove resource from previous tenant before adopting to new tenant",
 			)
+
+			// Clear any previous problems and add the ownership conflict
+			res.SetProblems([]string{
+				fmt.Sprintf("resource %s/%s is already owned by tenant %s", res.GetNamespace(), res.GetName(), currentTenant),
+			})
+			res.SetState(state.StateFailed)
+			if err := t.Status().Update(ctx, res); err != nil {
+				t.Error(err, fmt.Sprintf("failed to update resource (%s/%s) state", res.GetNamespace(), res.GetName()))
+			}
 			continue
 		}
 
@@ -188,43 +196,52 @@ func (t *TenantResourceManager) DisownResources(ctx context.Context, resourceToD
 	}
 }
 
-// ValidateSubscriptionOutputs validates the output references of a subscription
-func (t *TenantResourceManager) ValidateSubscriptionOutputs(ctx context.Context, subscription *v1alpha1.Subscription) []v1alpha1.NamespacedName {
-	validOutputs := []v1alpha1.NamespacedName{}
-	invalidOutputs := []v1alpha1.NamespacedName{}
+// ValidateSubscriptionReferencedOutputsWithCache validates outputs referenced by the subscription using a provided cache of outputs
+func (t *TenantResourceManager) ValidateSubscriptionReferencedOutputsWithCache(ctx context.Context, subscription *v1alpha1.Subscription, outputMap map[v1alpha1.NamespacedName]*v1alpha1.Output) ([]v1alpha1.NamespacedName, []string) {
+	var validOutputs []v1alpha1.NamespacedName
+	var invalidOutputs []string
 	for _, outputRef := range subscription.Spec.Outputs {
-		checkedOutput := &v1alpha1.Output{}
-		if err := t.Get(ctx, types.NamespacedName(outputRef), checkedOutput); err != nil {
-			t.Error(err, "referred output invalid", "output", outputRef.String())
-
-			invalidOutputs = append(invalidOutputs, outputRef)
+		invalid := false
+		checkedOutput, exists := outputMap[outputRef]
+		if !exists {
+			// Output doesn't exist or isn't owned by this tenant
+			t.Error(errors.New("output not found"), "referred output invalid", "output", outputRef.String())
+			invalidOutputs = append(invalidOutputs, outputRef.String())
 			continue
 		}
 
 		// ensure the output belongs to the same tenant
 		if checkedOutput.Status.Tenant != subscription.Status.Tenant {
+			invalid = true
 			t.Error(errors.New("output and subscription tenants mismatch"),
 				"output and subscription tenants mismatch",
 				"output", checkedOutput.NamespacedName().String(),
 				"output's tenant", checkedOutput.Status.Tenant,
 				"subscription", subscription.NamespacedName().String(),
 				"subscription's tenant", subscription.Status.Tenant)
-
-			invalidOutputs = append(invalidOutputs, outputRef)
-			continue
+			checkedOutput.AddProblem(fmt.Sprintf("output %s does not belong to the same tenant as subscription %s", outputRef.String(), subscription.NamespacedName().String()))
 		}
 
 		// validate output secret
 		if checkedOutput.Spec.Authentication != nil {
 			if err := components.QueryOutputSecret(ctx, t.Client, checkedOutput); err != nil {
+				invalid = true
 				t.Error(err, "failed to query output secret", "output", checkedOutput.NamespacedName().String())
+				checkedOutput.AddProblem(fmt.Sprintf("output %s secret could not be queried: %v", outputRef.String(), err))
+			}
+		}
 
-				invalidOutputs = append(invalidOutputs, outputRef)
-				continue
+		if invalid {
+			checkedOutput.Status.State = state.StateFailed
+			if updateErr := t.Status().Update(ctx, checkedOutput); updateErr != nil {
+				t.Error(updateErr, fmt.Sprintf("failed to update output (%s/%s) state", checkedOutput.GetNamespace(), checkedOutput.GetName()))
 			}
+			invalidOutputs = append(invalidOutputs, checkedOutput.NamespacedName().String())
+			continue
 		}
 
 		// update the output state if validation was successful
+		checkedOutput.ClearProblems()
 		checkedOutput.SetState(state.StateReady)
 		if updateErr := t.Status().Update(ctx, checkedOutput); updateErr != nil {
 			checkedOutput.SetState(state.StateFailed)
@@ -234,11 +251,7 @@ func (t *TenantResourceManager) ValidateSubscriptionOutputs(ctx context.Context,
 		validOutputs = append(validOutputs, outputRef)
 	}
 
-	if len(invalidOutputs) > 0 {
-		t.Error(errors.New("some outputs are invalid"), "some outputs are invalid", "invalidOutputs", invalidOutputs, "subscription", subscription.NamespacedName().String())
-	}
-
-	return validOutputs
+	return validOutputs, invalidOutputs
 }
 
 // getNamespacesForSelectorSlice returns a list of namespaces that match the given label selectors