mirror of
https://github.com/stakater/Reloader.git
synced 2026-08-27 14:37:17 +00:00
feat: Introduce a Generic ResourceReconciler and a generic BaseWorkload to de-duplicate a lot of code
This commit is contained in:
TheiLLeniumStudios
@@ -2,13 +2,10 @@ 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"
|
||||
)
|
||||
@@ -48,33 +45,31 @@ func UpdateObjectWithRetry(
|
||||
|
||||
// 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.
|
||||
// Workloads use their UpdateStrategy to determine how they're updated:
|
||||
// - UpdateStrategyPatch: uses strategic merge patch with retry (most workloads)
|
||||
// - UpdateStrategyRecreate: deletes and recreates (Jobs)
|
||||
// - UpdateStrategyCreateNew: creates a new resource from template (CronJobs)
|
||||
// Deployments have additional pause handling for paused rollouts.
|
||||
func UpdateWorkloadWithRetry(
|
||||
ctx context.Context,
|
||||
c client.Client,
|
||||
reloadService *reload.Service,
|
||||
pauseHandler *reload.PauseHandler,
|
||||
wl workload.WorkloadAccessor,
|
||||
wl workload.Workload,
|
||||
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)
|
||||
case workload.KindDeployment:
|
||||
return updateDeploymentWithPause(ctx, c, reloadService, pauseHandler, wl, resourceName, resourceType, namespace, hash, autoReload)
|
||||
switch wl.UpdateStrategy() {
|
||||
case workload.UpdateStrategyRecreate, workload.UpdateStrategyCreateNew:
|
||||
return updateWithSpecialStrategy(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
|
||||
default:
|
||||
// UpdateStrategyPatch: use standard retry logic with special handling for Deployments
|
||||
if wl.Kind() == workload.KindDeployment {
|
||||
return updateDeploymentWithPause(ctx, c, reloadService, pauseHandler, wl, resourceName, resourceType, namespace, hash, autoReload)
|
||||
}
|
||||
return updateStandardWorkload(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload)
|
||||
}
|
||||
}
|
||||
@@ -85,7 +80,7 @@ func retryWithReload(
|
||||
ctx context.Context,
|
||||
c client.Client,
|
||||
reloadService *reload.Service,
|
||||
wl workload.WorkloadAccessor,
|
||||
wl workload.Workload,
|
||||
resourceName string,
|
||||
resourceType reload.ResourceType,
|
||||
namespace string,
|
||||
@@ -133,7 +128,7 @@ func updateStandardWorkload(
|
||||
ctx context.Context,
|
||||
c client.Client,
|
||||
reloadService *reload.Service,
|
||||
wl workload.WorkloadAccessor,
|
||||
wl workload.Workload,
|
||||
resourceName string,
|
||||
resourceType reload.ResourceType,
|
||||
namespace string,
|
||||
@@ -154,7 +149,7 @@ func updateDeploymentWithPause(
|
||||
c client.Client,
|
||||
reloadService *reload.Service,
|
||||
pauseHandler *reload.PauseHandler,
|
||||
wl workload.WorkloadAccessor,
|
||||
wl workload.Workload,
|
||||
resourceName string,
|
||||
resourceType reload.ResourceType,
|
||||
namespace string,
|
||||
@@ -176,25 +171,19 @@ func updateDeploymentWithPause(
|
||||
)
|
||||
}
|
||||
|
||||
// updateJobWithRecreate deletes the Job and recreates it with the updated spec.
|
||||
// Jobs are immutable after creation, so we must delete and recreate.
|
||||
func updateJobWithRecreate(
|
||||
// updateWithSpecialStrategy handles workloads that don't use standard patch.
|
||||
// It applies reload changes, then delegates to the workload's PerformSpecialUpdate.
|
||||
func updateWithSpecialStrategy(
|
||||
ctx context.Context,
|
||||
c client.Client,
|
||||
reloadService *reload.Service,
|
||||
wl workload.WorkloadAccessor,
|
||||
wl workload.Workload,
|
||||
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,
|
||||
@@ -212,145 +201,5 @@ func updateJobWithRecreate(
|
||||
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(workload.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()
|
||||
|
||||
annotations := make(map[string]string)
|
||||
annotations["cronjob.kubernetes.io/instantiate"] = "manual"
|
||||
maps.Copy(annotations, cronJob.Spec.JobTemplate.Annotations)
|
||||
|
||||
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(workload.FieldManager)); err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
savedAnnotations := maps.Clone(cronJob.Spec.JobTemplate.Spec.Template.Annotations)
|
||||
|
||||
err = UpdateObjectWithRetry(
|
||||
ctx, c, cronJob, func() (bool, error) {
|
||||
if cronJob.Spec.JobTemplate.Spec.Template.Annotations == nil {
|
||||
cronJob.Spec.JobTemplate.Spec.Template.Annotations = make(map[string]string)
|
||||
}
|
||||
maps.Copy(cronJob.Spec.JobTemplate.Spec.Template.Annotations, savedAnnotations)
|
||||
return true, nil
|
||||
},
|
||||
)
|
||||
|
||||
if 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
|
||||
}
|
||||
|
||||
return retryWithReload(
|
||||
ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload,
|
||||
func() error {
|
||||
return rolloutWl.Update(ctx, c)
|
||||
},
|
||||
)
|
||||
return wl.PerformSpecialUpdate(ctx, c)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user