feat: Migration to new implementation

internal/pkg/controller/configmap_reconciler.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
+	"github.com/stakater/Reloader/internal/pkg/alerting"
 	"github.com/stakater/Reloader/internal/pkg/config"
 	"github.com/stakater/Reloader/internal/pkg/events"
 	"github.com/stakater/Reloader/internal/pkg/metrics"
@@ -33,6 +34,7 @@ type ConfigMapReconciler struct {
 	Collectors    *metrics.Collectors
 	EventRecorder *events.Recorder
 	WebhookClient *webhook.Client
+	Alerter       alerting.Alerter
 
 	initialized bool
 	initOnce    sync.Once
@@ -131,6 +133,19 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 				"workload", decision.Workload.GetName(),
 				"kind", decision.Workload.Kind(),
 			)
+
+			// Send alert notification
+			if err := r.Alerter.Send(ctx, alerting.AlertMessage{
+				WorkloadKind:      string(decision.Workload.Kind()),
+				WorkloadName:      decision.Workload.GetName(),
+				WorkloadNamespace: decision.Workload.GetNamespace(),
+				ResourceKind:      "ConfigMap",
+				ResourceName:      cm.Name,
+				ResourceNamespace: cm.Namespace,
+				Timestamp:         time.Now(),
+			}); err != nil {
+				log.Error(err, "failed to send alert")
+			}
 		}
 	}
 
@@ -202,6 +217,19 @@ func (r *ConfigMapReconciler) handleDelete(ctx context.Context, req ctrl.Request
 		if updated {
 			r.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), "ConfigMap", req.Name)
 			r.recordMetrics(true, req.Namespace)
+
+			// Send alert notification
+			if err := r.Alerter.Send(ctx, alerting.AlertMessage{
+				WorkloadKind:      string(decision.Workload.Kind()),
+				WorkloadName:      decision.Workload.GetName(),
+				WorkloadNamespace: decision.Workload.GetNamespace(),
+				ResourceKind:      "ConfigMap",
+				ResourceName:      req.Name,
+				ResourceNamespace: req.Namespace,
+				Timestamp:         time.Now(),
+			}); err != nil {
+				log.Error(err, "failed to send alert")
+			}
 		}
 	}
 

internal/pkg/controller/controller.go

@@ -1,282 +0,0 @@
-package controller
-
-import (
-	"fmt"
-	"time"
-
-	"github.com/sirupsen/logrus"
-	"github.com/stakater/Reloader/internal/pkg/handler"
-	"github.com/stakater/Reloader/internal/pkg/metrics"
-	"github.com/stakater/Reloader/internal/pkg/options"
-	"github.com/stakater/Reloader/internal/pkg/util"
-	"github.com/stakater/Reloader/pkg/kube"
-	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
-	"k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
-	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
-	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/tools/record"
-	"k8s.io/client-go/util/workqueue"
-	"k8s.io/kubectl/pkg/scheme"
-	"k8s.io/utils/strings/slices"
-)
-
-// Controller for checking events
-type Controller struct {
-	client            kubernetes.Interface
-	indexer           cache.Indexer
-	queue             workqueue.TypedRateLimitingInterface[any]
-	informer          cache.Controller
-	namespace         string
-	resource          string
-	ignoredNamespaces util.List
-	collectors        metrics.Collectors
-	recorder          record.EventRecorder
-	namespaceSelector string
-	resourceSelector  string
-}
-
-// controllerInitialized flag determines whether controlled is being initialized
-var secretControllerInitialized bool = false
-var configmapControllerInitialized bool = false
-var selectedNamespacesCache []string
-
-// NewController for initializing a Controller
-func NewController(
-	client kubernetes.Interface, resource string, namespace string, ignoredNamespaces []string, namespaceLabelSelector string, resourceLabelSelector string, collectors metrics.Collectors) (*Controller, error) {
-
-	if options.SyncAfterRestart {
-		secretControllerInitialized = true
-		configmapControllerInitialized = true
-	}
-
-	c := Controller{
-		client:            client,
-		namespace:         namespace,
-		ignoredNamespaces: ignoredNamespaces,
-		namespaceSelector: namespaceLabelSelector,
-		resourceSelector:  resourceLabelSelector,
-		resource:          resource,
-	}
-	eventBroadcaster := record.NewBroadcaster()
-	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{
-		Interface: client.CoreV1().Events(""),
-	})
-	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: fmt.Sprintf("reloader-%s", resource)})
-
-	queue := workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[any]())
-
-	optionsModifier := func(options *metav1.ListOptions) {
-		if resource == "namespaces" {
-			options.LabelSelector = c.namespaceSelector
-		} else if len(c.resourceSelector) > 0 {
-			options.LabelSelector = c.resourceSelector
-		} else {
-			options.FieldSelector = fields.Everything().String()
-		}
-	}
-
-	listWatcher := cache.NewFilteredListWatchFromClient(client.CoreV1().RESTClient(), resource, namespace, optionsModifier)
-
-	_, informer := cache.NewInformerWithOptions(cache.InformerOptions{
-		ListerWatcher: listWatcher,
-		ObjectType:    kube.ResourceMap[resource],
-		ResyncPeriod:  0,
-		Handler: cache.ResourceEventHandlerFuncs{
-			AddFunc:    c.Add,
-			UpdateFunc: c.Update,
-			DeleteFunc: c.Delete,
-		},
-		Indexers: cache.Indexers{},
-	})
-	c.informer = informer
-	c.queue = queue
-	c.collectors = collectors
-	c.recorder = recorder
-
-	logrus.Infof("created controller for: %s", resource)
-	return &c, nil
-}
-
-// Add function to add a new object to the queue in case of creating a resource
-func (c *Controller) Add(obj interface{}) {
-
-	switch object := obj.(type) {
-	case *v1.Namespace:
-		c.addSelectedNamespaceToCache(*object)
-		return
-	}
-
-	if options.ReloadOnCreate == "true" {
-		if !c.resourceInIgnoredNamespace(obj) && c.resourceInSelectedNamespaces(obj) && secretControllerInitialized && configmapControllerInitialized {
-			c.queue.Add(handler.ResourceCreatedHandler{
-				Resource:   obj,
-				Collectors: c.collectors,
-				Recorder:   c.recorder,
-			})
-		}
-	}
-}
-
-func (c *Controller) resourceInIgnoredNamespace(raw interface{}) bool {
-	switch object := raw.(type) {
-	case *v1.ConfigMap:
-		return c.ignoredNamespaces.Contains(object.Namespace)
-	case *v1.Secret:
-		return c.ignoredNamespaces.Contains(object.Namespace)
-	}
-	return false
-}
-
-func (c *Controller) resourceInSelectedNamespaces(raw interface{}) bool {
-	if len(c.namespaceSelector) == 0 {
-		return true
-	}
-
-	switch object := raw.(type) {
-	case *v1.ConfigMap:
-		if slices.Contains(selectedNamespacesCache, object.GetNamespace()) {
-			return true
-		}
-	case *v1.Secret:
-		if slices.Contains(selectedNamespacesCache, object.GetNamespace()) {
-			return true
-		}
-	}
-	return false
-}
-
-func (c *Controller) addSelectedNamespaceToCache(namespace v1.Namespace) {
-	selectedNamespacesCache = append(selectedNamespacesCache, namespace.GetName())
-	logrus.Infof("added namespace to be watched: %s", namespace.GetName())
-}
-
-func (c *Controller) removeSelectedNamespaceFromCache(namespace v1.Namespace) {
-	for i, v := range selectedNamespacesCache {
-		if v == namespace.GetName() {
-			selectedNamespacesCache = append(selectedNamespacesCache[:i], selectedNamespacesCache[i+1:]...)
-			logrus.Infof("removed namespace from watch: %s", namespace.GetName())
-			return
-		}
-	}
-}
-
-// Update function to add an old object and a new object to the queue in case of updating a resource
-func (c *Controller) Update(old interface{}, new interface{}) {
-	switch new.(type) {
-	case *v1.Namespace:
-		return
-	}
-
-	if !c.resourceInIgnoredNamespace(new) && c.resourceInSelectedNamespaces(new) {
-		c.queue.Add(handler.ResourceUpdatedHandler{
-			Resource:    new,
-			OldResource: old,
-			Collectors:  c.collectors,
-			Recorder:    c.recorder,
-		})
-	}
-}
-
-// Delete function to add an object to the queue in case of deleting a resource
-func (c *Controller) Delete(old interface{}) {
-
-	if options.ReloadOnDelete == "true" {
-		if !c.resourceInIgnoredNamespace(old) && c.resourceInSelectedNamespaces(old) && secretControllerInitialized && configmapControllerInitialized {
-			c.queue.Add(handler.ResourceDeleteHandler{
-				Resource:   old,
-				Collectors: c.collectors,
-				Recorder:   c.recorder,
-			})
-		}
-	}
-
-	switch object := old.(type) {
-	case *v1.Namespace:
-		c.removeSelectedNamespaceFromCache(*object)
-		return
-	}
-}
-
-// Run function for controller which handles the queue
-func (c *Controller) Run(threadiness int, stopCh chan struct{}) {
-	defer runtime.HandleCrash()
-
-	// Let the workers stop when we are done
-	defer c.queue.ShutDown()
-
-	go c.informer.Run(stopCh)
-
-	// Wait for all involved caches to be synced, before processing items from the queue is started
-	if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
-		runtime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
-		return
-	}
-
-	for i := 0; i < threadiness; i++ {
-		go wait.Until(c.runWorker, time.Second, stopCh)
-	}
-
-	<-stopCh
-	logrus.Infof("Stopping Controller")
-}
-
-func (c *Controller) runWorker() {
-	// At this point the controller is fully initialized and we can start processing the resources
-	if c.resource == string(v1.ResourceSecrets) {
-		secretControllerInitialized = true
-	} else if c.resource == string(v1.ResourceConfigMaps) {
-		configmapControllerInitialized = true
-	}
-
-	for c.processNextItem() {
-	}
-}
-
-func (c *Controller) processNextItem() bool {
-	// Wait until there is a new item in the working queue
-	resourceHandler, quit := c.queue.Get()
-	if quit {
-		return false
-	}
-	// Tell the queue that we are done with processing this key. This unblocks the key for other workers
-	// This allows safe parallel processing because two events with the same key are never processed in
-	// parallel.
-	defer c.queue.Done(resourceHandler)
-
-	// Invoke the method containing the business logic
-	err := resourceHandler.(handler.ResourceHandler).Handle()
-	// Handle the error if something went wrong during the execution of the business logic
-	c.handleErr(err, resourceHandler)
-	return true
-}
-
-// handleErr checks if an error happened and makes sure we will retry later.
-func (c *Controller) handleErr(err error, key interface{}) {
-	if err == nil {
-		// Forget about the #AddRateLimited history of the key on every successful synchronization.
-		// This ensures that future processing of updates for this key is not delayed because of
-		// an outdated error history.
-		c.queue.Forget(key)
-		return
-	}
-
-	// This controller retries 5 times if something goes wrong. After that, it stops trying.
-	if c.queue.NumRequeues(key) < 5 {
-		logrus.Errorf("Error syncing events: %v", err)
-
-		// Re-enqueue the key rate limited. Based on the rate limiter on the
-		// queue and the re-enqueue history, the key will be processed later again.
-		c.queue.AddRateLimited(key)
-		return
-	}
-
-	c.queue.Forget(key)
-	// Report to an external entity that, even after several retries, we could not successfully process this key
-	runtime.HandleError(err)
-	logrus.Errorf("Dropping key out of the queue: %v", err)
-	logrus.Debugf("Dropping the key %q out of the queue: %v", key, err)
-}

internal/pkg/controller/manager.go

@@ -3,10 +3,10 @@ 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/alerting"
 	"github.com/stakater/Reloader/internal/pkg/config"
 	"github.com/stakater/Reloader/internal/pkg/events"
 	"github.com/stakater/Reloader/internal/pkg/metrics"
@@ -36,12 +36,10 @@ type ManagerOptions struct {
 }
 
 // NewManager creates a new controller-runtime manager with the given options.
+// This follows controller-runtime and operator-sdk conventions for leader election.
 func NewManager(opts ManagerOptions) (ctrl.Manager, error) {
 	cfg := opts.Config
-
-	leaseDuration := 15 * time.Second
-	renewDeadline := 10 * time.Second
-	retryPeriod := 2 * time.Second
+	le := cfg.LeaderElection
 
 	mgrOpts := ctrl.Options{
 		Scheme: runtimeScheme,
@@ -49,11 +47,19 @@ func NewManager(opts ManagerOptions) (ctrl.Manager, error) {
 			BindAddress: cfg.MetricsAddr,
 		},
 		HealthProbeBindAddress: cfg.HealthAddr,
-		LeaderElection:         cfg.EnableHA,
-		LeaderElectionID:       "reloader-leader-election",
-		LeaseDuration:          &leaseDuration,
-		RenewDeadline:          &renewDeadline,
-		RetryPeriod:            &retryPeriod,
+
+		// Leader election configuration following operator-sdk best practices:
+		// - LeaderElection enables/disables leader election
+		// - LeaderElectionID is the name of the lease resource
+		// - LeaderElectionNamespace where the lease is created (defaults to pod namespace)
+		// - LeaderElectionReleaseOnCancel allows faster failover by releasing the lock on shutdown
+		LeaderElection:                cfg.EnableHA,
+		LeaderElectionID:              le.LockName,
+		LeaderElectionNamespace:       le.Namespace,
+		LeaderElectionReleaseOnCancel: le.ReleaseOnCancel,
+		LeaseDuration:                 &le.LeaseDuration,
+		RenewDeadline:                 &le.RenewDeadline,
+		RetryPeriod:                   &le.RetryPeriod,
 	}
 
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), mgrOpts)
@@ -61,6 +67,10 @@ func NewManager(opts ManagerOptions) (ctrl.Manager, error) {
 		return nil, fmt.Errorf("creating manager: %w", err)
 	}
 
+	// Add health and readiness probes.
+	// The healthz probe reports whether the manager is running.
+	// The readyz probe reports whether the manager is ready to serve requests.
+	// When leader election is enabled, readyz will fail until this instance becomes leader.
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
 		return nil, fmt.Errorf("setting up health check: %w", err)
 	}
@@ -76,6 +86,13 @@ func SetupReconcilers(mgr ctrl.Manager, cfg *config.Config, log logr.Logger, col
 	registry := workload.NewRegistry(cfg.ArgoRolloutsEnabled)
 	reloadService := reload.NewService(cfg)
 	eventRecorder := events.NewRecorder(mgr.GetEventRecorderFor("reloader"))
+	pauseHandler := reload.NewPauseHandler(cfg)
+
+	// Create alerter based on configuration
+	alerter := alerting.NewAlerter(cfg)
+	if cfg.Alerting.Enabled {
+		log.Info("alerting enabled", "sink", cfg.Alerting.Sink)
+	}
 
 	// Create webhook client if URL is configured
 	var webhookClient *webhook.Client
@@ -84,6 +101,7 @@ func SetupReconcilers(mgr ctrl.Manager, cfg *config.Config, log logr.Logger, col
 		log.Info("webhook mode enabled", "url", cfg.WebhookURL)
 	}
 
+	// Setup ConfigMap reconciler
 	if !cfg.IsResourceIgnored("configmaps") {
 		if err := (&ConfigMapReconciler{
 			Client:        mgr.GetClient(),
@@ -94,11 +112,13 @@ func SetupReconcilers(mgr ctrl.Manager, cfg *config.Config, log logr.Logger, col
 			Collectors:    collectors,
 			EventRecorder: eventRecorder,
 			WebhookClient: webhookClient,
+			Alerter:       alerter,
 		}).SetupWithManager(mgr); err != nil {
 			return fmt.Errorf("setting up configmap reconciler: %w", err)
 		}
 	}
 
+	// Setup Secret reconciler
 	if !cfg.IsResourceIgnored("secrets") {
 		if err := (&SecretReconciler{
 			Client:        mgr.GetClient(),
@@ -109,11 +129,36 @@ func SetupReconcilers(mgr ctrl.Manager, cfg *config.Config, log logr.Logger, col
 			Collectors:    collectors,
 			EventRecorder: eventRecorder,
 			WebhookClient: webhookClient,
+			Alerter:       alerter,
 		}).SetupWithManager(mgr); err != nil {
 			return fmt.Errorf("setting up secret reconciler: %w", err)
 		}
 	}
 
+	// Setup Namespace reconciler if namespace selectors are configured
+	if len(cfg.NamespaceSelectors) > 0 {
+		nsCache := NewNamespaceCache(true)
+		if err := (&NamespaceReconciler{
+			Client: mgr.GetClient(),
+			Log:    log.WithName("namespace-reconciler"),
+			Config: cfg,
+			Cache:  nsCache,
+		}).SetupWithManager(mgr); err != nil {
+			return fmt.Errorf("setting up namespace reconciler: %w", err)
+		}
+		log.Info("namespace reconciler enabled for label selector filtering")
+	}
+
+	// Setup Deployment reconciler for pause handling
+	if err := (&DeploymentReconciler{
+		Client:       mgr.GetClient(),
+		Log:          log.WithName("deployment-reconciler"),
+		Config:       cfg,
+		PauseHandler: pauseHandler,
+	}).SetupWithManager(mgr); err != nil {
+		return fmt.Errorf("setting up deployment reconciler: %w", err)
+	}
+
 	return nil
 }
 

internal/pkg/controller/secret_reconciler.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
+	"github.com/stakater/Reloader/internal/pkg/alerting"
 	"github.com/stakater/Reloader/internal/pkg/config"
 	"github.com/stakater/Reloader/internal/pkg/events"
 	"github.com/stakater/Reloader/internal/pkg/metrics"
@@ -33,6 +34,7 @@ type SecretReconciler struct {
 	Collectors    *metrics.Collectors
 	EventRecorder *events.Recorder
 	WebhookClient *webhook.Client
+	Alerter       alerting.Alerter
 
 	initialized bool
 	initOnce    sync.Once
@@ -131,6 +133,19 @@ func (r *SecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 				"workload", decision.Workload.GetName(),
 				"kind", decision.Workload.Kind(),
 			)
+
+			// Send alert notification
+			if err := r.Alerter.Send(ctx, alerting.AlertMessage{
+				WorkloadKind:      string(decision.Workload.Kind()),
+				WorkloadName:      decision.Workload.GetName(),
+				WorkloadNamespace: decision.Workload.GetNamespace(),
+				ResourceKind:      "Secret",
+				ResourceName:      secret.Name,
+				ResourceNamespace: secret.Namespace,
+				Timestamp:         time.Now(),
+			}); err != nil {
+				log.Error(err, "failed to send alert")
+			}
 		}
 	}
 
@@ -199,6 +214,19 @@ func (r *SecretReconciler) handleDelete(ctx context.Context, req ctrl.Request, l
 		if updated {
 			r.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), "Secret", req.Name)
 			r.recordMetrics(true, req.Namespace)
+
+			// Send alert notification
+			if err := r.Alerter.Send(ctx, alerting.AlertMessage{
+				WorkloadKind:      string(decision.Workload.Kind()),
+				WorkloadName:      decision.Workload.GetName(),
+				WorkloadNamespace: decision.Workload.GetNamespace(),
+				ResourceKind:      "Secret",
+				ResourceName:      req.Name,
+				ResourceNamespace: req.Namespace,
+				Timestamp:         time.Now(),
+			}); err != nil {
+				log.Error(err, "failed to send alert")
+			}
 		}
 	}