mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

PLT-6916: Redesign the jobs package and Jobserver. (#6733)

Просмотр исходного кода 
This commit redesigns the jobserver to be based around an architecture
of "workers", which carry out jobs of a particular type, and "jobs"
which are a unit of work carried by a particular worker. It also
introduces "schedulers" which are responsible for scheduling jobs of a
particular type automatically (jobs can also be scheduled manually when
apropriate).

Workers may be run many times, either in instances of the platform
binary, or the standalone jobserver binary. In any mattermost cluster,
only one instance of platform OR jobserver must run the schedulers. At
the moment this is controlled by a config variable, but in future will
be controlled through the cluster leader election process.
Этот коммит содержится в:
George Goldberg
2017-07-07 15:21:02 +01:00
коммит произвёл GitHub
родитель 6e0f5f0969
Коммит 0495a51949
31 изменённых файлов: 1370 добавлений и 583 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

136
jobs/jobs.go
Просмотреть файл

@@ -4,71 +4,111 @@
package jobs
import (
"sync"
"context"
"time"
l4g "github.com/alecthomas/log4go"
ejobs "github.com/mattermost/platform/einterfaces/jobs"
"github.com/mattermost/platform/model"
"github.com/mattermost/platform/store"
"github.com/mattermost/platform/utils"
)
type Jobs struct {
startOnce sync.Once
const (
CANCEL_WATCHER_POLLING_INTERVAL = 5000
)
DataRetention model.Job
// SearchIndexing model.Job
listenerId string
}
func InitJobs(s store.Store) *Jobs {
jobs := &Jobs{
// SearchIndexing: MakeTestJob(s, "SearchIndexing"),
func CreateJob(jobType string, jobData map[string]interface{}) (*model.Job, *model.AppError) {
job := model.Job{
Id: model.NewId(),
Type: jobType,
CreateAt: model.GetMillis(),
Status: model.JOB_STATUS_PENDING,
Data: jobData,
}
if dataRetentionInterface := ejobs.GetDataRetentionInterface(); dataRetentionInterface != nil {
jobs.DataRetention = dataRetentionInterface.MakeJob(s)
if result := <-Srv.Store.Job().Save(&job); result.Err != nil {
return nil, result.Err
}
return jobs
return &job, nil
}
func (jobs *Jobs) Start() *Jobs {
l4g.Info("Starting jobs")
func ClaimJob(job *model.Job) (bool, *model.AppError) {
if result := <-Srv.Store.Job().UpdateStatusOptimistically(job.Id, model.JOB_STATUS_PENDING, model.JOB_STATUS_IN_PROGRESS); result.Err != nil {
return false, result.Err
} else {
success := result.Data.(bool)
return success, nil
}
}
jobs.startOnce.Do(func() {
if jobs.DataRetention != nil && *utils.Cfg.DataRetentionSettings.Enable {
go jobs.DataRetention.Run()
func SetJobProgress(jobId string, progress int64) (bool, *model.AppError) {
var job *model.Job
if result := <-Srv.Store.Job().Get(jobId); result.Err != nil {
return false, result.Err
} else {
job = result.Data.(*model.Job)
}
job.Status = model.JOB_STATUS_IN_PROGRESS
job.Progress = progress
if result := <-Srv.Store.Job().UpdateOptimistically(job, model.JOB_STATUS_IN_PROGRESS); result.Err != nil {
return false, result.Err
} else {
if !result.Data.(bool) {
return false, nil
}
}
// go jobs.SearchIndexing.Run()
})
jobs.listenerId = utils.AddConfigListener(jobs.handleConfigChange)
return jobs
return true, nil
}
func (jobs *Jobs) handleConfigChange(oldConfig *model.Config, newConfig *model.Config) {
if jobs.DataRetention != nil {
if !*oldConfig.DataRetentionSettings.Enable && *newConfig.DataRetentionSettings.Enable {
go jobs.DataRetention.Run()
} else if *oldConfig.DataRetentionSettings.Enable && !*newConfig.DataRetentionSettings.Enable {
jobs.DataRetention.Stop()
func SetJobSuccess(job *model.Job) *model.AppError {
result := <-Srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_SUCCESS)
return result.Err
}
func SetJobError(job *model.Job) *model.AppError {
result := <-Srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_ERROR)
return result.Err
}
func SetJobCanceled(job *model.Job) *model.AppError {
result := <-Srv.Store.Job().UpdateStatus(job.Id, model.JOB_STATUS_CANCELED)
return result.Err
}
func RequestCancellation(job *model.Job) *model.AppError {
if result := <-Srv.Store.Job().UpdateStatusOptimistically(job.Id, model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED); result.Err != nil {
return result.Err
} else if result.Data.(bool) {
return nil
}
if result := <-Srv.Store.Job().UpdateStatusOptimistically(job.Id, model.JOB_STATUS_IN_PROGRESS, model.JOB_STATUS_CANCEL_REQUESTED); result.Err != nil {
return result.Err
} else if result.Data.(bool) {
return nil
}
return model.NewLocAppError("Jobs.RequestCancellation", "jobs.request_cancellation.status.error", nil, "id=" + job.Id)
}
func CancellationWatcher(ctx context.Context, jobId string, cancelChan chan interface{}) {
for {
select {
case <-ctx.Done():
l4g.Debug("CancellationWatcher for Job: %v Aborting as job has finished.", jobId)
return
case <-time.After(CANCEL_WATCHER_POLLING_INTERVAL * time.Millisecond):
l4g.Debug("CancellationWatcher for Job: %v polling.", jobId)
if result := <-Srv.Store.Job().Get(jobId); result.Err == nil {
jobStatus := result.Data.(*model.Job)
if jobStatus.Status == model.JOB_STATUS_CANCEL_REQUESTED {
close(cancelChan)
return
}
}
}
}
}
func (jobs *Jobs) Stop() *Jobs {
utils.RemoveConfigListener(jobs.listenerId)
if jobs.DataRetention != nil && *utils.Cfg.DataRetentionSettings.Enable {
jobs.DataRetention.Stop()
}
// jobs.SearchIndexing.Stop()
l4g.Info("Stopped jobs")
return jobs
}

85
jobs/jobs_watcher.go Обычный файл
Просмотреть файл

@@ -0,0 +1,85 @@
// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package jobs
import (
"math/rand"
"time"
l4g "github.com/alecthomas/log4go"
"github.com/mattermost/platform/model"
)
const (
WATCHER_POLLING_INTERVAL = 15000
)
type Watcher struct {
workers *Workers
stop chan bool
stopped chan bool
}
func MakeWatcher(workers *Workers) *Watcher {
return &Watcher{
stop: make(chan bool, 1),
stopped: make(chan bool, 1),
workers: workers,
}
}
func (watcher *Watcher) Start() {
l4g.Debug("Watcher Started")
// Delay for some random number of milliseconds before starting to ensure that multiple
// instances of the jobserver don't poll at a time too close to each other.
rand.Seed(time.Now().UTC().UnixNano())
_ = <-time.After(time.Duration(rand.Intn(WATCHER_POLLING_INTERVAL)) * time.Millisecond)
defer func(){
l4g.Debug("Watcher Finished")
watcher.stopped <- true
}()
for {
select {
case <-watcher.stop:
l4g.Debug("Watcher: Received stop signal")
return
case <-time.After(WATCHER_POLLING_INTERVAL * time.Millisecond):
watcher.PollAndNotify()
}
}
}
func (watcher *Watcher) Stop() {
l4g.Debug("Watcher Stopping")
watcher.stop <- true
<-watcher.stopped
}
func (watcher *Watcher) PollAndNotify() {
if result := <-Srv.Store.Job().GetAllByStatus(model.JOB_STATUS_PENDING); result.Err != nil {
l4g.Error("Error occured getting all pending statuses: %v", result.Err.Error())
} else {
jobStatuses := result.Data.([]*model.Job)
for _, js := range jobStatuses {
j := model.Job{
Type: js.Type,
Id: js.Id,
}
if js.Type == model.JOB_TYPE_DATA_RETENTION {
if watcher.workers.DataRetention != nil {
select {
case watcher.workers.DataRetention.JobChannel() <- j:
default:
}
}
}
}
}
}

15
jobs/jobserver/jobserver.go
Просмотреть файл

@@ -16,22 +16,20 @@ import (
_ "github.com/mattermost/platform/imports"
)
var Srv jobs.JobServer
func main() {
// Initialize
utils.InitAndLoadConfig("config.json")
defer l4g.Close()
Srv.Store = store.NewLayeredStore()
defer Srv.Store.Close()
jobs.Srv.Store = store.NewLayeredStore()
defer jobs.Srv.Store.Close()
Srv.LoadLicense()
jobs.Srv.LoadLicense()
// Run jobs
l4g.Info("Starting Mattermost job server")
Srv.Jobs = jobs.InitJobs(Srv.Store)
Srv.Jobs.Start()
jobs.Srv.StartWorkers()
jobs.Srv.StartSchedulers()
var signalChan chan os.Signal = make(chan os.Signal)
signal.Notify(signalChan, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
@@ -40,7 +38,8 @@ func main() {
// Cleanup anything that isn't handled by a defer statement
l4g.Info("Stopping Mattermost job server")
Srv.Jobs.Stop()
jobs.Srv.StopSchedulers()
jobs.Srv.StopWorkers()
l4g.Info("Stopped Mattermost job server")
}

68
jobs/schedulers.go Обычный файл
Просмотреть файл

@@ -0,0 +1,68 @@
// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package jobs
import (
"sync"
l4g "github.com/alecthomas/log4go"
ejobs "github.com/mattermost/platform/einterfaces/jobs"
"github.com/mattermost/platform/model"
"github.com/mattermost/platform/utils"
)
type Schedulers struct {
startOnce sync.Once
DataRetention model.Scheduler
listenerId string
}
func InitSchedulers() *Schedulers {
schedulers := &Schedulers{}
if dataRetentionInterface := ejobs.GetDataRetentionInterface(); dataRetentionInterface != nil {
schedulers.DataRetention = dataRetentionInterface.MakeScheduler()
}
return schedulers
}
func (schedulers *Schedulers) Start() *Schedulers {
l4g.Info("Starting schedulers")
schedulers.startOnce.Do(func() {
if schedulers.DataRetention != nil && *utils.Cfg.DataRetentionSettings.Enable {
go schedulers.DataRetention.Run()
}
})
schedulers.listenerId = utils.AddConfigListener(schedulers.handleConfigChange)
return schedulers
}
func (schedulers *Schedulers) handleConfigChange(oldConfig *model.Config, newConfig *model.Config) {
if schedulers.DataRetention != nil {
if !*oldConfig.DataRetentionSettings.Enable && *newConfig.DataRetentionSettings.Enable {
go schedulers.DataRetention.Run()
} else if *oldConfig.DataRetentionSettings.Enable && !*newConfig.DataRetentionSettings.Enable {
schedulers.DataRetention.Stop()
}
}
}
func (schedulers *Schedulers) Stop() *Schedulers {
utils.RemoveConfigListener(schedulers.listenerId)
if schedulers.DataRetention != nil && *utils.Cfg.DataRetentionSettings.Enable {
schedulers.DataRetention.Stop()
}
l4g.Info("Stopped schedulers")
return schedulers
}

31
jobs/server.go
Просмотреть файл

@@ -11,10 +11,13 @@ import (
)
type JobServer struct {
Store store.Store
Jobs *Jobs
Store store.Store
Workers *Workers
Schedulers *Schedulers
}
var Srv JobServer
func (server *JobServer) LoadLicense() {
licenseId := ""
if result := <-server.Store.System().Get(); result.Err == nil {
@@ -44,3 +47,27 @@ func (server *JobServer) LoadLicense() {
l4g.Info(utils.T("mattermost.load_license.find.warn"))
}
}
func (server *JobServer) StartWorkers() {
if *utils.Cfg.JobSettings.RunJobs {
Srv.Workers = InitWorkers().Start()
}
}
func (server *JobServer) StartSchedulers() {
if *utils.Cfg.JobSettings.RunJobs {
Srv.Schedulers = InitSchedulers().Start()
}
}
func (server *JobServer) StopWorkers() {
if Srv.Workers != nil {
Srv.Workers.Stop()
}
}
func (server *JobServer) StopSchedulers() {
if Srv.Schedulers != nil {
Srv.Schedulers.Stop()
}
}

54
jobs/testjob.go
Просмотреть файл

@@ -1,54 +0,0 @@
// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package jobs
import (
"time"
l4g "github.com/alecthomas/log4go"
"github.com/mattermost/platform/store"
)
type TestJob struct {
store store.Store
name string
stop chan bool
stopped chan bool
}
func MakeTestJob(s store.Store, name string) *TestJob {
return &TestJob{
store: s,
name: name,
stop: make(chan bool, 1),
stopped: make(chan bool, 1),
}
}
func (job *TestJob) Run() {
l4g.Debug("Job %v: Started", job.name)
running := true
for running {
l4g.Debug("Job %v: Tick", job.name)
select {
case <-job.stop:
l4g.Debug("Job %v: Received stop signal", job.name)
running = false
case <-time.After(10 * time.Second):
continue
}
}
l4g.Debug("Job %v: Finished", job.name)
job.stopped <- true
}
func (job *TestJob) Stop() {
l4g.Debug("Job %v: Stopping", job.name)
job.stop <- true
<-job.stopped
}

58
jobs/testscheduler.go Обычный файл
Просмотреть файл

@@ -0,0 +1,58 @@
// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package jobs
import (
"time"
l4g "github.com/alecthomas/log4go"
)
type TestScheduler struct {
name string
jobType string
stop chan bool
stopped chan bool
}
func MakeTestScheduler(name string, jobType string) *TestScheduler {
return &TestScheduler{
name: name,
jobType: jobType,
stop: make(chan bool, 1),
stopped: make(chan bool, 1),
}
}
func (scheduler *TestScheduler) Run() {
l4g.Debug("Scheduler %v: Started", scheduler.name)
defer func(){
l4g.Debug("Scheduler %v: Finished", scheduler.name)
scheduler.stopped <- true
}()
for {
select {
case <-scheduler.stop:
l4g.Debug("Scheduler %v: Received stop signal", scheduler.name)
return
case <-time.After(86400 * time.Second):
l4g.Debug("Scheduler: %v: Scheduling new job", scheduler.name)
scheduler.AddJob()
}
}
}
func (scheduler *TestScheduler) AddJob() {
if _, err := CreateJob(scheduler.jobType, nil); err != nil {
l4g.Error("Scheduler %v: failed to create job: %v", scheduler.name, err)
}
}
func (scheduler *TestScheduler) Stop() {
l4g.Debug("Scheduler %v: Stopping", scheduler.name)
scheduler.stop <- true
<-scheduler.stopped
}

104
jobs/testworker.go Обычный файл
Просмотреть файл

@@ -0,0 +1,104 @@
// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package jobs
import (
"context"
"time"
l4g "github.com/alecthomas/log4go"
"github.com/mattermost/platform/model"
)
type TestWorker struct {
name string
stop chan bool
stopped chan bool
jobs chan model.Job
}
func MakeTestWorker(name string) *TestWorker {
return &TestWorker{
name: name,
stop: make(chan bool, 1),
stopped: make(chan bool, 1),
jobs: make(chan model.Job),
}
}
func (worker *TestWorker) Run() {
l4g.Debug("Worker %v: Started", worker.name)
defer func() {
l4g.Debug("Worker %v: Finished", worker.name)
worker.stopped <- true
}()
for {
select {
case <-worker.stop:
l4g.Debug("Worker %v: Received stop signal", worker.name)
return
case job := <-worker.jobs:
l4g.Debug("Worker %v: Received a new candidate job.", worker.name)
worker.DoJob(&job)
}
}
}
func (worker *TestWorker) DoJob(job *model.Job) {
if claimed, err := ClaimJob(job); err != nil {
l4g.Error("Job: %v: Error occurred while trying to claim job: %v", job.Id, err.Error())
return
} else if !claimed {
return
}
cancelCtx, cancelCancelWatcher := context.WithCancel(context.Background())
cancelWatcherChan := make(chan interface{}, 1)
go CancellationWatcher(cancelCtx, job.Id, cancelWatcherChan)
defer cancelCancelWatcher()
counter := 0
for {
select {
case <-cancelWatcherChan:
l4g.Debug("Job %v: Job has been canceled via CancellationWatcher.", job.Id)
if err := SetJobCanceled(job); err != nil {
l4g.Error("Failed to mark job: %v as canceled. Error: %v", job.Id, err.Error())
}
return
case <-worker.stop:
l4g.Debug("Job %v: Job has been canceled via Worker Stop.", job.Id)
if err := SetJobCanceled(job); err != nil {
l4g.Error("Failed to mark job: %v as canceled. Error: %v", job.Id, err.Error())
}
return
case <-time.After(5 * time.Second):
counter++
if counter > 10 {
l4g.Debug("Job %v: Job completed.", job.Id)
if err := SetJobSuccess(job); err != nil {
l4g.Error("Failed to mark job: %v as succeeded. Error: %v", job.Id, err.Error())
}
return
} else {
if _, err := SetJobProgress(job.Id, int64(counter*10)); err != nil {
l4g.Error("Job: %v: an error occured while trying to set job progress: %v", job.Id, err.Error())
}
}
}
}
}
func (worker *TestWorker) Stop() {
l4g.Debug("Worker %v: Stopping", worker.name)
worker.stop <- true
<-worker.stopped
}
func (worker *TestWorker) JobChannel() chan<- model.Job {
return worker.jobs
}

79
jobs/workers.go Обычный файл
Просмотреть файл

@@ -0,0 +1,79 @@
// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package jobs
import (
"sync"
l4g "github.com/alecthomas/log4go"
ejobs "github.com/mattermost/platform/einterfaces/jobs"
"github.com/mattermost/platform/model"
"github.com/mattermost/platform/utils"
)
type Workers struct {
startOnce sync.Once
watcher *Watcher
DataRetention model.Worker
// SearchIndexing model.Job
listenerId string
}
func InitWorkers() *Workers {
workers := &Workers{
// SearchIndexing: MakeTestJob(s, "SearchIndexing"),
}
workers.watcher = MakeWatcher(workers)
if dataRetentionInterface := ejobs.GetDataRetentionInterface(); dataRetentionInterface != nil {
workers.DataRetention = dataRetentionInterface.MakeWorker()
}
return workers
}
func (workers *Workers) Start() *Workers {
l4g.Info("Starting workers")
workers.startOnce.Do(func() {
if workers.DataRetention != nil && *utils.Cfg.DataRetentionSettings.Enable {
go workers.DataRetention.Run()
}
// go workers.SearchIndexing.Run()
go workers.watcher.Start()
})
workers.listenerId = utils.AddConfigListener(workers.handleConfigChange)
return workers
}
func (workers *Workers) handleConfigChange(oldConfig *model.Config, newConfig *model.Config) {
if workers.DataRetention != nil {
if !*oldConfig.DataRetentionSettings.Enable && *newConfig.DataRetentionSettings.Enable {
go workers.DataRetention.Run()
} else if *oldConfig.DataRetentionSettings.Enable && !*newConfig.DataRetentionSettings.Enable {
workers.DataRetention.Stop()
}
}
}
func (workers *Workers) Stop() *Workers {
utils.RemoveConfigListener(workers.listenerId)
workers.watcher.Stop()
if workers.DataRetention != nil && *utils.Cfg.DataRetentionSettings.Enable {
workers.DataRetention.Stop()
}
// workers.SearchIndexing.Stop()
l4g.Info("Stopped workers")
return workers
}