From 3812c2ee9c00a3f4cbff046edb9f7971fd8979c2 Mon Sep 17 00:00:00 2001 From: luantranminh Date: Mon, 26 Oct 2026 16:16:51 +0000 Subject: [PATCH] internal/controller: share a single permanentError across controllers Signed-off-by: luantranminh --- .../controller/atlasdriftcheck_controller.go | 77 ++++++------------- .../atlasdriftcheck_controller_test.go | 6 +- .../atlassecurityscan_controller.go | 42 ++++------ internal/controller/common.go | 15 ++++ 4 files changed, 58 insertions(+), 82 deletions(-) diff --git a/internal/controller/atlasdriftcheck_controller.go b/internal/controller/atlasdriftcheck_controller.go index b8f15e06..6642f27f 100644 --- a/internal/controller/atlasdriftcheck_controller.go +++ b/internal/controller/atlasdriftcheck_controller.go @@ -86,12 +86,6 @@ type ( wd *atlasexec.WorkingDir cli AtlasExec } - // checkError describes how a failed check should be reported. - checkError struct { - reason string - message string - permanent bool - } ) // errTargetBusy means the target is applying a migration. @@ -100,9 +94,6 @@ var errTargetBusy = errors.New("migration apply in progress") // errLocked means another client holds the database lock. var errLocked = errors.New("acquiring database lock") -// Error implements the error interface. -func (e *checkError) Error() string { return e.message } - func NewAtlasDriftCheckReconciler(mgr Manager, _ bool) *AtlasDriftCheckReconciler { return &AtlasDriftCheckReconciler{ Client: mgr.GetClient(), @@ -172,7 +163,7 @@ func (r *AtlasDriftCheckReconciler) reconcile(ctx context.Context, res *dbv1alph defer s.close() interval := cmp.Or(res.Spec.Interval.Duration, defaultDriftInterval) err := s.run(ctx) - e, isCheckErr := errors.AsType[*checkError](err) + perr, isPermanent := errors.AsType[*permanentError](err) switch { case err == nil: // The check already updated the status. @@ -184,12 +175,9 @@ func (r *AtlasDriftCheckReconciler) reconcile(ctx context.Context, res *dbv1alph // Keep the previous result and retry soon. s.log.Info("database is locked by another client, skipping this check") return ctrl.Result{RequeueAfter: r.jitter(busyDriftRetry)} - case isCheckErr && e.permanent: - res.SetCheckFailed(e.reason, e.message, true) + case isPermanent: + res.SetCheckFailed(perr.reason, perr.Error(), true) return ctrl.Result{RequeueAfter: r.jitter(interval)} - case isCheckErr: - res.SetCheckFailed(e.reason, e.message, false) - return ctrl.Result{RequeueAfter: r.jitter(min(interval, transientDriftRetry))} default: res.SetCheckFailed(dbv1alpha1.ReasonCheckFailed, err.Error(), false) return ctrl.Result{RequeueAfter: r.jitter(min(interval, transientDriftRetry))} @@ -232,14 +220,11 @@ func (s *driftRun) loadTarget(ctx context.Context) error { key := types.NamespacedName{Namespace: s.res.Namespace, Name: s.res.Spec.TargetRef.Name} switch err := s.r.Get(ctx, key, target); { case apierrors.IsNotFound(err): - return &checkError{ - reason: dbv1alpha1.ReasonTargetNotFound, - permanent: true, - message: fmt.Sprintf("AtlasMigration %q not found in namespace %q; note that an operator "+ - "installed with --label-selector only sees the targets carrying those labels", key.Name, key.Namespace), - } + return permanent(dbv1alpha1.ReasonTargetNotFound, fmt.Errorf("AtlasMigration %q not found in namespace %q; "+ + "note that an operator installed with --label-selector only sees the targets carrying those labels", + key.Name, key.Namespace)) case err != nil: - return &checkError{reason: dbv1alpha1.ReasonCheckFailed, message: err.Error()} + return err } // A stalled target is still checked because its database may have drifted. if target.IsReconciling() { @@ -252,22 +237,17 @@ func (s *driftRun) loadTarget(ctx context.Context) error { // extractData reads the migration data and checks that the expected state can be resolved. func (s *driftRun) extractData(ctx context.Context) error { data, err := extractMigrationData(ctx, s.r, s.target, s.r.allowCustomConfig) - if err != nil { - return &checkError{ - reason: dbv1alpha1.ReasonCheckFailed, - message: err.Error(), - permanent: !isTransient(err), - } + switch { + case isTransient(err): + return err + case err != nil: + return permanent(dbv1alpha1.ReasonCheckFailed, err) } // A remote directory contains the expected state. A local directory needs a dev database to build it. if !data.hasRemoteDir() && !data.hasConfigRepo() && !data.hasDevURL() { - return &checkError{ - reason: dbv1alpha1.ReasonCheckFailed, - permanent: true, - message: fmt.Sprintf("local migration directory requires a dev database to replay: "+ - "set spec.devURL or spec.devURLFrom on AtlasMigration/%s, or push the directory to the Atlas Registry", - s.target.Name), - } + return permanent(dbv1alpha1.ReasonCheckFailed, fmt.Errorf("local migration directory requires a dev database to replay: "+ + "set spec.devURL or spec.devURLFrom on AtlasMigration/%s, or push the directory to the Atlas Registry", + s.target.Name)) } s.data = data return nil @@ -310,11 +290,8 @@ func (s *driftRun) login(ctx context.Context) error { func (s *driftRun) whoAmI(ctx context.Context) error { _, err := s.cli.WhoAmI(ctx, &atlasexec.WhoAmIParams{Vars: s.data.Vars}) if errors.Is(err, atlasexec.ErrRequireLogin) { - return &checkError{ - reason: dbv1alpha1.ReasonCheckFailed, - permanent: true, - message: fmt.Sprintf("drift checks require Atlas Cloud: set spec.cloud.tokenFrom on AtlasMigration/%s", s.target.Name), - } + return permanent(dbv1alpha1.ReasonCheckFailed, fmt.Errorf( + "drift checks require Atlas Cloud: set spec.cloud.tokenFrom on AtlasMigration/%s", s.target.Name)) } return err } @@ -341,11 +318,8 @@ func (s *driftRun) check(ctx context.Context) error { return s.classifyError(ctx, err) } if len(reports) != 1 { - return &checkError{ - reason: dbv1alpha1.ReasonCheckFailed, - permanent: true, - message: fmt.Sprintf("unexpected number of reports: %d; for_each targets are not supported", len(reports)), - } + return permanent(dbv1alpha1.ReasonCheckFailed, fmt.Errorf( + "unexpected number of reports: %d; for_each targets are not supported", len(reports))) } if rep := reports[0]; rep.Drifted { s.res.SetDrifted(rep, cmp.Or(s.res.Spec.OnDrift, dbv1alpha1.DriftActionReport)) @@ -382,16 +356,13 @@ func (s *driftRun) classifyError(ctx context.Context, err error) error { msg = cmp.Or(strings.TrimSpace(msg), "the drift check could not be completed") switch { case ctx.Err() != nil || errors.Is(err, context.DeadlineExceeded): - return &checkError{ - reason: dbv1alpha1.ReasonCheckFailed, - message: fmt.Sprintf("the drift check timed out: %s", msg), - } + return fmt.Errorf("the drift check timed out: %s", msg) case strings.Contains(msg, errLocked.Error()): return errLocked case errors.Is(err, atlasexec.ErrRequireLogin): - return &checkError{reason: dbv1alpha1.ReasonCheckFailed, message: msg, permanent: true} + return permanent(dbv1alpha1.ReasonCheckFailed, errors.New(msg)) case strings.Contains(msg, "no migration history found"): - return &checkError{reason: dbv1alpha1.ReasonNoMigrationHistory, message: msg, permanent: true} + return permanent(dbv1alpha1.ReasonNoMigrationHistory, errors.New(msg)) } // These errors will not change by retrying. for _, p := range []string{ @@ -401,10 +372,10 @@ func (s *driftRun) classifyError(ctx context.Context, err error) error { "is available only to Atlas Pro users", } { if strings.Contains(msg, p) { - return &checkError{reason: dbv1alpha1.ReasonCheckFailed, message: msg, permanent: true} + return permanent(dbv1alpha1.ReasonCheckFailed, errors.New(msg)) } } - return &checkError{reason: dbv1alpha1.ReasonCheckFailed, message: msg} + return errors.New(msg) } // recordTransitions emits events when drift or a check failure changes. diff --git a/internal/controller/atlasdriftcheck_controller_test.go b/internal/controller/atlasdriftcheck_controller_test.go index edae202a..67e301bf 100644 --- a/internal/controller/atlasdriftcheck_controller_test.go +++ b/internal/controller/atlasdriftcheck_controller_test.go @@ -353,9 +353,9 @@ func TestDriftCheck_Timeout(t *testing.T) { require.Equal(t, "the drift check timed out: signal: killed", driftCond(t, res, "Ready").Message) } -// TestDriftCheck_StepFailure covers a step that fails with a bare error rather -// than a *checkError: it is reported as a transient failure, so an unexpected -// failure keeps retrying instead of parking the check on a verdict. +// TestDriftCheck_StepFailure covers a step that fails with an error not marked +// permanent: it is reported as a transient failure, so an unexpected failure +// keeps retrying instead of parking the check on a verdict. func TestDriftCheck_StepFailure(t *testing.T) { var ( check = driftCheckObj() diff --git a/internal/controller/atlassecurityscan_controller.go b/internal/controller/atlassecurityscan_controller.go index 775c2ca3..47d94c5f 100644 --- a/internal/controller/atlassecurityscan_controller.go +++ b/internal/controller/atlassecurityscan_controller.go @@ -117,14 +117,8 @@ type ( waivers []string hash string } - // permanentError is an input no scan can fix. The resource stalls on it. - permanentError struct { - reason, message string - } ) -func (e *permanentError) Error() string { return e.message } - func NewAtlasSecurityScanReconciler(mgr Manager) *AtlasSecurityScanReconciler { return &AtlasSecurityScanReconciler{ Client: mgr.GetClient(), @@ -176,10 +170,10 @@ func (r *AtlasSecurityScanReconciler) Reconcile(ctx context.Context, req ctrl.Re } sched, err := r.validate(res, now) if perr, ok := errors.AsType[*permanentError](err); ok { - closeOpenAttempt(res, now, perr.message) - res.SetStalled(perr.reason, perr.message) + closeOpenAttempt(res, now, perr.Error()) + res.SetStalled(perr.reason, perr.Error()) res.Status.NextScheduleTime = nil - r.recorder.Event(res, corev1.EventTypeWarning, perr.reason, perr.message) + r.recorder.Event(res, corev1.EventTypeWarning, perr.reason, perr.Error()) return ctrl.Result{}, nil } snap, err := r.snapshot(ctx, res, sched, now) @@ -219,7 +213,7 @@ func (r *AtlasSecurityScanReconciler) Reconcile(ctx context.Context, req ctrl.Re log.Info("scanning the database for security issues", "trigger", trigger, "by", by) data, err := r.extractData(ctx, res) if perr, ok := errors.AsType[*permanentError](err); ok { - return r.stall(res, sched, snap, perr.reason, perr.message), nil + return r.stall(res, sched, snap, perr.reason, perr.Error()), nil } if err != nil { return r.fail(ctx, res, sched, snap, dbv1alpha1.ReasonReadingInputs, err), nil @@ -415,35 +409,35 @@ func (r *AtlasSecurityScanReconciler) validate(res *dbv1alpha1.AtlasSecurityScan s := res.Spec tz := cmp.Or(s.TimeZone, "UTC") if _, err := time.LoadLocation(tz); err != nil { - return nil, &permanentError{dbv1alpha1.ReasonInvalidTimeZone, "unknown time zone"} + return nil, permanent(dbv1alpha1.ReasonInvalidTimeZone, errors.New("unknown time zone")) } var sched cron.Schedule if s.Schedule != "" { switch { case strings.HasPrefix(s.Schedule, "TZ="), strings.HasPrefix(s.Schedule, "CRON_TZ="): - return nil, &permanentError{dbv1alpha1.ReasonInvalidSchedule, "use spec.timeZone instead of a TZ= prefix"} + return nil, permanent(dbv1alpha1.ReasonInvalidSchedule, errors.New("use spec.timeZone instead of a TZ= prefix")) case strings.HasPrefix(s.Schedule, "@every"): - return nil, &permanentError{dbv1alpha1.ReasonInvalidSchedule, "@every is interval-based and drifts"} + return nil, permanent(dbv1alpha1.ReasonInvalidSchedule, errors.New("@every is interval-based and drifts")) } // The zone goes through the parser's own prefix, which spec.schedule may not carry. parsed, err := cron.ParseStandard("CRON_TZ=" + tz + " " + s.Schedule) if err != nil { - return nil, &permanentError{dbv1alpha1.ReasonInvalidSchedule, "schedule does not parse"} + return nil, permanent(dbv1alpha1.ReasonInvalidSchedule, errors.New("schedule does not parse")) } sched = parsed if sched.Next(now).IsZero() { - return nil, &permanentError{dbv1alpha1.ReasonInvalidSchedule, "schedule does not fire"} + return nil, permanent(dbv1alpha1.ReasonInvalidSchedule, errors.New("schedule does not fire")) } } hasTarget := s.URL != "" || s.URLFrom.SecretKeyRef != nil || s.Credentials.Scheme != "" hasConfig := s.Config != "" || s.ConfigFrom.SecretKeyRef != nil switch { case !hasTarget && !hasConfig: - return nil, &permanentError{dbv1alpha1.ReasonInvalidTarget, "no target database or project configuration"} + return nil, permanent(dbv1alpha1.ReasonInvalidTarget, errors.New("no target database or project configuration")) case hasConfig && !r.allowCustomConfig: - return nil, &permanentError{dbv1alpha1.ReasonInvalidTarget, "install the operator with allowCustomConfig=true to use a custom atlas.hcl"} + return nil, permanent(dbv1alpha1.ReasonInvalidTarget, errors.New("install the operator with allowCustomConfig=true to use a custom atlas.hcl")) case hasConfig && s.EnvName == "": - return nil, &permanentError{dbv1alpha1.ReasonInvalidTarget, "envName must be set when using a custom atlas.hcl"} + return nil, permanent(dbv1alpha1.ReasonInvalidTarget, errors.New("envName must be set when using a custom atlas.hcl")) } return sched, nil } @@ -803,10 +797,8 @@ func checkPolicy(cfg *hclwrite.File, envName string) error { // never sent to the Security Graph: the scan then reports nothing and // still succeeds, which no failure path would catch. if b.Body().GetAttribute("exclude") != nil { - return &permanentError{ - dbv1alpha1.ReasonInvalidTarget, - "exclude in the custom atlas.hcl would hide extensions from the scan", - } + return permanent(dbv1alpha1.ReasonInvalidTarget, + errors.New("exclude in the custom atlas.hcl would hide extensions from the scan")) } for _, n := range b.Body().Blocks() { if n.Type() != "security" { @@ -827,10 +819,8 @@ func checkPolicy(cfg *hclwrite.File, envName string) error { func checkPolicyBlock(b *hclwrite.Block, path string) error { for _, name := range policyAttrs[path] { if b.Body().GetAttribute(name) != nil { - return &permanentError{ - dbv1alpha1.ReasonInvalidTarget, - fmt.Sprintf("%s.%s is set by spec.policy and must not appear in the custom atlas.hcl", path, name), - } + return permanent(dbv1alpha1.ReasonInvalidTarget, + fmt.Errorf("%s.%s is set by spec.policy and must not appear in the custom atlas.hcl", path, name)) } } for _, n := range b.Body().Blocks() { diff --git a/internal/controller/common.go b/internal/controller/common.go index 033ad21e..6ac76b5f 100644 --- a/internal/controller/common.go +++ b/internal/controller/common.go @@ -177,6 +177,21 @@ func isTransient(err error) bool { return errors.As(err, &t) } +// permanentError is an error that retrying cannot fix, such as an invalid +// spec. Controllers stall the resource with its reason. +type permanentError struct { + reason string + err error +} + +func (p *permanentError) Error() string { return p.err.Error() } +func (p *permanentError) Unwrap() error { return p.err } + +// permanent wraps an error to indicate that retrying cannot fix it. +func permanent(reason string, err error) error { + return &permanentError{reason: reason, err: err} +} + // result returns a ctrl.Result and an error. If the error is transient, the // task will be requeued after seconds defined by the error. // Permanent errors are not returned as errors because they cause