From e217d792b247f02723ef826c9ba2c6164ef156ec Mon Sep 17 00:00:00 2001 From: prahaladdarkin Date: Wed, 18 May 2022 17:59:50 +0530 Subject: [PATCH] Introduce system artifact manager cleanup job Signed-off-by: prahaladdarkin --- src/controller/systemartifact/callback.go | 26 +++ src/controller/systemartifact/execution.go | 152 ++++++++++++++++++ .../systemartifact/execution_test.go | 128 +++++++++++++++ src/core/main.go | 3 + .../job/impl/systemartifact/cleanup.go | 45 ++++++ .../job/impl/systemartifact/cleanup_test.go | 36 +++++ src/jobservice/job/known_jobs.go | 2 + src/jobservice/runtime/bootstrap.go | 9 +- 8 files changed, 398 insertions(+), 3 deletions(-) create mode 100644 src/controller/systemartifact/callback.go create mode 100644 src/controller/systemartifact/execution.go create mode 100644 src/controller/systemartifact/execution_test.go create mode 100644 src/jobservice/job/impl/systemartifact/cleanup.go create mode 100644 src/jobservice/job/impl/systemartifact/cleanup_test.go diff --git a/src/controller/systemartifact/callback.go b/src/controller/systemartifact/callback.go new file mode 100644 index 000000000000..ff46b3196c95 --- /dev/null +++ b/src/controller/systemartifact/callback.go @@ -0,0 +1,26 @@ +package systemartifact + +import ( + "context" + "github.com/goharbor/harbor/src/jobservice/logger" + "github.com/goharbor/harbor/src/lib/log" + "github.com/goharbor/harbor/src/pkg/scheduler" + "github.com/goharbor/harbor/src/pkg/task" +) + +const ( + SystemArtifactCleanupCallback = "SYSTEM_ARTIFACT_CLEANUP" +) + +func init() { + if err := scheduler.RegisterCallbackFunc(SystemArtifactCleanupCallback, cleanupCallBack); err != nil { + log.Fatalf("failed to register the callback for the system artifact cleanup schedule, error %v", err) + } +} + +func cleanupCallBack(ctx context.Context, param string) error { + + err := Ctl.Start(ctx, true, task.ExecutionTriggerSchedule) + logger.Errorf("System artifact cleanup job encountered errors: %v", err) + return err +} diff --git a/src/controller/systemartifact/execution.go b/src/controller/systemartifact/execution.go new file mode 100644 index 000000000000..8433c067e3a6 --- /dev/null +++ b/src/controller/systemartifact/execution.go @@ -0,0 +1,152 @@ +package systemartifact + +import ( + "context" + "github.com/goharbor/harbor/src/jobservice/job" + "github.com/goharbor/harbor/src/jobservice/logger" + "github.com/goharbor/harbor/src/lib/log" + "github.com/goharbor/harbor/src/lib/orm" + "github.com/goharbor/harbor/src/lib/q" + "github.com/goharbor/harbor/src/lib/retry" + "github.com/goharbor/harbor/src/pkg/scheduler" + "github.com/goharbor/harbor/src/pkg/systemartifact" + "github.com/goharbor/harbor/src/pkg/task" + "time" +) + +const ( + VendorTypeSystemArtifactCleanup = "SYSTEM_ARTIFACT_CLEANUP" + cronTypeDaily = "Daily" + cronSpec = "0 0 0 * * *" +) + +func init() { + task.SetExecutionSweeperCount(VendorTypeSystemArtifactCleanup, 50) +} + +var Ctl = NewController() + +type Controller interface { + Start(ctx context.Context, async bool, trigger string) error +} + +func NewController() Controller { + return &controller{ + execMgr: task.ExecMgr, + taskMgr: task.Mgr, + systemArtifactMgr: systemartifact.Mgr, + makeCtx: orm.Context, + } +} + +type controller struct { + execMgr task.ExecutionManager + taskMgr task.Manager + systemArtifactMgr systemartifact.Manager + makeCtx func() context.Context +} + +func (c *controller) Start(ctx context.Context, async bool, trigger string) error { + execId, err := c.execMgr.Create(ctx, VendorTypeSystemArtifactCleanup, 0, trigger) + if err != nil { + return err + } + // cleanup job would always be scheduled in async mode in production + // allowing for sync mode execution only for test mode purposes + // if there are any trigger settings then pass them to the cleanup manager first + jobParams := job.Parameters{} + + if !async { + err := c.createCleanupTask(ctx, jobParams, execId) + if err != nil { + log.Errorf("failed to create system artifact clean-up task: %v", err) + return err + } + + logger.Info("Created job for scan data export successfully") + return nil + } + go func(ctx context.Context) { + err := retry.Retry(func() error { + _, err := c.execMgr.Get(ctx, execId) + return err + }) + if err != nil { + log.Errorf("failed to get the execution %d for the export data cleanup job", execId) + return + } + err = c.createCleanupTask(ctx, jobParams, execId) + if err != nil { + logger.Errorf("Encountered error in scan data artifact cleanup : %v", err) + return + } + }(c.makeCtx()) + + return nil +} + +func (c *controller) createCleanupTask(ctx context.Context, jobParams job.Parameters, execId int64) error { + j := &task.Job{ + Name: job.SystemArtifactCleanup, + Metadata: &job.Metadata{ + JobKind: job.KindGeneric, + }, + Parameters: jobParams, + } + + _, err := c.taskMgr.Create(ctx, execId, j) + + if err != nil { + logger.Errorf("Unable to create a scan data export job in clean-up mode : %v", err) + c.markError(ctx, execId, err) + return err + } + return nil +} + +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 { + 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) + } +} + +// ScheduleCleanupTask schedules a system artifact cleanup task +func ScheduleCleanupTask(ctx context.Context) { + scheduleSystemArtifactCleanJob(ctx) +} + +func scheduleSystemArtifactCleanJob(ctx context.Context) { + schedule, err := getSystemArtifactCleanupSchedule(ctx) + if err != nil { + return + } + if schedule != nil { + logger.Debugf(" Export data cleanup job already scheduled with ID : %v.", schedule.ID) + return + } + scheduleId, err := scheduler.Sched.Schedule(ctx, VendorTypeSystemArtifactCleanup, 0, cronTypeDaily, cronSpec, SystemArtifactCleanupCallback, nil, nil) + if err != nil { + log.Errorf("Encountered error when scheduling scan data export cleanup job : %v", err) + return + } + log.Infof("Scheduled scan data export cleanup job with ID : %v", scheduleId) +} + +func getSystemArtifactCleanupSchedule(ctx context.Context) (*scheduler.Schedule, error) { + query := q.New(map[string]interface{}{"vendor_type": VendorTypeSystemArtifactCleanup}) + schedules, err := scheduler.Sched.ListSchedules(ctx, query) + if err != nil { + logger.Errorf("Unable to check if export data cleanup job is already scheduled : %v", err) + return nil, err + } + if len(schedules) > 0 { + logger.Infof("Found export data cleanup job with schedule id : %v", schedules[0].ID) + return schedules[0], nil + } + return nil, nil +} diff --git a/src/controller/systemartifact/execution_test.go b/src/controller/systemartifact/execution_test.go new file mode 100644 index 000000000000..95f4c54f4b2d --- /dev/null +++ b/src/controller/systemartifact/execution_test.go @@ -0,0 +1,128 @@ +package systemartifact + +import ( + "context" + "github.com/goharbor/harbor/src/lib/orm" + "github.com/goharbor/harbor/src/pkg/task" + ormtesting "github.com/goharbor/harbor/src/testing/lib/orm" + "github.com/goharbor/harbor/src/testing/mock" + "github.com/goharbor/harbor/src/testing/pkg/systemartifact" + testingTask "github.com/goharbor/harbor/src/testing/pkg/task" + "github.com/pkg/errors" + testifymock "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/suite" + "testing" +) + +type SystemArtifactCleanupTestSuite struct { + suite.Suite + execMgr *testingTask.ExecutionManager + taskMgr *testingTask.Manager + cleanupMgr *systemartifact.Manager + ctl *controller +} + +func (suite *SystemArtifactCleanupTestSuite) SetupSuite() { +} + +func (suite *SystemArtifactCleanupTestSuite) TestStartCleanup() { + suite.taskMgr = &testingTask.Manager{} + suite.execMgr = &testingTask.ExecutionManager{} + suite.cleanupMgr = &systemartifact.Manager{} + suite.ctl = &controller{ + execMgr: suite.execMgr, + taskMgr: suite.taskMgr, + systemArtifactMgr: suite.cleanupMgr, + makeCtx: func() context.Context { return orm.NewContext(nil, &ormtesting.FakeOrmer{}) }, + } + + { + + ctx := context.TODO() + + executionID := int64(1) + taskId := int64(1) + + suite.execMgr.On("Create", ctx, "SYSTEM_ARTIFACT_CLEANUP", int64(0), "SCHEDULE").Return(executionID, nil).Once() + + suite.taskMgr.On("Create", ctx, executionID, mock.Anything).Return(taskId, nil).Once() + + suite.execMgr.On("MarkDone", ctx, executionID, mock.Anything).Return(nil).Once() + + err := suite.ctl.Start(ctx, false, "SCHEDULE") + suite.NoError(err) + jobMatcher := testifymock.MatchedBy(func(j *task.Job) bool { + return "SYSTEM_ARTIFACT_CLEANUP" == j.Name + }) + suite.taskMgr.AssertCalled(suite.T(), "Create", ctx, executionID, jobMatcher) + } +} + +func (suite *SystemArtifactCleanupTestSuite) TestStartCleanupErrorDuringCreate() { + suite.taskMgr = &testingTask.Manager{} + suite.execMgr = &testingTask.ExecutionManager{} + suite.cleanupMgr = &systemartifact.Manager{} + suite.ctl = &controller{ + execMgr: suite.execMgr, + taskMgr: suite.taskMgr, + systemArtifactMgr: suite.cleanupMgr, + makeCtx: func() context.Context { return orm.NewContext(nil, &ormtesting.FakeOrmer{}) }, + } + + { + + ctx := context.TODO() + + executionID := int64(1) + + suite.execMgr.On( + "Create", ctx, "SYSTEM_ARTIFACT_CLEANUP", int64(0), "SCHEDULE", + ).Return(int64(0), errors.New("test error")).Once() + + suite.execMgr.On("MarkDone", ctx, executionID, mock.Anything).Return(nil).Once() + + err := suite.ctl.Start(ctx, false, "SCHEDULE") + suite.Error(err) + } +} + +func (suite *SystemArtifactCleanupTestSuite) TestStartCleanupErrorDuringTaskCreate() { + suite.taskMgr = &testingTask.Manager{} + suite.execMgr = &testingTask.ExecutionManager{} + suite.cleanupMgr = &systemartifact.Manager{} + suite.ctl = &controller{ + execMgr: suite.execMgr, + taskMgr: suite.taskMgr, + systemArtifactMgr: suite.cleanupMgr, + makeCtx: func() context.Context { return orm.NewContext(nil, &ormtesting.FakeOrmer{}) }, + } + + { + + ctx := context.TODO() + + executionID := int64(1) + taskId := int64(0) + + suite.execMgr.On( + "Create", ctx, "SYSTEM_ARTIFACT_CLEANUP", int64(0), "SCHEDULE", + ).Return(executionID, nil).Once() + + suite.taskMgr.On("Create", ctx, executionID, mock.Anything).Return(taskId, errors.New("test error")).Once() + + suite.execMgr.On("MarkError", ctx, executionID, mock.Anything).Return(nil).Once() + suite.execMgr.On("StopAndWait", ctx, executionID, mock.Anything).Return(nil).Once() + + err := suite.ctl.Start(ctx, false, "SCHEDULE") + suite.Error(err) + } +} + +func (suite *SystemArtifactCleanupTestSuite) TearDownSuite() { + suite.execMgr = nil + suite.taskMgr = nil +} + +func TestScanDataExportExecutionTestSuite(t *testing.T) { + suite.Run(t, &SystemArtifactCleanupTestSuite{}) +} diff --git a/src/core/main.go b/src/core/main.go index 2d4956febd80..08f6f7d4ce18 100755 --- a/src/core/main.go +++ b/src/core/main.go @@ -25,6 +25,8 @@ import ( "syscall" "time" + "github.com/goharbor/harbor/src/controller/systemartifact" + "github.com/beego/beego" "github.com/goharbor/harbor/src/core/session" @@ -224,6 +226,7 @@ func main() { log.Info("Fix empty subiss for meta info data.") oidc.FixEmptySubIss(orm.Context()) + systemartifact.ScheduleCleanupTask(ctx) beego.RunWithMiddleWares("", middlewares.MiddleWares()...) } diff --git a/src/jobservice/job/impl/systemartifact/cleanup.go b/src/jobservice/job/impl/systemartifact/cleanup.go new file mode 100644 index 000000000000..1ea7f366d5c4 --- /dev/null +++ b/src/jobservice/job/impl/systemartifact/cleanup.go @@ -0,0 +1,45 @@ +package systemartifact + +import ( + "github.com/goharbor/harbor/src/jobservice/job" + "github.com/goharbor/harbor/src/pkg/systemartifact" +) + +type Cleanup struct { + sysArtifactManager systemartifact.Manager +} + +func (c *Cleanup) MaxFails() uint { + return 1 +} + +func (c *Cleanup) MaxCurrency() uint { + return 1 +} + +func (c *Cleanup) ShouldRetry() bool { + return true +} + +func (c *Cleanup) Validate(params job.Parameters) error { + return nil +} + +func (c *Cleanup) Run(ctx job.Context, params job.Parameters) error { + logger := ctx.GetLogger() + logger.Infof("Running system data artifact cleanup job...") + c.init() + numRecordsDeleted, totalSizeReclaimed, err := c.sysArtifactManager.Cleanup(ctx.SystemContext()) + if err != nil { + logger.Errorf("Error when executing system artifact cleanup job: %v", err) + return err + } + logger.Infof("Num System artifacts cleaned up: %d, Total space reclaimed: %d.", numRecordsDeleted, totalSizeReclaimed) + return nil +} + +func (c *Cleanup) init() { + if c.sysArtifactManager == nil { + c.sysArtifactManager = systemartifact.NewManager() + } +} diff --git a/src/jobservice/job/impl/systemartifact/cleanup_test.go b/src/jobservice/job/impl/systemartifact/cleanup_test.go new file mode 100644 index 000000000000..4abcbe1ae31e --- /dev/null +++ b/src/jobservice/job/impl/systemartifact/cleanup_test.go @@ -0,0 +1,36 @@ +package systemartifact + +import ( + "github.com/goharbor/harbor/src/jobservice/job" + mockjobservice "github.com/goharbor/harbor/src/testing/jobservice" + "github.com/goharbor/harbor/src/testing/mock" + "github.com/goharbor/harbor/src/testing/pkg/systemartifact" + "github.com/stretchr/testify/suite" + "testing" +) + +type SystemArtifactCleanupSuite struct { + suite.Suite + sysArtifactMgr *systemartifact.Manager + job *Cleanup +} + +func (suite *SystemArtifactCleanupSuite) SetupTest() { + suite.sysArtifactMgr = &systemartifact.Manager{} + suite.job = &Cleanup{sysArtifactManager: suite.sysArtifactMgr} +} + +func (suite *SystemArtifactCleanupSuite) TestRun() { + mock.OnAnything(suite.sysArtifactMgr, "Cleanup").Return(int64(100), int64(100), nil) + params := job.Parameters{} + ctx := &mockjobservice.MockJobContext{} + + err := suite.job.Run(ctx, params) + suite.NoError(err) + // assert that job manager is invoked in this mode + suite.sysArtifactMgr.AssertCalled(suite.T(), "Cleanup", mock.Anything) +} + +func TestSystemArtifactCleanupSuite(t *testing.T) { + suite.Run(t, &SystemArtifactCleanupSuite{}) +} diff --git a/src/jobservice/job/known_jobs.go b/src/jobservice/job/known_jobs.go index ec67ed09534d..69710bef9ae2 100644 --- a/src/jobservice/job/known_jobs.go +++ b/src/jobservice/job/known_jobs.go @@ -36,4 +36,6 @@ const ( P2PPreheat = "P2P_PREHEAT" // PurgeAudit : the name of purge audit job PurgeAudit = "PURGE_AUDIT" + // SystemArtifactCleanup : the name of the SystemArtifact cleanup job + SystemArtifactCleanup = "SYSTEM_ARTIFACT_CLEANUP" ) diff --git a/src/jobservice/runtime/bootstrap.go b/src/jobservice/runtime/bootstrap.go index b5d00267f519..32b8f83ae842 100644 --- a/src/jobservice/runtime/bootstrap.go +++ b/src/jobservice/runtime/bootstrap.go @@ -24,6 +24,8 @@ import ( "syscall" "time" + "github.com/goharbor/harbor/src/jobservice/job/impl/systemartifact" + "github.com/goharbor/harbor/src/jobservice/api" "github.com/goharbor/harbor/src/jobservice/common/utils" "github.com/goharbor/harbor/src/jobservice/config" @@ -319,9 +321,10 @@ func (bs *Bootstrap) loadAndRunRedisWorkerPool( // In v2.2 we migrate the scheduled replication, garbage collection and scan all to // the scheduler mechanism, the following three jobs are kept for the legacy jobs // and they can be removed after several releases - "IMAGE_REPLICATE": (*legacy.ReplicationScheduler)(nil), - "IMAGE_GC": (*legacy.GarbageCollectionScheduler)(nil), - "IMAGE_SCAN_ALL": (*legacy.ScanAllScheduler)(nil), + "IMAGE_REPLICATE": (*legacy.ReplicationScheduler)(nil), + "IMAGE_GC": (*legacy.GarbageCollectionScheduler)(nil), + "IMAGE_SCAN_ALL": (*legacy.ScanAllScheduler)(nil), + job.SystemArtifactCleanup: (*systemartifact.Cleanup)(nil), }); err != nil { // exit return nil, err