refactor: refactor the old goroutine execution sweep

src/controller/gc/controller.go

@@ -10,11 +10,6 @@ import (
 	"github.com/goharbor/harbor/src/pkg/task"
 )
 
-func init() {
-	// keep only the latest created 50 gc execution records
-	task.SetExecutionSweeperCount(job.GarbageCollectionVendorType, 50)
-}
-
 var (
 	// Ctl is a global garbage collection controller instance
 	Ctl = NewController()

src/controller/p2p/preheat/enforcer.go

@@ -46,11 +46,6 @@ import (
 	"github.com/goharbor/harbor/src/pkg/task"
 )
 
-func init() {
-	// keep only the latest created 50 p2p preheat execution records
-	task.SetExecutionSweeperCount(job.P2PPreheatVendorType, 50)
-}
-
 const (
 	defaultSeverityCode     = 99
 	extraAttrTotal          = "totalCount"

src/controller/replication/execution.go

@@ -35,11 +35,6 @@ import (
 	"github.com/goharbor/harbor/src/pkg/task"
 )
 
-func init() {
-	// keep only the latest created 50 replication execution records
-	task.SetExecutionSweeperCount(job.ReplicationVendorType, 50)
-}
-
 // Ctl is a global replication controller instance
 var Ctl = NewController()
 

src/controller/retention/controller.go

@@ -31,11 +31,6 @@ import (
 	"github.com/goharbor/harbor/src/pkg/task"
 )
 
-func init() {
-	// keep only the latest created 50 retention execution records
-	task.SetExecutionSweeperCount(job.RetentionVendorType, 50)
-}
-
 // go:generate mockery -name Controller -case snake
 
 // Controller to handle the requests related with retention

src/controller/scan/base_controller.go

@@ -55,8 +55,6 @@ var DefaultController = NewController()
 
 // const definitions
 const (
-	VendorTypeScanAll = "SCAN_ALL"
-
 	configRegistryEndpoint = "registryEndpoint"
 	configCoreInternalAddr = "coreInternalAddr"
 
@@ -69,11 +67,6 @@ const (
 	robotIDKey     = "robot_id"
 )
 
-func init() {
-	// keep only the latest created 5 scan all execution records
-	task.SetExecutionSweeperCount(VendorTypeScanAll, 5)
-}
-
 // uuidGenerator is a func template which is for generating UUID.
 type uuidGenerator func() (string, error)
 
@@ -346,7 +339,7 @@ func (bc *basicController) Stop(ctx context.Context, artifact *ar.Artifact) erro
 }
 
 func (bc *basicController) ScanAll(ctx context.Context, trigger string, async bool) (int64, error) {
-	executionID, err := bc.execMgr.Create(ctx, VendorTypeScanAll, 0, trigger)
+	executionID, err := bc.execMgr.Create(ctx, job.ScanAllVendorType, 0, trigger)
 	if err != nil {
 		return 0, err
 	}

src/controller/scan/callback.go

@@ -46,7 +46,7 @@ func init() {
 	}
 
 	// NOTE: the vendor type of execution for the scan job trigger by the scan all is VendorTypeScanAll
-	if err := task.RegisterTaskStatusChangePostFunc(VendorTypeScanAll, scanTaskStatusChange); err != nil {
+	if err := task.RegisterTaskStatusChangePostFunc(job.ScanAllVendorType, scanTaskStatusChange); err != nil {
 		log.Fatalf("failed to register the task status change post for the scan all job, error %v", err)
 	}
 

src/controller/scandataexport/execution.go

@@ -16,10 +16,6 @@ import (
 	"github.com/goharbor/harbor/src/pkg/task"
 )
 
-func init() {
-	task.SetExecutionSweeperCount(job.ScanDataExportVendorType, 50)
-}
-
 var Ctl = NewController()
 
 type Controller interface {

src/controller/systemartifact/execution.go

@@ -24,10 +24,6 @@ var (
 	sched = scheduler.Sched
 )
 
-func init() {
-	task.SetExecutionSweeperCount(job.SystemArtifactCleanupVendorType, 50)
-}
-
 var Ctl = NewController()
 
 type Controller interface {

src/controller/task/sweep.go

@@ -0,0 +1,143 @@
+//  Copyright Project Harbor Authors
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+package task
+
+import (
+	"context"
+
+	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/lib/log"
+	"github.com/goharbor/harbor/src/lib/q"
+	"github.com/goharbor/harbor/src/pkg/scheduler"
+	"github.com/goharbor/harbor/src/pkg/task"
+)
+
+var (
+	// SweepCtl is the global sweep controller
+	SweepCtl = NewSweepController()
+)
+
+type SweepParams struct {
+	// ExecRetainCounts records the retained execution counts for different vendor type
+	ExecRetainCounts map[string]int64
+}
+
+const (
+	// SchedulerCallback ...
+	SchedulerCallback = "EXECUTION_SWEEP_CALLBACK"
+	// systemVendorID represents the id for system job.
+	systemVendorID = -1
+
+	cronTypeCustom = "Custom"
+	// run for every hour
+	cronSpec = "0 0 * * * *"
+)
+
+func init() {
+	err := scheduler.RegisterCallbackFunc(SchedulerCallback, sweepCallback)
+	if err != nil {
+		log.Fatalf("failed to register execution sweep job callback, error: %v", err)
+	}
+}
+
+func sweepCallback(ctx context.Context, p string) error {
+	params := &SweepParams{ExecRetainCounts: job.GetExecutionSweeperCount()}
+	return SweepCtl.Start(ctx, params, task.ExecutionTriggerSchedule)
+}
+
+type SweepController interface {
+	Start(ctx context.Context, params *SweepParams, trigger string) error
+}
+
+type sweepController struct {
+	execMgr task.ExecutionManager
+	taskMgr task.Manager
+}
+
+func (sc *sweepController) Start(ctx context.Context, params *SweepParams, trigger string) error {
+	jobParams := make(map[string]interface{})
+	jobParams[task.ExecRetainCounts] = params.ExecRetainCounts
+
+	execID, err := sc.execMgr.Create(ctx, job.ExecSweepVendorType, systemVendorID, trigger, jobParams)
+	if err != nil {
+		log.Errorf("failed to create execution for %s, error: %v", job.ExecSweepVendorType, err)
+		return err
+	}
+
+	_, err = sc.taskMgr.Create(ctx, execID, &task.Job{
+		Name: job.ExecSweepVendorType,
+		Metadata: &job.Metadata{
+			JobKind: job.KindGeneric,
+		},
+		Parameters: jobParams,
+	})
+	if err != nil {
+		log.Errorf("failed to create task for %s, error: %v", job.ExecSweepVendorType, err)
+		return err
+	}
+
+	return nil
+}
+
+func NewSweepController() SweepController {
+	return &sweepController{
+		execMgr: task.ExecMgr,
+		taskMgr: task.Mgr,
+	}
+}
+
+// ScheduleSweepJob schedules the system execution sweep job.
+func ScheduleSweepJob(ctx context.Context) error {
+	sched, err := getScheduledSweepJob(ctx)
+	if err != nil {
+		return err
+	}
+	// unschedule the job if the cron changed
+	if sched != nil {
+		if sched.CRON != cronSpec {
+			log.Debugf("reschedule the system execution job because the cron changed, old: %s, new: %s", sched.CRON, cronSpec)
+			if err = scheduler.Sched.UnScheduleByID(ctx, sched.ID); err != nil {
+				return err
+			}
+		} else {
+			log.Debug("skip to schedule the system execution job because the old one existed and cron not changed")
+			return nil
+		}
+	}
+
+	// schedule a job if no schedule found or cron changed
+	scheduleID, err := scheduler.Sched.Schedule(ctx, job.ExecSweepVendorType, systemVendorID, cronTypeCustom, cronSpec, SchedulerCallback, nil, nil)
+	if err != nil {
+		return err
+	}
+
+	log.Debugf("scheduled the system execution sweep job, id: %d", scheduleID)
+	return nil
+}
+
+// getScheduledSweepJob gets sweep job which already scheduled.
+func getScheduledSweepJob(ctx context.Context) (*scheduler.Schedule, error) {
+	query := q.New(map[string]interface{}{"vendor_type": job.ExecSweepVendorType})
+	schedules, err := scheduler.Sched.ListSchedules(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(schedules) > 0 {
+		return schedules[0], nil
+	}
+
+	return nil, nil
+}

src/core/main.go

@@ -35,6 +35,7 @@ import (
 	"github.com/goharbor/harbor/src/controller/health"
 	"github.com/goharbor/harbor/src/controller/registry"
 	"github.com/goharbor/harbor/src/controller/systemartifact"
+	"github.com/goharbor/harbor/src/controller/task"
 	"github.com/goharbor/harbor/src/core/api"
 	_ "github.com/goharbor/harbor/src/core/auth/authproxy"
 	_ "github.com/goharbor/harbor/src/core/auth/db"
@@ -258,7 +259,13 @@ func main() {
 			log.Errorf("failed to check the jobservice health status: timeout, error: %v", err)
 			return
 		}
+
+		// schedule system artifact cleanup job
 		systemartifact.ScheduleCleanupTask(ctx)
+		// schedule system execution sweep job
+		if err := task.ScheduleSweepJob(ctx); err != nil {
+			log.Errorf("failed to schedule system execution sweep job, error: %v", err)
+		}
 	}()
 	web.RunWithMiddleWares("", middlewares.MiddleWares()...)
 }

src/jobservice/job/known_jobs.go

@@ -40,4 +40,29 @@ const (
 	SystemArtifactCleanupVendorType = "SYSTEM_ARTIFACT_CLEANUP"
 	// ScanDataExportVendorType : the name of the scan data export job
 	ScanDataExportVendorType = "SCAN_DATA_EXPORT"
+	// ExecSweepVendorType: the name of the execution sweep job
+	ExecSweepVendorType = "EXECUTION_SWEEP"
+	// ScanAllVendorType: the name of the scan all job
+	ScanAllVendorType = "SCAN_ALL"
 )
+
+var (
+	// executionSweeperCount stores the count for execution retained
+	executionSweeperCount = map[string]int64{
+		ScanAllVendorType:               5,
+		ExecSweepVendorType:             10,
+		GarbageCollectionVendorType:     50,
+		SlackJobVendorType:              50,
+		WebhookJobVendorType:            50,
+		ReplicationVendorType:           50,
+		ScanDataExportVendorType:        50,
+		SystemArtifactCleanupVendorType: 50,
+		P2PPreheatVendorType:            50,
+		RetentionVendorType:             50,
+	}
+)
+
+// GetExecutionSweeperCount gets the count of execution records retained by the sweeper
+func GetExecutionSweeperCount() map[string]int64 {
+	return executionSweeperCount
+}

src/jobservice/runner/redis.go

@@ -161,7 +161,7 @@ func (rj *RedisJob) Run(j *work.Job) (err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			// Log the stack
-			buf := make([]byte, 1<<10)
+			buf := make([]byte, 1<<20)
 			size := runtime.Stack(buf, false)
 			err = errors.Errorf("runtime error: %s; stack: %s", r, buf[0:size])
 			logger.Errorf("Run job %s:%s error: %s", j.Name, j.ID, err)

src/jobservice/runtime/bootstrap.go

@@ -331,6 +331,7 @@ func (bs *Bootstrap) loadAndRunRedisWorkerPool(
 			"IMAGE_GC":                          (*legacy.GarbageCollectionScheduler)(nil),
 			"IMAGE_SCAN_ALL":                    (*legacy.ScanAllScheduler)(nil),
 			job.SystemArtifactCleanupVendorType: (*systemartifact.Cleanup)(nil),
+			job.ExecSweepVendorType:             (*task.SweepJob)(nil),
 		}); err != nil {
 		// exit
 		return nil, err