diff options
Diffstat (limited to 'jobs')
| -rw-r--r-- | jobs/interfaces/migrations_interface.go | 11 | ||||
| -rw-r--r-- | jobs/jobs.go | 7 | ||||
| -rw-r--r-- | jobs/jobs_watcher.go | 7 | ||||
| -rw-r--r-- | jobs/schedulers.go | 4 | ||||
| -rw-r--r-- | jobs/server.go | 2 | ||||
| -rw-r--r-- | jobs/workers.go | 13 |
6 files changed, 44 insertions, 0 deletions
diff --git a/jobs/interfaces/migrations_interface.go b/jobs/interfaces/migrations_interface.go new file mode 100644 index 000000000..48dc9f579 --- /dev/null +++ b/jobs/interfaces/migrations_interface.go @@ -0,0 +1,11 @@ +// Copyright (c) 2018-present Mattermost, Inc. All Rights Reserved. +// See License.txt for license information. + +package interfaces + +import "github.com/mattermost/mattermost-server/model" + +type MigrationsJobInterface interface { + MakeWorker() model.Worker + MakeScheduler() model.Scheduler +} diff --git a/jobs/jobs.go b/jobs/jobs.go index 850491403..ddbc4489b 100644 --- a/jobs/jobs.go +++ b/jobs/jobs.go @@ -106,6 +106,13 @@ func (srv *JobServer) SetJobCanceled(job *model.Job) *model.AppError { return result.Err } +func (srv *JobServer) UpdateInProgressJobData(job *model.Job) *model.AppError { + job.Status = model.JOB_STATUS_IN_PROGRESS + job.LastActivityAt = model.GetMillis() + result := <-srv.Store.Job().UpdateOptimistically(job, model.JOB_STATUS_IN_PROGRESS) + return result.Err +} + func (srv *JobServer) RequestCancellation(jobId string) *model.AppError { if result := <-srv.Store.Job().UpdateStatusOptimistically(jobId, model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED); result.Err != nil { return result.Err diff --git a/jobs/jobs_watcher.go b/jobs/jobs_watcher.go index 07979442d..01d0a8d0f 100644 --- a/jobs/jobs_watcher.go +++ b/jobs/jobs_watcher.go @@ -107,6 +107,13 @@ func (watcher *Watcher) PollAndNotify() { default: } } + } else if job.Type == model.JOB_TYPE_MIGRATIONS { + if watcher.workers.Migrations != nil { + select { + case watcher.workers.Migrations.JobChannel() <- *job: + default: + } + } } } } diff --git a/jobs/schedulers.go b/jobs/schedulers.go index 2823036df..96aa2b635 100644 --- a/jobs/schedulers.go +++ b/jobs/schedulers.go @@ -50,6 +50,10 @@ func (srv *JobServer) InitSchedulers() *Schedulers { schedulers.schedulers = append(schedulers.schedulers, ldapSyncInterface.MakeScheduler()) } + if migrationsInterface := srv.Migrations; migrationsInterface != nil { + schedulers.schedulers = append(schedulers.schedulers, migrationsInterface.MakeScheduler()) + } + schedulers.nextRunTimes = make([]*time.Time, len(schedulers.schedulers)) return schedulers } diff --git a/jobs/server.go b/jobs/server.go index 01cf821dc..10ea9a46f 100644 --- a/jobs/server.go +++ b/jobs/server.go @@ -5,6 +5,7 @@ package jobs import ( ejobs "github.com/mattermost/mattermost-server/einterfaces/jobs" + tjobs "github.com/mattermost/mattermost-server/jobs/interfaces" "github.com/mattermost/mattermost-server/model" "github.com/mattermost/mattermost-server/store" ) @@ -34,6 +35,7 @@ type JobServer struct { ElasticsearchAggregator ejobs.ElasticsearchAggregatorInterface ElasticsearchIndexer ejobs.ElasticsearchIndexerInterface LdapSync ejobs.LdapSyncInterface + Migrations tjobs.MigrationsJobInterface } func NewJobServer(configService ConfigService, store store.Store) *JobServer { diff --git a/jobs/workers.go b/jobs/workers.go index 57a255013..67ab43241 100644 --- a/jobs/workers.go +++ b/jobs/workers.go @@ -20,6 +20,7 @@ type Workers struct { ElasticsearchIndexing model.Worker ElasticsearchAggregation model.Worker LdapSync model.Worker + Migrations model.Worker listenerId string } @@ -50,6 +51,10 @@ func (srv *JobServer) InitWorkers() *Workers { workers.LdapSync = ldapSyncInterface.MakeWorker() } + if migrationsInterface := srv.Migrations; migrationsInterface != nil { + workers.Migrations = migrationsInterface.MakeWorker() + } + return workers } @@ -77,6 +82,10 @@ func (workers *Workers) Start() *Workers { go workers.LdapSync.Run() } + if workers.Migrations != nil { + go workers.Migrations.Run() + } + go workers.Watcher.Start() }) @@ -152,6 +161,10 @@ func (workers *Workers) Stop() *Workers { workers.LdapSync.Stop() } + if workers.Migrations != nil { + workers.Migrations.Stop() + } + mlog.Info("Stopped workers") return workers |