diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 8861fb5b..6288d5f9 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -82,6 +82,9 @@ type Config struct { // Metrics configuration MetricsAddr string // Address to serve metrics on (default :9090) + // Health probe configuration + HealthAddr string // Address to serve health probes on (default :8081) + // Profiling configuration EnablePProf bool PProfAddr string @@ -170,6 +173,7 @@ func NewDefault() *Config { LogFormat: "", LogLevel: "info", MetricsAddr: ":9090", + HealthAddr: ":8081", EnablePProf: false, PProfAddr: ":6060", Alerting: AlertingConfig{}, diff --git a/internal/pkg/config/flags.go b/internal/pkg/config/flags.go index bf242bca..ef423082 100644 --- a/internal/pkg/config/flags.go +++ b/internal/pkg/config/flags.go @@ -78,6 +78,10 @@ func BindFlags(fs *pflag.FlagSet, cfg *Config) { fs.StringVar(&cfg.MetricsAddr, "metrics-addr", cfg.MetricsAddr, "Address to serve metrics on") + // Health probes + fs.StringVar(&cfg.HealthAddr, "health-addr", cfg.HealthAddr, + "Address to serve health probes on") + // Profiling fs.BoolVar(&cfg.EnablePProf, "enable-pprof", cfg.EnablePProf, "Enable pprof profiling server") diff --git a/internal/pkg/controller/configmap_reconciler.go b/internal/pkg/controller/configmap_reconciler.go index 2dcd3338..cb78a7ba 100644 --- a/internal/pkg/controller/configmap_reconciler.go +++ b/internal/pkg/controller/configmap_reconciler.go @@ -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. diff --git a/internal/pkg/controller/manager.go b/internal/pkg/controller/manager.go new file mode 100644 index 00000000..d4914cdd --- /dev/null +++ b/internal/pkg/controller/manager.go @@ -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) +} diff --git a/internal/pkg/controller/retry.go b/internal/pkg/controller/retry.go new file mode 100644 index 00000000..f81e8e92 --- /dev/null +++ b/internal/pkg/controller/retry.go @@ -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 +} diff --git a/internal/pkg/controller/secret_reconciler.go b/internal/pkg/controller/secret_reconciler.go index 54e4aa8e..0d68ae13 100644 --- a/internal/pkg/controller/secret_reconciler.go +++ b/internal/pkg/controller/secret_reconciler.go @@ -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. diff --git a/internal/pkg/events/recorder.go b/internal/pkg/events/recorder.go new file mode 100644 index 00000000..1f3eef58 --- /dev/null +++ b/internal/pkg/events/recorder.go @@ -0,0 +1,60 @@ +package events + +import ( + "fmt" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/tools/record" +) + +const ( + // EventTypeNormal represents a normal event. + EventTypeNormal = corev1.EventTypeNormal + // EventTypeWarning represents a warning event. + EventTypeWarning = corev1.EventTypeWarning + + // ReasonReloaded indicates a workload was successfully reloaded. + ReasonReloaded = "Reloaded" + // ReasonReloadFailed indicates a workload reload failed. + ReasonReloadFailed = "ReloadFailed" +) + +// Recorder wraps the Kubernetes event recorder. +type Recorder struct { + recorder record.EventRecorder +} + +// NewRecorder creates a new event Recorder. +func NewRecorder(recorder record.EventRecorder) *Recorder { + if recorder == nil { + return nil + } + return &Recorder{recorder: recorder} +} + +// ReloadSuccess records a successful reload event. +func (r *Recorder) ReloadSuccess(object runtime.Object, resourceType, resourceName string) { + if r == nil || r.recorder == nil { + return + } + r.recorder.Event( + object, + EventTypeNormal, + ReasonReloaded, + fmt.Sprintf("Reloaded due to %s %s change", resourceType, resourceName), + ) +} + +// ReloadFailed records a failed reload event. +func (r *Recorder) ReloadFailed(object runtime.Object, resourceType, resourceName string, err error) { + if r == nil || r.recorder == nil { + return + } + r.recorder.Event( + object, + EventTypeWarning, + ReasonReloadFailed, + fmt.Sprintf("Failed to reload due to %s %s change: %v", resourceType, resourceName, err), + ) +} diff --git a/internal/pkg/metrics/prometheus.go b/internal/pkg/metrics/prometheus.go index 94153eac..f78ef03d 100644 --- a/internal/pkg/metrics/prometheus.go +++ b/internal/pkg/metrics/prometheus.go @@ -8,9 +8,32 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" ) +// Collectors holds Prometheus metrics collectors for Reloader. type Collectors struct { Reloaded *prometheus.CounterVec ReloadedByNamespace *prometheus.CounterVec + countByNamespace bool +} + +// RecordReload records a reload event with the given success status and namespace. +func (c *Collectors) RecordReload(success bool, namespace string) { + if c == nil { + return + } + + successLabel := "false" + if success { + successLabel = "true" + } + + c.Reloaded.With(prometheus.Labels{"success": successLabel}).Inc() + + if c.countByNamespace { + c.ReloadedByNamespace.With(prometheus.Labels{ + "success": successLabel, + "namespace": namespace, + }).Inc() + } } func NewCollectors() Collectors { @@ -43,6 +66,7 @@ func NewCollectors() Collectors { return Collectors{ Reloaded: reloaded, ReloadedByNamespace: reloaded_by_namespace, + countByNamespace: os.Getenv("METRICS_COUNT_BY_NAMESPACE") == "enabled", } } diff --git a/internal/pkg/webhook/webhook.go b/internal/pkg/webhook/webhook.go new file mode 100644 index 00000000..a40c5a5c --- /dev/null +++ b/internal/pkg/webhook/webhook.go @@ -0,0 +1,95 @@ +// Package webhook handles sending reload notifications to external endpoints. +// When --webhook-url is set, Reloader sends HTTP POST requests instead of modifying workloads. +package webhook + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "time" + + "github.com/go-logr/logr" +) + +// Payload represents the data sent to the webhook endpoint. +type Payload struct { + Kind string `json:"kind"` + Namespace string `json:"namespace"` + ResourceName string `json:"resourceName"` + ResourceType string `json:"resourceType"` + Hash string `json:"hash"` + Timestamp time.Time `json:"timestamp"` + + // Workloads contains the list of workloads that would be reloaded. + Workloads []WorkloadInfo `json:"workloads"` +} + +// WorkloadInfo describes a workload that would be reloaded. +type WorkloadInfo struct { + Kind string `json:"kind"` + Name string `json:"name"` + Namespace string `json:"namespace"` +} + +// Client sends reload notifications to webhook endpoints. +type Client struct { + httpClient *http.Client + url string + log logr.Logger +} + +// NewClient creates a new webhook client. +func NewClient(url string, log logr.Logger) *Client { + return &Client{ + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + url: url, + log: log, + } +} + +// Send posts the payload to the configured webhook URL. +func (c *Client) Send(ctx context.Context, payload Payload) error { + if c.url == "" { + return nil + } + + data, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshaling payload: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url, bytes.NewReader(data)) + if err != nil { + return fmt.Errorf("creating request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "Reloader/2.0") + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("sending request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("webhook returned status %d", resp.StatusCode) + } + + c.log.V(1).Info("webhook notification sent", + "url", c.url, + "resourceType", payload.ResourceType, + "resourceName", payload.ResourceName, + "workloadCount", len(payload.Workloads), + ) + + return nil +} + +// IsConfigured returns true if the webhook URL is set. +func (c *Client) IsConfigured() bool { + return c != nil && c.url != "" +}