From 3defc8bb297d7fa7255e56b22868c6cdb435f836 Mon Sep 17 00:00:00 2001 From: TheiLLeniumStudios <104288623+TheiLLeniumStudios@users.noreply.github.com> Date: Sun, 28 Dec 2025 08:47:55 +0100 Subject: [PATCH] refactor: Move all common reconcile logic to lister and reload handler --- internal/pkg/alerting/alerter_test.go | 392 ++++++++---------- internal/pkg/config/config_test.go | 86 ++-- internal/pkg/config/validation.go | 52 +-- internal/pkg/config/validation_test.go | 256 ++++++------ .../pkg/controller/configmap_reconciler.go | 374 ++--------------- internal/pkg/controller/handler.go | 187 +++++++++ internal/pkg/controller/secret_reconciler.go | 374 ++--------------- internal/pkg/reload/service.go | 11 + internal/pkg/workload/lister.go | 111 +++++ 9 files changed, 752 insertions(+), 1091 deletions(-) create mode 100644 internal/pkg/controller/handler.go create mode 100644 internal/pkg/workload/lister.go diff --git a/internal/pkg/alerting/alerter_test.go b/internal/pkg/alerting/alerter_test.go index 6e5724f4..74cc95e6 100644 --- a/internal/pkg/alerting/alerter_test.go +++ b/internal/pkg/alerting/alerter_test.go @@ -3,6 +3,7 @@ package alerting import ( "context" "encoding/json" + "io" "net/http" "net/http/httptest" "testing" @@ -11,97 +12,11 @@ import ( "github.com/stakater/Reloader/internal/pkg/config" ) -func TestNewAlerter_Disabled(t *testing.T) { - cfg := config.NewDefault() - cfg.Alerting.Enabled = false - - alerter := NewAlerter(cfg) - if _, ok := alerter.(*NoOpAlerter); !ok { - t.Error("Expected NoOpAlerter when alerting is disabled") - } -} - -func TestNewAlerter_NoWebhookURL(t *testing.T) { - cfg := config.NewDefault() - cfg.Alerting.Enabled = true - cfg.Alerting.WebhookURL = "" - - alerter := NewAlerter(cfg) - if _, ok := alerter.(*NoOpAlerter); !ok { - t.Error("Expected NoOpAlerter when webhook URL is empty") - } -} - -func TestNewAlerter_Slack(t *testing.T) { - cfg := config.NewDefault() - cfg.Alerting.Enabled = true - cfg.Alerting.WebhookURL = "http://example.com/webhook" - cfg.Alerting.Sink = "slack" - - alerter := NewAlerter(cfg) - if _, ok := alerter.(*SlackAlerter); !ok { - t.Error("Expected SlackAlerter for sink=slack") - } -} - -func TestNewAlerter_Teams(t *testing.T) { - cfg := config.NewDefault() - cfg.Alerting.Enabled = true - cfg.Alerting.WebhookURL = "http://example.com/webhook" - cfg.Alerting.Sink = "teams" - - alerter := NewAlerter(cfg) - if _, ok := alerter.(*TeamsAlerter); !ok { - t.Error("Expected TeamsAlerter for sink=teams") - } -} - -func TestNewAlerter_GChat(t *testing.T) { - cfg := config.NewDefault() - cfg.Alerting.Enabled = true - cfg.Alerting.WebhookURL = "http://example.com/webhook" - cfg.Alerting.Sink = "gchat" - - alerter := NewAlerter(cfg) - if _, ok := alerter.(*GChatAlerter); !ok { - t.Error("Expected GChatAlerter for sink=gchat") - } -} - -func TestNewAlerter_Raw(t *testing.T) { - cfg := config.NewDefault() - cfg.Alerting.Enabled = true - cfg.Alerting.WebhookURL = "http://example.com/webhook" - cfg.Alerting.Sink = "raw" - - alerter := NewAlerter(cfg) - if _, ok := alerter.(*RawAlerter); !ok { - t.Error("Expected RawAlerter for sink=raw") - } -} - -func TestNewAlerter_DefaultIsRaw(t *testing.T) { - cfg := config.NewDefault() - cfg.Alerting.Enabled = true - cfg.Alerting.WebhookURL = "http://example.com/webhook" - cfg.Alerting.Sink = "" // Empty sink should default to raw - - alerter := NewAlerter(cfg) - if _, ok := alerter.(*RawAlerter); !ok { - t.Error("Expected RawAlerter for empty sink") - } -} - -func TestNoOpAlerter_Send(t *testing.T) { - alerter := &NoOpAlerter{} - err := alerter.Send(context.Background(), AlertMessage{}) - if err != nil { - t.Errorf("NoOpAlerter.Send() error = %v, want nil", err) - } -} - -func TestSlackAlerter_Send(t *testing.T) { - var receivedBody []byte +// testServer creates a test HTTP server that captures the request body. +// Returns the server and a function to retrieve the captured body. +func testServer(t *testing.T) (*httptest.Server, func() []byte) { + t.Helper() + var body []byte server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { t.Errorf("Expected POST request, got %s", r.Method) @@ -109,14 +24,15 @@ func TestSlackAlerter_Send(t *testing.T) { if r.Header.Get("Content-Type") != "application/json" { t.Errorf("Expected Content-Type application/json, got %s", r.Header.Get("Content-Type")) } - receivedBody = make([]byte, r.ContentLength) - r.Body.Read(receivedBody) + body, _ = io.ReadAll(r.Body) w.WriteHeader(http.StatusOK) })) - defer server.Close() + return server, func() []byte { return body } +} - alerter := NewSlackAlerter(server.URL, "", "Test Cluster") - msg := AlertMessage{ +// testAlertMessage returns a standard AlertMessage for testing. +func testAlertMessage() AlertMessage { + return AlertMessage{ WorkloadKind: "Deployment", WorkloadName: "nginx", WorkloadNamespace: "default", @@ -125,143 +41,201 @@ func TestSlackAlerter_Send(t *testing.T) { ResourceNamespace: "default", Timestamp: time.Now(), } - - err := alerter.Send(context.Background(), msg) - if err != nil { - t.Fatalf("SlackAlerter.Send() error = %v", err) - } - - var slackMsg slackMessage - if err := json.Unmarshal(receivedBody, &slackMsg); err != nil { - t.Fatalf("Failed to unmarshal slack message: %v", err) - } - - if slackMsg.Text == "" { - t.Error("Expected non-empty text in slack message") - } } -func TestTeamsAlerter_Send(t *testing.T) { - var receivedBody []byte - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - receivedBody = make([]byte, r.ContentLength) - r.Body.Read(receivedBody) - w.WriteHeader(http.StatusOK) - })) - defer server.Close() - - alerter := NewTeamsAlerter(server.URL, "", "") - msg := AlertMessage{ - WorkloadKind: "Deployment", - WorkloadName: "nginx", - WorkloadNamespace: "default", - ResourceKind: "ConfigMap", - ResourceName: "nginx-config", - ResourceNamespace: "default", - Timestamp: time.Now(), +func TestNewAlerter(t *testing.T) { + tests := []struct { + name string + setup func(*config.Config) + wantType string + }{ + { + name: "disabled", + setup: func(cfg *config.Config) { + cfg.Alerting.Enabled = false + }, + wantType: "*alerting.NoOpAlerter", + }, + { + name: "no webhook URL", + setup: func(cfg *config.Config) { + cfg.Alerting.Enabled = true + cfg.Alerting.WebhookURL = "" + }, + wantType: "*alerting.NoOpAlerter", + }, + { + name: "slack", + setup: func(cfg *config.Config) { + cfg.Alerting.Enabled = true + cfg.Alerting.WebhookURL = "http://example.com/webhook" + cfg.Alerting.Sink = "slack" + }, + wantType: "*alerting.SlackAlerter", + }, + { + name: "teams", + setup: func(cfg *config.Config) { + cfg.Alerting.Enabled = true + cfg.Alerting.WebhookURL = "http://example.com/webhook" + cfg.Alerting.Sink = "teams" + }, + wantType: "*alerting.TeamsAlerter", + }, + { + name: "gchat", + setup: func(cfg *config.Config) { + cfg.Alerting.Enabled = true + cfg.Alerting.WebhookURL = "http://example.com/webhook" + cfg.Alerting.Sink = "gchat" + }, + wantType: "*alerting.GChatAlerter", + }, + { + name: "raw", + setup: func(cfg *config.Config) { + cfg.Alerting.Enabled = true + cfg.Alerting.WebhookURL = "http://example.com/webhook" + cfg.Alerting.Sink = "raw" + }, + wantType: "*alerting.RawAlerter", + }, + { + name: "empty sink defaults to raw", + setup: func(cfg *config.Config) { + cfg.Alerting.Enabled = true + cfg.Alerting.WebhookURL = "http://example.com/webhook" + cfg.Alerting.Sink = "" + }, + wantType: "*alerting.RawAlerter", + }, } - err := alerter.Send(context.Background(), msg) - if err != nil { - t.Fatalf("TeamsAlerter.Send() error = %v", err) - } - - var teamsMsg teamsMessage - if err := json.Unmarshal(receivedBody, &teamsMsg); err != nil { - t.Fatalf("Failed to unmarshal teams message: %v", err) - } - - if teamsMsg.Type != "MessageCard" { - t.Errorf("Expected @type=MessageCard, got %s", teamsMsg.Type) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := config.NewDefault() + tt.setup(cfg) + alerter := NewAlerter(cfg) + gotType := getTypeName(alerter) + if gotType != tt.wantType { + t.Errorf("NewAlerter() type = %s, want %s", gotType, tt.wantType) + } + }) } } -func TestGChatAlerter_Send(t *testing.T) { - var receivedBody []byte - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - receivedBody = make([]byte, r.ContentLength) - r.Body.Read(receivedBody) - w.WriteHeader(http.StatusOK) - })) - defer server.Close() - - alerter := NewGChatAlerter(server.URL, "", "") - msg := AlertMessage{ - WorkloadKind: "Deployment", - WorkloadName: "nginx", - WorkloadNamespace: "default", - ResourceKind: "ConfigMap", - ResourceName: "nginx-config", - ResourceNamespace: "default", - Timestamp: time.Now(), - } - - err := alerter.Send(context.Background(), msg) - if err != nil { - t.Fatalf("GChatAlerter.Send() error = %v", err) - } - - var gchatMsg gchatMessage - if err := json.Unmarshal(receivedBody, &gchatMsg); err != nil { - t.Fatalf("Failed to unmarshal gchat message: %v", err) - } - - if len(gchatMsg.Cards) != 1 { - t.Errorf("Expected 1 card, got %d", len(gchatMsg.Cards)) +func getTypeName(a Alerter) string { + switch a.(type) { + case *NoOpAlerter: + return "*alerting.NoOpAlerter" + case *SlackAlerter: + return "*alerting.SlackAlerter" + case *TeamsAlerter: + return "*alerting.TeamsAlerter" + case *GChatAlerter: + return "*alerting.GChatAlerter" + case *RawAlerter: + return "*alerting.RawAlerter" + default: + return "unknown" } } -func TestRawAlerter_Send(t *testing.T) { - var receivedBody []byte - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - receivedBody = make([]byte, r.ContentLength) - r.Body.Read(receivedBody) - w.WriteHeader(http.StatusOK) - })) - defer server.Close() +func TestNoOpAlerter_Send(t *testing.T) { + alerter := &NoOpAlerter{} + if err := alerter.Send(context.Background(), AlertMessage{}); err != nil { + t.Errorf("NoOpAlerter.Send() error = %v, want nil", err) + } +} - alerter := NewRawAlerter(server.URL, "", "custom-info") - msg := AlertMessage{ - WorkloadKind: "Deployment", - WorkloadName: "nginx", - WorkloadNamespace: "default", - ResourceKind: "ConfigMap", - ResourceName: "nginx-config", - ResourceNamespace: "default", - Timestamp: time.Now(), +func TestAlerter_Send(t *testing.T) { + tests := []struct { + name string + newAlert func(url string) Alerter + validate func(t *testing.T, body []byte) + }{ + { + name: "slack", + newAlert: func(url string) Alerter { return NewSlackAlerter(url, "", "Test Cluster") }, + validate: func(t *testing.T, body []byte) { + var msg slackMessage + if err := json.Unmarshal(body, &msg); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + if msg.Text == "" { + t.Error("Expected non-empty text") + } + }, + }, + { + name: "teams", + newAlert: func(url string) Alerter { return NewTeamsAlerter(url, "", "") }, + validate: func(t *testing.T, body []byte) { + var msg teamsMessage + if err := json.Unmarshal(body, &msg); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + if msg.Type != "MessageCard" { + t.Errorf("@type = %s, want MessageCard", msg.Type) + } + }, + }, + { + name: "gchat", + newAlert: func(url string) Alerter { return NewGChatAlerter(url, "", "") }, + validate: func(t *testing.T, body []byte) { + var msg gchatMessage + if err := json.Unmarshal(body, &msg); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + if len(msg.Cards) != 1 { + t.Errorf("cards = %d, want 1", len(msg.Cards)) + } + }, + }, + { + name: "raw", + newAlert: func(url string) Alerter { return NewRawAlerter(url, "", "custom-info") }, + validate: func(t *testing.T, body []byte) { + var msg rawMessage + if err := json.Unmarshal(body, &msg); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + if msg.Event != "reload" { + t.Errorf("event = %s, want reload", msg.Event) + } + if msg.WorkloadName != "nginx" { + t.Errorf("workloadName = %s, want nginx", msg.WorkloadName) + } + if msg.Additional != "custom-info" { + t.Errorf("additional = %s, want custom-info", msg.Additional) + } + }, + }, } - err := alerter.Send(context.Background(), msg) - if err != nil { - t.Fatalf("RawAlerter.Send() error = %v", err) - } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server, getBody := testServer(t) + defer server.Close() - var rawMsg rawMessage - if err := json.Unmarshal(receivedBody, &rawMsg); err != nil { - t.Fatalf("Failed to unmarshal raw message: %v", err) - } - - if rawMsg.Event != "reload" { - t.Errorf("Expected event=reload, got %s", rawMsg.Event) - } - if rawMsg.WorkloadName != "nginx" { - t.Errorf("Expected workloadName=nginx, got %s", rawMsg.WorkloadName) - } - if rawMsg.Additional != "custom-info" { - t.Errorf("Expected additional=custom-info, got %s", rawMsg.Additional) + alerter := tt.newAlert(server.URL) + if err := alerter.Send(context.Background(), testAlertMessage()); err != nil { + t.Fatalf("Send() error = %v", err) + } + tt.validate(t, getBody()) + }) } } func TestAlerter_WebhookError(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) - w.Write([]byte("Internal Server Error")) })) defer server.Close() alerter := NewRawAlerter(server.URL, "", "") - err := alerter.Send(context.Background(), AlertMessage{}) - if err == nil { + if err := alerter.Send(context.Background(), AlertMessage{}); err == nil { t.Error("Expected error for non-2xx response") } } diff --git a/internal/pkg/config/config_test.go b/internal/pkg/config/config_test.go index d9f740a7..38b7ab53 100644 --- a/internal/pkg/config/config_test.go +++ b/internal/pkg/config/config_test.go @@ -84,11 +84,13 @@ func TestDefaultAnnotations(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if tt.got != tt.want { - t.Errorf("%s = %q, want %q", tt.name, tt.got, tt.want) - } - }) + t.Run( + tt.name, func(t *testing.T) { + if tt.got != tt.want { + t.Errorf("%s = %q, want %q", tt.name, tt.got, tt.want) + } + }, + ) } } @@ -134,12 +136,14 @@ func TestConfig_IsResourceIgnored(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := cfg.IsResourceIgnored(tt.resource) - if got != tt.want { - t.Errorf("IsResourceIgnored(%q) = %v, want %v", tt.resource, got, tt.want) - } - }) + t.Run( + tt.name, func(t *testing.T) { + got := cfg.IsResourceIgnored(tt.resource) + if got != tt.want { + t.Errorf("IsResourceIgnored(%q) = %v, want %v", tt.resource, got, tt.want) + } + }, + ) } } @@ -159,12 +163,14 @@ func TestConfig_IsWorkloadIgnored(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := cfg.IsWorkloadIgnored(tt.workload) - if got != tt.want { - t.Errorf("IsWorkloadIgnored(%q) = %v, want %v", tt.workload, got, tt.want) - } - }) + t.Run( + tt.name, func(t *testing.T) { + got := cfg.IsWorkloadIgnored(tt.workload) + if got != tt.want { + t.Errorf("IsWorkloadIgnored(%q) = %v, want %v", tt.workload, got, tt.want) + } + }, + ) } } @@ -184,12 +190,14 @@ func TestConfig_IsNamespaceIgnored(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := cfg.IsNamespaceIgnored(tt.namespace) - if got != tt.want { - t.Errorf("IsNamespaceIgnored(%q) = %v, want %v", tt.namespace, got, tt.want) - } - }) + t.Run( + tt.name, func(t *testing.T) { + got := cfg.IsNamespaceIgnored(tt.namespace) + if got != tt.want { + t.Errorf("IsNamespaceIgnored(%q) = %v, want %v", tt.namespace, got, tt.want) + } + }, + ) } } @@ -209,29 +217,13 @@ func TestEqualFold(t *testing.T) { } for _, tt := range tests { - t.Run(tt.s+"_"+tt.t, func(t *testing.T) { - got := equalFold(tt.s, tt.t) - if got != tt.want { - t.Errorf("equalFold(%q, %q) = %v, want %v", tt.s, tt.t, got, tt.want) - } - }) - } -} - -func TestReloadStrategy_String(t *testing.T) { - if string(ReloadStrategyEnvVars) != "env-vars" { - t.Errorf("ReloadStrategyEnvVars = %q, want %q", ReloadStrategyEnvVars, "env-vars") - } - if string(ReloadStrategyAnnotations) != "annotations" { - t.Errorf("ReloadStrategyAnnotations = %q, want %q", ReloadStrategyAnnotations, "annotations") - } -} - -func TestArgoRolloutStrategy_String(t *testing.T) { - if string(ArgoRolloutStrategyRestart) != "restart" { - t.Errorf("ArgoRolloutStrategyRestart = %q, want %q", ArgoRolloutStrategyRestart, "restart") - } - if string(ArgoRolloutStrategyRollout) != "rollout" { - t.Errorf("ArgoRolloutStrategyRollout = %q, want %q", ArgoRolloutStrategyRollout, "rollout") + t.Run( + tt.s+"_"+tt.t, func(t *testing.T) { + got := equalFold(tt.s, tt.t) + if got != tt.want { + t.Errorf("equalFold(%q, %q) = %v, want %v", tt.s, tt.t, got, tt.want) + } + }, + ) } } diff --git a/internal/pkg/config/validation.go b/internal/pkg/config/validation.go index 8a3bbfe5..0ebce400 100644 --- a/internal/pkg/config/validation.go +++ b/internal/pkg/config/validation.go @@ -48,10 +48,12 @@ func (c *Config) Validate() error { case "": c.ReloadStrategy = ReloadStrategyEnvVars default: - errs = append(errs, ValidationError{ - Field: "ReloadStrategy", - Message: fmt.Sprintf("invalid value %q, must be %q or %q", c.ReloadStrategy, ReloadStrategyEnvVars, ReloadStrategyAnnotations), - }) + errs = append( + errs, ValidationError{ + Field: "ReloadStrategy", + Message: fmt.Sprintf("invalid value %q, must be %q or %q", c.ReloadStrategy, ReloadStrategyEnvVars, ReloadStrategyAnnotations), + }, + ) } // Validate ArgoRolloutStrategy @@ -61,10 +63,14 @@ func (c *Config) Validate() error { case "": c.ArgoRolloutStrategy = ArgoRolloutStrategyRollout default: - errs = append(errs, ValidationError{ - Field: "ArgoRolloutStrategy", - Message: fmt.Sprintf("invalid value %q, must be %q or %q", c.ArgoRolloutStrategy, ArgoRolloutStrategyRestart, ArgoRolloutStrategyRollout), - }) + errs = append( + errs, ValidationError{ + Field: "ArgoRolloutStrategy", + Message: fmt.Sprintf( + "invalid value %q, must be %q or %q", c.ArgoRolloutStrategy, ArgoRolloutStrategyRestart, ArgoRolloutStrategyRollout, + ), + }, + ) } // Validate LogLevel @@ -72,10 +78,12 @@ func (c *Config) Validate() error { case "trace", "debug", "info", "warn", "warning", "error", "fatal", "panic", "": // valid default: - errs = append(errs, ValidationError{ - Field: "LogLevel", - Message: fmt.Sprintf("invalid log level %q", c.LogLevel), - }) + errs = append( + errs, ValidationError{ + Field: "LogLevel", + Message: fmt.Sprintf("invalid log level %q", c.LogLevel), + }, + ) } // Validate LogFormat @@ -83,10 +91,12 @@ func (c *Config) Validate() error { case "json", "": // valid default: - errs = append(errs, ValidationError{ - Field: "LogFormat", - Message: fmt.Sprintf("invalid log format %q, must be \"json\" or empty", c.LogFormat), - }) + errs = append( + errs, ValidationError{ + Field: "LogFormat", + Message: fmt.Sprintf("invalid log format %q, must be \"json\" or empty", c.LogFormat), + }, + ) } // Normalize IgnoredResources to lowercase for consistent comparison @@ -136,13 +146,3 @@ func ParseSelectors(selectorStrings []string) ([]labels.Selector, error) { } return selectors, nil } - -// MustParseSelectors parses selectors and panics on error. -// Use only when selectors are known to be valid (e.g., from validated config). -func MustParseSelectors(selectorStrings []string) []labels.Selector { - selectors, err := ParseSelectors(selectorStrings) - if err != nil { - panic(err) - } - return selectors -} diff --git a/internal/pkg/config/validation_test.go b/internal/pkg/config/validation_test.go index 2972333c..54e0a47d 100644 --- a/internal/pkg/config/validation_test.go +++ b/internal/pkg/config/validation_test.go @@ -19,28 +19,30 @@ func TestConfig_Validate_ReloadStrategy(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - cfg := NewDefault() - cfg.ReloadStrategy = tt.strategy + t.Run( + tt.name, func(t *testing.T) { + cfg := NewDefault() + cfg.ReloadStrategy = tt.strategy - err := cfg.Validate() + err := cfg.Validate() - if tt.wantErr { - if err == nil { - t.Error("Validate() should return error for invalid strategy") + if tt.wantErr { + if err == nil { + t.Error("Validate() should return error for invalid strategy") + } + return } - return - } - if err != nil { - t.Errorf("Validate() error = %v", err) - return - } + if err != nil { + t.Errorf("Validate() error = %v", err) + return + } - if cfg.ReloadStrategy != tt.wantVal { - t.Errorf("ReloadStrategy = %v, want %v", cfg.ReloadStrategy, tt.wantVal) - } - }) + if cfg.ReloadStrategy != tt.wantVal { + t.Errorf("ReloadStrategy = %v, want %v", cfg.ReloadStrategy, tt.wantVal) + } + }, + ) } } @@ -58,51 +60,57 @@ func TestConfig_Validate_ArgoRolloutStrategy(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - cfg := NewDefault() - cfg.ArgoRolloutStrategy = tt.strategy + t.Run( + tt.name, func(t *testing.T) { + cfg := NewDefault() + cfg.ArgoRolloutStrategy = tt.strategy - err := cfg.Validate() + err := cfg.Validate() - if tt.wantErr { - if err == nil { - t.Error("Validate() should return error for invalid strategy") + if tt.wantErr { + if err == nil { + t.Error("Validate() should return error for invalid strategy") + } + return } - return - } - if err != nil { - t.Errorf("Validate() error = %v", err) - return - } + if err != nil { + t.Errorf("Validate() error = %v", err) + return + } - if cfg.ArgoRolloutStrategy != tt.wantVal { - t.Errorf("ArgoRolloutStrategy = %v, want %v", cfg.ArgoRolloutStrategy, tt.wantVal) - } - }) + if cfg.ArgoRolloutStrategy != tt.wantVal { + t.Errorf("ArgoRolloutStrategy = %v, want %v", cfg.ArgoRolloutStrategy, tt.wantVal) + } + }, + ) } } func TestConfig_Validate_LogLevel(t *testing.T) { validLevels := []string{"trace", "debug", "info", "warn", "warning", "error", "fatal", "panic", ""} for _, level := range validLevels { - t.Run("valid_"+level, func(t *testing.T) { - cfg := NewDefault() - cfg.LogLevel = level - if err := cfg.Validate(); err != nil { - t.Errorf("Validate() error for level %q: %v", level, err) - } - }) + t.Run( + "valid_"+level, func(t *testing.T) { + cfg := NewDefault() + cfg.LogLevel = level + if err := cfg.Validate(); err != nil { + t.Errorf("Validate() error for level %q: %v", level, err) + } + }, + ) } - t.Run("invalid level", func(t *testing.T) { - cfg := NewDefault() - cfg.LogLevel = "invalid" - err := cfg.Validate() - if err == nil { - t.Error("Validate() should return error for invalid log level") - } - }) + t.Run( + "invalid level", func(t *testing.T) { + cfg := NewDefault() + cfg.LogLevel = "invalid" + err := cfg.Validate() + if err == nil { + t.Error("Validate() should return error for invalid log level") + } + }, + ) } func TestConfig_Validate_LogFormat(t *testing.T) { @@ -117,14 +125,16 @@ func TestConfig_Validate_LogFormat(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - cfg := NewDefault() - cfg.LogFormat = tt.format - err := cfg.Validate() - if (err != nil) != tt.wantErr { - t.Errorf("Validate() error = %v, wantErr %v", err, tt.wantErr) - } - }) + t.Run( + tt.name, func(t *testing.T) { + cfg := NewDefault() + cfg.LogFormat = tt.format + err := cfg.Validate() + if (err != nil) != tt.wantErr { + t.Errorf("Validate() error = %v, wantErr %v", err, tt.wantErr) + } + }, + ) } } @@ -203,35 +213,41 @@ func TestValidationError_Error(t *testing.T) { } func TestValidationErrors_Error(t *testing.T) { - t.Run("empty", func(t *testing.T) { - var errs ValidationErrors - if errs.Error() != "" { - t.Errorf("Empty errors should return empty string, got %q", errs.Error()) - } - }) + t.Run( + "empty", func(t *testing.T) { + var errs ValidationErrors + if errs.Error() != "" { + t.Errorf("Empty errors should return empty string, got %q", errs.Error()) + } + }, + ) - t.Run("single error", func(t *testing.T) { - errs := ValidationErrors{ - {Field: "Field1", Message: "error1"}, - } - if !strings.Contains(errs.Error(), "Field1") { - t.Errorf("Error() should contain field name, got %q", errs.Error()) - } - }) + t.Run( + "single error", func(t *testing.T) { + errs := ValidationErrors{ + {Field: "Field1", Message: "error1"}, + } + if !strings.Contains(errs.Error(), "Field1") { + t.Errorf("Error() should contain field name, got %q", errs.Error()) + } + }, + ) - t.Run("multiple errors", func(t *testing.T) { - errs := ValidationErrors{ - {Field: "Field1", Message: "error1"}, - {Field: "Field2", Message: "error2"}, - } - errStr := errs.Error() - if !strings.Contains(errStr, "multiple configuration errors") { - t.Errorf("Error() should mention multiple errors, got %q", errStr) - } - if !strings.Contains(errStr, "Field1") || !strings.Contains(errStr, "Field2") { - t.Errorf("Error() should contain all field names, got %q", errStr) - } - }) + t.Run( + "multiple errors", func(t *testing.T) { + errs := ValidationErrors{ + {Field: "Field1", Message: "error1"}, + {Field: "Field2", Message: "error2"}, + } + errStr := errs.Error() + if !strings.Contains(errStr, "multiple configuration errors") { + t.Errorf("Error() should mention multiple errors, got %q", errStr) + } + if !strings.Contains(errStr, "Field1") || !strings.Contains(errStr, "Field2") { + t.Errorf("Error() should contain all field names, got %q", errStr) + } + }, + ) } func TestParseSelectors(t *testing.T) { @@ -252,37 +268,21 @@ func TestParseSelectors(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - selectors, err := ParseSelectors(tt.selectors) - if (err != nil) != tt.wantErr { - t.Errorf("ParseSelectors() error = %v, wantErr %v", err, tt.wantErr) - return - } - if !tt.wantErr && len(selectors) != tt.wantLen { - t.Errorf("ParseSelectors() returned %d selectors, want %d", len(selectors), tt.wantLen) - } - }) + t.Run( + tt.name, func(t *testing.T) { + selectors, err := ParseSelectors(tt.selectors) + if (err != nil) != tt.wantErr { + t.Errorf("ParseSelectors() error = %v, wantErr %v", err, tt.wantErr) + return + } + if !tt.wantErr && len(selectors) != tt.wantLen { + t.Errorf("ParseSelectors() returned %d selectors, want %d", len(selectors), tt.wantLen) + } + }, + ) } } -func TestMustParseSelectors(t *testing.T) { - t.Run("valid selectors", func(t *testing.T) { - selectors := MustParseSelectors([]string{"env=production"}) - if len(selectors) != 1 { - t.Errorf("MustParseSelectors() returned %d selectors, want 1", len(selectors)) - } - }) - - t.Run("panics on invalid", func(t *testing.T) { - defer func() { - if r := recover(); r == nil { - t.Error("MustParseSelectors() should panic on invalid selector") - } - }() - MustParseSelectors([]string{"env in (prod,staging"}) // missing closing paren - }) -} - func TestNormalizeToLower(t *testing.T) { tests := []struct { name string @@ -300,21 +300,23 @@ func TestNormalizeToLower(t *testing.T) { } for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := normalizeToLower(tt.input) - if tt.want == nil && got != nil { - t.Errorf("normalizeToLower() = %v, want nil", got) - return - } - if len(got) != len(tt.want) { - t.Errorf("normalizeToLower() length = %d, want %d", len(got), len(tt.want)) - return - } - for i := range got { - if got[i] != tt.want[i] { - t.Errorf("normalizeToLower()[%d] = %q, want %q", i, got[i], tt.want[i]) + t.Run( + tt.name, func(t *testing.T) { + got := normalizeToLower(tt.input) + if tt.want == nil && got != nil { + t.Errorf("normalizeToLower() = %v, want nil", got) + return } - } - }) + if len(got) != len(tt.want) { + t.Errorf("normalizeToLower() length = %d, want %d", len(got), len(tt.want)) + return + } + for i := range got { + if got[i] != tt.want[i] { + t.Errorf("normalizeToLower()[%d] = %q, want %q", i, got[i], tt.want[i]) + } + } + }, + ) } } diff --git a/internal/pkg/controller/configmap_reconciler.go b/internal/pkg/controller/configmap_reconciler.go index 447348c0..3f744ea5 100644 --- a/internal/pkg/controller/configmap_reconciler.go +++ b/internal/pkg/controller/configmap_reconciler.go @@ -3,7 +3,6 @@ package controller import ( "context" "sync" - "time" "github.com/go-logr/logr" "github.com/stakater/Reloader/internal/pkg/alerting" @@ -13,14 +12,10 @@ import ( "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" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" - "sigs.k8s.io/controller-runtime/pkg/event" - "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" ) @@ -36,6 +31,7 @@ type ConfigMapReconciler struct { WebhookClient *webhook.Client Alerter alerting.Alerter + handler *ReloadHandler initialized bool initOnce sync.Once } @@ -44,17 +40,14 @@ type ConfigMapReconciler struct { func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { log := r.Log.WithValues("configmap", req.NamespacedName) - // Mark as initialized after first reconcile (caches are synced at this point) r.initOnce.Do(func() { r.initialized = true log.Info("ConfigMap controller initialized") }) - // Fetch the ConfigMap var cm corev1.ConfigMap if err := r.Get(ctx, req.NamespacedName, &cm); err != nil { if errors.IsNotFound(err) { - // ConfigMap was deleted - handle if ReloadOnDelete is enabled if r.Config.ReloadOnDelete { return r.handleDelete(ctx, req, log) } @@ -64,364 +57,63 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) ( return ctrl.Result{}, err } - // Check if namespace should be ignored if r.Config.IsNamespaceIgnored(cm.Namespace) { log.V(1).Info("skipping ConfigMap in ignored namespace") return ctrl.Result{}, nil } - // Get all workloads in the same namespace - workloads, err := r.listWorkloads(ctx, cm.Namespace) - if err != nil { - log.Error(err, "failed to list workloads") - return ctrl.Result{}, err - } - - // Evaluate which workloads should be reloaded - change := reload.ConfigMapChange{ - ConfigMap: &cm, - EventType: reload.EventTypeUpdate, - } - decisions := r.ReloadService.ProcessConfigMap(change, workloads) - - // Collect workloads that should be reloaded - var workloadsToReload []reload.ReloadDecision - for _, decision := range decisions { - 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 := UpdateWorkloadWithRetry( - ctx, - r.Client, - r.ReloadService, - decision.Workload, - cm.Name, - reload.ResourceTypeConfigMap, - cm.Namespace, - decision.Hash, - decision.AutoReload, - ) - if err != nil { - 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 { - r.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), "ConfigMap", cm.Name) - r.recordMetrics(true, cm.Namespace) - log.Info("workload reloaded successfully", - "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") - } - } - } - - return ctrl.Result{}, nil + return r.reloadHandler().Process(ctx, cm.Namespace, cm.Name, "ConfigMap", reload.ResourceTypeConfigMap, + func(workloads []workload.WorkloadAccessor) []reload.ReloadDecision { + return r.ReloadService.ProcessConfigMap(reload.ConfigMapChange{ + ConfigMap: &cm, + EventType: reload.EventTypeUpdate, + }, workloads) + }, log) } // 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") - // Get all workloads in the namespace - workloads, err := r.listWorkloads(ctx, req.Namespace) - if err != nil { - log.Error(err, "failed to list workloads") - return ctrl.Result{}, err - } + cm := &corev1.ConfigMap{} + cm.Name = req.Name + cm.Namespace = req.Namespace - // For delete events, we create a change with empty ConfigMap - change := reload.ConfigMapChange{ - ConfigMap: &corev1.ConfigMap{}, - EventType: reload.EventTypeDelete, - } - change.ConfigMap.Name = req.Name - change.ConfigMap.Namespace = req.Namespace + return r.reloadHandler().Process(ctx, req.Namespace, req.Name, "ConfigMap", reload.ResourceTypeConfigMap, + func(workloads []workload.WorkloadAccessor) []reload.ReloadDecision { + return r.ReloadService.ProcessConfigMap(reload.ConfigMapChange{ + ConfigMap: cm, + EventType: reload.EventTypeDelete, + }, workloads) + }, log) +} - decisions := r.ReloadService.ProcessConfigMap(change, workloads) - - // Collect workloads that should be reloaded - var workloadsToReload []reload.ReloadDecision - for _, decision := range decisions { - if decision.ShouldReload { - workloadsToReload = append(workloadsToReload, decision) +func (r *ConfigMapReconciler) reloadHandler() *ReloadHandler { + if r.handler == nil { + r.handler = &ReloadHandler{ + Client: r.Client, + Lister: workload.NewLister(r.Client, r.Registry, r.Config), + ReloadService: r.ReloadService, + WebhookClient: r.WebhookClient, + Collectors: r.Collectors, + EventRecorder: r.EventRecorder, + Alerter: r.Alerter, } } - - // 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 := UpdateWorkloadWithRetry( - ctx, - r.Client, - r.ReloadService, - decision.Workload, - req.Name, - reload.ResourceTypeConfigMap, - req.Namespace, - decision.Hash, - decision.AutoReload, - ) - if err != nil { - 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 { - 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") - } - } - } - - return ctrl.Result{}, nil -} - -// listWorkloads returns all workloads in the given namespace. -func (r *ConfigMapReconciler) listWorkloads(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var result []workload.WorkloadAccessor - - for _, kind := range r.Registry.SupportedKinds() { - // Skip ignored workload types - if r.Config.IsWorkloadIgnored(string(kind)) { - continue - } - - workloads, err := r.listWorkloadsByKind(ctx, namespace, kind) - if err != nil { - return nil, err - } - result = append(result, workloads...) - } - - return result, nil -} - -// listWorkloadsByKind lists workloads of a specific kind in the namespace. -func (r *ConfigMapReconciler) listWorkloadsByKind(ctx context.Context, namespace string, kind workload.Kind) ([]workload.WorkloadAccessor, error) { - switch kind { - case workload.KindDeployment: - return r.listDeployments(ctx, namespace) - case workload.KindDaemonSet: - return r.listDaemonSets(ctx, namespace) - case workload.KindStatefulSet: - return r.listStatefulSets(ctx, namespace) - case workload.KindJob: - return r.listJobs(ctx, namespace) - case workload.KindCronJob: - return r.listCronJobs(ctx, namespace) - default: - return nil, nil - } -} - -func (r *ConfigMapReconciler) listDeployments(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list appsv1.DeploymentList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewDeploymentWorkload(&list.Items[i]) - } - return result, nil -} - -func (r *ConfigMapReconciler) listDaemonSets(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list appsv1.DaemonSetList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewDaemonSetWorkload(&list.Items[i]) - } - return result, nil -} - -func (r *ConfigMapReconciler) listStatefulSets(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list appsv1.StatefulSetList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewStatefulSetWorkload(&list.Items[i]) - } - return result, nil -} - -func (r *ConfigMapReconciler) listJobs(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list batchv1.JobList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewJobWorkload(&list.Items[i]) - } - return result, nil -} - -func (r *ConfigMapReconciler) listCronJobs(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list batchv1.CronJobList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewCronJobWorkload(&list.Items[i]) - } - return result, nil -} - -// recordMetrics records reload metrics. -func (r *ConfigMapReconciler) recordMetrics(success bool, namespace string) { - 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 - } - } - - 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 + return r.handler } // SetupWithManager sets up the controller with the Manager. func (r *ConfigMapReconciler) SetupWithManager(mgr ctrl.Manager) error { - hasher := r.ReloadService.Hasher() - return ctrl.NewControllerManagedBy(mgr). For(&corev1.ConfigMap{}). - WithEventFilter(predicate.And( - reload.ConfigMapPredicates(r.Config, hasher), - reload.NamespaceFilterPredicate(r.Config), - reload.LabelSelectorPredicate(r.Config), - reload.IgnoreAnnotationPredicate(r.Config), - r.createEventFilter(), + WithEventFilter(BuildEventFilter( + reload.ConfigMapPredicates(r.Config, r.ReloadService.Hasher()), + r.Config, &r.initialized, )). Complete(r) } -// createEventFilter filters create events based on initialization state. -func (r *ConfigMapReconciler) createEventFilter() predicate.Predicate { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - // During startup, skip create events unless SyncAfterRestart is enabled - if !r.initialized && !r.Config.SyncAfterRestart { - return false - } - // After initialization, only process creates if ReloadOnCreate is enabled - return r.Config.ReloadOnCreate - }, - UpdateFunc: func(e event.UpdateEvent) bool { - return true - }, - DeleteFunc: func(e event.DeleteEvent) bool { - return r.Config.ReloadOnDelete - }, - GenericFunc: func(e event.GenericEvent) bool { - return false - }, - } -} - -// Ensure ConfigMapReconciler implements reconcile.Reconciler var _ reconcile.Reconciler = &ConfigMapReconciler{} diff --git a/internal/pkg/controller/handler.go b/internal/pkg/controller/handler.go new file mode 100644 index 00000000..b3ac4d2f --- /dev/null +++ b/internal/pkg/controller/handler.go @@ -0,0 +1,187 @@ +package controller + +import ( + "context" + "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" + "github.com/stakater/Reloader/internal/pkg/reload" + "github.com/stakater/Reloader/internal/pkg/webhook" + "github.com/stakater/Reloader/internal/pkg/workload" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/predicate" +) + +// ReloadHandler handles the common reload workflow. +type ReloadHandler struct { + Client client.Client + Lister *workload.Lister + ReloadService *reload.Service + WebhookClient *webhook.Client + Collectors *metrics.Collectors + EventRecorder *events.Recorder + Alerter alerting.Alerter +} + +// Process handles the reload workflow: list workloads, get decisions, webhook or apply. +func (h *ReloadHandler) Process( + ctx context.Context, + namespace, resourceName, resourceKind string, + resourceType reload.ResourceType, + getDecisions func([]workload.WorkloadAccessor) []reload.ReloadDecision, + log logr.Logger, +) (ctrl.Result, error) { + workloads, err := h.Lister.List(ctx, namespace) + if err != nil { + log.Error(err, "failed to list workloads") + return ctrl.Result{}, err + } + + decisions := reload.FilterDecisions(getDecisions(workloads)) + + if h.WebhookClient.IsConfigured() && len(decisions) > 0 { + return h.sendWebhook(ctx, resourceName, namespace, resourceType, decisions, log) + } + + h.applyReloads(ctx, resourceName, namespace, resourceKind, resourceType, decisions, log) + return ctrl.Result{}, nil +} + +func (h *ReloadHandler) sendWebhook( + 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 + } + } + + payload := webhook.Payload{ + Kind: string(resourceType), + Namespace: namespace, + ResourceName: resourceName, + ResourceType: string(resourceType), + Hash: hash, + Timestamp: time.Now().UTC(), + Workloads: workloads, + } + + if err := h.WebhookClient.Send(ctx, payload); err != nil { + log.Error(err, "failed to send webhook notification") + h.Collectors.RecordReload(false, namespace) + return ctrl.Result{}, err + } + + log.Info("webhook notification sent", + "resource", resourceName, + "workloadCount", len(workloads), + ) + h.Collectors.RecordReload(true, namespace) + return ctrl.Result{}, nil +} + +func (h *ReloadHandler) applyReloads( + ctx context.Context, + resourceName, resourceNamespace, resourceKind string, + resourceType reload.ResourceType, + decisions []reload.ReloadDecision, + log logr.Logger, +) { + for _, decision := range decisions { + log.Info("reloading workload", + "workload", decision.Workload.GetName(), + "kind", decision.Workload.Kind(), + "reason", decision.Reason, + ) + + updated, err := UpdateWorkloadWithRetry( + ctx, + h.Client, + h.ReloadService, + decision.Workload, + resourceName, + resourceType, + resourceNamespace, + decision.Hash, + decision.AutoReload, + ) + if err != nil { + log.Error(err, "failed to update workload", + "workload", decision.Workload.GetName(), + "kind", decision.Workload.Kind(), + ) + h.EventRecorder.ReloadFailed(decision.Workload.GetObject(), resourceKind, resourceName, err) + h.Collectors.RecordReload(false, resourceNamespace) + continue + } + + if updated { + h.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), resourceKind, resourceName) + h.Collectors.RecordReload(true, resourceNamespace) + 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: resourceKind, + ResourceName: resourceName, + ResourceNamespace: resourceNamespace, + Timestamp: time.Now(), + }); err != nil { + log.Error(err, "failed to send alert") + } + } + } +} + +// BuildEventFilter combines a resource-specific predicate with common filters. +func BuildEventFilter(resourcePredicate predicate.Predicate, cfg *config.Config, initialized *bool) predicate.Predicate { + return predicate.And( + resourcePredicate, + reload.NamespaceFilterPredicate(cfg), + reload.LabelSelectorPredicate(cfg), + reload.IgnoreAnnotationPredicate(cfg), + createEventPredicate(cfg, initialized), + ) +} + +func createEventPredicate(cfg *config.Config, initialized *bool) predicate.Predicate { + return predicate.Funcs{ + CreateFunc: func(e event.CreateEvent) bool { + if !*initialized && !cfg.SyncAfterRestart { + return false + } + return cfg.ReloadOnCreate + }, + UpdateFunc: func(e event.UpdateEvent) bool { + return true + }, + DeleteFunc: func(e event.DeleteEvent) bool { + return cfg.ReloadOnDelete + }, + GenericFunc: func(e event.GenericEvent) bool { + return false + }, + } +} diff --git a/internal/pkg/controller/secret_reconciler.go b/internal/pkg/controller/secret_reconciler.go index 8c5dbd2b..c6c79131 100644 --- a/internal/pkg/controller/secret_reconciler.go +++ b/internal/pkg/controller/secret_reconciler.go @@ -3,7 +3,6 @@ package controller import ( "context" "sync" - "time" "github.com/go-logr/logr" "github.com/stakater/Reloader/internal/pkg/alerting" @@ -13,14 +12,10 @@ import ( "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" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" - "sigs.k8s.io/controller-runtime/pkg/event" - "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" ) @@ -36,6 +31,7 @@ type SecretReconciler struct { WebhookClient *webhook.Client Alerter alerting.Alerter + handler *ReloadHandler initialized bool initOnce sync.Once } @@ -44,17 +40,14 @@ type SecretReconciler struct { func (r *SecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { log := r.Log.WithValues("secret", req.NamespacedName) - // Mark as initialized after first reconcile (caches are synced at this point) r.initOnce.Do(func() { r.initialized = true log.Info("Secret controller initialized") }) - // Fetch the Secret var secret corev1.Secret if err := r.Get(ctx, req.NamespacedName, &secret); err != nil { if errors.IsNotFound(err) { - // Secret was deleted - handle if ReloadOnDelete is enabled if r.Config.ReloadOnDelete { return r.handleDelete(ctx, req, log) } @@ -64,361 +57,60 @@ func (r *SecretReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr return ctrl.Result{}, err } - // Check if namespace should be ignored if r.Config.IsNamespaceIgnored(secret.Namespace) { log.V(1).Info("skipping Secret in ignored namespace") return ctrl.Result{}, nil } - // Get all workloads in the same namespace - workloads, err := r.listWorkloads(ctx, secret.Namespace) - if err != nil { - log.Error(err, "failed to list workloads") - return ctrl.Result{}, err - } - - // Evaluate which workloads should be reloaded - change := reload.SecretChange{ - Secret: &secret, - EventType: reload.EventTypeUpdate, - } - decisions := r.ReloadService.ProcessSecret(change, workloads) - - // Collect workloads that should be reloaded - var workloadsToReload []reload.ReloadDecision - for _, decision := range decisions { - 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 := UpdateWorkloadWithRetry( - ctx, - r.Client, - r.ReloadService, - decision.Workload, - secret.Name, - reload.ResourceTypeSecret, - secret.Namespace, - decision.Hash, - decision.AutoReload, - ) - if err != nil { - 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 { - r.EventRecorder.ReloadSuccess(decision.Workload.GetObject(), "Secret", secret.Name) - r.recordMetrics(true, secret.Namespace) - log.Info("workload reloaded successfully", - "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") - } - } - } - - return ctrl.Result{}, nil + return r.reloadHandler().Process(ctx, secret.Namespace, secret.Name, "Secret", reload.ResourceTypeSecret, + func(workloads []workload.WorkloadAccessor) []reload.ReloadDecision { + return r.ReloadService.ProcessSecret(reload.SecretChange{ + Secret: &secret, + EventType: reload.EventTypeUpdate, + }, workloads) + }, log) } -// handleDelete handles Secret deletion events. func (r *SecretReconciler) handleDelete(ctx context.Context, req ctrl.Request, log logr.Logger) (ctrl.Result, error) { log.Info("handling Secret deletion") - // Get all workloads in the namespace - workloads, err := r.listWorkloads(ctx, req.Namespace) - if err != nil { - log.Error(err, "failed to list workloads") - return ctrl.Result{}, err - } + secret := &corev1.Secret{} + secret.Name = req.Name + secret.Namespace = req.Namespace - // For delete events, we create a change with empty Secret - change := reload.SecretChange{ - Secret: &corev1.Secret{}, - EventType: reload.EventTypeDelete, - } - change.Secret.Name = req.Name - change.Secret.Namespace = req.Namespace + return r.reloadHandler().Process(ctx, req.Namespace, req.Name, "Secret", reload.ResourceTypeSecret, + func(workloads []workload.WorkloadAccessor) []reload.ReloadDecision { + return r.ReloadService.ProcessSecret(reload.SecretChange{ + Secret: secret, + EventType: reload.EventTypeDelete, + }, workloads) + }, log) +} - decisions := r.ReloadService.ProcessSecret(change, workloads) - - // Collect workloads that should be reloaded - var workloadsToReload []reload.ReloadDecision - for _, decision := range decisions { - if decision.ShouldReload { - workloadsToReload = append(workloadsToReload, decision) +func (r *SecretReconciler) reloadHandler() *ReloadHandler { + if r.handler == nil { + r.handler = &ReloadHandler{ + Client: r.Client, + Lister: workload.NewLister(r.Client, r.Registry, r.Config), + ReloadService: r.ReloadService, + WebhookClient: r.WebhookClient, + Collectors: r.Collectors, + EventRecorder: r.EventRecorder, + Alerter: r.Alerter, } } - - // 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 := UpdateWorkloadWithRetry( - ctx, - r.Client, - r.ReloadService, - decision.Workload, - req.Name, - reload.ResourceTypeSecret, - req.Namespace, - decision.Hash, - decision.AutoReload, - ) - if err != nil { - 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 { - 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") - } - } - } - - return ctrl.Result{}, nil -} - -// listWorkloads returns all workloads in the given namespace. -func (r *SecretReconciler) listWorkloads(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var result []workload.WorkloadAccessor - - for _, kind := range r.Registry.SupportedKinds() { - // Skip ignored workload types - if r.Config.IsWorkloadIgnored(string(kind)) { - continue - } - - workloads, err := r.listWorkloadsByKind(ctx, namespace, kind) - if err != nil { - return nil, err - } - result = append(result, workloads...) - } - - return result, nil -} - -// listWorkloadsByKind lists workloads of a specific kind in the namespace. -func (r *SecretReconciler) listWorkloadsByKind(ctx context.Context, namespace string, kind workload.Kind) ([]workload.WorkloadAccessor, error) { - switch kind { - case workload.KindDeployment: - return r.listDeployments(ctx, namespace) - case workload.KindDaemonSet: - return r.listDaemonSets(ctx, namespace) - case workload.KindStatefulSet: - return r.listStatefulSets(ctx, namespace) - case workload.KindJob: - return r.listJobs(ctx, namespace) - case workload.KindCronJob: - return r.listCronJobs(ctx, namespace) - default: - return nil, nil - } -} - -func (r *SecretReconciler) listDeployments(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list appsv1.DeploymentList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewDeploymentWorkload(&list.Items[i]) - } - return result, nil -} - -func (r *SecretReconciler) listDaemonSets(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list appsv1.DaemonSetList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewDaemonSetWorkload(&list.Items[i]) - } - return result, nil -} - -func (r *SecretReconciler) listStatefulSets(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list appsv1.StatefulSetList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewStatefulSetWorkload(&list.Items[i]) - } - return result, nil -} - -func (r *SecretReconciler) listJobs(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list batchv1.JobList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewJobWorkload(&list.Items[i]) - } - return result, nil -} - -func (r *SecretReconciler) listCronJobs(ctx context.Context, namespace string) ([]workload.WorkloadAccessor, error) { - var list batchv1.CronJobList - if err := r.List(ctx, &list, client.InNamespace(namespace)); err != nil { - return nil, err - } - result := make([]workload.WorkloadAccessor, len(list.Items)) - for i := range list.Items { - result[i] = workload.NewCronJobWorkload(&list.Items[i]) - } - return result, nil -} - -// recordMetrics records reload metrics. -func (r *SecretReconciler) recordMetrics(success bool, namespace string) { - 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 - } - } - - 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 + return r.handler } // SetupWithManager sets up the controller with the Manager. func (r *SecretReconciler) SetupWithManager(mgr ctrl.Manager) error { - hasher := r.ReloadService.Hasher() - return ctrl.NewControllerManagedBy(mgr). For(&corev1.Secret{}). - WithEventFilter(predicate.And( - reload.SecretPredicates(r.Config, hasher), - reload.NamespaceFilterPredicate(r.Config), - reload.LabelSelectorPredicate(r.Config), - reload.IgnoreAnnotationPredicate(r.Config), - r.createEventFilter(), + WithEventFilter(BuildEventFilter( + reload.SecretPredicates(r.Config, r.ReloadService.Hasher()), + r.Config, &r.initialized, )). Complete(r) } -// createEventFilter filters create events based on initialization state. -func (r *SecretReconciler) createEventFilter() predicate.Predicate { - return predicate.Funcs{ - CreateFunc: func(e event.CreateEvent) bool { - // During startup, skip create events unless SyncAfterRestart is enabled - if !r.initialized && !r.Config.SyncAfterRestart { - return false - } - // After initialization, only process creates if ReloadOnCreate is enabled - return r.Config.ReloadOnCreate - }, - UpdateFunc: func(e event.UpdateEvent) bool { - return true - }, - DeleteFunc: func(e event.DeleteEvent) bool { - return r.Config.ReloadOnDelete - }, - GenericFunc: func(e event.GenericEvent) bool { - return false - }, - } -} - -// Ensure SecretReconciler implements reconcile.Reconciler var _ reconcile.Reconciler = &SecretReconciler{} diff --git a/internal/pkg/reload/service.go b/internal/pkg/reload/service.go index 6d7825fe..e9ff12b6 100644 --- a/internal/pkg/reload/service.go +++ b/internal/pkg/reload/service.go @@ -66,6 +66,17 @@ type ReloadDecision struct { Hash string } +// FilterDecisions returns only decisions where ShouldReload is true. +func FilterDecisions(decisions []ReloadDecision) []ReloadDecision { + var result []ReloadDecision + for _, d := range decisions { + if d.ShouldReload { + result = append(result, d) + } + } + return result +} + // ProcessConfigMap evaluates all workloads to determine which should be reloaded. func (s *Service) ProcessConfigMap(change ConfigMapChange, workloads []workload.WorkloadAccessor) []ReloadDecision { if change.ConfigMap == nil { diff --git a/internal/pkg/workload/lister.go b/internal/pkg/workload/lister.go new file mode 100644 index 00000000..a1487bb7 --- /dev/null +++ b/internal/pkg/workload/lister.go @@ -0,0 +1,111 @@ +package workload + +import ( + "context" + + appsv1 "k8s.io/api/apps/v1" + batchv1 "k8s.io/api/batch/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// IgnoreChecker checks if a workload kind should be ignored. +type IgnoreChecker interface { + IsWorkloadIgnored(kind string) bool +} + +// Lister lists workloads from the cluster. +type Lister struct { + Client client.Client + Registry *Registry + Checker IgnoreChecker +} + +// NewLister creates a new workload lister. +func NewLister(c client.Client, registry *Registry, checker IgnoreChecker) *Lister { + return &Lister{ + Client: c, + Registry: registry, + Checker: checker, + } +} + +// List returns all workloads in the given namespace. +func (l *Lister) List(ctx context.Context, namespace string) ([]WorkloadAccessor, error) { + var result []WorkloadAccessor + + for _, kind := range l.Registry.SupportedKinds() { + if l.Checker != nil && l.Checker.IsWorkloadIgnored(string(kind)) { + continue + } + + workloads, err := l.listByKind(ctx, namespace, kind) + if err != nil { + return nil, err + } + result = append(result, workloads...) + } + + return result, nil +} + +func (l *Lister) listByKind(ctx context.Context, namespace string, kind Kind) ([]WorkloadAccessor, error) { + switch kind { + case KindDeployment: + var list appsv1.DeploymentList + if err := l.Client.List(ctx, &list, client.InNamespace(namespace)); err != nil { + return nil, err + } + result := make([]WorkloadAccessor, len(list.Items)) + for i := range list.Items { + result[i] = NewDeploymentWorkload(&list.Items[i]) + } + return result, nil + + case KindDaemonSet: + var list appsv1.DaemonSetList + if err := l.Client.List(ctx, &list, client.InNamespace(namespace)); err != nil { + return nil, err + } + result := make([]WorkloadAccessor, len(list.Items)) + for i := range list.Items { + result[i] = NewDaemonSetWorkload(&list.Items[i]) + } + return result, nil + + case KindStatefulSet: + var list appsv1.StatefulSetList + if err := l.Client.List(ctx, &list, client.InNamespace(namespace)); err != nil { + return nil, err + } + result := make([]WorkloadAccessor, len(list.Items)) + for i := range list.Items { + result[i] = NewStatefulSetWorkload(&list.Items[i]) + } + return result, nil + + case KindJob: + var list batchv1.JobList + if err := l.Client.List(ctx, &list, client.InNamespace(namespace)); err != nil { + return nil, err + } + result := make([]WorkloadAccessor, len(list.Items)) + for i := range list.Items { + result[i] = NewJobWorkload(&list.Items[i]) + } + return result, nil + + case KindCronJob: + var list batchv1.CronJobList + if err := l.Client.List(ctx, &list, client.InNamespace(namespace)); err != nil { + return nil, err + } + result := make([]WorkloadAccessor, len(list.Items)) + for i := range list.Items { + result[i] = NewCronJobWorkload(&list.Items[i]) + } + return result, nil + + default: + return nil, nil + } +}