Reloader/internal/pkg/controller/retry.go

package controller

import (
	"context"
	"maps"

	"github.com/stakater/Reloader/internal/pkg/reload"
	"github.com/stakater/Reloader/internal/pkg/workload"
	batchv1 "k8s.io/api/batch/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/util/retry"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// UpdateWorkloadWithRetry updates a workload with exponential backoff on conflict.
// On conflict, it re-fetches the object, re-applies the reload changes, and retries.
// For Jobs and CronJobs, special handling is applied:
// - Jobs are deleted and recreated with the same spec
// - CronJobs create a new Job from their template
// For Argo Rollouts, special handling is applied based on the rollout strategy annotation.
func UpdateWorkloadWithRetry(
	ctx context.Context,
	c client.Client,
	reloadService *reload.Service,
	wl workload.WorkloadAccessor,
	resourceName string,
	resourceType reload.ResourceType,
	namespace string,
	hash string,
	autoReload bool,
) (bool, error) {
	// Handle special workload types
	switch wl.Kind() {
	case workload.KindJob:
		return updateJobWithRecreate(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
	case workload.KindCronJob:
		return updateCronJobWithNewJob(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
	case workload.KindArgoRollout:
		return updateArgoRollout(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
	default:
		return updateStandardWorkload(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
	}
}

// updateStandardWorkload updates Deployments, DaemonSets, StatefulSets, etc.
func updateStandardWorkload(
	ctx context.Context,
	c client.Client,
	reloadService *reload.Service,
	wl workload.WorkloadAccessor,
	resourceName string,
	resourceType reload.ResourceType,
	namespace string,
	hash string,
	autoReload bool,
) (bool, error) {
	var updated bool
	isFirstAttempt := true

	err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		// On retry, re-fetch the object to get the latest ResourceVersion
		if !isFirstAttempt {
			obj := wl.GetObject()
			key := client.ObjectKeyFromObject(obj)
			if err := c.Get(ctx, key, obj); err != nil {
				if errors.IsNotFound(err) {
					// Object was deleted, nothing to update
					return nil
				}
				return err
			}
		}
		isFirstAttempt = false

		// Apply reload changes (this modifies the workload in-place)
		var applyErr error
		updated, applyErr = reloadService.ApplyReload(
			ctx,
			wl,
			resourceName,
			resourceType,
			namespace,
			hash,
			autoReload,
		)
		if applyErr != nil {
			return applyErr
		}

		if !updated {
			return nil
		}

		// Attempt update with field ownership
		return c.Update(ctx, wl.GetObject(), client.FieldOwner(FieldManager))
	})

	return updated, err
}

// updateJobWithRecreate deletes the Job and recreates it with the updated spec.
// Jobs are immutable after creation, so we must delete and recreate.
func updateJobWithRecreate(
	ctx context.Context,
	c client.Client,
	reloadService *reload.Service,
	wl workload.WorkloadAccessor,
	resourceName string,
	resourceType reload.ResourceType,
	namespace string,
	hash string,
	autoReload bool,
) (bool, error) {
	jobWl, ok := wl.(*workload.JobWorkload)
	if !ok {
		return false, nil
	}

	// Apply reload changes to the workload
	updated, err := reloadService.ApplyReload(
		ctx,
		wl,
		resourceName,
		resourceType,
		namespace,
		hash,
		autoReload,
	)
	if err != nil {
		return false, err
	}

	if !updated {
		return false, nil
	}

	oldJob := jobWl.GetJob()
	newJob := oldJob.DeepCopy()

	// Delete the old job with background propagation
	policy := metav1.DeletePropagationBackground
	if err := c.Delete(ctx, oldJob, &client.DeleteOptions{
		PropagationPolicy: &policy,
	}); err != nil {
		if !errors.IsNotFound(err) {
			return false, err
		}
	}

	// Clear fields that should not be specified when creating a new Job
	newJob.ResourceVersion = ""
	newJob.UID = ""
	newJob.CreationTimestamp = metav1.Time{}
	newJob.Status = batchv1.JobStatus{}

	// Remove problematic labels that are auto-generated
	delete(newJob.Spec.Template.Labels, "controller-uid")
	delete(newJob.Spec.Template.Labels, batchv1.ControllerUidLabel)
	delete(newJob.Spec.Template.Labels, batchv1.JobNameLabel)
	delete(newJob.Spec.Template.Labels, "job-name")

	// Remove the selector to allow it to be auto-generated
	newJob.Spec.Selector = nil

	// Create the new job with same spec
	if err := c.Create(ctx, newJob, client.FieldOwner(FieldManager)); err != nil {
		return false, err
	}

	return true, nil
}

// updateCronJobWithNewJob creates a new Job from the CronJob's template.
// CronJobs don't get updated directly; instead, a new Job is triggered.
func updateCronJobWithNewJob(
	ctx context.Context,
	c client.Client,
	reloadService *reload.Service,
	wl workload.WorkloadAccessor,
	resourceName string,
	resourceType reload.ResourceType,
	namespace string,
	hash string,
	autoReload bool,
) (bool, error) {
	cronJobWl, ok := wl.(*workload.CronJobWorkload)
	if !ok {
		return false, nil
	}

	// Apply reload changes to get the updated spec
	updated, err := reloadService.ApplyReload(
		ctx,
		wl,
		resourceName,
		resourceType,
		namespace,
		hash,
		autoReload,
	)
	if err != nil {
		return false, err
	}

	if !updated {
		return false, nil
	}

	cronJob := cronJobWl.GetCronJob()

	// Build annotations for the new Job
	annotations := make(map[string]string)
	annotations["cronjob.kubernetes.io/instantiate"] = "manual"
	maps.Copy(annotations, cronJob.Spec.JobTemplate.Annotations)

	// Create a new Job from the CronJob template
	job := &batchv1.Job{
		ObjectMeta: metav1.ObjectMeta{
			GenerateName: cronJob.Name + "-",
			Namespace:    cronJob.Namespace,
			Annotations:  annotations,
			Labels:       cronJob.Spec.JobTemplate.Labels,
			OwnerReferences: []metav1.OwnerReference{
				*metav1.NewControllerRef(cronJob, batchv1.SchemeGroupVersion.WithKind("CronJob")),
			},
		},
		Spec: cronJob.Spec.JobTemplate.Spec,
	}

	if err := c.Create(ctx, job, client.FieldOwner(FieldManager)); err != nil {
		return false, err
	}

	return true, nil
}

// updateArgoRollout updates an Argo Rollout using its custom Update method.
// This handles the rollout strategy annotation to determine whether to do
// a standard rollout or set the restartAt field.
func updateArgoRollout(
	ctx context.Context,
	c client.Client,
	reloadService *reload.Service,
	wl workload.WorkloadAccessor,
	resourceName string,
	resourceType reload.ResourceType,
	namespace string,
	hash string,
	autoReload bool,
) (bool, error) {
	rolloutWl, ok := wl.(*workload.RolloutWorkload)
	if !ok {
		return false, nil
	}

	var updated bool
	isFirstAttempt := true

	err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
		// On retry, re-fetch the object to get the latest ResourceVersion
		if !isFirstAttempt {
			obj := rolloutWl.GetObject()
			key := client.ObjectKeyFromObject(obj)
			if err := c.Get(ctx, key, obj); err != nil {
				if errors.IsNotFound(err) {
					// Object was deleted, nothing to update
					return nil
				}
				return err
			}
		}
		isFirstAttempt = false

		// Apply reload changes (this modifies the workload in-place)
		var applyErr error
		updated, applyErr = reloadService.ApplyReload(
			ctx,
			wl,
			resourceName,
			resourceType,
			namespace,
			hash,
			autoReload,
		)
		if applyErr != nil {
			return applyErr
		}

		if !updated {
			return nil
		}

		// Use the RolloutWorkload's Update method which handles the rollout strategy
		return rolloutWl.Update(ctx, c)
	})

	return updated, err
}