Propagate errors from cluster cache down to affected leaf applications

Author: jcogilvie

controller/appcontroller.go

@@ -1823,33 +1823,82 @@ func (ctrl *ApplicationController) processAppRefreshQueueItem() (processNext boo
 		app.Status.ReconciledAt = &now
 	}
 	app.Status.Sync = *compareResult.syncStatus
-	// Check both for conversion webhook errors in app state and compareResult conditions
-	hasConversionIssues := hasConversionWebhookErrors(app)
+	// Check for cluster cache issues that affect this application
+	hasCacheIssues := false
+	usesTaintedResources := false
 
-	// Also check if compareResult detected conversion webhook errors during refresh
-	// This helps identify conversion webhook errors during refresh operations
+	// 1. First check if the app itself reports conversion webhook errors
+	hasCacheIssues = hasClusterCacheIssues(app)
+	
+	// 2. Check app conditions for cache-related errors during refresh
 	for _, condition := range app.Status.Conditions {
 		if condition.Type == appv1.ApplicationConditionComparisonError && 
 		   (strings.Contains(condition.Message, "conversion webhook") || 
 		    strings.Contains(condition.Message, "known conversion webhook failures") || 
-		    strings.Contains(condition.Message, "unavailable resource types")) {
-			hasConversionIssues = true
+		    strings.Contains(condition.Message, "unavailable resource types") ||
+		    strings.Contains(condition.Message, "failed to list resources") ||
+		    strings.Contains(condition.Message, "Expired: too old resource version")) {
+			hasCacheIssues = true
 			break
 		}
 	}
 
-	// Also check resources for conversion webhook errors
+	// 3. Check resources for errors
 	for _, res := range app.Status.Resources {
 		if res.Health != nil && res.Health.Message != "" && 
 		   (strings.Contains(res.Health.Message, "conversion webhook") || 
-		    strings.Contains(res.Health.Message, "unavailable resource types")) {
-			hasConversionIssues = true
+		    strings.Contains(res.Health.Message, "unavailable resource types") ||
+		    strings.Contains(res.Health.Message, "failed to get resource")) {
+			hasCacheIssues = true
 			break
 		}
 	}
 
+	// 4. Check if the app directly uses any tainted resources
+	clusterURL := app.Spec.Destination.Server
+	// Check if cluster is tainted directly using our cluster taint tracking
+	statecache.ClusterTaintLock.RLock()
+	taints, exists := statecache.GetClusterTaints()[clusterURL]
+	statecache.ClusterTaintLock.RUnlock()
+	
+	// If the cluster is tainted, check if the app uses any of the affected resource types
+	if exists && len(taints) > 0 {
+		// Check if app directly uses any of the tainted GVKs
+		failedGVKs := statecache.GetTaintedGVKs(clusterURL)
+		
+		resLoop:
+		for _, res := range app.Status.Resources {
+			gvkStr := fmt.Sprintf("%s/%s, Kind=%s", res.Group, res.Version, res.Kind)
+			gvkWildcard := fmt.Sprintf("%s/*, Kind=%s", res.Group, res.Kind)
+			
+			for _, failedGVK := range failedGVKs {
+				if failedGVK == gvkStr || failedGVK == gvkWildcard {
+					usesTaintedResources = true
+					break resLoop
+				}
+			}
+		}
+	}
+	
 	// If the application has conversion webhook errors, set health status to degraded
-	if hasConversionIssues {
+	// Set health status based on the severity of detected issues
+	if usesTaintedResources {
+		// Application directly uses tainted resources - mark as degraded
+		app.Status.Health.Status = health.HealthStatusDegraded
+		// Add a condition explaining why it's degraded
+		now := metav1.Now()
+		app.Status.SetConditions(
+			[]appv1.ApplicationCondition{
+				{
+					Type:    appv1.ApplicationConditionComparisonError,
+					Message: "Application directly uses resources with known issues in the cluster cache",
+					LastTransitionTime: &now,
+				},
+			},
+			map[appv1.ApplicationConditionType]bool{appv1.ApplicationConditionComparisonError: true},
+		)
+	} else if hasCacheIssues {
+		// Application has cache issues but doesn't directly use tainted resources
 		app.Status.Health.Status = health.HealthStatusDegraded
 	} else {
 		app.Status.Health.Status = compareResult.healthStatus
@@ -2014,18 +2063,26 @@ func (ctrl *ApplicationController) needRefreshAppStatus(app *appv1.Application,
 	return false, refreshType, compareWith
 }
 
-// hasConversionWebhookErrors checks if the application has conversion webhook errors
-// in either its sync results or conditions
-func hasConversionWebhookErrors(app *appv1.Application) bool {
+// hasClusterCacheIssues checks if the application has cluster cache issues
+// like conversion webhook errors, pagination token expiration, etc.
+func hasClusterCacheIssues(app *appv1.Application) bool {
 	// Check operation state for conversion webhook errors
 	if app.Status.OperationState != nil {
-		if strings.Contains(app.Status.OperationState.Message, "conversion webhook") || strings.Contains(app.Status.OperationState.Message, "known conversion webhook failures") || strings.Contains(app.Status.OperationState.Message, "unavailable resource types") {
+		if strings.Contains(app.Status.OperationState.Message, "conversion webhook") || 
+		   strings.Contains(app.Status.OperationState.Message, "known conversion webhook failures") || 
+		   strings.Contains(app.Status.OperationState.Message, "unavailable resource types") ||
+		   strings.Contains(app.Status.OperationState.Message, "failed to list resources") ||
+		   strings.Contains(app.Status.OperationState.Message, "Expired: too old resource version") {
 			return true
 		}
 		// Also check sync result resources
 		if app.Status.OperationState.SyncResult != nil {
 			for _, res := range app.Status.OperationState.SyncResult.Resources {
-				if strings.Contains(res.Message, "conversion webhook") || strings.Contains(res.Message, "known conversion webhook failures") || strings.Contains(res.Message, "unavailable resource types") {
+				if strings.Contains(res.Message, "conversion webhook") || 
+				   strings.Contains(res.Message, "known conversion webhook failures") || 
+				   strings.Contains(res.Message, "unavailable resource types") ||
+				   strings.Contains(res.Message, "failed to list resources") ||
+				   strings.Contains(res.Message, "Expired: too old resource version") {
 					return true
 				}
 			}
@@ -2036,7 +2093,11 @@ func hasConversionWebhookErrors(app *appv1.Application) bool {
 	for _, condition := range app.Status.Conditions {
 		if (condition.Type == appv1.ApplicationConditionComparisonError || 
 		    condition.Type == appv1.ApplicationConditionSyncError) && 
-		    (strings.Contains(condition.Message, "conversion webhook") || strings.Contains(condition.Message, "known conversion webhook failures") || strings.Contains(condition.Message, "unavailable resource types")) {
+		    (strings.Contains(condition.Message, "conversion webhook") || 
+		     strings.Contains(condition.Message, "known conversion webhook failures") || 
+		     strings.Contains(condition.Message, "unavailable resource types") ||
+		     strings.Contains(condition.Message, "failed to list resources") ||
+		     strings.Contains(condition.Message, "Expired: too old resource version")) {
 			return true
 		}
 	}

controller/cache/cache.go

@@ -655,14 +655,42 @@ func (c *liveStateCache) getSyncedCluster(server *appv1.Cluster) (clustercache.C
 
 	err = clusterCache.EnsureSynced()
 	if err != nil {
-		if isConversionWebhookError(err) {
-			log.WithField("cluster", server.Server).Warnf("Conversion webhook error during cluster sync, cluster cache may be incomplete: %v", err)
+		if isClusterCacheError(err) {
+			log.WithField("cluster", server.Server).Warnf("Cluster cache sync error detected, cluster cache may be incomplete: %v", err)
+			
 			// Extract the GVK from the error and track it as a failed GVK
-			gvkStr := extractGVKFromConversionWebhookError(err)
+			gvkStr := extractGVKFromCacheError(err)
 			if gvkStr != "" {
 				c.trackFailedResourceGVK(server.Server, gvkStr)
+				
+				// Also mark the cluster as tainted
+				errorType := "unknown"
+				errStr := err.Error()
+				if strings.Contains(errStr, "conversion webhook") {
+					errorType = "conversion_webhook"
+				} else if strings.Contains(errStr, "Expired: too old resource version") {
+					errorType = "pagination_token_expired"
+				} else if strings.Contains(errStr, "connection") {
+					errorType = "connection_issue"
+				}
+				
+				markClusterTainted(server.Server, errStr, gvkStr, errorType)
+			}
+			
+			// Mark the cluster as tainted
+			errorType := "unknown"
+			errStr := err.Error()
+			if strings.Contains(errStr, "conversion webhook") {
+				errorType = "conversion_webhook"
+			} else if strings.Contains(errStr, "Expired: too old resource version") {
+				errorType = "pagination_token_expired"
+			} else if strings.Contains(errStr, "connection") {
+				errorType = "connection_issue"
 			}
-			// For conversion webhook errors, we return the cluster cache without an error
+			
+			markClusterTainted(server.Server, errStr, gvkStr, errorType)
+			
+			// For known cache errors, we return the cluster cache without an error
 			// but log the issue. This allows applications to continue with
 			// whatever state is available rather than failing completely.
 			// We're handling partial state in each of the methods that use getSyncedCluster
@@ -1068,18 +1096,40 @@ func (c *liveStateCache) GetClustersInfo() []clustercache.ClusterInfo {
 			for gvk := range failedGVKs {
 				failedGVKList = append(failedGVKList, gvk)
 			}
+		}
+		c.failedGVKLock.RUnlock()
+		
+		// Update cluster info with failed resource GVKs from our tracking
+		
+		// Check if this cluster is tainted
+		ClusterTaintLock.RLock()
+		taints, exists := clusterTaints[server]
+		ClusterTaintLock.RUnlock()
+		
+		// Update cluster info with taint status and reason
+		if exists && len(taints) > 0 {
+			info.IsTainted = true
 
-			// If there are failed GVKs, modify SyncError to include this information
-			if len(failedGVKList) > 0 {
-				msg := fmt.Sprintf("Cluster has %d unavailable resource types due to conversion webhook errors", len(failedGVKList))
-				if info.SyncError == nil {
-					info.SyncError = fmt.Errorf(msg)
-				} else {
-					info.SyncError = fmt.Errorf("%s; %s", info.SyncError.Error(), msg)
+			// Add any tainted GVKs that aren't already in the list
+			for gvk := range taints {
+				if !contains(failedGVKList, gvk) {
+					info.FailedResourceGVKs = append(info.FailedResourceGVKs, gvk)
 				}
 			}
+			
+			// Set taint reason
+			info.TaintReason = "Cluster has tainted resource types"
+		}
+		
+		// If there are failed GVKs, modify SyncError to include this information
+		if len(failedGVKList) > 0 {
+			msg := fmt.Sprintf("Cluster has %d unavailable resource types", len(failedGVKList))
+			if info.SyncError == nil {
+				info.SyncError = errors.New(msg)
+			} else {
+				info.SyncError = errors.New(info.SyncError.Error() + "; " + msg)
+			}
 		}
-		c.failedGVKLock.RUnlock()
 
 		res = append(res, info)
 	}
@@ -1095,7 +1145,42 @@ func (c *liveStateCache) UpdateShard(shard int) bool {
 	return c.clusterSharding.UpdateShard(shard)
 }
 
+// isClusterCacheError checks if an error is a known cluster cache error
+// This handles a broader class of errors that can happen when syncing cluster cache
+func isClusterCacheError(err error) bool {
+	if err == nil {
+		return false
+	}
+	errStr := err.Error()
+	
+	// Define all the error patterns we want to catch
+	errorPatterns := []string{
+		// Conversion webhook errors
+		"conversion webhook",
+		
+		// Pagination token expiration errors
+		"Expired: too old resource version",
+		
+		// General resource list errors
+		"failed to list resources",
+		
+		// Connection issues
+		"connection refused",
+		"connection reset by peer",
+		"i/o timeout",
+	}
+	
+	for _, pattern := range errorPatterns {
+		if strings.Contains(errStr, pattern) {
+			return true
+		}
+	}
+	
+	return false
+}
+
 // isConversionWebhookError checks if an error is related to conversion webhooks
+// Kept for backward compatibility
 func isConversionWebhookError(err error) bool {
 	if err == nil {
 		return false
@@ -1104,9 +1189,9 @@ func isConversionWebhookError(err error) bool {
 	return strings.Contains(errStr, "conversion webhook")
 }
 
-// extractGVKFromConversionWebhookError attempts to extract the GroupVersionKind from a conversion webhook error
+// extractGVKFromCacheError attempts to extract the GroupVersionKind from various cache errors
 // Returns an empty string if no GVK could be extracted
-func extractGVKFromConversionWebhookError(err error) string {
+func extractGVKFromCacheError(err error) string {
 	if err == nil {
 		return ""
 	}
@@ -1152,6 +1237,53 @@ func (c *liveStateCache) trackFailedResourceGVK(server string, gvkStr string) {
 	}).Infof("Tracked failed resource GVK due to conversion webhook error")
 }
 
+// Map of server URL to tainted resource GVKs and error types
+var clusterTaints = make(map[string]map[string]string) // server -> gvk -> error type
+var ClusterTaintLock = sync.RWMutex{}
+
+// markClusterTainted marks a cluster as having tainted cache state
+func markClusterTainted(server string, reason string, gvk string, errorType string) {
+	ClusterTaintLock.Lock()
+	defer ClusterTaintLock.Unlock()
+	
+	// Initialize if not exists
+	_, exists := clusterTaints[server]
+	if !exists {
+		clusterTaints[server] = make(map[string]string)
+	}
+	
+	// Store the GVK and error type
+	if gvk != "" {
+		clusterTaints[server][gvk] = errorType
+	}
+}
+
+// IsClusterTainted checks if a cluster is in tainted state
+func IsClusterTainted(server string) bool {
+	ClusterTaintLock.RLock()
+	defer ClusterTaintLock.RUnlock()
+	
+	taints, exists := clusterTaints[server]
+	return exists && len(taints) > 0
+}
+
+// GetTaintedGVKs returns a list of tainted GVKs for a cluster
+func GetTaintedGVKs(server string) []string {
+	ClusterTaintLock.RLock()
+	defer ClusterTaintLock.RUnlock()
+	
+	taints, exists := clusterTaints[server]
+	if !exists {
+		return nil
+	}
+	
+	gvks := make([]string, 0, len(taints))
+	for gvk := range taints {
+		gvks = append(gvks, gvk)
+	}
+	return gvks
+}
+
 // isResourceGVKFailed checks if a resource GVK is in the failed resources map
 func (c *liveStateCache) isResourceGVKFailed(server string, gvkStr string) bool {
 	c.failedGVKLock.RLock()
@@ -1176,6 +1308,21 @@ func (c *liveStateCache) clearFailedResourceGVKs(server string) {
 	delete(c.failedResourceGVKs, server)
 }
 
+// contains checks if a string is in a slice
+func contains(slice []string, str string) bool {
+	for _, s := range slice {
+		if s == str {
+			return true
+		}
+	}
+	return false
+}
+
+// GetClusterTaints returns the map of cluster taints
+func GetClusterTaints() map[string]map[string]string {
+	return clusterTaints
+}
+
 // cleanupExpiredFailedGVKs removes expired entries from the failed GVKs cache
 // This should be called periodically to avoid the cache growing too large
 func (c *liveStateCache) cleanupExpiredFailedGVKs() {