From 662ac3356eab392024b94a03e367094af58b1035 Mon Sep 17 00:00:00 2001 From: stonezdj Date: Sat, 16 Mar 2024 00:37:34 -0300 Subject: [PATCH] Refine execution and task to avoid pannic when execution has error fixes #20129 --- src/controller/replication/execution.go | 7 ++----- src/controller/retention/controller.go | 7 ++----- src/controller/systemartifact/execution.go | 7 ++----- src/pkg/task/dao/task.go | 22 ++++++++++++++++++--- src/pkg/task/dao/task_test.go | 23 ++++++++++++++++++++++ src/pkg/task/execution.go | 14 +++++++++++++ src/testing/pkg/task/execution_manager.go | 14 +++++++++++++ 7 files changed, 76 insertions(+), 18 deletions(-) diff --git a/src/controller/replication/execution.go b/src/controller/replication/execution.go index f6facc02a0a9..95136d9d8494 100644 --- a/src/controller/replication/execution.go +++ b/src/controller/replication/execution.go @@ -154,11 +154,8 @@ func (c *controller) Start(ctx context.Context, policy *replicationmodel.Policy, func (c *controller) markError(ctx context.Context, executionID int64, err error) { logger := log.GetLogger(ctx) // try to stop the execution first in case that some tasks are already created - if err := c.execMgr.StopAndWait(ctx, executionID, 10*time.Second); err != nil { - logger.Errorf("failed to stop the execution %d: %v", executionID, err) - } - if err := c.execMgr.MarkError(ctx, executionID, err.Error()); err != nil { - logger.Errorf("failed to mark error for the execution %d: %v", executionID, err) + if e := c.execMgr.StopAndWaitWithError(ctx, executionID, 10*time.Second, err); e != nil { + logger.Errorf("failed to stop the execution %d: %v", executionID, e) } } diff --git a/src/controller/retention/controller.go b/src/controller/retention/controller.go index 4ec607a40cb2..bcf4fc4200d1 100644 --- a/src/controller/retention/controller.go +++ b/src/controller/retention/controller.go @@ -280,13 +280,10 @@ func (r *defaultController) TriggerRetentionExec(ctx context.Context, policyID i if num, err := r.launcher.Launch(ctx, p, id, dryRun); err != nil { logger.Errorf("failed to launch the retention jobs, err: %v", err) - if err = r.execMgr.StopAndWait(ctx, id, 10*time.Second); err != nil { - logger.Errorf("failed to stop the retention execution %d: %v", id, err) + if e := r.execMgr.StopAndWaitWithError(ctx, id, 10*time.Second, err); e != nil { + logger.Errorf("failed to stop the retention execution %d: %v", id, e) } - if err = r.execMgr.MarkError(ctx, id, err.Error()); err != nil { - logger.Errorf("failed to mark error for the retention execution %d: %v", id, err) - } } else if num == 0 { // no candidates, mark the execution as done directly if err := r.execMgr.MarkDone(ctx, id, "no resources for retention"); err != nil { diff --git a/src/controller/systemartifact/execution.go b/src/controller/systemartifact/execution.go index c9f6c896d714..7a575e076c00 100644 --- a/src/controller/systemartifact/execution.go +++ b/src/controller/systemartifact/execution.go @@ -119,11 +119,8 @@ func (c *controller) createCleanupTask(ctx context.Context, jobParams job.Parame func (c *controller) markError(ctx context.Context, executionID int64, err error) { // try to stop the execution first in case that some tasks are already created - if err := c.execMgr.StopAndWait(ctx, executionID, 10*time.Second); err != nil { - log.Errorf("failed to stop the execution %d: %v", executionID, err) - } - if err := c.execMgr.MarkError(ctx, executionID, err.Error()); err != nil { - log.Errorf("failed to mark error for the execution %d: %v", executionID, err) + if e := c.execMgr.StopAndWaitWithError(ctx, executionID, 10*time.Second, err); e != nil { + log.Errorf("failed to stop the execution %d: %v", executionID, e) } } diff --git a/src/pkg/task/dao/task.go b/src/pkg/task/dao/task.go index de3c71bd22bf..95f585047b84 100644 --- a/src/pkg/task/dao/task.go +++ b/src/pkg/task/dao/task.go @@ -25,6 +25,8 @@ import ( "github.com/goharbor/harbor/src/lib/log" "github.com/goharbor/harbor/src/lib/orm" "github.com/goharbor/harbor/src/lib/q" + + "github.com/google/uuid" ) // TaskDAO is the data access object interface for task @@ -91,22 +93,36 @@ func (t *taskDAO) List(ctx context.Context, query *q.Query) ([]*Task, error) { return tasks, nil } +func isValidUUID(id string) bool { + if len(id) == 0 { + return false + } + if _, err := uuid.Parse(id); err != nil { + return false + } + return true +} + func (t *taskDAO) ListScanTasksByReportUUID(ctx context.Context, uuid string) ([]*Task, error) { ormer, err := orm.FromContext(ctx) if err != nil { return nil, err } + if !isValidUUID(uuid) { + return nil, errors.BadRequestError(fmt.Errorf("invalid UUID %v", uuid)) + } + tasks := []*Task{} // Due to the limitation of the beego's orm, the SQL cannot be converted by orm framework, // so we can only execute the query by raw SQL, the SQL filters the task contains the report uuid in the column extra_attrs, // consider from performance side which can using indexes to speed up queries. - sql := fmt.Sprintf(`SELECT * FROM task WHERE extra_attrs::jsonb->'report_uuids' @> '["%s"]'`, uuid) - _, err = ormer.Raw(sql).QueryRows(&tasks) + param := fmt.Sprintf(`{"report_uuids":["%s"]}`, uuid) + sql := `SELECT * FROM task WHERE extra_attrs::jsonb @> cast( ? as jsonb )` + _, err = ormer.Raw(sql, param).QueryRows(&tasks) if err != nil { return nil, err } - return tasks, nil } diff --git a/src/pkg/task/dao/task_test.go b/src/pkg/task/dao/task_test.go index aeca41a1ca99..f848a1737eb0 100644 --- a/src/pkg/task/dao/task_test.go +++ b/src/pkg/task/dao/task_test.go @@ -299,6 +299,29 @@ func (t *taskDAOTestSuite) TestExecutionIDsByVendorAndStatus() { defer t.taskDAO.Delete(t.ctx, tid) } +func TestIsValidUUID(t *testing.T) { + tests := []struct { + name string + uuid string + expected bool + }{ + {"Valid UUID", "7f20b1b9-6117-4a2e-820b-e4cc0401f15f", true}, + {"Invalid UUID - Short", "7f20b1b9-6117-4a2e-820b", false}, + {"Invalid UUID - Long", "7f20b1b9-6117-4a2e-820b-e4cc0401f15f-extra", false}, + {"Invalid UUID - Invalid Characters", "7f20b1b9-6117-4z2e-820b-e4cc0401f15f", false}, + {"Empty String", "", false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + result := isValidUUID(test.uuid) + if result != test.expected { + t.Errorf("Expected isValidUUID(%s) to be %t, got %t", test.uuid, test.expected, result) + } + }) + } +} + func TestTaskDAOSuite(t *testing.T) { suite.Run(t, &taskDAOTestSuite{}) } diff --git a/src/pkg/task/execution.go b/src/pkg/task/execution.go index 49fee7f96fd1..2be32fcc52b4 100644 --- a/src/pkg/task/execution.go +++ b/src/pkg/task/execution.go @@ -59,6 +59,10 @@ type ExecutionManager interface { // StopAndWait stops all linked tasks of the specified execution and waits until all tasks are stopped // or get an error StopAndWait(ctx context.Context, id int64, timeout time.Duration) (err error) + + // StopAndWaitWithError calls the StopAndWait if it doesn't return error, then it call MarkError if the origError is not empty + StopAndWaitWithError(ctx context.Context, id int64, timeout time.Duration, origError error) (err error) + // Delete the specified execution and its tasks Delete(ctx context.Context, id int64) (err error) // Delete all executions and tasks of the specific vendor. They can be deleted only when all the executions/tasks @@ -250,6 +254,16 @@ func (e *executionManager) StopAndWait(ctx context.Context, id int64, timeout ti } } +func (e *executionManager) StopAndWaitWithError(ctx context.Context, id int64, timeout time.Duration, origError error) error { + if err := e.StopAndWait(ctx, id, timeout); err != nil { + return err + } + if origError != nil { + return e.MarkError(ctx, id, origError.Error()) + } + return nil +} + func (e *executionManager) Delete(ctx context.Context, id int64) error { tasks, err := e.taskDAO.List(ctx, &q.Query{ Keywords: map[string]interface{}{ diff --git a/src/testing/pkg/task/execution_manager.go b/src/testing/pkg/task/execution_manager.go index 17e912fccf6d..890aa6d4bd22 100644 --- a/src/testing/pkg/task/execution_manager.go +++ b/src/testing/pkg/task/execution_manager.go @@ -209,6 +209,20 @@ func (_m *ExecutionManager) StopAndWait(ctx context.Context, id int64, timeout t return r0 } +// StopAndWaitWithError provides a mock function with given fields: ctx, id, timeout, origError +func (_m *ExecutionManager) StopAndWaitWithError(ctx context.Context, id int64, timeout time.Duration, origError error) error { + ret := _m.Called(ctx, id, timeout, origError) + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, int64, time.Duration, error) error); ok { + r0 = rf(ctx, id, timeout, origError) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // UpdateExtraAttrs provides a mock function with given fields: ctx, id, extraAttrs func (_m *ExecutionManager) UpdateExtraAttrs(ctx context.Context, id int64, extraAttrs map[string]interface{}) error { ret := _m.Called(ctx, id, extraAttrs)