diff --git a/app/app.go b/app/app.go index 152f50056f..d29d1aa91e 100644 --- a/app/app.go +++ b/app/app.go @@ -122,6 +122,13 @@ func (a *App) initJobs() { if jobsImportDeleteInterface != nil { a.srv.Jobs.ImportDelete = jobsImportDeleteInterface(a) } + if jobsExportDeleteInterface != nil { + a.srv.Jobs.ExportDelete = jobsExportDeleteInterface(a) + } + + if jobsExportProcessInterface != nil { + a.srv.Jobs.ExportProcess = jobsExportProcessInterface(a) + } if jobsExportProcessInterface != nil { a.srv.Jobs.ExportProcess = jobsExportProcessInterface(a) diff --git a/app/enterprise.go b/app/enterprise.go index 754d37da1f..f61f7e3ea4 100644 --- a/app/enterprise.go +++ b/app/enterprise.go @@ -126,6 +126,12 @@ func RegisterJobsExportProcessInterface(f func(*App) tjobs.ExportProcessInterfac jobsExportProcessInterface = f } +var jobsExportDeleteInterface func(*App) tjobs.ExportDeleteInterface + +func RegisterJobsExportDeleteInterface(f func(*App) tjobs.ExportDeleteInterface) { + jobsExportDeleteInterface = f +} + var productNoticesJobInterface func(*App) tjobs.ProductNoticesJobInterface func RegisterProductNoticesJobInterface(f func(*App) tjobs.ProductNoticesJobInterface) { diff --git a/i18n/en.json b/i18n/en.json index 08546a01e8..bc63a92e4b 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -4096,7 +4096,7 @@ }, { "id": "app.export.export_attachment.create_file.error", - "translation": "Failed to created file during export." + "translation": "Failed to create file during export." }, { "id": "app.export.export_attachment.mkdirall.error", diff --git a/imports/placeholder.go b/imports/placeholder.go index ba3d6966db..5b482e803a 100644 --- a/imports/placeholder.go +++ b/imports/placeholder.go @@ -30,4 +30,7 @@ import ( // This is a placeholder so this package can be imported in Team Edition when it will be otherwise empty. _ "github.com/mattermost/mattermost-server/v5/jobs/export_process" + + // This is a placeholder so this package can be imported in Team Edition when it will be otherwise empty. + _ "github.com/mattermost/mattermost-server/v5/jobs/export_delete" ) diff --git a/jobs/export_delete/scheduler.go b/jobs/export_delete/scheduler.go new file mode 100644 index 0000000000..0e7cb9a2b0 --- /dev/null +++ b/jobs/export_delete/scheduler.go @@ -0,0 +1,51 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package export_delete + +import ( + "time" + + "github.com/mattermost/mattermost-server/v5/app" + "github.com/mattermost/mattermost-server/v5/model" +) + +const ( + jobName = "ExportDelete" + schedFrequency = 24 * time.Hour +) + +type Scheduler struct { + app *app.App +} + +func (i *ExportDeleteInterfaceImpl) MakeScheduler() model.Scheduler { + return &Scheduler{i.app} +} + +func (scheduler *Scheduler) Name() string { + return jobName + "Scheduler" +} + +func (scheduler *Scheduler) JobType() string { + return model.JOB_TYPE_EXPORT_DELETE +} + +func (scheduler *Scheduler) Enabled(cfg *model.Config) bool { + return *cfg.ExportSettings.Directory != "" && *cfg.ExportSettings.RetentionDays > 0 +} + +func (scheduler *Scheduler) NextScheduleTime(cfg *model.Config, now time.Time, pendingJobs bool, lastSuccessfulJob *model.Job) *time.Time { + nextTime := time.Now().Add(schedFrequency) + return &nextTime +} + +func (scheduler *Scheduler) ScheduleJob(cfg *model.Config, pendingJobs bool, lastSuccessfulJob *model.Job) (*model.Job, *model.AppError) { + data := map[string]string{} + + job, err := scheduler.app.Srv().Jobs.CreateJob(model.JOB_TYPE_EXPORT_DELETE, data) + if err != nil { + return nil, err + } + return job, nil +} diff --git a/jobs/export_delete/worker.go b/jobs/export_delete/worker.go new file mode 100644 index 0000000000..1e25fa1d04 --- /dev/null +++ b/jobs/export_delete/worker.go @@ -0,0 +1,137 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package export_delete + +import ( + "path/filepath" + "time" + + "github.com/mattermost/mattermost-server/v5/app" + "github.com/mattermost/mattermost-server/v5/jobs" + tjobs "github.com/mattermost/mattermost-server/v5/jobs/interfaces" + "github.com/mattermost/mattermost-server/v5/mlog" + "github.com/mattermost/mattermost-server/v5/model" +) + +func init() { + app.RegisterJobsExportDeleteInterface(func(a *app.App) tjobs.ExportDeleteInterface { + return &ExportDeleteInterfaceImpl{a} + }) +} + +type ExportDeleteInterfaceImpl struct { + app *app.App +} + +type ExportDeleteWorker struct { + name string + stopChan chan struct{} + stoppedChan chan struct{} + jobsChan chan model.Job + jobServer *jobs.JobServer + app *app.App +} + +func (i *ExportDeleteInterfaceImpl) MakeWorker() model.Worker { + return &ExportDeleteWorker{ + name: "ExportDelete", + stopChan: make(chan struct{}), + stoppedChan: make(chan struct{}), + jobsChan: make(chan model.Job), + jobServer: i.app.Srv().Jobs, + app: i.app, + } +} + +func (w *ExportDeleteWorker) JobChannel() chan<- model.Job { + return w.jobsChan +} + +func (w *ExportDeleteWorker) Run() { + mlog.Debug("Worker started", mlog.String("worker", w.name)) + + defer func() { + mlog.Debug("Worker finished", mlog.String("worker", w.name)) + close(w.stoppedChan) + }() + + for { + select { + case <-w.stopChan: + mlog.Debug("Worker received stop signal", mlog.String("worker", w.name)) + return + case job := <-w.jobsChan: + mlog.Debug("Worker received a new candidate job.", mlog.String("worker", w.name)) + w.doJob(&job) + } + } +} + +func (w *ExportDeleteWorker) Stop() { + mlog.Debug("Worker stopping", mlog.String("worker", w.name)) + close(w.stopChan) + <-w.stoppedChan +} + +func (w *ExportDeleteWorker) doJob(job *model.Job) { + if claimed, err := w.jobServer.ClaimJob(job); err != nil { + mlog.Warn("Worker experienced an error while trying to claim job", + mlog.String("worker", w.name), + mlog.String("job_id", job.Id), + mlog.String("error", err.Error())) + return + } else if !claimed { + return + } + + exportPath := *w.app.Config().ExportSettings.Directory + retentionTime := time.Duration(*w.app.Config().ExportSettings.RetentionDays) * 24 * time.Hour + exports, appErr := w.app.ListDirectory(exportPath) + if appErr != nil { + w.setJobError(job, appErr) + return + } + + var hasErrs bool + for i := range exports { + filename := filepath.Base(exports[i]) + modTime, appErr := w.app.FileModTime(filepath.Join(exportPath, filename)) + if appErr != nil { + mlog.Debug("Worker: Failed to get file modification time", + mlog.Err(appErr), mlog.String("export", exports[i])) + hasErrs = true + continue + } + + if time.Now().After(modTime.Add(retentionTime)) { + // remove file data from storage. + if appErr := w.app.RemoveFile(exports[i]); appErr != nil { + mlog.Debug("Worker: Failed to remove file", + mlog.Err(appErr), mlog.String("export", exports[i])) + hasErrs = true + continue + } + } + } + + if hasErrs { + mlog.Warn("Worker: errors occurred") + } + + mlog.Info("Worker: Job is complete", mlog.String("worker", w.name), mlog.String("job_id", job.Id)) + w.setJobSuccess(job) +} + +func (w *ExportDeleteWorker) setJobSuccess(job *model.Job) { + if err := w.app.Srv().Jobs.SetJobSuccess(job); err != nil { + mlog.Error("Worker: Failed to set success for job", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error())) + w.setJobError(job, err) + } +} + +func (w *ExportDeleteWorker) setJobError(job *model.Job, appError *model.AppError) { + if err := w.app.Srv().Jobs.SetJobError(job, appError); err != nil { + mlog.Error("Worker: Failed to set job error", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error())) + } +} diff --git a/jobs/interfaces/export_delete_interface.go b/jobs/interfaces/export_delete_interface.go new file mode 100644 index 0000000000..5912b85f9b --- /dev/null +++ b/jobs/interfaces/export_delete_interface.go @@ -0,0 +1,11 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package interfaces + +import "github.com/mattermost/mattermost-server/v5/model" + +type ExportDeleteInterface interface { + MakeWorker() model.Worker + MakeScheduler() model.Scheduler +} diff --git a/jobs/jobs_watcher.go b/jobs/jobs_watcher.go index ab5dc3ee30..b146728ee4 100644 --- a/jobs/jobs_watcher.go +++ b/jobs/jobs_watcher.go @@ -170,6 +170,13 @@ func (watcher *Watcher) PollAndNotify() { default: } } + } else if job.Type == model.JOB_TYPE_EXPORT_DELETE { + if watcher.workers.ExportDelete != nil { + select { + case watcher.workers.ExportDelete.JobChannel() <- *job: + default: + } + } } else if job.Type == model.JOB_TYPE_CLOUD { if watcher.workers.Cloud != nil { select { diff --git a/jobs/schedulers.go b/jobs/schedulers.go index 9a2f479634..2583a28e58 100644 --- a/jobs/schedulers.go +++ b/jobs/schedulers.go @@ -82,6 +82,10 @@ func (srv *JobServer) InitSchedulers() *Schedulers { schedulers.schedulers = append(schedulers.schedulers, importDeleteInterface.MakeScheduler()) } + if exportDeleteInterface := srv.ExportDelete; exportDeleteInterface != nil { + schedulers.schedulers = append(schedulers.schedulers, exportDeleteInterface.MakeScheduler()) + } + schedulers.nextRunTimes = make([]*time.Time, len(schedulers.schedulers)) return schedulers } diff --git a/jobs/server.go b/jobs/server.go index 5235fe3d9d..76f6a91af3 100644 --- a/jobs/server.go +++ b/jobs/server.go @@ -33,6 +33,7 @@ type JobServer struct { ImportProcess tjobs.ImportProcessInterface ImportDelete tjobs.ImportDeleteInterface ExportProcess tjobs.ExportProcessInterface + ExportDelete tjobs.ExportDeleteInterface Cloud ejobs.CloudJobInterface } diff --git a/jobs/workers.go b/jobs/workers.go index bbb91c6659..b67e11d25e 100644 --- a/jobs/workers.go +++ b/jobs/workers.go @@ -30,6 +30,7 @@ type Workers struct { ImportProcess model.Worker ImportDelete model.Worker ExportProcess model.Worker + ExportDelete model.Worker Cloud model.Worker listenerId string @@ -97,6 +98,10 @@ func (srv *JobServer) InitWorkers() *Workers { workers.ExportProcess = exportProcessInterface.MakeWorker() } + if exportDeleteInterface := srv.ExportDelete; exportDeleteInterface != nil { + workers.ExportDelete = exportDeleteInterface.MakeWorker() + } + if cloudInterface := srv.Cloud; cloudInterface != nil { workers.Cloud = cloudInterface.MakeWorker() } @@ -164,6 +169,10 @@ func (workers *Workers) Start() *Workers { go workers.ExportProcess.Run() } + if workers.ExportDelete != nil { + go workers.ExportDelete.Run() + } + if workers.Cloud != nil { go workers.Cloud.Run() } @@ -289,6 +298,10 @@ func (workers *Workers) Stop() *Workers { workers.ExportProcess.Stop() } + if workers.ExportDelete != nil { + workers.ExportDelete.Stop() + } + if workers.Cloud != nil { workers.Cloud.Stop() } diff --git a/model/job.go b/model/job.go index 31312add7a..95fe972b5e 100644 --- a/model/job.go +++ b/model/job.go @@ -25,6 +25,7 @@ const ( JOB_TYPE_IMPORT_PROCESS = "import_process" JOB_TYPE_IMPORT_DELETE = "import_delete" JOB_TYPE_EXPORT_PROCESS = "export_process" + JOB_TYPE_EXPORT_DELETE = "export_delete" JOB_TYPE_CLOUD = "cloud" JOB_STATUS_PENDING = "pending" @@ -72,6 +73,7 @@ func (j *Job) IsValid() *AppError { case JOB_TYPE_IMPORT_PROCESS: case JOB_TYPE_IMPORT_DELETE: case JOB_TYPE_EXPORT_PROCESS: + case JOB_TYPE_EXPORT_DELETE: case JOB_TYPE_CLOUD: default: return NewAppError("Job.IsValid", "model.job.is_valid.type.app_error", nil, "id="+j.Id, http.StatusBadRequest)