Add functionality to cleanup old jobs (#18646)

* Add functionality to cleanup old jobs

Historically, we never cleaned up old jobs from the DB
leading to them being accumulated forever.

This PR introduces functionality to cleanup old jobs
older than a defined threshold.

The functionality is set to false by default and has
to be enabled for it to work.

```release-note
2 new config settings were added.

JobSettings.CleanupOldJobs: This indicates whether to clean up old jobs
from the DB or not. Default is false.

JobSettings.CleanupJobsThresholdHours: This defines the time gap in hours beyond
which older jobs will be removed. This has no effect if the above config
setting is set to false. Default is -1
```

* fix copy pasta

```release-note
NONE
```

* address review comments

```release-note
NONE
```

* Fix lint

```release-note
NONE
```

* Use single config option for everything

```release-note
NONE
```

Co-authored-by: Mattermod <mattermod@users.noreply.github.com>

app/server.go

@@ -760,6 +760,9 @@ func (s *Server) runJobs() {
 	s.Go(func() {
 		runSessionCleanupJob(s)
 	})
+	s.Go(func() {
+		runJobsCleanupJob(s)
+	})
 	s.Go(func() {
 		runTokenCleanupJob(s)
 	})
@@ -1504,6 +1507,13 @@ func runSessionCleanupJob(s *Server) {
 	}, time.Hour*24)
 }
 
+func runJobsCleanupJob(s *Server) {
+	doJobsCleanup(s)
+	model.CreateRecurringTask("Job Cleanup", func() {
+		doJobsCleanup(s)
+	}, time.Hour*24)
+}
+
 func (s *Server) runLicenseExpirationCheckJob() {
 	s.doLicenseExpirationCheck()
 	model.CreateRecurringTask("License Expiration Check", func() {
@@ -1549,6 +1559,7 @@ func doCommandWebhookCleanup(s *Server) {
 
 const (
 	sessionsCleanupBatchSize = 1000
+	jobsCleanupBatchSize     = 1000
 )
 
 func doSessionCleanup(s *Server) {
@@ -1559,6 +1570,20 @@ func doSessionCleanup(s *Server) {
 	}
 }
 
+func doJobsCleanup(s *Server) {
+	if *s.Config().JobSettings.CleanupJobsThresholdDays < 0 {
+		return
+	}
+	mlog.Debug("Cleaning up jobs store.")
+
+	dur := time.Duration(*s.Config().JobSettings.CleanupJobsThresholdDays) * time.Hour * 24
+	expiry := model.GetMillisForTime(time.Now().Add(-dur))
+	err := s.Store.Job().Cleanup(expiry, jobsCleanupBatchSize)
+	if err != nil {
+		mlog.Warn("Error while cleaning up jobs", mlog.Err(err))
+	}
+}
+
 func doCheckAdminSupportStatus(a *App, c *request.Context) {
 	isE0Edition := model.BuildEnterpriseReady == "true"
 

model/config.go

@@ -2615,8 +2615,9 @@ func (s *DataRetentionSettings) SetDefaults() {
 }
 
 type JobSettings struct {
-	RunJobs      *bool `access:"write_restrictable,cloud_restrictable"`
-	RunScheduler *bool `access:"write_restrictable,cloud_restrictable"`
+	RunJobs                  *bool `access:"write_restrictable,cloud_restrictable"`
+	RunScheduler             *bool `access:"write_restrictable,cloud_restrictable"`
+	CleanupJobsThresholdDays *int  `access:"write_restrictable,cloud_restrictable"`
 }
 
 func (s *JobSettings) SetDefaults() {
@@ -2627,6 +2628,10 @@ func (s *JobSettings) SetDefaults() {
 	if s.RunScheduler == nil {
 		s.RunScheduler = NewBool(true)
 	}
+
+	if s.CleanupJobsThresholdDays == nil {
+		s.CleanupJobsThresholdDays = NewInt(-1)
+	}
 }
 
 type CloudSettings struct {

store/opentracinglayer/opentracinglayer.go

@@ -4177,6 +4177,24 @@ func (s *OpenTracingLayerGroupStore) UpsertMember(groupID string, userID string)
 	return result, err
 }
 
+func (s *OpenTracingLayerJobStore) Cleanup(expiryTime int64, batchSize int) error {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "JobStore.Cleanup")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	err := s.JobStore.Cleanup(expiryTime, batchSize)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return err
+}
+
 func (s *OpenTracingLayerJobStore) Delete(id string) (string, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "JobStore.Delete")

store/retrylayer/retrylayer.go

@@ -4506,6 +4506,26 @@ func (s *RetryLayerGroupStore) UpsertMember(groupID string, userID string) (*mod
 
 }
 
+func (s *RetryLayerJobStore) Cleanup(expiryTime int64, batchSize int) error {
+
+	tries := 0
+	for {
+		err := s.JobStore.Cleanup(expiryTime, batchSize)
+		if err == nil {
+			return nil
+		}
+		if !isRepeatableError(err) {
+			return err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return err
+		}
+	}
+
+}
+
 func (s *RetryLayerJobStore) Delete(id string) (string, error) {
 
 	tries := 0

store/sqlstore/job_store.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/mattermost/gorp"
@@ -17,6 +18,10 @@ import (
 	"github.com/mattermost/mattermost-server/v6/store"
 )
 
+const (
+	jobsCleanupDelay = 100 * time.Millisecond
+)
+
 type SqlJobStore struct {
 	*SqlStore
 }
@@ -286,3 +291,31 @@ func (jss SqlJobStore) Delete(id string) (string, error) {
 	}
 	return id, nil
 }
+
+func (jss SqlJobStore) Cleanup(expiryTime int64, batchSize int) error {
+	var query string
+	if jss.DriverName() == model.DatabaseDriverPostgres {
+		query = "DELETE FROM Jobs WHERE Id IN (SELECT Id FROM Jobs WHERE CreateAt < ? AND (Status != ? AND Status != ?) ORDER BY CreateAt ASC LIMIT ?)"
+	} else {
+		query = "DELETE FROM Jobs WHERE CreateAt < ? AND (Status != ? AND Status != ?) ORDER BY CreateAt ASC LIMIT ?"
+	}
+
+	var rowsAffected int64 = 1
+
+	for rowsAffected > 0 {
+		sqlResult, err := jss.GetMasterX().Exec(query,
+			expiryTime, model.JobStatusInProgress, model.JobStatusPending, batchSize)
+		if err != nil {
+			return errors.Wrap(err, "unable to delete jobs")
+		}
+		var rowErr error
+		rowsAffected, rowErr = sqlResult.RowsAffected()
+		if rowErr != nil {
+			return errors.Wrap(err, "unable to delete jobs")
+		}
+
+		time.Sleep(jobsCleanupDelay)
+	}
+
+	return nil
+}

store/store.go

@@ -681,6 +681,7 @@ type JobStore interface {
 	GetNewestJobByStatusesAndType(statuses []string, jobType string) (*model.Job, error)
 	GetCountByStatusAndType(status string, jobType string) (int64, error)
 	Delete(id string) (string, error)
+	Cleanup(expiryTime int64, batchSize int) error
 }
 
 type UserAccessTokenStore interface {

store/storetest/job_store.go

@@ -29,6 +29,7 @@ func TestJobStore(t *testing.T, ss store.Store) {
 	t.Run("JobUpdateOptimistically", func(t *testing.T) { testJobUpdateOptimistically(t, ss) })
 	t.Run("JobUpdateStatusUpdateStatusOptimistically", func(t *testing.T) { testJobUpdateStatusUpdateStatusOptimistically(t, ss) })
 	t.Run("JobDelete", func(t *testing.T) { testJobDelete(t, ss) })
+	t.Run("JobCleanup", func(t *testing.T) { testJobCleanup(t, ss) })
 }
 
 func testJobSaveGet(t *testing.T, ss store.Store) {
@@ -552,3 +553,43 @@ func testJobDelete(t *testing.T, ss store.Store) {
 	_, err = ss.Job().Delete(job.Id)
 	assert.NoError(t, err)
 }
+
+func testJobCleanup(t *testing.T, ss store.Store) {
+	now := model.GetMillis()
+	ids := make([]string, 0, 10)
+	for i := 0; i < 10; i++ {
+		job, err := ss.Job().Save(&model.Job{
+			Id:       model.NewId(),
+			CreateAt: now - int64(i),
+			Status:   model.JobStatusPending,
+		})
+		require.NoError(t, err)
+		ids = append(ids, job.Id)
+		defer ss.Job().Delete(job.Id)
+	}
+
+	jobs, err := ss.Job().GetAllByStatus(model.JobStatusPending)
+	require.NoError(t, err)
+	assert.Len(t, jobs, 10)
+
+	err = ss.Job().Cleanup(now+1, 5)
+	require.NoError(t, err)
+
+	// Should not clean up pending jobs
+	jobs, err = ss.Job().GetAllByStatus(model.JobStatusPending)
+	require.NoError(t, err)
+	assert.Len(t, jobs, 10)
+
+	for _, id := range ids {
+		_, err = ss.Job().UpdateStatus(id, model.JobStatusSuccess)
+		require.NoError(t, err)
+	}
+
+	err = ss.Job().Cleanup(now+1, 5)
+	require.NoError(t, err)
+
+	// Should clean up now
+	jobs, err = ss.Job().GetAllByStatus(model.JobStatusSuccess)
+	require.NoError(t, err)
+	assert.Len(t, jobs, 0)
+}

store/storetest/mocks/JobStore.go

@@ -14,6 +14,20 @@ type JobStore struct {
 	mock.Mock
 }
 
+// Cleanup provides a mock function with given fields: expiryTime, batchSize
+func (_m *JobStore) Cleanup(expiryTime int64, batchSize int) error {
+	ret := _m.Called(expiryTime, batchSize)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(int64, int) error); ok {
+		r0 = rf(expiryTime, batchSize)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // Delete provides a mock function with given fields: id
 func (_m *JobStore) Delete(id string) (string, error) {
 	ret := _m.Called(id)

store/timerlayer/timerlayer.go

@@ -3803,6 +3803,22 @@ func (s *TimerLayerGroupStore) UpsertMember(groupID string, userID string) (*mod
 	return result, err
 }
 
+func (s *TimerLayerJobStore) Cleanup(expiryTime int64, batchSize int) error {
+	start := timemodule.Now()
+
+	err := s.JobStore.Cleanup(expiryTime, batchSize)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("JobStore.Cleanup", success, elapsed)
+	}
+	return err
+}
+
 func (s *TimerLayerJobStore) Delete(id string) (string, error) {
 	start := timemodule.Now()