mirror of
https://github.com/stakater/Reloader.git
synced 2026-08-27 14:37:17 +00:00
feat: Argo rollouts workload + refactor of alerting
This commit is contained in:
TheiLLeniumStudios
@@ -0,0 +1,89 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/go-logr/logr"
|
||||
"github.com/stakater/Reloader/internal/pkg/config"
|
||||
"github.com/stakater/Reloader/internal/pkg/reload"
|
||||
appsv1 "k8s.io/api/apps/v1"
|
||||
"k8s.io/apimachinery/pkg/api/errors"
|
||||
ctrl "sigs.k8s.io/controller-runtime"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
"sigs.k8s.io/controller-runtime/pkg/predicate"
|
||||
)
|
||||
|
||||
// DeploymentReconciler reconciles Deployment objects to handle pause expiration.
|
||||
// This reconciler watches for deployments that were paused by Reloader and
|
||||
// unpauses them when the pause period expires.
|
||||
type DeploymentReconciler struct {
|
||||
client.Client
|
||||
Log logr.Logger
|
||||
Config *config.Config
|
||||
PauseHandler *reload.PauseHandler
|
||||
}
|
||||
|
||||
// Reconcile handles Deployment pause expiration.
|
||||
func (r *DeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
|
||||
log := r.Log.WithValues("deployment", req.NamespacedName)
|
||||
|
||||
var deploy appsv1.Deployment
|
||||
if err := r.Get(ctx, req.NamespacedName, &deploy); err != nil {
|
||||
if errors.IsNotFound(err) {
|
||||
return ctrl.Result{}, nil
|
||||
}
|
||||
return ctrl.Result{}, err
|
||||
}
|
||||
|
||||
// Check if this deployment was paused by Reloader
|
||||
if !r.PauseHandler.IsPausedByReloader(&deploy) {
|
||||
return ctrl.Result{}, nil
|
||||
}
|
||||
|
||||
// Check if pause period has expired
|
||||
expired, remainingTime, err := r.PauseHandler.CheckPauseExpired(&deploy)
|
||||
if err != nil {
|
||||
log.Error(err, "Failed to check pause expiration")
|
||||
return ctrl.Result{}, err
|
||||
}
|
||||
|
||||
if !expired {
|
||||
// Still within pause period - requeue to check again
|
||||
log.V(1).Info("Deployment pause not yet expired", "remaining", remainingTime)
|
||||
return ctrl.Result{RequeueAfter: remainingTime}, nil
|
||||
}
|
||||
|
||||
// Pause period has expired - unpause the deployment
|
||||
log.Info("Unpausing deployment after pause period expired")
|
||||
r.PauseHandler.ClearPause(&deploy)
|
||||
|
||||
if err := r.Update(ctx, &deploy, client.FieldOwner(FieldManager)); err != nil {
|
||||
log.Error(err, "Failed to unpause deployment")
|
||||
return ctrl.Result{}, err
|
||||
}
|
||||
|
||||
return ctrl.Result{}, nil
|
||||
}
|
||||
|
||||
// SetupWithManager sets up the DeploymentReconciler with the manager.
|
||||
func (r *DeploymentReconciler) SetupWithManager(mgr ctrl.Manager) error {
|
||||
return ctrl.NewControllerManagedBy(mgr).
|
||||
For(&appsv1.Deployment{}).
|
||||
WithEventFilter(r.pausedByReloaderPredicate()).
|
||||
Complete(r)
|
||||
}
|
||||
|
||||
// pausedByReloaderPredicate returns a predicate that only selects deployments
|
||||
// that have been paused by Reloader (have the paused-at annotation).
|
||||
func (r *DeploymentReconciler) pausedByReloaderPredicate() predicate.Predicate {
|
||||
return predicate.NewPredicateFuncs(func(obj client.Object) bool {
|
||||
annotations := obj.GetAnnotations()
|
||||
if annotations == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Only process if deployment has our paused-at annotation
|
||||
_, hasPausedAt := annotations[r.Config.Annotations.PausedAt]
|
||||
return hasPausedAt
|
||||
})
|
||||
}
|
||||
@@ -2,16 +2,23 @@ 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,
|
||||
@@ -22,6 +29,31 @@ func UpdateWorkloadWithRetry(
|
||||
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
|
||||
@@ -66,3 +98,202 @@ func UpdateWorkloadWithRetry(
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user