Merge pull request #417 from fabriziosestito/fix/error-handling

fix(handlers): ack messages when an error occurs and fix status update conflict

Author: flavio

internal/handlers/create_catalog.go

@@ -22,6 +22,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -215,18 +216,29 @@ func (h *CreateCatalogHandler) Handle(ctx context.Context, message []byte) error
 		return fmt.Errorf("cannot delete obsolete images in registry %s: %w", registry.Name, err)
 	}
 
-	if len(discoveredImages) == 0 {
-		h.logger.DebugContext(ctx, "No images to process", "scanjob", scanJob.Name, "namespace", scanJob.Namespace)
+	// It is possible that the controller is slow to set the status condition "Scheduled" to true,
+	// so we might encounter conflicts when setting the status conditions.
+	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		if err = h.k8sClient.Get(ctx, types.NamespacedName{
+			Name:      scanJob.Name,
+			Namespace: scanJob.Namespace,
+		}, scanJob); err != nil {
+			return fmt.Errorf("cannot get scan job %s/%s while updating status: %w", scanJob.Namespace, scanJob.Name, err)
+		}
 
-		scanJob.MarkComplete(v1alpha1.ReasonNoImagesToScan, "No images to process")
-	} else {
-		h.logger.DebugContext(ctx, "Images to process", "count", len(discoveredImages))
-		scanJob.MarkInProgress(v1alpha1.ReasonSBOMGenerationInProgress, "SBOM generation in progress")
-		scanJob.Status.ImagesCount = len(discoveredImages)
-		scanJob.Status.ScannedImagesCount = 0
-	}
+		if len(discoveredImages) == 0 {
+			h.logger.DebugContext(ctx, "No images to process", "scanjob", scanJob.Name, "namespace", scanJob.Namespace)
+			scanJob.MarkComplete(v1alpha1.ReasonNoImagesToScan, "No images to process")
+		} else {
+			h.logger.DebugContext(ctx, "Images to process", "count", len(discoveredImages))
+			scanJob.MarkInProgress(v1alpha1.ReasonSBOMGenerationInProgress, "SBOM generation in progress")
+			scanJob.Status.ImagesCount = len(discoveredImages)
+			scanJob.Status.ScannedImagesCount = 0
+		}
 
-	if err = h.k8sClient.Status().Update(ctx, scanJob); err != nil {
+		return h.k8sClient.Status().Update(ctx, scanJob)
+	})
+	if err != nil {
 		return fmt.Errorf("cannot update scan job status %s/%s: %w", createCatalogMessage.ScanJob.Namespace, createCatalogMessage.ScanJob.Name, err)
 	}
 

internal/handlers/scanjob_failure.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	sbombasticv1alpha1 "github.com/rancher/sbombastic/api/v1alpha1"
@@ -34,40 +35,40 @@ func (h *ScanJobFailureHandler) HandleFailure(ctx context.Context, message []byt
 	if err := json.Unmarshal(message, baseMessage); err != nil {
 		return fmt.Errorf("failed to unmarshal base message: %w", err)
 	}
-
 	h.logger.DebugContext(ctx, "Handling ScanJob failure",
 		"scanjob", baseMessage.ScanJob.Name,
 		"namespace", baseMessage.ScanJob.Namespace,
 		"error", errorMessage,
 	)
 
 	scanJob := &sbombasticv1alpha1.ScanJob{}
-	err := h.k8sClient.Get(ctx, client.ObjectKey{
-		Name:      baseMessage.ScanJob.Name,
-		Namespace: baseMessage.ScanJob.Namespace,
-	}, scanJob)
-	if err != nil {
-		return fmt.Errorf("failed to get ScanJob %s/%s: %w",
-			baseMessage.ScanJob.Namespace, baseMessage.ScanJob.Name, err)
-	}
 
-	original := scanJob.DeepCopy()
-	scanJob.MarkFailed(sbombasticv1alpha1.ReasonInternalError, errorMessage)
+	// It is possible that the controller is slow to set the status condition "Scheduled" to true,
+	// so we might encounter conflicts when setting the status conditions.
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		if err := h.k8sClient.Get(ctx, client.ObjectKey{
+			Name:      baseMessage.ScanJob.Name,
+			Namespace: baseMessage.ScanJob.Namespace,
+		}, scanJob); err != nil {
+			return fmt.Errorf("failed to get ScanJob %s/%s: %w", scanJob.Namespace, scanJob.Name, err)
+		}
 
-	if err := h.k8sClient.Status().Patch(ctx, scanJob, client.MergeFrom(original)); err != nil {
+		scanJob.MarkFailed(sbombasticv1alpha1.ReasonInternalError, errorMessage)
+		return h.k8sClient.Status().Update(ctx, scanJob)
+	})
+	if err != nil {
 		h.logger.ErrorContext(ctx, "Failed to update ScanJob status with failure",
 			"scanjob", scanJob.Name,
 			"namespace", scanJob.Namespace,
 			"error", err,
 		)
-		return fmt.Errorf("failed to update ScanJob %s/%s status: %w", baseMessage.ScanJob.Namespace, baseMessage.ScanJob.Name, err)
+		return fmt.Errorf("failed to update ScanJob %s/%s status: %w", scanJob.Namespace, scanJob.Name, err)
 	}
 
 	h.logger.DebugContext(ctx, "ScanJob marked as failed",
 		"scanjob", scanJob.Name,
 		"namespace", scanJob.Namespace,
 		"error_message", errorMessage,
 	)
-
 	return nil
 }

internal/messaging/subscriber.go

@@ -78,11 +78,11 @@ func (s *NatsSubscriber) Run(ctx context.Context) error {
 							"error", err,
 						)
 					}
-				}
 
-				// Return early to avoid acking a message that failed processing.
-				// This allows the message to be retried later even if the nak fails.
-				return
+					// Return early to avoid acking a message that failed processing.
+					// This allows the message to be retried later even if the nak fails.
+					return
+				}
 			}
 
 			if err := msg.Ack(); err != nil {