github/Reloader
1
0
Fork 0
mirror of https://github.com/stakater/Reloader.git synced 2026-08-27 14:37:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Use strategic merge to patch workloads and add metrics for load testing

Browse Source
This commit is contained in:
TheiLLeniumStudios
2026-01-04 14:01:14 +01:00
parent da6f33c059
commit c785067a44
17 changed files with 537 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/pkg/controller/configmap_reconciler.go
+26 -5
View File
@@ -3,6 +3,7 @@ package controller
import (
"context"
"sync"
"time"
"github.com/go-logr/logr"
"github.com/stakater/Reloader/internal/pkg/alerting"
@@ -39,6 +40,7 @@ type ConfigMapReconciler struct {
// Reconcile handles ConfigMap events and triggers workload reloads as needed.
func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
startTime := time.Now()
log := r.Log.WithValues("configmap", req.NamespacedName)
r.initOnce.Do(func() {
@@ -46,34 +48,53 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
log.Info("ConfigMap controller initialized")
})
r.Collectors.RecordEventReceived("reconcile", "configmap")
var cm corev1.ConfigMap
if err := r.Get(ctx, req.NamespacedName, &cm); err != nil {
if errors.IsNotFound(err) {
if r.Config.ReloadOnDelete {
return r.handleDelete(ctx, req, log)
r.Collectors.RecordEventReceived("delete", "configmap")
result, err := r.handleDelete(ctx, req, log)
if err != nil {
r.Collectors.RecordReconcile("error", time.Since(startTime))
} else {
r.Collectors.RecordReconcile("success", time.Since(startTime))
}
return result, err
}
r.Collectors.RecordSkipped("not_found")
r.Collectors.RecordReconcile("success", time.Since(startTime))
return ctrl.Result{}, nil
}
log.Error(err, "failed to get ConfigMap")
r.Collectors.RecordError("get_configmap")
r.Collectors.RecordReconcile("error", time.Since(startTime))
return ctrl.Result{}, err
}
if r.Config.IsNamespaceIgnored(cm.Namespace) {
log.V(1).Info("skipping ConfigMap in ignored namespace")
r.Collectors.RecordSkipped("ignored_namespace")
r.Collectors.RecordReconcile("success", time.Since(startTime))
return ctrl.Result{}, nil
}
return r.reloadHandler().Process(ctx, cm.Namespace, cm.Name, reload.ResourceTypeConfigMap,
result, err := r.reloadHandler().Process(ctx, cm.Namespace, cm.Name, reload.ResourceTypeConfigMap,
func(workloads []workload.WorkloadAccessor) []reload.ReloadDecision {
return r.ReloadService.Process(reload.ConfigMapChange{
ConfigMap: &cm,
EventType: reload.EventTypeUpdate,
}, workloads)
}, log)
}
// FieldManager is the field manager name used for server-side apply.
const FieldManager = "reloader"
if err != nil {
r.Collectors.RecordReconcile("error", time.Since(startTime))
} else {
r.Collectors.RecordReconcile("success", time.Since(startTime))
}
return result, err
}
func (r *ConfigMapReconciler) handleDelete(ctx context.Context, req ctrl.Request, log logr.Logger) (ctrl.Result, error) {
log.Info("handling ConfigMap deletion")
internal/pkg/controller/handler.go
+59 -18
View File
@@ -38,11 +38,32 @@ func (h *ReloadHandler) Process(
workloads, err := h.Lister.List(ctx, namespace)
if err != nil {
log.Error(err, "failed to list workloads")
h.Collectors.RecordError("list_workloads")
return ctrl.Result{}, err
}
workloadsByKind := make(map[string]int)
for _, w := range workloads {
workloadsByKind[string(w.Kind())]++
}
for kind, count := range workloadsByKind {
h.Collectors.RecordWorkloadsScanned(kind, count)
}
decisions := reload.FilterDecisions(getDecisions(workloads))
matchedByKind := make(map[string]int)
for _, d := range decisions {
matchedByKind[string(d.Workload.Kind())]++
}
for kind, count := range matchedByKind {
h.Collectors.RecordWorkloadsMatched(kind, count)
}
if len(decisions) == 0 {
h.Collectors.RecordSkipped("no_match")
}
if h.WebhookClient.IsConfigured() && len(decisions) > 0 {
return h.sendWebhook(ctx, resourceName, namespace, resourceType, decisions, log)
}
@@ -61,11 +82,13 @@ func (h *ReloadHandler) sendWebhook(
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(),
})
workloads = append(
workloads, webhook.WorkloadInfo{
Kind: string(d.Workload.Kind()),
Name: d.Workload.GetName(),
Namespace: d.Workload.GetNamespace(),
},
)
if hash == "" {
hash = d.Hash
}
@@ -81,17 +104,22 @@ func (h *ReloadHandler) sendWebhook(
Workloads: workloads,
}
actionStartTime := time.Now()
if err := h.WebhookClient.Send(ctx, payload); err != nil {
log.Error(err, "failed to send webhook notification")
h.Collectors.RecordReload(false, namespace)
h.Collectors.RecordAction("webhook", "error", time.Since(actionStartTime))
h.Collectors.RecordError("webhook_send")
return ctrl.Result{}, err
}
log.Info("webhook notification sent",
log.Info(
"webhook notification sent",
"resource", resourceName,
"workloadCount", len(workloads),
)
h.Collectors.RecordReload(true, namespace)
h.Collectors.RecordAction("webhook", "success", time.Since(actionStartTime))
return ctrl.Result{}, nil
}
@@ -103,12 +131,14 @@ func (h *ReloadHandler) applyReloads(
log logr.Logger,
) {
for _, decision := range decisions {
log.Info("reloading workload",
log.Info(
"reloading workload",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
"reason", decision.Reason,
)
actionStartTime := time.Now()
updated, err := UpdateWorkloadWithRetry(
ctx,
h.Client,
@@ -121,35 +151,46 @@ func (h *ReloadHandler) applyReloads(
decision.Hash,
decision.AutoReload,
)
actionLatency := time.Since(actionStartTime)
if err != nil {
log.Error(err, "failed to update workload",
log.Error(
err, "failed to update workload",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
)
h.EventRecorder.ReloadFailed(decision.Workload.GetObject(), resourceType.Kind(), resourceName, err)
h.Collectors.RecordReload(false, resourceNamespace)
h.Collectors.RecordAction(string(decision.Workload.Kind()), "error", actionLatency)
h.Collectors.RecordError("update_workload")
continue
}
if updated {
h.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), resourceType.Kind(), resourceName)
h.Collectors.RecordReload(true, resourceNamespace)
log.Info("workload reloaded successfully",
h.Collectors.RecordAction(string(decision.Workload.Kind()), "success", actionLatency)
log.Info(
"workload reloaded successfully",
"workload", decision.Workload.GetName(),
"kind", decision.Workload.Kind(),
)
if err := h.Alerter.Send(ctx, alerting.AlertMessage{
WorkloadKind: string(decision.Workload.Kind()),
WorkloadName: decision.Workload.GetName(),
WorkloadNamespace: decision.Workload.GetNamespace(),
ResourceKind: resourceType.Kind(),
ResourceName: resourceName,
ResourceNamespace: resourceNamespace,
Timestamp: time.Now(),
}); err != nil {
if err := h.Alerter.Send(
ctx, alerting.AlertMessage{
WorkloadKind: string(decision.Workload.Kind()),
WorkloadName: decision.Workload.GetName(),
WorkloadNamespace: decision.Workload.GetNamespace(),
ResourceKind: resourceType.Kind(),
ResourceName: resourceName,
ResourceNamespace: resourceNamespace,
Timestamp: time.Now(),
},
); err != nil {
log.Error(err, "failed to send alert")
}
} else {
h.Collectors.RecordAction(string(decision.Workload.Kind()), "no_change", actionLatency)
}
}
}
internal/pkg/controller/manager.go
+16 -6
View File
@@ -27,8 +27,16 @@ var runtimeScheme = runtime.NewScheme()
func init() {
utilruntime.Must(clientgoscheme.AddToScheme(runtimeScheme))
utilruntime.Must(argorolloutsv1alpha1.AddToScheme(runtimeScheme))
utilruntime.Must(openshiftv1.AddToScheme(runtimeScheme))
}
// AddOptionalSchemes adds optional workload type schemes if enabled.
func AddOptionalSchemes(argoRolloutsEnabled, deploymentConfigEnabled bool) {
if argoRolloutsEnabled {
utilruntime.Must(argorolloutsv1alpha1.AddToScheme(runtimeScheme))
}
if deploymentConfigEnabled {
utilruntime.Must(openshiftv1.AddToScheme(runtimeScheme))
}
}
// ManagerOptions contains options for creating a new Manager.
@@ -117,10 +125,12 @@ func NewManagerWithRestConfig(opts ManagerOptions, restConfig *rest.Config) (ctr
// 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(workload.RegistryOptions{
ArgoRolloutsEnabled: cfg.ArgoRolloutsEnabled,
DeploymentConfigEnabled: cfg.DeploymentConfigEnabled,
})
registry := workload.NewRegistry(
workload.RegistryOptions{
ArgoRolloutsEnabled: cfg.ArgoRolloutsEnabled,
DeploymentConfigEnabled: cfg.DeploymentConfigEnabled,
},
)
reloadService := reload.NewService(cfg)
eventRecorder := events.NewRecorder(mgr.GetEventRecorderFor("reloader"))
pauseHandler := reload.NewPauseHandler(cfg)
internal/pkg/controller/retry.go
+6 -5
View File
@@ -41,7 +41,7 @@ func UpdateObjectWithRetry(
return nil
}
return c.Update(ctx, obj, client.FieldOwner(FieldManager))
return c.Update(ctx, obj, client.FieldOwner(workload.FieldManager))
},
)
}
@@ -107,6 +107,7 @@ func retryWithReload(
}
return err
}
wl.ResetOriginal()
}
isFirstAttempt = false
@@ -142,7 +143,7 @@ func updateStandardWorkload(
return retryWithReload(
ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload,
func() error {
return c.Update(ctx, wl.GetObject(), client.FieldOwner(FieldManager))
return wl.Update(ctx, c)
},
)
}
@@ -170,7 +171,7 @@ func updateDeploymentWithPause(
return err
}
}
return c.Update(ctx, wl.GetObject(), client.FieldOwner(FieldManager))
return wl.Update(ctx, c)
},
)
}
@@ -242,7 +243,7 @@ func updateJobWithRecreate(
newJob.Spec.Selector = nil
// Create the new job with same spec
if err := c.Create(ctx, newJob, client.FieldOwner(FieldManager)); err != nil {
if err := c.Create(ctx, newJob, client.FieldOwner(workload.FieldManager)); err != nil {
return false, err
}
@@ -304,7 +305,7 @@ func updateCronJobWithNewJob(
Spec: cronJob.Spec.JobTemplate.Spec,
}
if err := c.Create(ctx, job, client.FieldOwner(FieldManager)); err != nil {
if err := c.Create(ctx, job, client.FieldOwner(workload.FieldManager)); err != nil {
return false, err
}
internal/pkg/controller/secret_reconciler.go
+26 -2
View File
@@ -3,6 +3,7 @@ package controller
import (
"context"
"sync"
"time"
"github.com/go-logr/logr"
"github.com/stakater/Reloader/internal/pkg/alerting"
@@ -39,6 +40,7 @@ type SecretReconciler struct {
// Reconcile handles Secret events and triggers workload reloads as needed.
func (r *SecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
startTime := time.Now()
log := r.Log.WithValues("secret", req.NamespacedName)
r.initOnce.Do(func() {
@@ -46,30 +48,52 @@ func (r *SecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
log.Info("Secret controller initialized")
})
r.Collectors.RecordEventReceived("reconcile", "secret")
var secret corev1.Secret
if err := r.Get(ctx, req.NamespacedName, &secret); err != nil {
if errors.IsNotFound(err) {
if r.Config.ReloadOnDelete {
return r.handleDelete(ctx, req, log)
r.Collectors.RecordEventReceived("delete", "secret")
result, err := r.handleDelete(ctx, req, log)
if err != nil {
r.Collectors.RecordReconcile("error", time.Since(startTime))
} else {
r.Collectors.RecordReconcile("success", time.Since(startTime))
}
return result, err
}
r.Collectors.RecordSkipped("not_found")
r.Collectors.RecordReconcile("success", time.Since(startTime))
return ctrl.Result{}, nil
}
log.Error(err, "failed to get Secret")
r.Collectors.RecordError("get_secret")
r.Collectors.RecordReconcile("error", time.Since(startTime))
return ctrl.Result{}, err
}
if r.Config.IsNamespaceIgnored(secret.Namespace) {
log.V(1).Info("skipping Secret in ignored namespace")
r.Collectors.RecordSkipped("ignored_namespace")
r.Collectors.RecordReconcile("success", time.Since(startTime))
return ctrl.Result{}, nil
}
return r.reloadHandler().Process(ctx, secret.Namespace, secret.Name, reload.ResourceTypeSecret,
result, err := r.reloadHandler().Process(ctx, secret.Namespace, secret.Name, reload.ResourceTypeSecret,
func(workloads []workload.WorkloadAccessor) []reload.ReloadDecision {
return r.ReloadService.Process(reload.SecretChange{
Secret: &secret,
EventType: reload.EventTypeUpdate,
}, workloads)
}, log)
if err != nil {
r.Collectors.RecordReconcile("error", time.Since(startTime))
} else {
r.Collectors.RecordReconcile("success", time.Since(startTime))
}
return result, err
}
func (r *SecretReconciler) handleDelete(ctx context.Context, req ctrl.Request, log logr.Logger) (ctrl.Result, error) {