From 94972b84fa324796472c12df6ccfb8dacd8ec6c5 Mon Sep 17 00:00:00 2001 From: Jeff Ortel Date: Thu, 18 Apr 2024 09:25:17 -0700 Subject: [PATCH 1/4] Fix handling of quota exceeded. Signed-off-by: Jeff Ortel --- task/manager.go | 69 +++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 56 insertions(+), 13 deletions(-) diff --git a/task/manager.go b/task/manager.go index c2e66fa1d..9ae945726 100644 --- a/task/manager.go +++ b/task/manager.go @@ -6,6 +6,7 @@ import ( "fmt" "path" "strconv" + "strings" "time" "github.com/golang-jwt/jwt/v4" @@ -64,6 +65,36 @@ func (e *AddonNotFound) Is(err error) (matched bool) { return } +// QuotaExceeded report quota exceeded. +type QuotaExceeded struct { + Reason string +} + +// Match returns true when the error is Forbidden due to quota exceeded. +func (e *QuotaExceeded) Match(err error) (matched bool) { + if k8serr.IsForbidden(err) { + matched = true + e.Reason = err.Error() + for _, s := range []string{"quota", "exceeded"} { + matched = strings.Contains(e.Reason, s) + if !matched { + break + } + } + } + return +} + +func (e *QuotaExceeded) Error() (s string) { + return e.Reason +} + +func (e *QuotaExceeded) Is(err error) (matched bool) { + var inst *QuotaExceeded + matched = errors.As(err, &inst) + return +} + // Manager provides task management. type Manager struct { // DB @@ -146,9 +177,6 @@ func (m *Manager) startReady() { Log.Error(sErr, "") continue } - if ready.Retries == 0 { - metrics.TasksInitiated.Inc() - } rt := Task{ready} err := rt.Run(m.Client) if err != nil { @@ -161,9 +189,12 @@ func (m *Manager) startReady() { Log.Error(err, "") continue } - Log.Info("Task started.", "id", ready.ID) - err = m.DB.Save(ready).Error - Log.Error(err, "") + if ready.State == Pending { + Log.Info("Task started.", "id", ready.ID) + if ready.Retries == 0 { + metrics.TasksInitiated.Inc() + } + } default: // Ignored. // Other states included to support @@ -259,11 +290,17 @@ type Task struct { func (r *Task) Run(client k8s.Client) (err error) { mark := time.Now() defer func() { - if err != nil { - r.Error("Error", err.Error()) - r.Terminated = &mark - r.State = Failed + if err == nil { + return + } + if errors.Is(err, &QuotaExceeded{}) { + Log.V(1).Info(err.Error()) + err = nil + return } + r.Error("Error", err.Error()) + r.Terminated = &mark + r.State = Failed }() addon, err := r.findAddon(client, r.Addon) if err != nil { @@ -288,6 +325,10 @@ func (r *Task) Run(client k8s.Client) (err error) { pod := r.pod(addon, owner, &secret) err = client.Create(context.TODO(), &pod) if err != nil { + qe := &QuotaExceeded{err.Error()} + if qe.Match(err) { + err = qe + } err = liberr.Wrap(err) return } @@ -329,10 +370,12 @@ func (r *Task) Reflect(client k8s.Client) (err error) { pod) if err != nil { if k8serr.IsNotFound(err) { - err = r.Run(client) - } else { - err = liberr.Wrap(err) + r.Pod = "" + r.State = Ready + err = nil + return } + err = liberr.Wrap(err) return } mark := time.Now() From 867ab8fe74d9d5f33538149aa3850bd16fa9500b Mon Sep 17 00:00:00 2001 From: Jeff Ortel Date: Thu, 18 Apr 2024 09:41:42 -0700 Subject: [PATCH 2/4] checkpoint Signed-off-by: Jeff Ortel --- task/manager.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/task/manager.go b/task/manager.go index 9ae945726..1f56f879e 100644 --- a/task/manager.go +++ b/task/manager.go @@ -178,7 +178,7 @@ func (m *Manager) startReady() { continue } rt := Task{ready} - err := rt.Run(m.Client) + started, err := rt.Run(m.Client) if err != nil { if errors.Is(err, &AddonNotFound{}) { ready.Error("Error", err.Error()) @@ -189,7 +189,7 @@ func (m *Manager) startReady() { Log.Error(err, "") continue } - if ready.State == Pending { + if started { Log.Info("Task started.", "id", ready.ID) if ready.Retries == 0 { metrics.TasksInitiated.Inc() @@ -287,7 +287,7 @@ type Task struct { } // Run the specified task. -func (r *Task) Run(client k8s.Client) (err error) { +func (r *Task) Run(client k8s.Client) (started bool, err error) { mark := time.Now() defer func() { if err == nil { @@ -350,6 +350,7 @@ func (r *Task) Run(client k8s.Client) (err error) { err = liberr.Wrap(err) return } + started = true r.Started = &mark r.State = Pending r.Pod = path.Join( From 404de0df0886b4d2aab9e3777f73eebeb8af2e91 Mon Sep 17 00:00:00 2001 From: Jeff Ortel Date: Thu, 18 Apr 2024 12:28:47 -0700 Subject: [PATCH 3/4] checkpoint Signed-off-by: Jeff Ortel --- task/manager.go | 26 +++++++++++++++----------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/task/manager.go b/task/manager.go index 1f56f879e..fd2c55669 100644 --- a/task/manager.go +++ b/task/manager.go @@ -180,15 +180,15 @@ func (m *Manager) startReady() { rt := Task{ready} started, err := rt.Run(m.Client) if err != nil { - if errors.Is(err, &AddonNotFound{}) { - ready.Error("Error", err.Error()) - ready.State = Failed - sErr := m.DB.Save(ready).Error - Log.Error(sErr, "") - } Log.Error(err, "") + ready.Error("Error", err.Error()) + ready.State = Failed + sErr := m.DB.Save(ready).Error + Log.Error(sErr, "") continue } + err = m.DB.Save(ready).Error + Log.Error(err, "") if started { Log.Info("Task started.", "id", ready.ID) if ready.Retries == 0 { @@ -298,9 +298,13 @@ func (r *Task) Run(client k8s.Client) (started bool, err error) { err = nil return } - r.Error("Error", err.Error()) - r.Terminated = &mark - r.State = Failed + if errors.Is(err, &AddonNotFound{}) { + r.Error("Error", err.Error()) + r.Terminated = &mark + r.State = Failed + err = nil + return + } }() addon, err := r.findAddon(client, r.Addon) if err != nil { @@ -374,9 +378,9 @@ func (r *Task) Reflect(client k8s.Client) (err error) { r.Pod = "" r.State = Ready err = nil - return + } else { + err = liberr.Wrap(err) } - err = liberr.Wrap(err) return } mark := time.Now() From ea4932a2886a2156fe12bc73dc7ed83c352c7be0 Mon Sep 17 00:00:00 2001 From: Jeff Ortel Date: Thu, 18 Apr 2024 12:33:42 -0700 Subject: [PATCH 4/4] checkpoint Signed-off-by: Jeff Ortel --- task/manager.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/task/manager.go b/task/manager.go index fd2c55669..96ca6d7f7 100644 --- a/task/manager.go +++ b/task/manager.go @@ -173,8 +173,8 @@ func (m *Manager) startReady() { if m.postpone(ready, list) { ready.State = Postponed Log.Info("Task postponed.", "id", ready.ID) - sErr := m.DB.Save(ready).Error - Log.Error(sErr, "") + err := m.DB.Save(ready).Error + Log.Error(err, "") continue } rt := Task{ready} @@ -183,8 +183,8 @@ func (m *Manager) startReady() { Log.Error(err, "") ready.Error("Error", err.Error()) ready.State = Failed - sErr := m.DB.Save(ready).Error - Log.Error(sErr, "") + err = m.DB.Save(ready).Error + Log.Error(err, "") continue } err = m.DB.Save(ready).Error