From c785067a44b0d69df128c550972c85c36a5a5f60 Mon Sep 17 00:00:00 2001 From: TheiLLeniumStudios <104288623+TheiLLeniumStudios@users.noreply.github.com> Date: Sun, 4 Jan 2026 14:01:14 +0100 Subject: [PATCH] feat: Use strategic merge to patch workloads and add metrics for load testing --- cmd/reloader/main.go | 22 +- go.mod | 1 - go.sum | 6 - .../pkg/controller/configmap_reconciler.go | 31 +- internal/pkg/controller/handler.go | 77 ++++- internal/pkg/controller/manager.go | 22 +- internal/pkg/controller/retry.go | 11 +- internal/pkg/controller/secret_reconciler.go | 28 +- internal/pkg/metrics/prometheus.go | 316 +++++++++++++++++- internal/pkg/workload/cronjob.go | 4 + internal/pkg/workload/daemonset.go | 17 +- internal/pkg/workload/deployment.go | 17 +- internal/pkg/workload/deploymentconfig.go | 19 +- internal/pkg/workload/interface.go | 9 + internal/pkg/workload/job.go | 4 + internal/pkg/workload/rollout.go | 22 +- internal/pkg/workload/statefulset.go | 17 +- 17 files changed, 537 insertions(+), 86 deletions(-) diff --git a/cmd/reloader/main.go b/cmd/reloader/main.go index da23618c..b603df65 100644 --- a/cmd/reloader/main.go +++ b/cmd/reloader/main.go @@ -106,6 +106,18 @@ func run(cmd *cobra.Command, args []string) error { collectors := metrics.SetupPrometheusEndpoint() + if config.ShouldAutoDetectOpenShift() { + restConfig := controllerruntime.GetConfigOrDie() + discoveryClient, err := discovery.NewDiscoveryClientForConfig(restConfig) + if err != nil { + log.V(1).Info("Failed to create discovery client for DeploymentConfig detection", "error", err) + } else if openshift.HasDeploymentConfigSupport(discoveryClient, log) { + cfg.DeploymentConfigEnabled = true + } + } + + controller.AddOptionalSchemes(cfg.ArgoRolloutsEnabled, cfg.DeploymentConfigEnabled) + mgr, err := controller.NewManager( controller.ManagerOptions{ Config: cfg, @@ -117,16 +129,6 @@ func run(cmd *cobra.Command, args []string) error { return fmt.Errorf("creating manager: %w", err) } - if config.ShouldAutoDetectOpenShift() { - restConfig := controllerruntime.GetConfigOrDie() - discoveryClient, err := discovery.NewDiscoveryClientForConfig(restConfig) - if err != nil { - log.V(1).Info("Failed to create discovery client for DeploymentConfig detection", "error", err) - } else if openshift.HasDeploymentConfigSupport(discoveryClient, log) { - cfg.DeploymentConfigEnabled = true - } - } - if err := controller.SetupReconcilers(mgr, cfg, log, &collectors); err != nil { return fmt.Errorf("setting up reconcilers: %w", err) } diff --git a/go.mod b/go.mod index ece1cd8a..ad21ed63 100644 --- a/go.mod +++ b/go.mod @@ -60,7 +60,6 @@ require ( github.com/prometheus/common v0.67.4 // indirect github.com/prometheus/procfs v0.19.2 // indirect github.com/sagikazarmark/locafero v0.12.0 // indirect - github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/spf13/afero v1.15.0 // indirect github.com/spf13/cast v1.10.0 // indirect github.com/subosito/gotenv v1.6.0 // indirect diff --git a/go.sum b/go.sum index caceb32c..07c36025 100644 --- a/go.sum +++ b/go.sum @@ -117,8 +117,6 @@ github.com/onsi/ginkgo/v2 v2.27.2 h1:LzwLj0b89qtIy6SSASkzlNvX6WktqurSHwkk2ipF/Ns github.com/onsi/ginkgo/v2 v2.27.2/go.mod h1:ArE1D/XhNXBXCBkKOLkbsb2c81dQHCRcF5zwn/ykDRo= github.com/onsi/gomega v1.38.2 h1:eZCjf2xjZAqe+LeWvKb5weQ+NcPwX84kqJ0cZNxok2A= github.com/onsi/gomega v1.38.2/go.mod h1:W2MJcYxRGV63b418Ai34Ud0hEdTVXq9NW9+Sx6uXf3k= -github.com/openshift/api v0.0.0-20251222154221-d4b2fef98af2 h1:q7fp/9fnJuXWCzpBzbHZne6aMLGYPKhzPy1uULfaJqA= -github.com/openshift/api v0.0.0-20251222154221-d4b2fef98af2/go.mod h1:d5uzF0YN2nQQFA0jIEWzzOZ+edmo6wzlGLvx5Fhz4uY= github.com/openshift/api v0.0.0-20251223163548-3f584b29ee4a h1:lz22938uOBlzTHjGpobGeVWkcxGu6fDQ7oZWheClTHE= github.com/openshift/api v0.0.0-20251223163548-3f584b29ee4a/go.mod h1:d5uzF0YN2nQQFA0jIEWzzOZ+edmo6wzlGLvx5Fhz4uY= github.com/openshift/client-go v0.0.0-20251223102348-558b0eef16bc h1:nIlRaJfr/yGjPV15MNF5eVHLAGyXFjcUzO+hXeWDDk8= @@ -143,12 +141,8 @@ github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY= github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= -github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= -github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik= github.com/sagikazarmark/locafero v0.12.0 h1:/NQhBAkUb4+fH1jivKHWusDYFjMOOKU88eegjfxfHb4= github.com/sagikazarmark/locafero v0.12.0/go.mod h1:sZh36u/YSZ918v0Io+U9ogLYQJ9tLLBmM4eneO6WwsI= -github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= -github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U= github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= github.com/spf13/afero v1.15.0/go.mod h1:NC2ByUVxtQs4b3sIUphxK0NioZnmxgyCrfzeuq8lxMg= github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= diff --git a/internal/pkg/controller/configmap_reconciler.go b/internal/pkg/controller/configmap_reconciler.go index 6736a518..e14c14bf 100644 --- a/internal/pkg/controller/configmap_reconciler.go +++ b/internal/pkg/controller/configmap_reconciler.go @@ -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") diff --git a/internal/pkg/controller/handler.go b/internal/pkg/controller/handler.go index 423f2432..f4065ce8 100644 --- a/internal/pkg/controller/handler.go +++ b/internal/pkg/controller/handler.go @@ -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) } } } diff --git a/internal/pkg/controller/manager.go b/internal/pkg/controller/manager.go index e6a0bcd7..aa1ab6cd 100644 --- a/internal/pkg/controller/manager.go +++ b/internal/pkg/controller/manager.go @@ -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) diff --git a/internal/pkg/controller/retry.go b/internal/pkg/controller/retry.go index a7f20810..fec0daac 100644 --- a/internal/pkg/controller/retry.go +++ b/internal/pkg/controller/retry.go @@ -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 } diff --git a/internal/pkg/controller/secret_reconciler.go b/internal/pkg/controller/secret_reconciler.go index 46723803..e7b2481b 100644 --- a/internal/pkg/controller/secret_reconciler.go +++ b/internal/pkg/controller/secret_reconciler.go @@ -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) { diff --git a/internal/pkg/metrics/prometheus.go b/internal/pkg/metrics/prometheus.go index 87f9148b..b95731c8 100644 --- a/internal/pkg/metrics/prometheus.go +++ b/internal/pkg/metrics/prometheus.go @@ -3,16 +3,48 @@ package metrics import ( "net/http" "os" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" + ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" ) -// Collectors holds Prometheus metrics collectors for Reloader. +// Collectors holds all Prometheus metrics collectors for Reloader. type Collectors struct { Reloaded *prometheus.CounterVec ReloadedByNamespace *prometheus.CounterVec countByNamespace bool + + // === Comprehensive metrics for load testing === + + // Reconcile/Handler metrics + ReconcileTotal *prometheus.CounterVec // Total reconcile calls by result + ReconcileDuration *prometheus.HistogramVec // Time spent in reconcile/handler + + // Action metrics + ActionTotal *prometheus.CounterVec // Total actions by workload kind and result + ActionLatency *prometheus.HistogramVec // Time from event to action applied + + // Skip metrics + SkippedTotal *prometheus.CounterVec // Skipped operations by reason + + // Queue metrics (controller-runtime exposes some automatically, but we add custom ones) + QueueDepth prometheus.Gauge // Current queue depth + QueueAdds prometheus.Counter // Total items added to queue + QueueLatency *prometheus.HistogramVec // Time spent in queue + + // Error and retry metrics + ErrorsTotal *prometheus.CounterVec // Errors by type + RetriesTotal prometheus.Counter // Total retries + + // Event processing metrics + EventsReceived *prometheus.CounterVec // Events received by type (add/update/delete) + EventsProcessed *prometheus.CounterVec // Events processed by type and result + + // Resource discovery metrics + WorkloadsScanned *prometheus.CounterVec // Workloads scanned by kind + WorkloadsMatched *prometheus.CounterVec // Workloads matched for reload by kind } // RecordReload records a reload event with the given success status and namespace. @@ -29,13 +61,113 @@ func (c *Collectors) RecordReload(success bool, namespace string) { c.Reloaded.With(prometheus.Labels{"success": successLabel}).Inc() if c.countByNamespace { - c.ReloadedByNamespace.With(prometheus.Labels{ - "success": successLabel, - "namespace": namespace, - }).Inc() + c.ReloadedByNamespace.With( + prometheus.Labels{ + "success": successLabel, + "namespace": namespace, + }, + ).Inc() } } +// RecordReconcile records a reconcile/handler invocation. +func (c *Collectors) RecordReconcile(result string, duration time.Duration) { + if c == nil { + return + } + c.ReconcileTotal.With(prometheus.Labels{"result": result}).Inc() + c.ReconcileDuration.With(prometheus.Labels{"result": result}).Observe(duration.Seconds()) +} + +// RecordAction records a reload action on a workload. +func (c *Collectors) RecordAction(workloadKind string, result string, latency time.Duration) { + if c == nil { + return + } + c.ActionTotal.With(prometheus.Labels{"workload_kind": workloadKind, "result": result}).Inc() + c.ActionLatency.With(prometheus.Labels{"workload_kind": workloadKind}).Observe(latency.Seconds()) +} + +// RecordSkipped records a skipped operation with reason. +func (c *Collectors) RecordSkipped(reason string) { + if c == nil { + return + } + c.SkippedTotal.With(prometheus.Labels{"reason": reason}).Inc() +} + +// RecordQueueAdd records an item being added to the queue. +func (c *Collectors) RecordQueueAdd() { + if c == nil { + return + } + c.QueueAdds.Inc() +} + +// SetQueueDepth sets the current queue depth. +func (c *Collectors) SetQueueDepth(depth int) { + if c == nil { + return + } + c.QueueDepth.Set(float64(depth)) +} + +// RecordQueueLatency records how long an item spent in the queue. +func (c *Collectors) RecordQueueLatency(latency time.Duration) { + if c == nil { + return + } + c.QueueLatency.With(prometheus.Labels{}).Observe(latency.Seconds()) +} + +// RecordError records an error by type. +func (c *Collectors) RecordError(errorType string) { + if c == nil { + return + } + c.ErrorsTotal.With(prometheus.Labels{"type": errorType}).Inc() +} + +// RecordRetry records a retry attempt. +func (c *Collectors) RecordRetry() { + if c == nil { + return + } + c.RetriesTotal.Inc() +} + +// RecordEventReceived records an event being received. +func (c *Collectors) RecordEventReceived(eventType string, resourceType string) { + if c == nil { + return + } + c.EventsReceived.With(prometheus.Labels{"event_type": eventType, "resource_type": resourceType}).Inc() +} + +// RecordEventProcessed records an event being processed. +func (c *Collectors) RecordEventProcessed(eventType string, resourceType string, result string) { + if c == nil { + return + } + c.EventsProcessed.With(prometheus.Labels{"event_type": eventType, "resource_type": resourceType, "result": result}).Inc() +} + +// RecordWorkloadsScanned records workloads scanned during a reconcile. +func (c *Collectors) RecordWorkloadsScanned(kind string, count int) { + if c == nil { + return + } + c.WorkloadsScanned.With(prometheus.Labels{"kind": kind}).Add(float64(count)) +} + +// RecordWorkloadsMatched records workloads matched for reload. +func (c *Collectors) RecordWorkloadsMatched(kind string, count int) { + if c == nil { + return + } + c.WorkloadsMatched.With(prometheus.Labels{"kind": kind}).Add(float64(count)) +} + func NewCollectors() Collectors { reloaded := prometheus.NewCounterVec( prometheus.CounterOpts{ @@ -43,11 +175,8 @@ func NewCollectors() Collectors { Name: "reload_executed_total", Help: "Counter of reloads executed by Reloader.", }, - []string{ - "success", - }, + []string{"success"}, ) - reloaded.With(prometheus.Labels{"success": "true"}).Add(0) reloaded.With(prometheus.Labels{"success": "false"}).Add(0) @@ -57,26 +186,185 @@ func NewCollectors() Collectors { Name: "reload_executed_total_by_namespace", Help: "Counter of reloads executed by Reloader by namespace.", }, - []string{ - "success", - "namespace", + []string{"success", "namespace"}, + ) + + // === Comprehensive metrics === + + reconcileTotal := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "reconcile_total", + Help: "Total number of reconcile/handler invocations by result.", + }, + []string{"result"}, + ) + + reconcileDuration := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "reloader", + Name: "reconcile_duration_seconds", + Help: "Time spent in reconcile/handler in seconds.", + Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + }, + []string{"result"}, + ) + + actionTotal := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "action_total", + Help: "Total number of reload actions by workload kind and result.", + }, + []string{"workload_kind", "result"}, + ) + + actionLatency := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "reloader", + Name: "action_latency_seconds", + Help: "Time from event received to action applied in seconds.", + Buckets: []float64{0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}, + }, + []string{"workload_kind"}, + ) + + skippedTotal := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "skipped_total", + Help: "Total number of skipped operations by reason.", + }, + []string{"reason"}, + ) + + queueDepth := prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: "reloader", + Name: "workqueue_depth", + Help: "Current depth of the work queue.", }, ) + + queueAdds := prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "workqueue_adds_total", + Help: "Total number of items added to the work queue.", + }, + ) + + queueLatency := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "reloader", + Name: "workqueue_latency_seconds", + Help: "Time spent in the work queue in seconds.", + Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5}, + }, + []string{}, + ) + + errorsTotal := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "errors_total", + Help: "Total number of errors by type.", + }, + []string{"type"}, + ) + + retriesTotal := prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "retries_total", + Help: "Total number of retry attempts.", + }, + ) + + eventsReceived := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "events_received_total", + Help: "Total number of events received by type and resource.", + }, + []string{"event_type", "resource_type"}, + ) + + eventsProcessed := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "events_processed_total", + Help: "Total number of events processed by type, resource, and result.", + }, + []string{"event_type", "resource_type", "result"}, + ) + + workloadsScanned := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "workloads_scanned_total", + Help: "Total number of workloads scanned by kind.", + }, + []string{"kind"}, + ) + + workloadsMatched := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "reloader", + Name: "workloads_matched_total", + Help: "Total number of workloads matched for reload by kind.", + }, + []string{"kind"}, + ) + return Collectors{ Reloaded: reloaded, ReloadedByNamespace: reloadedByNamespace, countByNamespace: os.Getenv("METRICS_COUNT_BY_NAMESPACE") == "enabled", + + ReconcileTotal: reconcileTotal, + ReconcileDuration: reconcileDuration, + ActionTotal: actionTotal, + ActionLatency: actionLatency, + SkippedTotal: skippedTotal, + QueueDepth: queueDepth, + QueueAdds: queueAdds, + QueueLatency: queueLatency, + ErrorsTotal: errorsTotal, + RetriesTotal: retriesTotal, + EventsReceived: eventsReceived, + EventsProcessed: eventsProcessed, + WorkloadsScanned: workloadsScanned, + WorkloadsMatched: workloadsMatched, } } func SetupPrometheusEndpoint() Collectors { collectors := NewCollectors() - prometheus.MustRegister(collectors.Reloaded) + + ctrlmetrics.Registry.MustRegister(collectors.Reloaded) + ctrlmetrics.Registry.MustRegister(collectors.ReconcileTotal) + ctrlmetrics.Registry.MustRegister(collectors.ReconcileDuration) + ctrlmetrics.Registry.MustRegister(collectors.ActionTotal) + ctrlmetrics.Registry.MustRegister(collectors.ActionLatency) + ctrlmetrics.Registry.MustRegister(collectors.SkippedTotal) + ctrlmetrics.Registry.MustRegister(collectors.QueueDepth) + ctrlmetrics.Registry.MustRegister(collectors.QueueAdds) + ctrlmetrics.Registry.MustRegister(collectors.QueueLatency) + ctrlmetrics.Registry.MustRegister(collectors.ErrorsTotal) + ctrlmetrics.Registry.MustRegister(collectors.RetriesTotal) + ctrlmetrics.Registry.MustRegister(collectors.EventsReceived) + ctrlmetrics.Registry.MustRegister(collectors.EventsProcessed) + ctrlmetrics.Registry.MustRegister(collectors.WorkloadsScanned) + ctrlmetrics.Registry.MustRegister(collectors.WorkloadsMatched) if os.Getenv("METRICS_COUNT_BY_NAMESPACE") == "enabled" { - prometheus.MustRegister(collectors.ReloadedByNamespace) + ctrlmetrics.Registry.MustRegister(collectors.ReloadedByNamespace) } + // Note: For controller-runtime based Reloader, the metrics are served + // by controller-runtime's metrics server. This http.Handle is kept for + // the legacy informer-based Reloader which uses its own HTTP server. http.Handle("/metrics", promhttp.Handler()) return collectors diff --git a/internal/pkg/workload/cronjob.go b/internal/pkg/workload/cronjob.go index 80d672e3..9f61b019 100644 --- a/internal/pkg/workload/cronjob.go +++ b/internal/pkg/workload/cronjob.go @@ -91,6 +91,10 @@ func (w *CronJobWorkload) DeepCopy() Workload { return &CronJobWorkload{cronjob: w.cronjob.DeepCopy()} } +// ResetOriginal is a no-op for CronJobs since they don't use strategic merge patch. +// CronJobs create new Jobs instead of being patched. +func (w *CronJobWorkload) ResetOriginal() {} + func (w *CronJobWorkload) GetEnvFromSources() []corev1.EnvFromSource { var sources []corev1.EnvFromSource for _, container := range w.cronjob.Spec.JobTemplate.Spec.Template.Spec.Containers { diff --git a/internal/pkg/workload/daemonset.go b/internal/pkg/workload/daemonset.go index 85ac7b05..c2294a4c 100644 --- a/internal/pkg/workload/daemonset.go +++ b/internal/pkg/workload/daemonset.go @@ -12,11 +12,15 @@ import ( // DaemonSetWorkload wraps a Kubernetes DaemonSet. type DaemonSetWorkload struct { daemonset *appsv1.DaemonSet + original *appsv1.DaemonSet } // NewDaemonSetWorkload creates a new DaemonSetWorkload. func NewDaemonSetWorkload(d *appsv1.DaemonSet) *DaemonSetWorkload { - return &DaemonSetWorkload{daemonset: d} + return &DaemonSetWorkload{ + daemonset: d, + original: d.DeepCopy(), + } } // Ensure DaemonSetWorkload implements WorkloadAccessor. @@ -77,11 +81,18 @@ func (w *DaemonSetWorkload) GetVolumes() []corev1.Volume { } func (w *DaemonSetWorkload) Update(ctx context.Context, c client.Client) error { - return c.Update(ctx, w.daemonset) + return c.Patch(ctx, w.daemonset, client.StrategicMergeFrom(w.original), client.FieldOwner(FieldManager)) } func (w *DaemonSetWorkload) DeepCopy() Workload { - return &DaemonSetWorkload{daemonset: w.daemonset.DeepCopy()} + return &DaemonSetWorkload{ + daemonset: w.daemonset.DeepCopy(), + original: w.original.DeepCopy(), + } +} + +func (w *DaemonSetWorkload) ResetOriginal() { + w.original = w.daemonset.DeepCopy() } func (w *DaemonSetWorkload) GetEnvFromSources() []corev1.EnvFromSource { diff --git a/internal/pkg/workload/deployment.go b/internal/pkg/workload/deployment.go index e4ebefb5..747e9945 100644 --- a/internal/pkg/workload/deployment.go +++ b/internal/pkg/workload/deployment.go @@ -12,11 +12,15 @@ import ( // DeploymentWorkload wraps a Kubernetes Deployment. type DeploymentWorkload struct { deployment *appsv1.Deployment + original *appsv1.Deployment } // NewDeploymentWorkload creates a new DeploymentWorkload. func NewDeploymentWorkload(d *appsv1.Deployment) *DeploymentWorkload { - return &DeploymentWorkload{deployment: d} + return &DeploymentWorkload{ + deployment: d, + original: d.DeepCopy(), + } } // Ensure DeploymentWorkload implements WorkloadAccessor. @@ -77,11 +81,18 @@ func (w *DeploymentWorkload) GetVolumes() []corev1.Volume { } func (w *DeploymentWorkload) Update(ctx context.Context, c client.Client) error { - return c.Update(ctx, w.deployment) + return c.Patch(ctx, w.deployment, client.StrategicMergeFrom(w.original), client.FieldOwner(FieldManager)) } func (w *DeploymentWorkload) DeepCopy() Workload { - return &DeploymentWorkload{deployment: w.deployment.DeepCopy()} + return &DeploymentWorkload{ + deployment: w.deployment.DeepCopy(), + original: w.original.DeepCopy(), + } +} + +func (w *DeploymentWorkload) ResetOriginal() { + w.original = w.deployment.DeepCopy() } func (w *DeploymentWorkload) GetEnvFromSources() []corev1.EnvFromSource { diff --git a/internal/pkg/workload/deploymentconfig.go b/internal/pkg/workload/deploymentconfig.go index 14c60469..680a78b6 100644 --- a/internal/pkg/workload/deploymentconfig.go +++ b/internal/pkg/workload/deploymentconfig.go @@ -11,12 +11,16 @@ import ( // DeploymentConfigWorkload wraps an OpenShift DeploymentConfig. type DeploymentConfigWorkload struct { - dc *openshiftv1.DeploymentConfig + dc *openshiftv1.DeploymentConfig + original *openshiftv1.DeploymentConfig } // NewDeploymentConfigWorkload creates a new DeploymentConfigWorkload. func NewDeploymentConfigWorkload(dc *openshiftv1.DeploymentConfig) *DeploymentConfigWorkload { - return &DeploymentConfigWorkload{dc: dc} + return &DeploymentConfigWorkload{ + dc: dc, + original: dc.DeepCopy(), + } } // Ensure DeploymentConfigWorkload implements WorkloadAccessor. @@ -98,11 +102,18 @@ func (w *DeploymentConfigWorkload) GetVolumes() []corev1.Volume { } func (w *DeploymentConfigWorkload) Update(ctx context.Context, c client.Client) error { - return c.Update(ctx, w.dc) + return c.Patch(ctx, w.dc, client.StrategicMergeFrom(w.original), client.FieldOwner(FieldManager)) } func (w *DeploymentConfigWorkload) DeepCopy() Workload { - return &DeploymentConfigWorkload{dc: w.dc.DeepCopy()} + return &DeploymentConfigWorkload{ + dc: w.dc.DeepCopy(), + original: w.original.DeepCopy(), + } +} + +func (w *DeploymentConfigWorkload) ResetOriginal() { + w.original = w.dc.DeepCopy() } func (w *DeploymentConfigWorkload) GetEnvFromSources() []corev1.EnvFromSource { diff --git a/internal/pkg/workload/interface.go b/internal/pkg/workload/interface.go index e1d50a18..40249edb 100644 --- a/internal/pkg/workload/interface.go +++ b/internal/pkg/workload/interface.go @@ -14,6 +14,10 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" ) +// FieldManager is the field manager name used for server-side apply and patch operations. +// This identifies Reloader as the actor making changes to workload resources. +const FieldManager = "reloader" + // Kind represents the type of workload. type Kind string @@ -69,6 +73,11 @@ type Workload interface { // Update persists changes to the workload. Update(ctx context.Context, c client.Client) error + // ResetOriginal resets the original state to the current object state. + // This should be called after re-fetching the object (e.g., after a conflict) + // to ensure strategic merge patch diffs are calculated correctly. + ResetOriginal() + // DeepCopy returns a deep copy of the workload. DeepCopy() Workload } diff --git a/internal/pkg/workload/job.go b/internal/pkg/workload/job.go index 4e6c9fc6..291249f8 100644 --- a/internal/pkg/workload/job.go +++ b/internal/pkg/workload/job.go @@ -90,6 +90,10 @@ func (w *JobWorkload) DeepCopy() Workload { return &JobWorkload{job: w.job.DeepCopy()} } +// ResetOriginal is a no-op for Jobs since they don't use strategic merge patch. +// Jobs are deleted and recreated instead of being patched. +func (w *JobWorkload) ResetOriginal() {} + func (w *JobWorkload) GetEnvFromSources() []corev1.EnvFromSource { var sources []corev1.EnvFromSource for _, container := range w.job.Spec.Template.Spec.Containers { diff --git a/internal/pkg/workload/rollout.go b/internal/pkg/workload/rollout.go index f19c1713..8e78d3e9 100644 --- a/internal/pkg/workload/rollout.go +++ b/internal/pkg/workload/rollout.go @@ -27,12 +27,16 @@ const RolloutStrategyAnnotation = "reloader.stakater.com/rollout-strategy" // RolloutWorkload wraps an Argo Rollout. type RolloutWorkload struct { - rollout *argorolloutv1alpha1.Rollout + rollout *argorolloutv1alpha1.Rollout + original *argorolloutv1alpha1.Rollout } // NewRolloutWorkload creates a new RolloutWorkload. func NewRolloutWorkload(r *argorolloutv1alpha1.Rollout) *RolloutWorkload { - return &RolloutWorkload{rollout: r} + return &RolloutWorkload{ + rollout: r, + original: r.DeepCopy(), + } } // Ensure RolloutWorkload implements WorkloadAccessor. @@ -98,12 +102,11 @@ func (w *RolloutWorkload) Update(ctx context.Context, c client.Client) error { strategy := w.getStrategy() switch strategy { case RolloutStrategyRestart: - // Use merge patch to set restartAt field + // Set restartAt field to trigger a restart restartAt := metav1.NewTime(time.Now()) w.rollout.Spec.RestartAt = &restartAt } - // For both strategies, we update the rollout (annotations have already been set) - return c.Update(ctx, w.rollout) + return c.Patch(ctx, w.rollout, client.StrategicMergeFrom(w.original), client.FieldOwner(FieldManager)) } // getStrategy returns the rollout strategy from the annotation. @@ -122,7 +125,14 @@ func (w *RolloutWorkload) getStrategy() RolloutStrategy { } func (w *RolloutWorkload) DeepCopy() Workload { - return &RolloutWorkload{rollout: w.rollout.DeepCopy()} + return &RolloutWorkload{ + rollout: w.rollout.DeepCopy(), + original: w.original.DeepCopy(), + } +} + +func (w *RolloutWorkload) ResetOriginal() { + w.original = w.rollout.DeepCopy() } func (w *RolloutWorkload) GetEnvFromSources() []corev1.EnvFromSource { diff --git a/internal/pkg/workload/statefulset.go b/internal/pkg/workload/statefulset.go index 31dddeb2..ebec4a00 100644 --- a/internal/pkg/workload/statefulset.go +++ b/internal/pkg/workload/statefulset.go @@ -12,11 +12,15 @@ import ( // StatefulSetWorkload wraps a Kubernetes StatefulSet. type StatefulSetWorkload struct { statefulset *appsv1.StatefulSet + original *appsv1.StatefulSet } // NewStatefulSetWorkload creates a new StatefulSetWorkload. func NewStatefulSetWorkload(s *appsv1.StatefulSet) *StatefulSetWorkload { - return &StatefulSetWorkload{statefulset: s} + return &StatefulSetWorkload{ + statefulset: s, + original: s.DeepCopy(), + } } // Ensure StatefulSetWorkload implements WorkloadAccessor. @@ -77,11 +81,18 @@ func (w *StatefulSetWorkload) GetVolumes() []corev1.Volume { } func (w *StatefulSetWorkload) Update(ctx context.Context, c client.Client) error { - return c.Update(ctx, w.statefulset) + return c.Patch(ctx, w.statefulset, client.StrategicMergeFrom(w.original), client.FieldOwner(FieldManager)) } func (w *StatefulSetWorkload) DeepCopy() Workload { - return &StatefulSetWorkload{statefulset: w.statefulset.DeepCopy()} + return &StatefulSetWorkload{ + statefulset: w.statefulset.DeepCopy(), + original: w.original.DeepCopy(), + } +} + +func (w *StatefulSetWorkload) ResetOriginal() { + w.original = w.statefulset.DeepCopy() } func (w *StatefulSetWorkload) GetEnvFromSources() []corev1.EnvFromSource {