fix(backups): warn when a Kafka backupTimeout is rejected

kafkaRunDeadline silently substituted the 30m default for a backupTimeout
that time.ParseDuration rejects or that falls below the one-minute floor,
with no Event, condition or log line. An operator who set an unparseable
or sub-minute value, then raised it, saw every run keep dying at
DeadlineExceeded with nothing naming the discarded parameter.

Return the applied deadline together with a message when the configured
value is rejected, and record it as a BackupTimeoutInvalid Warning Event
on the BackupJob and RestoreJob (the reconcilers already hold a Recorder).
An honoured or unset value stays silent. TestKafkaRunDeadline now asserts
the warning fires for the rejected cases and not otherwise.

Signed-off-by: Andrey Kolkov <androndo@gmail.com>
Assisted-by: LLM
This commit is contained in:
Andrey Kolkov committed 2026-09-14 17:55:48 +04:00
1 parent caae13a60a
commit ca2a6d6429
2 files changed
+38 -17

No files matched your search

@@ -147,14 +147,24 @@ func kafkaStrategyParameters(b *backupsv1alpha1.Backup) map[string]string {
// kafkaRunDeadline resolves the wall-clock bound for a run from its parameters,
// falling back to kafkaDefaultBackupDeadline when unset, unparseable, or below
// the floor.
func kafkaRunDeadline(parameters map[string]string) time.Duration {
if v := parameters[kafkaBackupTimeoutParam]; v != "" {
if d, err := time.ParseDuration(v); err == nil && d >= kafkaMinBackupDeadline {
return d
}
// the floor. When a backupTimeout is set but rejected, the second return value is
// a message naming the rejected value and the deadline actually applied, so the
// silent fallback becomes a Warning Event the caller records; it is empty when the
// value is honoured or unset.
func kafkaRunDeadline(parameters map[string]string) (time.Duration, string) {
v := parameters[kafkaBackupTimeoutParam]
if v == "" {
return kafkaDefaultBackupDeadline, ""
}
d, err := time.ParseDuration(v)
switch {
case err != nil:
return kafkaDefaultBackupDeadline, fmt.Sprintf("backupTimeout %q is not a valid Go duration; applying the default %s", v, kafkaDefaultBackupDeadline)
case d < kafkaMinBackupDeadline:
return kafkaDefaultBackupDeadline, fmt.Sprintf("backupTimeout %q is below the %s floor; applying the default %s", v, kafkaMinBackupDeadline, kafkaDefaultBackupDeadline)
default:
return d, ""
}
return kafkaDefaultBackupDeadline
}
// kafkaRenderContext builds the template context. Unlike the Job strategy it
@@ -263,7 +273,10 @@ func (r *BackupJobReconciler) reconcileKafka(ctx context.Context, j *backupsv1al
return r.markBackupJobFailed(ctx, j, err.Error())
}
deadline := kafkaRunDeadline(resolved.Parameters)
deadline, deadlineWarn := kafkaRunDeadline(resolved.Parameters)
if deadlineWarn != "" && r.Recorder != nil {
r.Recorder.Event(j, corev1.EventTypeWarning, "BackupTimeoutInvalid", deadlineWarn)
}
// First-reconcile bookkeeping (refetch guards against a stale informer
// sliding StartedAt forward). Mirrors reconcileJob.
@@ -545,7 +558,10 @@ func (r *RestoreJobReconciler) reconcileKafkaRestore(ctx context.Context, restor
// The deadline round-trips from backup time via the Backup's parameters, so a
// restore is bounded by the same value the operator set for the backup.
deadline := kafkaRunDeadline(kafkaStrategyParameters(backup))
deadline, deadlineWarn := kafkaRunDeadline(kafkaStrategyParameters(backup))
if deadlineWarn != "" && r.Recorder != nil {
r.Recorder.Event(restoreJob, corev1.EventTypeWarning, "BackupTimeoutInvalid", deadlineWarn)
}
// A Backup with no recorded object can never restore; fail fast, before the
// readiness gate, so it does not wait out the whole deadline and then report a
@@ -551,20 +551,25 @@ func getKafkaJob(t *testing.T, c client.Client, namespace, name string) *batchv1
func TestKafkaRunDeadline(t *testing.T) {
cases := []struct {
name string
params map[string]string
want time.Duration
name string
params map[string]string
want time.Duration
wantWarn bool // a rejected value must surface a non-empty warning
}{
{"unset falls back to default", nil, kafkaDefaultBackupDeadline},
{"valid override honoured", map[string]string{kafkaBackupTimeoutParam: "2h"}, 2 * time.Hour},
{"below floor falls back", map[string]string{kafkaBackupTimeoutParam: "10s"}, kafkaDefaultBackupDeadline},
{"unparseable falls back", map[string]string{kafkaBackupTimeoutParam: "nope"}, kafkaDefaultBackupDeadline},
{"unset falls back to default", nil, kafkaDefaultBackupDeadline, false},
{"valid override honoured", map[string]string{kafkaBackupTimeoutParam: "2h"}, 2 * time.Hour, false},
{"below floor falls back and warns", map[string]string{kafkaBackupTimeoutParam: "10s"}, kafkaDefaultBackupDeadline, true},
{"unparseable falls back and warns", map[string]string{kafkaBackupTimeoutParam: "nope"}, kafkaDefaultBackupDeadline, true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := kafkaRunDeadline(tc.params); got != tc.want {
got, warn := kafkaRunDeadline(tc.params)
if got != tc.want {
t.Fatalf("kafkaRunDeadline(%v) = %s, want %s", tc.params, got, tc.want)
}
if (warn != "") != tc.wantWarn {
t.Fatalf("kafkaRunDeadline(%v) warn = %q, wantWarn %v", tc.params, warn, tc.wantWarn)
}
})
}
}