github/Reloader
1
0
Fork 0
mirror of https://github.com/stakater/Reloader.git synced 2026-08-23 22:16:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add webhook notification and conflict retry for Reloader v2

Browse Source
This commit is contained in:
TheiLLeniumStudios
2025-12-28 08:47:52 +01:00
parent 0a2aa122f5
commit ce1e7dfafb
9 changed files with 555 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/pkg/controller/configmap_reconciler.go
+90 -31
View File
@@ -3,11 +3,14 @@ package controller
import (
"context"
"sync"
"time"
"github.com/go-logr/logr"
"github.com/stakater/Reloader/internal/pkg/config"
"github.com/stakater/Reloader/internal/pkg/events"
"github.com/stakater/Reloader/internal/pkg/metrics"
"github.com/stakater/Reloader/internal/pkg/reload"
"github.com/stakater/Reloader/internal/pkg/webhook"
"github.com/stakater/Reloader/internal/pkg/workload"
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
@@ -28,9 +31,9 @@ type ConfigMapReconciler struct {
ReloadService *reload.Service
Registry *workload.Registry
Collectors *metrics.Collectors
EventRecorder *events.Recorder
WebhookClient *webhook.Client
// initialized tracks whether initial sync has completed.
// Used to skip create events during startup unless SyncAfterRestart is enabled.
initialized bool
initOnce sync.Once
}
@@ -79,20 +82,31 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
}
decisions := r.ReloadService.ProcessConfigMap(change, workloads)
// Apply reloads
// Collect workloads that should be reloaded
var workloadsToReload []reload.ReloadDecision
for _, decision := range decisions {
if !decision.ShouldReload {
continue
if decision.ShouldReload {
workloadsToReload = append(workloadsToReload, decision)
}
}
// If webhook is configured, send notification instead of modifying workloads
if r.WebhookClient.IsConfigured() && len(workloadsToReload) > 0 {
return r.sendWebhookNotification(ctx, cm.Name, cm.Namespace, reload.ResourceTypeConfigMap, workloadsToReload, log)
}
// Apply reloads with conflict retry
for _, decision := range workloadsToReload {
log.Info("reloading workload",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
"reason", decision.Reason,
)
updated, err := r.ReloadService.ApplyReload(
updated, err := UpdateWorkloadWithRetry(
ctx,
r.Client,
r.ReloadService,
decision.Workload,
cm.Name,
reload.ResourceTypeConfigMap,
@@ -101,24 +115,17 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
decision.AutoReload,
)
if err != nil {
log.Error(err, "failed to apply reload",
log.Error(err, "failed to update workload",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
)
r.EventRecorder.ReloadFailed(decision.Workload.GetObject(), "ConfigMap", cm.Name, err)
r.recordMetrics(false, cm.Namespace)
continue
}
if updated {
// Persist the changes
if err := r.Update(ctx, decision.Workload.GetObject()); err != nil {
log.Error(err, "failed to update workload",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
)
r.recordMetrics(false, cm.Namespace)
continue
}
r.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), "ConfigMap", cm.Name)
r.recordMetrics(true, cm.Namespace)
log.Info("workload reloaded successfully",
"workload", decision.Workload.GetName(),
@@ -130,6 +137,9 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
return ctrl.Result{}, nil
}
// FieldManager is the field manager name used for server-side apply.
const FieldManager = "reloader"
// handleDelete handles ConfigMap deletion events.
func (r *ConfigMapReconciler) handleDelete(ctx context.Context, req ctrl.Request, log logr.Logger) (ctrl.Result, error) {
log.Info("handling ConfigMap deletion")
@@ -141,8 +151,7 @@ func (r *ConfigMapReconciler) handleDelete(ctx context.Context, req ctrl.Request
return ctrl.Result{}, err
}
// For delete events, we create a change with nil ConfigMap
// The service will use an empty hash
// For delete events, we create a change with empty ConfigMap
change := reload.ConfigMapChange{
ConfigMap: &corev1.ConfigMap{},
EventType: reload.EventTypeDelete,
@@ -152,19 +161,30 @@ func (r *ConfigMapReconciler) handleDelete(ctx context.Context, req ctrl.Request
decisions := r.ReloadService.ProcessConfigMap(change, workloads)
// Apply reloads for delete
// Collect workloads that should be reloaded
var workloadsToReload []reload.ReloadDecision
for _, decision := range decisions {
if !decision.ShouldReload {
continue
if decision.ShouldReload {
workloadsToReload = append(workloadsToReload, decision)
}
}
// If webhook is configured, send notification instead of modifying workloads
if r.WebhookClient.IsConfigured() && len(workloadsToReload) > 0 {
return r.sendWebhookNotification(ctx, req.Name, req.Namespace, reload.ResourceTypeConfigMap, workloadsToReload, log)
}
// Apply reloads for delete with conflict retry
for _, decision := range workloadsToReload {
log.Info("reloading workload due to ConfigMap deletion",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
)
updated, err := r.ReloadService.ApplyReload(
updated, err := UpdateWorkloadWithRetry(
ctx,
r.Client,
r.ReloadService,
decision.Workload,
req.Name,
reload.ResourceTypeConfigMap,
@@ -173,17 +193,14 @@ func (r *ConfigMapReconciler) handleDelete(ctx context.Context, req ctrl.Request
decision.AutoReload,
)
if err != nil {
log.Error(err, "failed to apply reload for deletion")
log.Error(err, "failed to update workload")
r.EventRecorder.ReloadFailed(decision.Workload.GetObject(), "ConfigMap", req.Name, err)
r.recordMetrics(false, req.Namespace)
continue
}
if updated {
if err := r.Update(ctx, decision.Workload.GetObject()); err != nil {
log.Error(err, "failed to update workload")
r.recordMetrics(false, req.Namespace)
continue
}
r.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), "ConfigMap", req.Name)
r.recordMetrics(true, req.Namespace)
}
}
@@ -291,10 +308,52 @@ func (r *ConfigMapReconciler) listCronJobs(ctx context.Context, namespace string
// recordMetrics records reload metrics.
func (r *ConfigMapReconciler) recordMetrics(success bool, namespace string) {
if r.Collectors == nil {
return
r.Collectors.RecordReload(success, namespace)
}
// sendWebhookNotification sends a webhook notification instead of modifying workloads.
func (r *ConfigMapReconciler) sendWebhookNotification(
ctx context.Context,
resourceName, namespace string,
resourceType reload.ResourceType,
decisions []reload.ReloadDecision,
log logr.Logger,
) (ctrl.Result, error) {
var workloads []webhook.WorkloadInfo
var hash string
for _, d := range decisions {
workloads = append(workloads, webhook.WorkloadInfo{
Kind: string(d.Workload.Kind()),
Name: d.Workload.GetName(),
Namespace: d.Workload.GetNamespace(),
})
if hash == "" {
hash = d.Hash
}
}
// TODO: Integrate with existing metrics collectors
payload := webhook.Payload{
Kind: string(resourceType),
Namespace: namespace,
ResourceName: resourceName,
ResourceType: string(resourceType),
Hash: hash,
Timestamp: time.Now().UTC(),
Workloads: workloads,
}
if err := r.WebhookClient.Send(ctx, payload); err != nil {
log.Error(err, "failed to send webhook notification")
r.recordMetrics(false, namespace)
return ctrl.Result{}, err
}
log.Info("webhook notification sent",
"resource", resourceName,
"workloadCount", len(workloads),
)
r.recordMetrics(true, namespace)
return ctrl.Result{}, nil
}
// SetupWithManager sets up the controller with the Manager.
internal/pkg/controller/manager.go
+124
View File
@@ -0,0 +1,124 @@
package controller
import (
"context"
"fmt"
"time"
argorolloutsv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
"github.com/go-logr/logr"
"github.com/stakater/Reloader/internal/pkg/config"
"github.com/stakater/Reloader/internal/pkg/events"
"github.com/stakater/Reloader/internal/pkg/metrics"
"github.com/stakater/Reloader/internal/pkg/reload"
"github.com/stakater/Reloader/internal/pkg/webhook"
"github.com/stakater/Reloader/internal/pkg/workload"
"k8s.io/apimachinery/pkg/runtime"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
clientgoscheme "k8s.io/client-go/kubernetes/scheme"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/healthz"
ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics/server"
)
var runtimeScheme = runtime.NewScheme()
func init() {
utilruntime.Must(clientgoscheme.AddToScheme(runtimeScheme))
utilruntime.Must(argorolloutsv1alpha1.AddToScheme(runtimeScheme))
}
// ManagerOptions contains options for creating a new Manager.
type ManagerOptions struct {
Config *config.Config
Log logr.Logger
Collectors *metrics.Collectors
}
// NewManager creates a new controller-runtime manager with the given options.
func NewManager(opts ManagerOptions) (ctrl.Manager, error) {
cfg := opts.Config
leaseDuration := 15 * time.Second
renewDeadline := 10 * time.Second
retryPeriod := 2 * time.Second
mgrOpts := ctrl.Options{
Scheme: runtimeScheme,
Metrics: ctrlmetrics.Options{
BindAddress: cfg.MetricsAddr,
},
HealthProbeBindAddress: cfg.HealthAddr,
LeaderElection: cfg.EnableHA,
LeaderElectionID: "reloader-leader-election",
LeaseDuration: &leaseDuration,
RenewDeadline: &renewDeadline,
RetryPeriod: &retryPeriod,
}
mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), mgrOpts)
if err != nil {
return nil, fmt.Errorf("creating manager: %w", err)
}
if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
return nil, fmt.Errorf("setting up health check: %w", err)
}
if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
return nil, fmt.Errorf("setting up ready check: %w", err)
}
return mgr, nil
}
// SetupReconcilers sets up all reconcilers with the manager.
func SetupReconcilers(mgr ctrl.Manager, cfg *config.Config, log logr.Logger, collectors *metrics.Collectors) error {
registry := workload.NewRegistry(cfg.ArgoRolloutsEnabled)
reloadService := reload.NewService(cfg)
eventRecorder := events.NewRecorder(mgr.GetEventRecorderFor("reloader"))
// Create webhook client if URL is configured
var webhookClient *webhook.Client
if cfg.WebhookURL != "" {
webhookClient = webhook.NewClient(cfg.WebhookURL, log.WithName("webhook"))
log.Info("webhook mode enabled", "url", cfg.WebhookURL)
}
if !cfg.IsResourceIgnored("configmaps") {
if err := (&ConfigMapReconciler{
Client: mgr.GetClient(),
Log: log.WithName("configmap-reconciler"),
Config: cfg,
ReloadService: reloadService,
Registry: registry,
Collectors: collectors,
EventRecorder: eventRecorder,
WebhookClient: webhookClient,
}).SetupWithManager(mgr); err != nil {
return fmt.Errorf("setting up configmap reconciler: %w", err)
}
}
if !cfg.IsResourceIgnored("secrets") {
if err := (&SecretReconciler{
Client: mgr.GetClient(),
Log: log.WithName("secret-reconciler"),
Config: cfg,
ReloadService: reloadService,
Registry: registry,
Collectors: collectors,
EventRecorder: eventRecorder,
WebhookClient: webhookClient,
}).SetupWithManager(mgr); err != nil {
return fmt.Errorf("setting up secret reconciler: %w", err)
}
}
return nil
}
// RunManager starts the manager and blocks until it stops.
func RunManager(ctx context.Context, mgr ctrl.Manager, log logr.Logger) error {
log.Info("starting manager")
return mgr.Start(ctx)
}
internal/pkg/controller/retry.go
+68
View File
@@ -0,0 +1,68 @@
package controller
import (
"context"
"github.com/stakater/Reloader/internal/pkg/reload"
"github.com/stakater/Reloader/internal/pkg/workload"
"k8s.io/apimachinery/pkg/api/errors"
"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.
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) {
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
}
internal/pkg/controller/secret_reconciler.go
+86 -30
View File
@@ -3,11 +3,14 @@ package controller
import (
"context"
"sync"
"time"
"github.com/go-logr/logr"
"github.com/stakater/Reloader/internal/pkg/config"
"github.com/stakater/Reloader/internal/pkg/events"
"github.com/stakater/Reloader/internal/pkg/metrics"
"github.com/stakater/Reloader/internal/pkg/reload"
"github.com/stakater/Reloader/internal/pkg/webhook"
"github.com/stakater/Reloader/internal/pkg/workload"
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
@@ -28,9 +31,9 @@ type SecretReconciler struct {
ReloadService *reload.Service
Registry *workload.Registry
Collectors *metrics.Collectors
EventRecorder *events.Recorder
WebhookClient *webhook.Client
// initialized tracks whether initial sync has completed.
// Used to skip create events during startup unless SyncAfterRestart is enabled.
initialized bool
initOnce sync.Once
}
@@ -79,20 +82,31 @@ func (r *SecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
}
decisions := r.ReloadService.ProcessSecret(change, workloads)
// Apply reloads
// Collect workloads that should be reloaded
var workloadsToReload []reload.ReloadDecision
for _, decision := range decisions {
if !decision.ShouldReload {
continue
if decision.ShouldReload {
workloadsToReload = append(workloadsToReload, decision)
}
}
// If webhook is configured, send notification instead of modifying workloads
if r.WebhookClient.IsConfigured() && len(workloadsToReload) > 0 {
return r.sendWebhookNotification(ctx, secret.Name, secret.Namespace, reload.ResourceTypeSecret, workloadsToReload, log)
}
// Apply reloads with conflict retry
for _, decision := range workloadsToReload {
log.Info("reloading workload",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
"reason", decision.Reason,
)
updated, err := r.ReloadService.ApplyReload(
updated, err := UpdateWorkloadWithRetry(
ctx,
r.Client,
r.ReloadService,
decision.Workload,
secret.Name,
reload.ResourceTypeSecret,
@@ -101,24 +115,17 @@ func (r *SecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
decision.AutoReload,
)
if err != nil {
log.Error(err, "failed to apply reload",
log.Error(err, "failed to update workload",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
)
r.EventRecorder.ReloadFailed(decision.Workload.GetObject(), "Secret", secret.Name, err)
r.recordMetrics(false, secret.Namespace)
continue
}
if updated {
// Persist the changes
if err := r.Update(ctx, decision.Workload.GetObject()); err != nil {
log.Error(err, "failed to update workload",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
)
r.recordMetrics(false, secret.Namespace)
continue
}
r.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), "Secret", secret.Name)
r.recordMetrics(true, secret.Namespace)
log.Info("workload reloaded successfully",
"workload", decision.Workload.GetName(),
@@ -142,7 +149,6 @@ func (r *SecretReconciler) handleDelete(ctx context.Context, req ctrl.Request, l
}
// For delete events, we create a change with empty Secret
// The service will use an empty hash
change := reload.SecretChange{
Secret: &corev1.Secret{},
EventType: reload.EventTypeDelete,
@@ -152,19 +158,30 @@ func (r *SecretReconciler) handleDelete(ctx context.Context, req ctrl.Request, l
decisions := r.ReloadService.ProcessSecret(change, workloads)
// Apply reloads for delete
// Collect workloads that should be reloaded
var workloadsToReload []reload.ReloadDecision
for _, decision := range decisions {
if !decision.ShouldReload {
continue
if decision.ShouldReload {
workloadsToReload = append(workloadsToReload, decision)
}
}
// If webhook is configured, send notification instead of modifying workloads
if r.WebhookClient.IsConfigured() && len(workloadsToReload) > 0 {
return r.sendWebhookNotification(ctx, req.Name, req.Namespace, reload.ResourceTypeSecret, workloadsToReload, log)
}
// Apply reloads for delete with conflict retry
for _, decision := range workloadsToReload {
log.Info("reloading workload due to Secret deletion",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
)
updated, err := r.ReloadService.ApplyReload(
updated, err := UpdateWorkloadWithRetry(
ctx,
r.Client,
r.ReloadService,
decision.Workload,
req.Name,
reload.ResourceTypeSecret,
@@ -173,17 +190,14 @@ func (r *SecretReconciler) handleDelete(ctx context.Context, req ctrl.Request, l
decision.AutoReload,
)
if err != nil {
log.Error(err, "failed to apply reload for deletion")
log.Error(err, "failed to update workload")
r.EventRecorder.ReloadFailed(decision.Workload.GetObject(), "Secret", req.Name, err)
r.recordMetrics(false, req.Namespace)
continue
}
if updated {
if err := r.Update(ctx, decision.Workload.GetObject()); err != nil {
log.Error(err, "failed to update workload")
r.recordMetrics(false, req.Namespace)
continue
}
r.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), "Secret", req.Name)
r.recordMetrics(true, req.Namespace)
}
}
@@ -291,10 +305,52 @@ func (r *SecretReconciler) listCronJobs(ctx context.Context, namespace string) (
// recordMetrics records reload metrics.
func (r *SecretReconciler) recordMetrics(success bool, namespace string) {
if r.Collectors == nil {
return
r.Collectors.RecordReload(success, namespace)
}
// sendWebhookNotification sends a webhook notification instead of modifying workloads.
func (r *SecretReconciler) sendWebhookNotification(
ctx context.Context,
resourceName, namespace string,
resourceType reload.ResourceType,
decisions []reload.ReloadDecision,
log logr.Logger,
) (ctrl.Result, error) {
var workloads []webhook.WorkloadInfo
var hash string
for _, d := range decisions {
workloads = append(workloads, webhook.WorkloadInfo{
Kind: string(d.Workload.Kind()),
Name: d.Workload.GetName(),
Namespace: d.Workload.GetNamespace(),
})
if hash == "" {
hash = d.Hash
}
}
// TODO: Integrate with existing metrics collectors
payload := webhook.Payload{
Kind: string(resourceType),
Namespace: namespace,
ResourceName: resourceName,
ResourceType: string(resourceType),
Hash: hash,
Timestamp: time.Now().UTC(),
Workloads: workloads,
}
if err := r.WebhookClient.Send(ctx, payload); err != nil {
log.Error(err, "failed to send webhook notification")
r.recordMetrics(false, namespace)
return ctrl.Result{}, err
}
log.Info("webhook notification sent",
"resource", resourceName,
"workloadCount", len(workloads),
)
r.recordMetrics(true, namespace)
return ctrl.Result{}, nil
}
// SetupWithManager sets up the controller with the Manager.