feat: delete jobs in batches

Signed-off-by: Gaius <gaius.qi@gmail.com>

Author: gaius-qi

manager/config/config.go

@@ -321,6 +321,9 @@ type GCConfig struct {
 
 	// TTL is the ttl for job.
 	TTL time.Duration `yaml:"ttl" mapstructure:"ttl"`
+
+	// BatchSize is the batch size when operating gorm database.
+	BatchSize int `yaml:"batchSize" mapstructure:"batchSize"`
 }
 
 type PreheatConfig struct {
@@ -442,8 +445,9 @@ func New() *Config {
 		},
 		Job: JobConfig{
 			GC: GCConfig{
-				Interval: DefaultJobGCInterval,
-				TTL:      DefaultJobGCTTL,
+				Interval:  DefaultJobGCInterval,
+				TTL:       DefaultJobGCTTL,
+				BatchSize: DefaultJobGCBatchSize,
 			},
 			Preheat: PreheatConfig{
 				RegistryTimeout: DefaultJobPreheatRegistryTimeout,
@@ -629,6 +633,10 @@ func (cfg *Config) Validate() error {
 		return errors.New("gc requires parameter ttl")
 	}
 
+	if cfg.Job.GC.BatchSize == 0 {
+		return errors.New("gc requires parameter batchSize")
+	}
+
 	if cfg.Job.Preheat.RegistryTimeout == 0 {
 		return errors.New("preheat requires parameter registryTimeout")
 	}

manager/config/config_test.go

@@ -178,8 +178,9 @@ func TestConfig_Load(t *testing.T) {
 		},
 		Job: JobConfig{
 			GC: GCConfig{
-				Interval: 1 * time.Second,
-				TTL:      1 * time.Second,
+				Interval:  1 * time.Second,
+				TTL:       1 * time.Second,
+				BatchSize: 100,
 			},
 			Preheat: PreheatConfig{
 				RegistryTimeout: DefaultJobPreheatRegistryTimeout,
@@ -765,6 +766,21 @@ func TestConfig_Validate(t *testing.T) {
 				assert.EqualError(err, "gc requires parameter ttl")
 			},
 		},
+		{
+			name:   "gc requires parameter batchSize",
+			config: New(),
+			mock: func(cfg *Config) {
+				cfg.Auth.JWT = mockJWTConfig
+				cfg.Database.Type = DatabaseTypeMysql
+				cfg.Database.Mysql = mockMysqlConfig
+				cfg.Database.Redis = mockRedisConfig
+				cfg.Job.GC.BatchSize = 0
+			},
+			expect: func(t *testing.T, err error) {
+				assert := assert.New(t)
+				assert.EqualError(err, "gc requires parameter batchSize")
+			},
+		},
 		{
 			name:   "preheat requires parameter registryTimeout",
 			config: New(),

manager/config/constants.go

@@ -93,6 +93,9 @@ const (
 	// DefaultJobGCTTL is the default ttl for job.
 	DefaultJobGCTTL = 12 * time.Hour
 
+	// DefaultJobGCBatchSize is the default batch size for operating on the database in gc job.
+	DefaultJobGCBatchSize = 5000
+
 	// DefaultJobPreheatRegistryTimeout is the default timeout for requesting registry to get token and manifest.
 	DefaultJobPreheatRegistryTimeout = 1 * time.Minute
 

manager/config/testdata/manager.yaml

@@ -72,6 +72,7 @@ job:
   gc:
     interval: 1s
     ttl: 1s
+    batchSize: 100
   preheat:
     registryTimeout: 1m
     tls:

manager/job/gc.go

@@ -61,7 +61,7 @@ func (gc *gc) Serve() {
 		select {
 		case <-tick.C:
 			logger.Infof("gc job started")
-			if err := gc.db.WithContext(context.Background()).Where("created_at < ?", time.Now().Add(-gc.config.Job.GC.TTL)).Unscoped().Delete(&models.Job{}).Error; err != nil {
+			if err := gc.deleteInBatches(context.Background()); err != nil {
 				logger.Errorf("gc job failed: %v", err)
 			}
 		case <-gc.done:
@@ -74,3 +74,31 @@ func (gc *gc) Serve() {
 func (gc *gc) Stop() {
 	close(gc.done)
 }
+
+// deleteInBatches deletes jobs in batches.
+func (gc *gc) deleteInBatches(ctx context.Context) error {
+	for {
+		jobs := make([]models.Job, 0, gc.config.Job.GC.BatchSize)
+		if err := gc.db.WithContext(ctx).Where("created_at < ?", time.Now().Add(-gc.config.Job.GC.TTL)).Limit(gc.config.Job.GC.BatchSize).Find(&jobs).Error; err != nil {
+			return err
+		}
+
+		if len(jobs) == 0 {
+			logger.Infof("gc job finished")
+			break
+		}
+
+		var ids []uint
+		for _, job := range jobs {
+			ids = append(ids, job.ID)
+		}
+
+		if err := gc.db.WithContext(ctx).Where("id IN ?", ids).Unscoped().Delete(&models.Job{}).Error; err != nil {
+			return err
+		}
+
+		logger.Infof("deleted %d jobs", len(jobs))
+	}
+
+	return nil
+}