feat: clean up expired jobs to prevent performance problems (#3504)

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

Author: gaius-qi

manager/config/config.go

@@ -300,13 +300,25 @@ type JobConfig struct {
 	// RateLimit configuration.
 	RateLimit RateLimitConfig `yaml:"rateLimit" mapstructure:"rateLimit"`
 
+	// GC configuration, used to clean up expired jobs. If the count of the jobs is huge,
+	// it may cause performance problems.
+	GC GCConfig `yaml:"gc" mapstructure:"gc"`
+
 	// Preheat configuration.
 	Preheat PreheatConfig `yaml:"preheat" mapstructure:"preheat"`
 
 	// Sync peers configuration.
 	SyncPeers SyncPeersConfig `yaml:"syncPeers" mapstructure:"syncPeers"`
 }
 
+type GCConfig struct {
+	// Interval is the interval for gc.
+	Interval time.Duration `yaml:"interval" mapstructure:"interval"`
+
+	// TTL is the ttl for job.
+	TTL time.Duration `yaml:"ttl" mapstructure:"ttl"`
+}
+
 type PreheatConfig struct {
 	// RegistryTimeout is the timeout for requesting registry to get token and manifest.
 	RegistryTimeout time.Duration `yaml:"registryTimeout" mapstructure:"registryTimeout"`
@@ -466,6 +478,10 @@ func New() *Config {
 				Capacity:     DefaultJobRateLimitCapacity,
 				Quantum:      DefaultJobRateLimitQuantum,
 			},
+			GC: GCConfig{
+				Interval: DefaultJobGCInterval,
+				TTL:      DefaultJobGCTTL,
+			},
 			Preheat: PreheatConfig{
 				RegistryTimeout: DefaultJobPreheatRegistryTimeout,
 			},
@@ -647,6 +663,14 @@ func (cfg *Config) Validate() error {
 		return errors.New("rateLimit requires parameter quantum")
 	}
 
+	if cfg.Job.GC.Interval == 0 {
+		return errors.New("gc requires parameter interval")
+	}
+
+	if cfg.Job.GC.TTL == 0 {
+		return errors.New("gc requires parameter ttl")
+	}
+
 	if cfg.Job.Preheat.TLS != nil {
 		if cfg.Job.Preheat.TLS.CACert == "" {
 			return errors.New("preheat requires parameter caCert")

manager/config/config_test.go

@@ -192,6 +192,10 @@ func TestConfig_Load(t *testing.T) {
 				Capacity:     1000,
 				Quantum:      1000,
 			},
+			GC: GCConfig{
+				Interval: 1 * time.Second,
+				TTL:      1 * time.Second,
+			},
 			Preheat: PreheatConfig{
 				RegistryTimeout: DefaultJobPreheatRegistryTimeout,
 				TLS: &PreheatTLSClientConfig{
@@ -747,6 +751,36 @@ func TestConfig_Validate(t *testing.T) {
 				assert.EqualError(err, "rateLimit requires parameter quantum")
 			},
 		},
+		{
+			name:   "gc requires parameter interval",
+			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.Interval = 0
+			},
+			expect: func(t *testing.T, err error) {
+				assert := assert.New(t)
+				assert.EqualError(err, "gc requires parameter interval")
+			},
+		},
+		{
+			name:   "gc requires parameter ttl",
+			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.TTL = 0
+			},
+			expect: func(t *testing.T, err error) {
+				assert := assert.New(t)
+				assert.EqualError(err, "gc requires parameter ttl")
+			},
+		},
 		{
 			name:   "preheat requires parameter caCert",
 			config: New(),

manager/config/constants.go

@@ -96,6 +96,12 @@ const (
 	// DefaultJobRateLimitQuantum is the default quantum for job rate limit.
 	DefaultJobRateLimitQuantum = 10
 
+	// DefaultJobGCInterval is the default interval for gc job.
+	DefaultJobGCInterval = 24 * time.Hour
+
+	// DefaultJobGCTTL is the default ttl for job.
+	DefaultJobGCTTL = 7 * 24 * time.Hour
+
 	// DefaultJobPreheatRegistryTimeout is the default timeout for requesting registry to get token and manifest.
 	DefaultJobPreheatRegistryTimeout = 1 * time.Minute
 

manager/config/testdata/manager.yaml

@@ -70,6 +70,9 @@ job:
     fillInterval: 1s
     capacity: 1000
     quantum: 1000
+  gc:
+    interval: 1s
+    ttl: 1s
   preheat:
     registryTimeout: 1m
     tls:

manager/job/gc.go

@@ -0,0 +1,76 @@
+/*
+ *     Copyright 2024 The Dragonfly 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.
+ */
+
+//go:generate mockgen -destination mocks/gc_mock.go -source gc.go -package mocks
+
+package job
+
+import (
+	"context"
+	"time"
+
+	"gorm.io/gorm"
+
+	logger "d7y.io/dragonfly/v2/internal/dflog"
+	"d7y.io/dragonfly/v2/manager/config"
+	"d7y.io/dragonfly/v2/manager/models"
+)
+
+// GC is an interface for gc.
+type GC interface {
+	// Serve started gc server.
+	Serve()
+
+	// Stop gc server.
+	Stop()
+}
+
+// gc is an implementation of GC.
+type gc struct {
+	config *config.Config
+	db     *gorm.DB
+	done   chan struct{}
+}
+
+// newGC returns a new GC.
+func newGC(cfg *config.Config, gdb *gorm.DB) (GC, error) {
+	return &gc{
+		config: cfg,
+		db:     gdb,
+		done:   make(chan struct{}),
+	}, nil
+}
+
+// Serve started gc server.
+func (gc *gc) Serve() {
+	tick := time.NewTicker(gc.config.Job.GC.Interval)
+	for {
+		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 {
+				logger.Errorf("gc job failed: %v", err)
+			}
+		case <-gc.done:
+			return
+		}
+	}
+}
+
+// Stop gc server.
+func (gc *gc) Stop() {
+	close(gc.done)
+}

manager/job/job.go

@@ -41,6 +41,7 @@ type Job struct {
 	Preheat
 	SyncPeers
 	Task
+	GC
 }
 
 // New returns a new Job.
@@ -77,22 +78,30 @@ func New(cfg *config.Config, gdb *gorm.DB) (*Job, error) {
 		return nil, err
 	}
 
+	gc, err := newGC(cfg, gdb)
+	if err != nil {
+		return nil, err
+	}
+
 	task := newTask(j)
 	return &Job{
 		Job:       j,
 		Preheat:   preheat,
 		SyncPeers: syncPeers,
 		Task:      task,
+		GC:        gc,
 	}, nil
 }
 
 // Serve starts the job server.
 func (j *Job) Serve() {
-	j.SyncPeers.Serve()
+	go j.GC.Serve()
+	go j.SyncPeers.Serve()
 }
 
 // Stop stops the job server.
 func (j *Job) Stop() {
+	j.GC.Stop()
 	j.SyncPeers.Stop()
 }