From a5463c865195d0f286de63d57782ef997c270e93 Mon Sep 17 00:00:00 2001 From: Claudio Costa Date: Wed, 4 Aug 2021 11:10:53 +0200 Subject: [PATCH] Implement content extraction job (#18022) --- app/enterprise.go | 6 + app/job.go | 6 +- app/server.go | 4 + i18n/en.json | 12 ++ imports/placeholder.go | 3 + jobs/extract_content/worker.go | 176 +++++++++++++++++++ jobs/interfaces/extract_content_interface.go | 12 ++ jobs/jobs_watcher.go | 7 + jobs/server.go | 1 + jobs/workers.go | 13 ++ model/job.go | 3 + 11 files changed, 241 insertions(+), 2 deletions(-) create mode 100644 jobs/extract_content/worker.go create mode 100644 jobs/interfaces/extract_content_interface.go diff --git a/app/enterprise.go b/app/enterprise.go index 9fd035f2fa..ab434dadfa 100644 --- a/app/enterprise.go +++ b/app/enterprise.go @@ -139,6 +139,12 @@ func RegisterJobsExportDeleteInterface(f func(*Server) tjobs.ExportDeleteInterfa jobsExportDeleteInterface = f } +var jobsExtractContentInterface func(*Server) tjobs.ExtractContentInterface + +func RegisterJobsExtractContentInterface(f func(*Server) tjobs.ExtractContentInterface) { + jobsExtractContentInterface = f +} + var productNoticesJobInterface func(*Server) tjobs.ProductNoticesJobInterface func RegisterProductNoticesJobInterface(f func(*Server) tjobs.ProductNoticesJobInterface) { diff --git a/app/job.go b/app/job.go index 650a0eeb7b..5ba697f3e1 100644 --- a/app/job.go +++ b/app/job.go @@ -96,7 +96,8 @@ func (a *App) SessionHasPermissionToCreateJob(session model.Session, job *model. model.JobTypeImportDelete, model.JobTypeExportProcess, model.JobTypeExportDelete, - model.JobTypeCloud: + model.JobTypeCloud, + model.JobTypeExtractContent: return a.SessionHasPermissionTo(session, model.PermissionManageJobs), model.PermissionManageJobs } @@ -126,7 +127,8 @@ func (a *App) SessionHasPermissionToReadJob(session model.Session, jobType strin model.JobTypeImportDelete, model.JobTypeExportProcess, model.JobTypeExportDelete, - model.JobTypeCloud: + model.JobTypeCloud, + model.JobTypeExtractContent: return a.SessionHasPermissionTo(session, model.PermissionReadJobs), model.PermissionReadJobs } diff --git a/app/server.go b/app/server.go index af960d4494..c22cafe142 100644 --- a/app/server.go +++ b/app/server.go @@ -2061,6 +2061,10 @@ func (s *Server) initJobs() { s.Jobs.ResendInvitationEmails = jobsResendInvitationEmailInterface(s) } + if jobsExtractContentInterface != nil { + s.Jobs.ExtractContent = jobsExtractContentInterface(s) + } + s.Jobs.InitWorkers() s.Jobs.InitSchedulers() } diff --git a/i18n/en.json b/i18n/en.json index 6953877604..a3a2dfdad3 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -7494,6 +7494,18 @@ "id": "error", "translation": "Error" }, + { + "id": "extrac_content.worker.do_job.invalid_input.from", + "translation": "Invalid input value 'from'" + }, + { + "id": "extrac_content.worker.do_job.invalid_input.to", + "translation": "Invalid input value 'to'" + }, + { + "id": "extract_content.worker.do_job.file_info", + "translation": "Failed to get file information for content extraction." + }, { "id": "group_not_associated_to_synced_team", "translation": "Group cannot be associated to the channel until it is first associated to the parent group-synced team." diff --git a/imports/placeholder.go b/imports/placeholder.go index 410f12d034..46d29c79da 100644 --- a/imports/placeholder.go +++ b/imports/placeholder.go @@ -36,4 +36,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/v6/jobs/resend_invitation_email" + + // This is a placeholder so this package can be imported in Team Edition when it will be otherwise empty. + _ "github.com/mattermost/mattermost-server/v6/jobs/extract_content" ) diff --git a/jobs/extract_content/worker.go b/jobs/extract_content/worker.go new file mode 100644 index 0000000000..ed6a427246 --- /dev/null +++ b/jobs/extract_content/worker.go @@ -0,0 +1,176 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package extract_content + +import ( + "net/http" + "strconv" + + "github.com/mattermost/mattermost-server/v6/app" + "github.com/mattermost/mattermost-server/v6/app/request" + "github.com/mattermost/mattermost-server/v6/jobs" + tjobs "github.com/mattermost/mattermost-server/v6/jobs/interfaces" + "github.com/mattermost/mattermost-server/v6/model" + "github.com/mattermost/mattermost-server/v6/shared/mlog" +) + +var ignoredFiles = map[string]bool{ + "png": true, "jpg": true, "jpeg": true, "gif": true, "wmv": true, + "mpg": true, "mpeg": true, "mp3": true, "mp4": true, "ogg": true, + "ogv": true, "mov": true, "apk": true, "svg": true, "webm": true, + "mkv": true, +} + +func init() { + app.RegisterJobsExtractContentInterface(func(s *app.Server) tjobs.ExtractContentInterface { + a := app.New(app.ServerConnector(s)) + return &ExtractContentInterfaceImpl{a} + }) +} + +type ExtractContentInterfaceImpl struct { + app *app.App +} + +type ExtractContentWorker struct { + name string + stopChan chan struct{} + stoppedChan chan struct{} + jobsChan chan model.Job + jobServer *jobs.JobServer + app *app.App + appContext *request.Context +} + +func (i *ExtractContentInterfaceImpl) MakeWorker() model.Worker { + return &ExtractContentWorker{ + name: "ExtractContent", + stopChan: make(chan struct{}), + stoppedChan: make(chan struct{}), + jobsChan: make(chan model.Job), + jobServer: i.app.Srv().Jobs, + app: i.app, + appContext: &request.Context{}, + } +} + +func (w *ExtractContentWorker) JobChannel() chan<- model.Job { + return w.jobsChan +} + +func (w *ExtractContentWorker) 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 *ExtractContentWorker) Stop() { + mlog.Debug("Worker stopping", mlog.String("worker", w.name)) + close(w.stopChan) + <-w.stoppedChan +} + +func (w *ExtractContentWorker) 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 + } + + var err error + var fromTS int64 = 0 + var toTS int64 = model.GetMillis() + if fromStr, ok := job.Data["from"]; ok { + if fromTS, err = strconv.ParseInt(fromStr, 10, 64); err != nil { + w.setJobError(job, model.NewAppError("ExtractContentWorker", "extrac_content.worker.do_job.invalid_input.from", nil, "", http.StatusBadRequest)) + return + } + fromTS *= 1000 + } + if toStr, ok := job.Data["to"]; ok { + if toTS, err = strconv.ParseInt(toStr, 10, 64); err != nil { + w.setJobError(job, model.NewAppError("ExtractContentWorker", "extrac_content.worker.do_job.invalid_input.to", nil, "", http.StatusBadRequest)) + return + } + toTS *= 1000 + } + + var nFiles int + var nErrs int + for { + opts := model.GetFileInfosOptions{ + Since: fromTS, + SortBy: model.FileinfoSortByCreated, + IncludeDeleted: false, + } + fileInfos, err := w.app.Srv().Store.FileInfo().GetWithOptions(0, 1000, &opts) + if err != nil { + w.setJobError(job, model.NewAppError("ExtractContentWorker", "extract_content.worker.do_job.file_info", nil, err.Error(), http.StatusInternalServerError)) + return + } + if len(fileInfos) == 0 { + break + } + for _, fileInfo := range fileInfos { + if !ignoredFiles[fileInfo.Extension] { + mlog.Debug("extracting file", mlog.String("filename", fileInfo.Name), mlog.String("filepath", fileInfo.Path)) + err = w.app.ExtractContentFromFileInfo(fileInfo) + if err != nil { + mlog.Warn("Failed to extract file content", mlog.Err(err), mlog.String("file_info_id", fileInfo.Id)) + nErrs++ + } + nFiles++ + } + } + lastFileInfo := fileInfos[len(fileInfos)-1] + if lastFileInfo.CreateAt > toTS { + break + } + fromTS = lastFileInfo.CreateAt + 1 + } + + job.Data["errors"] = strconv.Itoa(nErrs) + job.Data["processed"] = strconv.Itoa(nFiles) + w.updateData(job) + + mlog.Info("Worker: Job is complete", mlog.String("worker", w.name), mlog.String("job_id", job.Id)) + w.setJobSuccess(job) +} + +func (w *ExtractContentWorker) 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 *ExtractContentWorker) 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())) + } +} + +func (w *ExtractContentWorker) updateData(job *model.Job) { + if err := w.app.Srv().Jobs.UpdateInProgressJobData(job); err != nil { + mlog.Error("Worker: Failed to update job data", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error())) + } +} diff --git a/jobs/interfaces/extract_content_interface.go b/jobs/interfaces/extract_content_interface.go new file mode 100644 index 0000000000..643df0c6bb --- /dev/null +++ b/jobs/interfaces/extract_content_interface.go @@ -0,0 +1,12 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package interfaces + +import ( + "github.com/mattermost/mattermost-server/v6/model" +) + +type ExtractContentInterface interface { + MakeWorker() model.Worker +} diff --git a/jobs/jobs_watcher.go b/jobs/jobs_watcher.go index e103196e13..d519c428e9 100644 --- a/jobs/jobs_watcher.go +++ b/jobs/jobs_watcher.go @@ -191,6 +191,13 @@ func (watcher *Watcher) PollAndNotify() { default: } } + } else if job.Type == model.JobTypeExtractContent { + if watcher.workers.ExtractContent != nil { + select { + case watcher.workers.ExtractContent.JobChannel() <- *job: + default: + } + } } } } diff --git a/jobs/server.go b/jobs/server.go index 73961c12d6..b34b073927 100644 --- a/jobs/server.go +++ b/jobs/server.go @@ -36,6 +36,7 @@ type JobServer struct { ExportDelete tjobs.ExportDeleteInterface Cloud ejobs.CloudJobInterface ResendInvitationEmails ejobs.ResendInvitationEmailJobInterface + ExtractContent tjobs.ExtractContentInterface // mut is used to protect the following fields from concurrent access. mut sync.Mutex diff --git a/jobs/workers.go b/jobs/workers.go index 2d13b45fde..bb34ed4b97 100644 --- a/jobs/workers.go +++ b/jobs/workers.go @@ -32,6 +32,7 @@ type Workers struct { ExportDelete model.Worker Cloud model.Worker ResendInvitationEmail model.Worker + ExtractContent model.Worker listenerId string running bool @@ -124,6 +125,10 @@ func (srv *JobServer) InitWorkers() error { workers.ResendInvitationEmail = resendInvitationEmailInterface.MakeWorker() } + if extractContentInterface := srv.ExtractContent; extractContentInterface != nil { + workers.ExtractContent = extractContentInterface.MakeWorker() + } + srv.workers = workers return nil @@ -202,6 +207,10 @@ func (workers *Workers) Start() { go workers.ResendInvitationEmail.Run() } + if workers.ExtractContent != nil { + go workers.ExtractContent.Run() + } + go workers.Watcher.Start() workers.listenerId = workers.ConfigService.AddConfigListener(workers.handleConfigChange) @@ -335,6 +344,10 @@ func (workers *Workers) Stop() { workers.ResendInvitationEmail.Stop() } + if workers.ExtractContent != nil { + workers.ExtractContent.Stop() + } + workers.running = false mlog.Info("Stopped workers") diff --git a/model/job.go b/model/job.go index e0e9103de0..5f0188df1f 100644 --- a/model/job.go +++ b/model/job.go @@ -28,6 +28,7 @@ const ( JobTypeExportDelete = "export_delete" JobTypeCloud = "cloud" JobTypeResendInvitationEmail = "resend_invitation_email" + JobTypeExtractContent = "extract_content" JobStatusPending = "pending" JobStatusInProgress = "in_progress" @@ -55,6 +56,7 @@ var AllJobTypes = [...]string{ JobTypeExportProcess, JobTypeExportDelete, JobTypeCloud, + JobTypeExtractContent, } type Job struct { @@ -96,6 +98,7 @@ func (j *Job) IsValid() *AppError { case JobTypeExportDelete: case JobTypeCloud: case JobTypeResendInvitationEmail: + case JobTypeExtractContent: default: return NewAppError("Job.IsValid", "model.job.is_valid.type.app_error", nil, "id="+j.Id, http.StatusBadRequest) }