diff --git a/internal/pkg/alerting/alerter.go b/internal/pkg/alerting/alerter.go new file mode 100644 index 00000000..5213d382 --- /dev/null +++ b/internal/pkg/alerting/alerter.go @@ -0,0 +1,51 @@ +package alerting + +import ( + "context" + "time" + + "github.com/stakater/Reloader/internal/pkg/config" +) + +// AlertMessage contains the details of a reload event to be sent as an alert. +type AlertMessage struct { + WorkloadKind string + WorkloadName string + WorkloadNamespace string + ResourceKind string + ResourceName string + ResourceNamespace string + Timestamp time.Time +} + +// Alerter is the interface for sending reload notifications. +type Alerter interface { + Send(ctx context.Context, message AlertMessage) error +} + +// NewAlerter creates an Alerter based on the configuration. +// Returns a NoOpAlerter if alerting is disabled. +func NewAlerter(cfg *config.Config) Alerter { + alertCfg := cfg.Alerting + if !alertCfg.Enabled || alertCfg.WebhookURL == "" { + return &NoOpAlerter{} + } + + switch alertCfg.Sink { + case "slack": + return NewSlackAlerter(alertCfg.WebhookURL, alertCfg.Proxy, alertCfg.Additional) + case "teams": + return NewTeamsAlerter(alertCfg.WebhookURL, alertCfg.Proxy, alertCfg.Additional) + case "gchat": + return NewGChatAlerter(alertCfg.WebhookURL, alertCfg.Proxy, alertCfg.Additional) + default: + return NewRawAlerter(alertCfg.WebhookURL, alertCfg.Proxy, alertCfg.Additional) + } +} + +// NoOpAlerter is an Alerter that does nothing. +type NoOpAlerter struct{} + +func (a *NoOpAlerter) Send(ctx context.Context, message AlertMessage) error { + return nil +} diff --git a/internal/pkg/alerting/alerter_test.go b/internal/pkg/alerting/alerter_test.go new file mode 100644 index 00000000..6e5724f4 --- /dev/null +++ b/internal/pkg/alerting/alerter_test.go @@ -0,0 +1,267 @@ +package alerting + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "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 + 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) + } + 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) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + alerter := NewSlackAlerter(server.URL, "", "Test Cluster") + 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("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(), + } + + 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) + } +} + +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 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() + + alerter := NewRawAlerter(server.URL, "", "custom-info") + 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("RawAlerter.Send() error = %v", err) + } + + 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) + } +} + +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 { + t.Error("Expected error for non-2xx response") + } +} diff --git a/internal/pkg/alerting/gchat.go b/internal/pkg/alerting/gchat.go new file mode 100644 index 00000000..73e2b303 --- /dev/null +++ b/internal/pkg/alerting/gchat.go @@ -0,0 +1,90 @@ +package alerting + +import ( + "context" + "encoding/json" + "fmt" +) + +// GChatAlerter sends alerts to Google Chat webhooks. +type GChatAlerter struct { + webhookURL string + additional string + client *httpClient +} + +// NewGChatAlerter creates a new GChatAlerter. +func NewGChatAlerter(webhookURL, proxyURL, additional string) *GChatAlerter { + return &GChatAlerter{ + webhookURL: webhookURL, + additional: additional, + client: newHTTPClient(proxyURL), + } +} + +// gchatMessage represents a Google Chat message. +type gchatMessage struct { + Text string `json:"text,omitempty"` + Cards []gchatCard `json:"cards,omitempty"` +} + +type gchatCard struct { + Header gchatHeader `json:"header"` + Sections []gchatSection `json:"sections"` +} + +type gchatHeader struct { + Title string `json:"title"` + Subtitle string `json:"subtitle,omitempty"` +} + +type gchatSection struct { + Widgets []gchatWidget `json:"widgets"` +} + +type gchatWidget struct { + KeyValue *gchatKeyValue `json:"keyValue,omitempty"` +} + +type gchatKeyValue struct { + TopLabel string `json:"topLabel"` + Content string `json:"content"` +} + +func (a *GChatAlerter) Send(ctx context.Context, message AlertMessage) error { + msg := a.buildMessage(message) + + body, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("marshaling gchat message: %w", err) + } + + return a.client.post(ctx, a.webhookURL, body) +} + +func (a *GChatAlerter) buildMessage(msg AlertMessage) gchatMessage { + widgets := []gchatWidget{ + {KeyValue: &gchatKeyValue{TopLabel: "Workload", Content: fmt.Sprintf("%s/%s (%s)", msg.WorkloadNamespace, msg.WorkloadName, msg.WorkloadKind)}}, + {KeyValue: &gchatKeyValue{TopLabel: "Resource", Content: fmt.Sprintf("%s/%s (%s)", msg.ResourceNamespace, msg.ResourceName, msg.ResourceKind)}}, + {KeyValue: &gchatKeyValue{TopLabel: "Time", Content: msg.Timestamp.Format("2006-01-02 15:04:05 UTC")}}, + } + + subtitle := "" + if a.additional != "" { + subtitle = a.additional + } + + return gchatMessage{ + Cards: []gchatCard{ + { + Header: gchatHeader{ + Title: "Reloader triggered reload", + Subtitle: subtitle, + }, + Sections: []gchatSection{ + {Widgets: widgets}, + }, + }, + }, + } +} diff --git a/internal/pkg/alerting/http.go b/internal/pkg/alerting/http.go new file mode 100644 index 00000000..827091e7 --- /dev/null +++ b/internal/pkg/alerting/http.go @@ -0,0 +1,57 @@ +package alerting + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "net/url" + "time" +) + +// httpClient wraps http.Client with common configuration. +type httpClient struct { + client *http.Client +} + +// newHTTPClient creates a new httpClient with optional proxy support. +func newHTTPClient(proxyURL string) *httpClient { + transport := &http.Transport{} + + if proxyURL != "" { + proxy, err := url.Parse(proxyURL) + if err == nil { + transport.Proxy = http.ProxyURL(proxy) + } + } + + return &httpClient{ + client: &http.Client{ + Transport: transport, + Timeout: 10 * time.Second, + }, + } +} + +// post sends a POST request with JSON body. +func (c *httpClient) post(ctx context.Context, url string, body []byte) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("creating request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("sending request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("unexpected status %d: %s", resp.StatusCode, string(body)) + } + + return nil +} diff --git a/internal/pkg/alerting/raw.go b/internal/pkg/alerting/raw.go new file mode 100644 index 00000000..ad0add08 --- /dev/null +++ b/internal/pkg/alerting/raw.go @@ -0,0 +1,57 @@ +package alerting + +import ( + "context" + "encoding/json" + "fmt" +) + +// RawAlerter sends alerts as raw JSON to a webhook. +type RawAlerter struct { + webhookURL string + additional string + client *httpClient +} + +// NewRawAlerter creates a new RawAlerter. +func NewRawAlerter(webhookURL, proxyURL, additional string) *RawAlerter { + return &RawAlerter{ + webhookURL: webhookURL, + additional: additional, + client: newHTTPClient(proxyURL), + } +} + +// rawMessage is the JSON payload for raw webhook alerts. +type rawMessage struct { + Event string `json:"event"` + WorkloadKind string `json:"workloadKind"` + WorkloadName string `json:"workloadName"` + WorkloadNamespace string `json:"workloadNamespace"` + ResourceKind string `json:"resourceKind"` + ResourceName string `json:"resourceName"` + ResourceNamespace string `json:"resourceNamespace"` + Timestamp string `json:"timestamp"` + Additional string `json:"additional,omitempty"` +} + +func (a *RawAlerter) Send(ctx context.Context, message AlertMessage) error { + msg := rawMessage{ + Event: "reload", + WorkloadKind: message.WorkloadKind, + WorkloadName: message.WorkloadName, + WorkloadNamespace: message.WorkloadNamespace, + ResourceKind: message.ResourceKind, + ResourceName: message.ResourceName, + ResourceNamespace: message.ResourceNamespace, + Timestamp: message.Timestamp.Format("2006-01-02T15:04:05Z07:00"), + Additional: a.additional, + } + + body, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("marshaling raw message: %w", err) + } + + return a.client.post(ctx, a.webhookURL, body) +} diff --git a/internal/pkg/alerting/slack.go b/internal/pkg/alerting/slack.go new file mode 100644 index 00000000..1b917118 --- /dev/null +++ b/internal/pkg/alerting/slack.go @@ -0,0 +1,57 @@ +package alerting + +import ( + "context" + "encoding/json" + "fmt" +) + +// SlackAlerter sends alerts to Slack webhooks. +type SlackAlerter struct { + webhookURL string + additional string + client *httpClient +} + +// NewSlackAlerter creates a new SlackAlerter. +func NewSlackAlerter(webhookURL, proxyURL, additional string) *SlackAlerter { + return &SlackAlerter{ + webhookURL: webhookURL, + additional: additional, + client: newHTTPClient(proxyURL), + } +} + +type slackMessage struct { + Text string `json:"text"` +} + +func (a *SlackAlerter) Send(ctx context.Context, message AlertMessage) error { + text := a.formatMessage(message) + msg := slackMessage{Text: text} + + body, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("marshaling slack message: %w", err) + } + + return a.client.post(ctx, a.webhookURL, body) +} + +func (a *SlackAlerter) formatMessage(msg AlertMessage) string { + text := fmt.Sprintf( + "Reloader triggered reload\n"+ + "*Workload:* %s/%s (%s)\n"+ + "*Resource:* %s/%s (%s)\n"+ + "*Time:* %s", + msg.WorkloadNamespace, msg.WorkloadName, msg.WorkloadKind, + msg.ResourceNamespace, msg.ResourceName, msg.ResourceKind, + msg.Timestamp.Format("2006-01-02 15:04:05 UTC"), + ) + + if a.additional != "" { + text = a.additional + "\n" + text + } + + return text +} diff --git a/internal/pkg/alerting/teams.go b/internal/pkg/alerting/teams.go new file mode 100644 index 00000000..99b08d5c --- /dev/null +++ b/internal/pkg/alerting/teams.go @@ -0,0 +1,81 @@ +package alerting + +import ( + "context" + "encoding/json" + "fmt" +) + +// TeamsAlerter sends alerts to Microsoft Teams webhooks. +type TeamsAlerter struct { + webhookURL string + additional string + client *httpClient +} + +// NewTeamsAlerter creates a new TeamsAlerter. +func NewTeamsAlerter(webhookURL, proxyURL, additional string) *TeamsAlerter { + return &TeamsAlerter{ + webhookURL: webhookURL, + additional: additional, + client: newHTTPClient(proxyURL), + } +} + +// teamsMessage represents a Microsoft Teams message card. +type teamsMessage struct { + Type string `json:"@type"` + Context string `json:"@context"` + ThemeColor string `json:"themeColor"` + Summary string `json:"summary"` + Sections []teamsSection `json:"sections"` +} + +type teamsSection struct { + ActivityTitle string `json:"activityTitle"` + ActivitySubtitle string `json:"activitySubtitle,omitempty"` + Facts []teamsFact `json:"facts"` +} + +type teamsFact struct { + Name string `json:"name"` + Value string `json:"value"` +} + +func (a *TeamsAlerter) Send(ctx context.Context, message AlertMessage) error { + msg := a.buildMessage(message) + + body, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("marshaling teams message: %w", err) + } + + return a.client.post(ctx, a.webhookURL, body) +} + +func (a *TeamsAlerter) buildMessage(msg AlertMessage) teamsMessage { + facts := []teamsFact{ + {Name: "Workload", Value: fmt.Sprintf("%s/%s (%s)", msg.WorkloadNamespace, msg.WorkloadName, msg.WorkloadKind)}, + {Name: "Resource", Value: fmt.Sprintf("%s/%s (%s)", msg.ResourceNamespace, msg.ResourceName, msg.ResourceKind)}, + {Name: "Time", Value: msg.Timestamp.Format("2006-01-02 15:04:05 UTC")}, + } + + subtitle := "" + if a.additional != "" { + subtitle = a.additional + } + + return teamsMessage{ + Type: "MessageCard", + Context: "http://schema.org/extensions", + ThemeColor: "0076D7", + Summary: "Reloader triggered reload", + Sections: []teamsSection{ + { + ActivityTitle: "Reloader triggered reload", + ActivitySubtitle: subtitle, + Facts: facts, + }, + }, + } +} diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 6288d5f9..9ce3e0ce 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -140,9 +140,20 @@ type AnnotationConfig struct { // AlertingConfig holds configuration for alerting integrations. type AlertingConfig struct { - SlackWebhookURL string - TeamsWebhookURL string - GChatWebhookURL string + // Enabled enables alerting notifications on reload events. + Enabled bool + + // WebhookURL is the webhook URL to send alerts to. + WebhookURL string + + // Sink determines the alert format: "slack", "teams", "gchat", or "raw" (default). + Sink string + + // Proxy is an optional HTTP proxy for webhook requests. + Proxy string + + // Additional is optional context prepended to alert messages. + Additional string } // LeaderElectionConfig holds configuration for leader election. diff --git a/internal/pkg/controller/deployment_reconciler.go b/internal/pkg/controller/deployment_reconciler.go new file mode 100644 index 00000000..d28272c5 --- /dev/null +++ b/internal/pkg/controller/deployment_reconciler.go @@ -0,0 +1,89 @@ +package controller + +import ( + "context" + + "github.com/go-logr/logr" + "github.com/stakater/Reloader/internal/pkg/config" + "github.com/stakater/Reloader/internal/pkg/reload" + appsv1 "k8s.io/api/apps/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/predicate" +) + +// DeploymentReconciler reconciles Deployment objects to handle pause expiration. +// This reconciler watches for deployments that were paused by Reloader and +// unpauses them when the pause period expires. +type DeploymentReconciler struct { + client.Client + Log logr.Logger + Config *config.Config + PauseHandler *reload.PauseHandler +} + +// Reconcile handles Deployment pause expiration. +func (r *DeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + log := r.Log.WithValues("deployment", req.NamespacedName) + + var deploy appsv1.Deployment + if err := r.Get(ctx, req.NamespacedName, &deploy); err != nil { + if errors.IsNotFound(err) { + return ctrl.Result{}, nil + } + return ctrl.Result{}, err + } + + // Check if this deployment was paused by Reloader + if !r.PauseHandler.IsPausedByReloader(&deploy) { + return ctrl.Result{}, nil + } + + // Check if pause period has expired + expired, remainingTime, err := r.PauseHandler.CheckPauseExpired(&deploy) + if err != nil { + log.Error(err, "Failed to check pause expiration") + return ctrl.Result{}, err + } + + if !expired { + // Still within pause period - requeue to check again + log.V(1).Info("Deployment pause not yet expired", "remaining", remainingTime) + return ctrl.Result{RequeueAfter: remainingTime}, nil + } + + // Pause period has expired - unpause the deployment + log.Info("Unpausing deployment after pause period expired") + r.PauseHandler.ClearPause(&deploy) + + if err := r.Update(ctx, &deploy, client.FieldOwner(FieldManager)); err != nil { + log.Error(err, "Failed to unpause deployment") + return ctrl.Result{}, err + } + + return ctrl.Result{}, nil +} + +// SetupWithManager sets up the DeploymentReconciler with the manager. +func (r *DeploymentReconciler) SetupWithManager(mgr ctrl.Manager) error { + return ctrl.NewControllerManagedBy(mgr). + For(&appsv1.Deployment{}). + WithEventFilter(r.pausedByReloaderPredicate()). + Complete(r) +} + +// pausedByReloaderPredicate returns a predicate that only selects deployments +// that have been paused by Reloader (have the paused-at annotation). +func (r *DeploymentReconciler) pausedByReloaderPredicate() predicate.Predicate { + return predicate.NewPredicateFuncs(func(obj client.Object) bool { + annotations := obj.GetAnnotations() + if annotations == nil { + return false + } + + // Only process if deployment has our paused-at annotation + _, hasPausedAt := annotations[r.Config.Annotations.PausedAt] + return hasPausedAt + }) +} diff --git a/internal/pkg/controller/retry.go b/internal/pkg/controller/retry.go index f81e8e92..f8af3de0 100644 --- a/internal/pkg/controller/retry.go +++ b/internal/pkg/controller/retry.go @@ -2,16 +2,23 @@ package controller import ( "context" + "maps" "github.com/stakater/Reloader/internal/pkg/reload" "github.com/stakater/Reloader/internal/pkg/workload" + batchv1 "k8s.io/api/batch/v1" "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "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. +// For Jobs and CronJobs, special handling is applied: +// - Jobs are deleted and recreated with the same spec +// - CronJobs create a new Job from their template +// For Argo Rollouts, special handling is applied based on the rollout strategy annotation. func UpdateWorkloadWithRetry( ctx context.Context, c client.Client, @@ -22,6 +29,31 @@ func UpdateWorkloadWithRetry( namespace string, hash string, autoReload bool, +) (bool, error) { + // Handle special workload types + switch wl.Kind() { + case workload.KindJob: + return updateJobWithRecreate(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload) + case workload.KindCronJob: + return updateCronJobWithNewJob(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload) + case workload.KindArgoRollout: + return updateArgoRollout(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload) + default: + return updateStandardWorkload(ctx, c, reloadService, wl, resourceName, resourceType, namespace, hash, autoReload) + } +} + +// updateStandardWorkload updates Deployments, DaemonSets, StatefulSets, etc. +func updateStandardWorkload( + 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 @@ -66,3 +98,202 @@ func UpdateWorkloadWithRetry( return updated, err } + +// updateJobWithRecreate deletes the Job and recreates it with the updated spec. +// Jobs are immutable after creation, so we must delete and recreate. +func updateJobWithRecreate( + 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) { + jobWl, ok := wl.(*workload.JobWorkload) + if !ok { + return false, nil + } + + // Apply reload changes to the workload + updated, err := reloadService.ApplyReload( + ctx, + wl, + resourceName, + resourceType, + namespace, + hash, + autoReload, + ) + if err != nil { + return false, err + } + + if !updated { + return false, nil + } + + oldJob := jobWl.GetJob() + newJob := oldJob.DeepCopy() + + // Delete the old job with background propagation + policy := metav1.DeletePropagationBackground + if err := c.Delete(ctx, oldJob, &client.DeleteOptions{ + PropagationPolicy: &policy, + }); err != nil { + if !errors.IsNotFound(err) { + return false, err + } + } + + // Clear fields that should not be specified when creating a new Job + newJob.ResourceVersion = "" + newJob.UID = "" + newJob.CreationTimestamp = metav1.Time{} + newJob.Status = batchv1.JobStatus{} + + // Remove problematic labels that are auto-generated + delete(newJob.Spec.Template.Labels, "controller-uid") + delete(newJob.Spec.Template.Labels, batchv1.ControllerUidLabel) + delete(newJob.Spec.Template.Labels, batchv1.JobNameLabel) + delete(newJob.Spec.Template.Labels, "job-name") + + // Remove the selector to allow it to be auto-generated + newJob.Spec.Selector = nil + + // Create the new job with same spec + if err := c.Create(ctx, newJob, client.FieldOwner(FieldManager)); err != nil { + return false, err + } + + return true, nil +} + +// updateCronJobWithNewJob creates a new Job from the CronJob's template. +// CronJobs don't get updated directly; instead, a new Job is triggered. +func updateCronJobWithNewJob( + 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) { + cronJobWl, ok := wl.(*workload.CronJobWorkload) + if !ok { + return false, nil + } + + // Apply reload changes to get the updated spec + updated, err := reloadService.ApplyReload( + ctx, + wl, + resourceName, + resourceType, + namespace, + hash, + autoReload, + ) + if err != nil { + return false, err + } + + if !updated { + return false, nil + } + + cronJob := cronJobWl.GetCronJob() + + // Build annotations for the new Job + annotations := make(map[string]string) + annotations["cronjob.kubernetes.io/instantiate"] = "manual" + maps.Copy(annotations, cronJob.Spec.JobTemplate.Annotations) + + // Create a new Job from the CronJob template + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + GenerateName: cronJob.Name + "-", + Namespace: cronJob.Namespace, + Annotations: annotations, + Labels: cronJob.Spec.JobTemplate.Labels, + OwnerReferences: []metav1.OwnerReference{ + *metav1.NewControllerRef(cronJob, batchv1.SchemeGroupVersion.WithKind("CronJob")), + }, + }, + Spec: cronJob.Spec.JobTemplate.Spec, + } + + if err := c.Create(ctx, job, client.FieldOwner(FieldManager)); err != nil { + return false, err + } + + return true, nil +} + +// updateArgoRollout updates an Argo Rollout using its custom Update method. +// This handles the rollout strategy annotation to determine whether to do +// a standard rollout or set the restartAt field. +func updateArgoRollout( + 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) { + rolloutWl, ok := wl.(*workload.RolloutWorkload) + if !ok { + return false, nil + } + + 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 := rolloutWl.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 + } + + // Use the RolloutWorkload's Update method which handles the rollout strategy + return rolloutWl.Update(ctx, c) + }) + + return updated, err +} diff --git a/internal/pkg/reload/pause.go b/internal/pkg/reload/pause.go new file mode 100644 index 00000000..1d00d313 --- /dev/null +++ b/internal/pkg/reload/pause.go @@ -0,0 +1,130 @@ +package reload + +import ( + "fmt" + "time" + + "github.com/stakater/Reloader/internal/pkg/config" + "github.com/stakater/Reloader/internal/pkg/workload" + appsv1 "k8s.io/api/apps/v1" +) + +// PauseHandler handles pause deployment logic. +type PauseHandler struct { + cfg *config.Config +} + +// NewPauseHandler creates a new PauseHandler. +func NewPauseHandler(cfg *config.Config) *PauseHandler { + return &PauseHandler{cfg: cfg} +} + +// ShouldPause checks if a deployment should be paused after reload. +func (h *PauseHandler) ShouldPause(wl workload.WorkloadAccessor) bool { + if wl.Kind() != workload.KindDeployment { + return false + } + + annotations := wl.GetAnnotations() + if annotations == nil { + return false + } + + pausePeriod := annotations[h.cfg.Annotations.PausePeriod] + return pausePeriod != "" +} + +// GetPausePeriod returns the configured pause period for a workload. +func (h *PauseHandler) GetPausePeriod(wl workload.WorkloadAccessor) (time.Duration, error) { + annotations := wl.GetAnnotations() + if annotations == nil { + return 0, fmt.Errorf("no annotations on workload") + } + + pausePeriodStr := annotations[h.cfg.Annotations.PausePeriod] + if pausePeriodStr == "" { + return 0, fmt.Errorf("no pause period annotation") + } + + return time.ParseDuration(pausePeriodStr) +} + +// ApplyPause pauses a deployment and sets the paused-at annotation. +func (h *PauseHandler) ApplyPause(wl workload.WorkloadAccessor) error { + deployWl, ok := wl.(*workload.DeploymentWorkload) + if !ok { + return fmt.Errorf("workload is not a deployment") + } + + deploy := deployWl.GetDeployment() + + // Set paused flag + deploy.Spec.Paused = true + + // Set paused-at annotation + if deploy.Annotations == nil { + deploy.Annotations = make(map[string]string) + } + deploy.Annotations[h.cfg.Annotations.PausedAt] = time.Now().UTC().Format(time.RFC3339) + + return nil +} + +// CheckPauseExpired checks if the pause period has expired for a deployment. +func (h *PauseHandler) CheckPauseExpired(deploy *appsv1.Deployment) (expired bool, remainingTime time.Duration, err error) { + annotations := deploy.GetAnnotations() + if annotations == nil { + return false, 0, fmt.Errorf("no annotations on deployment") + } + + pausePeriodStr := annotations[h.cfg.Annotations.PausePeriod] + if pausePeriodStr == "" { + return false, 0, fmt.Errorf("no pause period annotation") + } + + pausedAtStr := annotations[h.cfg.Annotations.PausedAt] + if pausedAtStr == "" { + return false, 0, fmt.Errorf("no paused-at annotation") + } + + pausePeriod, err := time.ParseDuration(pausePeriodStr) + if err != nil { + return false, 0, fmt.Errorf("invalid pause period %q: %w", pausePeriodStr, err) + } + + pausedAt, err := time.Parse(time.RFC3339, pausedAtStr) + if err != nil { + return false, 0, fmt.Errorf("invalid paused-at time %q: %w", pausedAtStr, err) + } + + elapsed := time.Since(pausedAt) + if elapsed >= pausePeriod { + return true, 0, nil + } + + return false, pausePeriod - elapsed, nil +} + +// ClearPause removes the pause from a deployment. +func (h *PauseHandler) ClearPause(deploy *appsv1.Deployment) { + deploy.Spec.Paused = false + delete(deploy.Annotations, h.cfg.Annotations.PausedAt) + // Keep pause-period annotation (user's config) +} + +// IsPausedByReloader checks if a deployment was paused by Reloader. +func (h *PauseHandler) IsPausedByReloader(deploy *appsv1.Deployment) bool { + if !deploy.Spec.Paused { + return false + } + + annotations := deploy.GetAnnotations() + if annotations == nil { + return false + } + + _, hasPausedAt := annotations[h.cfg.Annotations.PausedAt] + _, hasPausePeriod := annotations[h.cfg.Annotations.PausePeriod] + + return hasPausedAt && hasPausePeriod +} diff --git a/internal/pkg/reload/pause_test.go b/internal/pkg/reload/pause_test.go new file mode 100644 index 00000000..9c7992d0 --- /dev/null +++ b/internal/pkg/reload/pause_test.go @@ -0,0 +1,327 @@ +package reload + +import ( + "testing" + "time" + + "github.com/stakater/Reloader/internal/pkg/config" + "github.com/stakater/Reloader/internal/pkg/workload" + appsv1 "k8s.io/api/apps/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestPauseHandler_ShouldPause(t *testing.T) { + cfg := config.NewDefault() + handler := NewPauseHandler(cfg) + + tests := []struct { + name string + workload workload.WorkloadAccessor + want bool + }{ + { + name: "deployment with pause period", + workload: workload.NewDeploymentWorkload(&appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "5m", + }, + }, + }), + want: true, + }, + { + name: "deployment without pause period", + workload: workload.NewDeploymentWorkload(&appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{}, + }), + want: false, + }, + { + name: "daemonset with pause period (ignored)", + workload: workload.NewDaemonSetWorkload(&appsv1.DaemonSet{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "5m", + }, + }, + }), + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := handler.ShouldPause(tt.workload) + if got != tt.want { + t.Errorf("ShouldPause() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestPauseHandler_GetPausePeriod(t *testing.T) { + cfg := config.NewDefault() + handler := NewPauseHandler(cfg) + + tests := []struct { + name string + workload workload.WorkloadAccessor + wantPeriod time.Duration + wantErr bool + }{ + { + name: "valid pause period", + workload: workload.NewDeploymentWorkload(&appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "5m", + }, + }, + }), + wantPeriod: 5 * time.Minute, + wantErr: false, + }, + { + name: "invalid pause period", + workload: workload.NewDeploymentWorkload(&appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "invalid", + }, + }, + }), + wantErr: true, + }, + { + name: "no pause period annotation", + workload: workload.NewDeploymentWorkload(&appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{}, + }), + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := handler.GetPausePeriod(tt.workload) + if (err != nil) != tt.wantErr { + t.Errorf("GetPausePeriod() error = %v, wantErr %v", err, tt.wantErr) + return + } + if !tt.wantErr && got != tt.wantPeriod { + t.Errorf("GetPausePeriod() = %v, want %v", got, tt.wantPeriod) + } + }) + } +} + +func TestPauseHandler_ApplyPause(t *testing.T) { + cfg := config.NewDefault() + handler := NewPauseHandler(cfg) + + deploy := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-deploy", + }, + Spec: appsv1.DeploymentSpec{ + Paused: false, + }, + } + + wl := workload.NewDeploymentWorkload(deploy) + err := handler.ApplyPause(wl) + if err != nil { + t.Fatalf("ApplyPause() error = %v", err) + } + + if !deploy.Spec.Paused { + t.Error("Expected deployment to be paused") + } + + pausedAt := deploy.Annotations[cfg.Annotations.PausedAt] + if pausedAt == "" { + t.Error("Expected paused-at annotation to be set") + } + + // Verify the timestamp is valid + _, err = time.Parse(time.RFC3339, pausedAt) + if err != nil { + t.Errorf("Invalid paused-at timestamp: %v", err) + } +} + +func TestPauseHandler_CheckPauseExpired(t *testing.T) { + cfg := config.NewDefault() + handler := NewPauseHandler(cfg) + + tests := []struct { + name string + deploy *appsv1.Deployment + wantExpired bool + wantErr bool + }{ + { + name: "pause expired", + deploy: &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "1ms", + cfg.Annotations.PausedAt: time.Now().Add(-time.Second).UTC().Format(time.RFC3339), + }, + }, + Spec: appsv1.DeploymentSpec{Paused: true}, + }, + wantExpired: true, + wantErr: false, + }, + { + name: "pause not expired", + deploy: &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "1h", + cfg.Annotations.PausedAt: time.Now().UTC().Format(time.RFC3339), + }, + }, + Spec: appsv1.DeploymentSpec{Paused: true}, + }, + wantExpired: false, + wantErr: false, + }, + { + name: "no paused-at annotation", + deploy: &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "5m", + }, + }, + }, + wantErr: true, + }, + { + name: "invalid pause period", + deploy: &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "invalid", + cfg.Annotations.PausedAt: time.Now().UTC().Format(time.RFC3339), + }, + }, + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + expired, _, err := handler.CheckPauseExpired(tt.deploy) + if (err != nil) != tt.wantErr { + t.Errorf("CheckPauseExpired() error = %v, wantErr %v", err, tt.wantErr) + return + } + if !tt.wantErr && expired != tt.wantExpired { + t.Errorf("CheckPauseExpired() expired = %v, want %v", expired, tt.wantExpired) + } + }) + } +} + +func TestPauseHandler_ClearPause(t *testing.T) { + cfg := config.NewDefault() + handler := NewPauseHandler(cfg) + + deploy := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "5m", + cfg.Annotations.PausedAt: time.Now().UTC().Format(time.RFC3339), + }, + }, + Spec: appsv1.DeploymentSpec{ + Paused: true, + }, + } + + handler.ClearPause(deploy) + + if deploy.Spec.Paused { + t.Error("Expected deployment to be unpaused") + } + + if _, exists := deploy.Annotations[cfg.Annotations.PausedAt]; exists { + t.Error("Expected paused-at annotation to be removed") + } + + // Pause period should be preserved (user's config) + if deploy.Annotations[cfg.Annotations.PausePeriod] != "5m" { + t.Error("Expected pause-period annotation to be preserved") + } +} + +func TestPauseHandler_IsPausedByReloader(t *testing.T) { + cfg := config.NewDefault() + handler := NewPauseHandler(cfg) + + tests := []struct { + name string + deploy *appsv1.Deployment + want bool + }{ + { + name: "paused by reloader", + deploy: &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "5m", + cfg.Annotations.PausedAt: time.Now().UTC().Format(time.RFC3339), + }, + }, + Spec: appsv1.DeploymentSpec{Paused: true}, + }, + want: true, + }, + { + name: "paused but not by reloader (no paused-at)", + deploy: &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "5m", + }, + }, + Spec: appsv1.DeploymentSpec{Paused: true}, + }, + want: false, + }, + { + name: "not paused", + deploy: &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + cfg.Annotations.PausePeriod: "5m", + cfg.Annotations.PausedAt: time.Now().UTC().Format(time.RFC3339), + }, + }, + Spec: appsv1.DeploymentSpec{Paused: false}, + }, + want: false, + }, + { + name: "no annotations", + deploy: &appsv1.Deployment{ + Spec: appsv1.DeploymentSpec{Paused: true}, + }, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := handler.IsPausedByReloader(tt.deploy) + if got != tt.want { + t.Errorf("IsPausedByReloader() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/internal/pkg/workload/deployment.go b/internal/pkg/workload/deployment.go index 0ddb5b45..1b5ab5f8 100644 --- a/internal/pkg/workload/deployment.go +++ b/internal/pkg/workload/deployment.go @@ -192,3 +192,8 @@ func (w *DeploymentWorkload) UsesSecret(name string) bool { func (w *DeploymentWorkload) GetOwnerReferences() []metav1.OwnerReference { return w.deployment.OwnerReferences } + +// GetDeployment returns the underlying Deployment for special handling. +func (w *DeploymentWorkload) GetDeployment() *appsv1.Deployment { + return w.deployment +} diff --git a/internal/pkg/workload/registry.go b/internal/pkg/workload/registry.go index 55f55b2d..8525d562 100644 --- a/internal/pkg/workload/registry.go +++ b/internal/pkg/workload/registry.go @@ -3,6 +3,7 @@ package workload import ( "fmt" + argorolloutv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1" appsv1 "k8s.io/api/apps/v1" batchv1 "k8s.io/api/batch/v1" "sigs.k8s.io/controller-runtime/pkg/client" @@ -48,6 +49,11 @@ func (r *Registry) FromObject(obj client.Object) (WorkloadAccessor, error) { return NewJobWorkload(o), nil case *batchv1.CronJob: return NewCronJobWorkload(o), nil + case *argorolloutv1alpha1.Rollout: + if !r.argoRolloutsEnabled { + return nil, fmt.Errorf("Argo Rollouts support is not enabled") + } + return NewRolloutWorkload(o), nil default: return nil, fmt.Errorf("unsupported object type: %T", obj) } diff --git a/internal/pkg/workload/rollout.go b/internal/pkg/workload/rollout.go new file mode 100644 index 00000000..7ea9643d --- /dev/null +++ b/internal/pkg/workload/rollout.go @@ -0,0 +1,274 @@ +package workload + +import ( + "context" + "fmt" + "time" + + argorolloutv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// RolloutStrategy defines how Argo Rollouts are updated. +type RolloutStrategy string + +const ( + // RolloutStrategyRollout performs a standard rollout update. + RolloutStrategyRollout RolloutStrategy = "rollout" + + // RolloutStrategyRestart sets the restartAt field to trigger a restart. + RolloutStrategyRestart RolloutStrategy = "restart" +) + +// RolloutStrategyAnnotation is the annotation key for specifying the rollout strategy. +const RolloutStrategyAnnotation = "reloader.stakater.com/rollout-strategy" + +// RolloutWorkload wraps an Argo Rollout. +type RolloutWorkload struct { + rollout *argorolloutv1alpha1.Rollout +} + +// NewRolloutWorkload creates a new RolloutWorkload. +func NewRolloutWorkload(r *argorolloutv1alpha1.Rollout) *RolloutWorkload { + return &RolloutWorkload{rollout: r} +} + +// Ensure RolloutWorkload implements WorkloadAccessor. +var _ WorkloadAccessor = (*RolloutWorkload)(nil) + +func (w *RolloutWorkload) Kind() Kind { + return KindArgoRollout +} + +func (w *RolloutWorkload) GetObject() client.Object { + return w.rollout +} + +func (w *RolloutWorkload) GetName() string { + return w.rollout.Name +} + +func (w *RolloutWorkload) GetNamespace() string { + return w.rollout.Namespace +} + +func (w *RolloutWorkload) GetAnnotations() map[string]string { + return w.rollout.Annotations +} + +func (w *RolloutWorkload) GetPodTemplateAnnotations() map[string]string { + if w.rollout.Spec.Template.Annotations == nil { + w.rollout.Spec.Template.Annotations = make(map[string]string) + } + return w.rollout.Spec.Template.Annotations +} + +func (w *RolloutWorkload) SetPodTemplateAnnotation(key, value string) { + if w.rollout.Spec.Template.Annotations == nil { + w.rollout.Spec.Template.Annotations = make(map[string]string) + } + w.rollout.Spec.Template.Annotations[key] = value +} + +func (w *RolloutWorkload) GetContainers() []corev1.Container { + return w.rollout.Spec.Template.Spec.Containers +} + +func (w *RolloutWorkload) SetContainers(containers []corev1.Container) { + w.rollout.Spec.Template.Spec.Containers = containers +} + +func (w *RolloutWorkload) GetInitContainers() []corev1.Container { + return w.rollout.Spec.Template.Spec.InitContainers +} + +func (w *RolloutWorkload) SetInitContainers(containers []corev1.Container) { + w.rollout.Spec.Template.Spec.InitContainers = containers +} + +func (w *RolloutWorkload) GetVolumes() []corev1.Volume { + return w.rollout.Spec.Template.Spec.Volumes +} + +// Update updates the Rollout. It uses the rollout strategy annotation to determine +// whether to do a standard rollout or set the restartAt field. +func (w *RolloutWorkload) Update(ctx context.Context, c client.Client) error { + strategy := w.getStrategy() + switch strategy { + case RolloutStrategyRestart: + // Use merge patch to set restartAt field + restartAt := metav1.NewTime(time.Now()) + w.rollout.Spec.RestartAt = &restartAt + } + // For both strategies, we update the rollout (annotations have already been set) + return c.Update(ctx, w.rollout) +} + +// getStrategy returns the rollout strategy from the annotation. +func (w *RolloutWorkload) getStrategy() RolloutStrategy { + annotations := w.rollout.GetAnnotations() + if annotations == nil { + return RolloutStrategyRollout + } + strategy := annotations[RolloutStrategyAnnotation] + switch RolloutStrategy(strategy) { + case RolloutStrategyRestart: + return RolloutStrategyRestart + default: + return RolloutStrategyRollout + } +} + +func (w *RolloutWorkload) DeepCopy() Workload { + return &RolloutWorkload{rollout: w.rollout.DeepCopy()} +} + +func (w *RolloutWorkload) GetEnvFromSources() []corev1.EnvFromSource { + var sources []corev1.EnvFromSource + for _, container := range w.rollout.Spec.Template.Spec.Containers { + sources = append(sources, container.EnvFrom...) + } + for _, container := range w.rollout.Spec.Template.Spec.InitContainers { + sources = append(sources, container.EnvFrom...) + } + return sources +} + +func (w *RolloutWorkload) UsesConfigMap(name string) bool { + spec := &w.rollout.Spec.Template.Spec + + // Check volumes + for _, vol := range spec.Volumes { + if vol.ConfigMap != nil && vol.ConfigMap.Name == name { + return true + } + if vol.Projected != nil { + for _, source := range vol.Projected.Sources { + if source.ConfigMap != nil && source.ConfigMap.Name == name { + return true + } + } + } + } + + // Check containers + for _, container := range spec.Containers { + for _, envFrom := range container.EnvFrom { + if envFrom.ConfigMapRef != nil && envFrom.ConfigMapRef.Name == name { + return true + } + } + for _, env := range container.Env { + if env.ValueFrom != nil && env.ValueFrom.ConfigMapKeyRef != nil && env.ValueFrom.ConfigMapKeyRef.Name == name { + return true + } + } + } + + // Check init containers + for _, container := range spec.InitContainers { + for _, envFrom := range container.EnvFrom { + if envFrom.ConfigMapRef != nil && envFrom.ConfigMapRef.Name == name { + return true + } + } + for _, env := range container.Env { + if env.ValueFrom != nil && env.ValueFrom.ConfigMapKeyRef != nil && env.ValueFrom.ConfigMapKeyRef.Name == name { + return true + } + } + } + + return false +} + +func (w *RolloutWorkload) UsesSecret(name string) bool { + spec := &w.rollout.Spec.Template.Spec + + // Check volumes + for _, vol := range spec.Volumes { + if vol.Secret != nil && vol.Secret.SecretName == name { + return true + } + if vol.Projected != nil { + for _, source := range vol.Projected.Sources { + if source.Secret != nil && source.Secret.Name == name { + return true + } + } + } + } + + // Check containers + for _, container := range spec.Containers { + for _, envFrom := range container.EnvFrom { + if envFrom.SecretRef != nil && envFrom.SecretRef.Name == name { + return true + } + } + for _, env := range container.Env { + if env.ValueFrom != nil && env.ValueFrom.SecretKeyRef != nil && env.ValueFrom.SecretKeyRef.Name == name { + return true + } + } + } + + // Check init containers + for _, container := range spec.InitContainers { + for _, envFrom := range container.EnvFrom { + if envFrom.SecretRef != nil && envFrom.SecretRef.Name == name { + return true + } + } + for _, env := range container.Env { + if env.ValueFrom != nil && env.ValueFrom.SecretKeyRef != nil && env.ValueFrom.SecretKeyRef.Name == name { + return true + } + } + } + + return false +} + +func (w *RolloutWorkload) GetOwnerReferences() []metav1.OwnerReference { + return w.rollout.OwnerReferences +} + +// GetRollout returns the underlying Rollout for special handling. +func (w *RolloutWorkload) GetRollout() *argorolloutv1alpha1.Rollout { + return w.rollout +} + +// GetStrategy returns the configured rollout strategy. +func (w *RolloutWorkload) GetStrategy() RolloutStrategy { + return w.getStrategy() +} + +// String returns a string representation of the strategy. +func (s RolloutStrategy) String() string { + return string(s) +} + +// ToRolloutStrategy converts a string to RolloutStrategy. +func ToRolloutStrategy(s string) RolloutStrategy { + switch RolloutStrategy(s) { + case RolloutStrategyRestart: + return RolloutStrategyRestart + case RolloutStrategyRollout: + return RolloutStrategyRollout + default: + return RolloutStrategyRollout + } +} + +// Validate checks if the rollout strategy is valid. +func (s RolloutStrategy) Validate() error { + switch s { + case RolloutStrategyRollout, RolloutStrategyRestart: + return nil + default: + return fmt.Errorf("invalid rollout strategy: %s", s) + } +} diff --git a/internal/pkg/workload/workload_test.go b/internal/pkg/workload/workload_test.go index b7cb2510..b616e0c6 100644 --- a/internal/pkg/workload/workload_test.go +++ b/internal/pkg/workload/workload_test.go @@ -3,6 +3,7 @@ package workload import ( "testing" + argorolloutv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -691,4 +692,226 @@ func TestWorkloadInterface(t *testing.T) { var _ WorkloadAccessor = (*DeploymentWorkload)(nil) var _ WorkloadAccessor = (*DaemonSetWorkload)(nil) var _ WorkloadAccessor = (*StatefulSetWorkload)(nil) + var _ WorkloadAccessor = (*RolloutWorkload)(nil) +} + +// RolloutWorkload tests +func TestRolloutWorkload_BasicGetters(t *testing.T) { + rollout := &argorolloutv1alpha1.Rollout{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-rollout", + Namespace: "test-ns", + Annotations: map[string]string{ + "key": "value", + }, + }, + } + + w := NewRolloutWorkload(rollout) + + if w.Kind() != KindArgoRollout { + t.Errorf("Kind() = %v, want %v", w.Kind(), KindArgoRollout) + } + if w.GetName() != "test-rollout" { + t.Errorf("GetName() = %v, want test-rollout", w.GetName()) + } + if w.GetNamespace() != "test-ns" { + t.Errorf("GetNamespace() = %v, want test-ns", w.GetNamespace()) + } + if w.GetAnnotations()["key"] != "value" { + t.Errorf("GetAnnotations()[key] = %v, want value", w.GetAnnotations()["key"]) + } + if w.GetObject() != rollout { + t.Error("GetObject() should return the underlying rollout") + } +} + +func TestRolloutWorkload_PodTemplateAnnotations(t *testing.T) { + rollout := &argorolloutv1alpha1.Rollout{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + Spec: argorolloutv1alpha1.RolloutSpec{ + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + "existing": "annotation", + }, + }, + }, + }, + } + + w := NewRolloutWorkload(rollout) + + // Test get + annotations := w.GetPodTemplateAnnotations() + if annotations["existing"] != "annotation" { + t.Errorf("GetPodTemplateAnnotations()[existing] = %v, want annotation", annotations["existing"]) + } + + // Test set + w.SetPodTemplateAnnotation("new-key", "new-value") + if w.GetPodTemplateAnnotations()["new-key"] != "new-value" { + t.Error("SetPodTemplateAnnotation should add new annotation") + } +} + +func TestRolloutWorkload_GetStrategy_Default(t *testing.T) { + rollout := &argorolloutv1alpha1.Rollout{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + } + + w := NewRolloutWorkload(rollout) + + if w.GetStrategy() != RolloutStrategyRollout { + t.Errorf("GetStrategy() = %v, want %v (default)", w.GetStrategy(), RolloutStrategyRollout) + } +} + +func TestRolloutWorkload_GetStrategy_Restart(t *testing.T) { + rollout := &argorolloutv1alpha1.Rollout{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test", + Annotations: map[string]string{ + RolloutStrategyAnnotation: "restart", + }, + }, + } + + w := NewRolloutWorkload(rollout) + + if w.GetStrategy() != RolloutStrategyRestart { + t.Errorf("GetStrategy() = %v, want %v", w.GetStrategy(), RolloutStrategyRestart) + } +} + +func TestRolloutWorkload_UsesConfigMap_Volume(t *testing.T) { + rollout := &argorolloutv1alpha1.Rollout{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + Spec: argorolloutv1alpha1.RolloutSpec{ + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Volumes: []corev1.Volume{ + { + Name: "config-vol", + VolumeSource: corev1.VolumeSource{ + ConfigMap: &corev1.ConfigMapVolumeSource{ + LocalObjectReference: corev1.LocalObjectReference{ + Name: "rollout-config", + }, + }, + }, + }, + }, + }, + }, + }, + } + + w := NewRolloutWorkload(rollout) + + if !w.UsesConfigMap("rollout-config") { + t.Error("Rollout UsesConfigMap should return true for ConfigMap volume") + } + if w.UsesConfigMap("other-config") { + t.Error("Rollout UsesConfigMap should return false for non-existent ConfigMap") + } +} + +func TestRolloutWorkload_UsesSecret_EnvFrom(t *testing.T) { + rollout := &argorolloutv1alpha1.Rollout{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + Spec: argorolloutv1alpha1.RolloutSpec{ + Template: corev1.PodTemplateSpec{ + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Name: "main", + EnvFrom: []corev1.EnvFromSource{ + { + SecretRef: &corev1.SecretEnvSource{ + LocalObjectReference: corev1.LocalObjectReference{ + Name: "rollout-secret", + }, + }, + }, + }, + }, + }, + }, + }, + }, + } + + w := NewRolloutWorkload(rollout) + + if !w.UsesSecret("rollout-secret") { + t.Error("Rollout UsesSecret should return true for Secret envFrom") + } + if w.UsesSecret("other-secret") { + t.Error("Rollout UsesSecret should return false for non-existent Secret") + } +} + +func TestRolloutWorkload_DeepCopy(t *testing.T) { + rollout := &argorolloutv1alpha1.Rollout{ + ObjectMeta: metav1.ObjectMeta{Name: "test"}, + Spec: argorolloutv1alpha1.RolloutSpec{ + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Annotations: map[string]string{ + "original": "value", + }, + }, + }, + }, + } + + w := NewRolloutWorkload(rollout) + copy := w.DeepCopy() + + // Verify copy is independent + w.SetPodTemplateAnnotation("modified", "true") + + copyAnnotations := copy.(*RolloutWorkload).GetPodTemplateAnnotations() + if copyAnnotations["modified"] == "true" { + t.Error("DeepCopy should create independent copy") + } +} + +func TestRolloutStrategy_Validate(t *testing.T) { + tests := []struct { + strategy RolloutStrategy + wantErr bool + }{ + {RolloutStrategyRollout, false}, + {RolloutStrategyRestart, false}, + {RolloutStrategy("invalid"), true}, + {RolloutStrategy(""), true}, + } + + for _, tt := range tests { + err := tt.strategy.Validate() + if (err != nil) != tt.wantErr { + t.Errorf("Validate(%s) error = %v, wantErr %v", tt.strategy, err, tt.wantErr) + } + } +} + +func TestToRolloutStrategy(t *testing.T) { + tests := []struct { + input string + expected RolloutStrategy + }{ + {"rollout", RolloutStrategyRollout}, + {"restart", RolloutStrategyRestart}, + {"invalid", RolloutStrategyRollout}, // defaults to rollout + {"", RolloutStrategyRollout}, // defaults to rollout + } + + for _, tt := range tests { + result := ToRolloutStrategy(tt.input) + if result != tt.expected { + t.Errorf("ToRolloutStrategy(%s) = %v, want %v", tt.input, result, tt.expected) + } + } }