PLT-7644: Improve job scheduler architecture. (#7532)
Этот коммит содержится в:
George Goldberg
коммит произвёл
Corey Hulen
Corey Hulen
родитель
f263d2b951
Коммит
a06830b2f8
26
jobs/jobs.go
26
jobs/jobs.go
@@ -141,3 +141,29 @@ func CancellationWatcher(ctx context.Context, jobId string, cancelChan chan inte
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func GenerateNextStartDateTime(now time.Time, nextStartTime time.Time) *time.Time {
|
||||
nextTime := time.Date(now.Year(), now.Month(), now.Day(), nextStartTime.Hour(), nextStartTime.Minute(), 0, 0, time.Local)
|
||||
|
||||
if !now.Before(nextTime) {
|
||||
nextTime = nextTime.AddDate(0, 0, 1)
|
||||
}
|
||||
|
||||
return &nextTime
|
||||
}
|
||||
|
||||
func CheckForPendingJobsByType(jobType string) (bool, *model.AppError) {
|
||||
if result := <-Srv.Store.Job().GetCountByStatusAndType(model.JOB_STATUS_PENDING, jobType); result.Err != nil {
|
||||
return false, result.Err
|
||||
} else {
|
||||
return result.Data.(int64) > 0, nil
|
||||
}
|
||||
}
|
||||
|
||||
func GetLastSuccessfulJobByType(jobType string) (*model.Job, *model.AppError) {
|
||||
if result := <-Srv.Store.Job().GetNewestJobByStatusAndType(model.JOB_STATUS_SUCCESS, jobType); result.Err != nil {
|
||||
return nil, result.Err
|
||||
} else {
|
||||
return result.Data.(*model.Job), nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,106 +5,160 @@ package jobs
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
l4g "github.com/alecthomas/log4go"
|
||||
ejobs "github.com/mattermost/mattermost-server/einterfaces/jobs"
|
||||
|
||||
ejobs "github.com/mattermost/mattermost-server/einterfaces/jobs"
|
||||
"github.com/mattermost/mattermost-server/model"
|
||||
"github.com/mattermost/mattermost-server/utils"
|
||||
)
|
||||
|
||||
type Schedulers struct {
|
||||
startOnce sync.Once
|
||||
stop chan bool
|
||||
stopped chan bool
|
||||
configChanged chan *model.Config
|
||||
listenerId string
|
||||
startOnce sync.Once
|
||||
|
||||
DataRetention model.Scheduler
|
||||
ElasticsearchAggregation model.Scheduler
|
||||
LdapSync model.Scheduler
|
||||
|
||||
listenerId string
|
||||
schedulers []model.Scheduler
|
||||
nextRunTimes []*time.Time
|
||||
}
|
||||
|
||||
func InitSchedulers() *Schedulers {
|
||||
schedulers := &Schedulers{}
|
||||
l4g.Debug("Initialising schedulers.")
|
||||
schedulers := &Schedulers{
|
||||
stop: make(chan bool),
|
||||
stopped: make(chan bool),
|
||||
configChanged: make(chan *model.Config),
|
||||
}
|
||||
|
||||
if dataRetentionInterface := ejobs.GetDataRetentionInterface(); dataRetentionInterface != nil {
|
||||
schedulers.DataRetention = dataRetentionInterface.MakeScheduler()
|
||||
schedulers.schedulers = append(schedulers.schedulers, dataRetentionInterface.MakeScheduler())
|
||||
}
|
||||
|
||||
if elasticsearchAggregatorInterface := ejobs.GetElasticsearchAggregatorInterface(); elasticsearchAggregatorInterface != nil {
|
||||
schedulers.ElasticsearchAggregation = elasticsearchAggregatorInterface.MakeScheduler()
|
||||
schedulers.schedulers = append(schedulers.schedulers, elasticsearchAggregatorInterface.MakeScheduler())
|
||||
}
|
||||
|
||||
if ldaySyncInterface := ejobs.GetLdapSyncInterface(); ldaySyncInterface != nil {
|
||||
schedulers.LdapSync = ldaySyncInterface.MakeScheduler()
|
||||
if ldapSyncInterface := ejobs.GetLdapSyncInterface(); ldapSyncInterface != nil {
|
||||
schedulers.schedulers = append(schedulers.schedulers, ldapSyncInterface.MakeScheduler())
|
||||
}
|
||||
|
||||
schedulers.nextRunTimes = make([]*time.Time, len(schedulers.schedulers))
|
||||
return schedulers
|
||||
}
|
||||
|
||||
func (schedulers *Schedulers) Start() *Schedulers {
|
||||
l4g.Info("Starting schedulers")
|
||||
|
||||
schedulers.startOnce.Do(func() {
|
||||
if schedulers.DataRetention != nil && (*utils.Cfg.DataRetentionSettings.EnableMessageDeletion || *utils.Cfg.DataRetentionSettings.EnableFileDeletion) {
|
||||
go schedulers.DataRetention.Run()
|
||||
}
|
||||
|
||||
if schedulers.ElasticsearchAggregation != nil && *utils.Cfg.ElasticsearchSettings.EnableIndexing {
|
||||
go schedulers.ElasticsearchAggregation.Run()
|
||||
}
|
||||
|
||||
if schedulers.LdapSync != nil && *utils.Cfg.LdapSettings.Enable {
|
||||
go schedulers.LdapSync.Run()
|
||||
}
|
||||
})
|
||||
|
||||
schedulers.listenerId = utils.AddConfigListener(schedulers.handleConfigChange)
|
||||
|
||||
go func() {
|
||||
schedulers.startOnce.Do(func() {
|
||||
l4g.Info("Starting schedulers.")
|
||||
|
||||
defer func() {
|
||||
l4g.Info("Schedulers stopped.")
|
||||
close(schedulers.stopped)
|
||||
}()
|
||||
|
||||
now := time.Now()
|
||||
for idx, scheduler := range schedulers.schedulers {
|
||||
if !scheduler.Enabled(utils.Cfg) {
|
||||
schedulers.nextRunTimes[idx] = nil
|
||||
} else {
|
||||
schedulers.setNextRunTime(utils.Cfg, idx, now, false)
|
||||
}
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-schedulers.stop:
|
||||
l4g.Debug("Schedulers received stop signal.")
|
||||
return
|
||||
case now = <-time.After(1 * time.Minute):
|
||||
cfg := utils.Cfg
|
||||
|
||||
for idx, nextTime := range schedulers.nextRunTimes {
|
||||
if nextTime == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if time.Now().After(*nextTime) {
|
||||
scheduler := schedulers.schedulers[idx]
|
||||
if scheduler != nil {
|
||||
if scheduler.Enabled(cfg) {
|
||||
if _, err := schedulers.scheduleJob(cfg, scheduler); err != nil {
|
||||
l4g.Warn("Failed to schedule job with scheduler: %v", scheduler.Name())
|
||||
l4g.Error(err)
|
||||
} else {
|
||||
schedulers.setNextRunTime(cfg, idx, now, true)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
case newCfg := <-schedulers.configChanged:
|
||||
for idx, scheduler := range schedulers.schedulers {
|
||||
if !scheduler.Enabled(newCfg) {
|
||||
schedulers.nextRunTimes[idx] = nil
|
||||
} else {
|
||||
schedulers.setNextRunTime(newCfg, idx, now, false)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}()
|
||||
|
||||
return schedulers
|
||||
}
|
||||
|
||||
func (schedulers *Schedulers) handleConfigChange(oldConfig *model.Config, newConfig *model.Config) {
|
||||
if schedulers.DataRetention != nil {
|
||||
if (!*oldConfig.DataRetentionSettings.EnableMessageDeletion && !*oldConfig.DataRetentionSettings.EnableFileDeletion) && (*newConfig.DataRetentionSettings.EnableMessageDeletion || *newConfig.DataRetentionSettings.EnableFileDeletion) {
|
||||
go schedulers.DataRetention.Run()
|
||||
} else if (*oldConfig.DataRetentionSettings.EnableMessageDeletion || *oldConfig.DataRetentionSettings.EnableFileDeletion) && (!*newConfig.DataRetentionSettings.EnableMessageDeletion && !*newConfig.DataRetentionSettings.EnableFileDeletion) {
|
||||
schedulers.DataRetention.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
if schedulers.ElasticsearchAggregation != nil {
|
||||
if !*oldConfig.ElasticsearchSettings.EnableIndexing && *newConfig.ElasticsearchSettings.EnableIndexing {
|
||||
go schedulers.ElasticsearchAggregation.Run()
|
||||
} else if *oldConfig.ElasticsearchSettings.EnableIndexing && !*newConfig.ElasticsearchSettings.EnableIndexing {
|
||||
schedulers.ElasticsearchAggregation.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
if schedulers.LdapSync != nil {
|
||||
if !*oldConfig.LdapSettings.Enable && *newConfig.LdapSettings.Enable {
|
||||
go schedulers.LdapSync.Run()
|
||||
} else if *oldConfig.LdapSettings.Enable && !*newConfig.LdapSettings.Enable {
|
||||
schedulers.LdapSync.Stop()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (schedulers *Schedulers) Stop() *Schedulers {
|
||||
utils.RemoveConfigListener(schedulers.listenerId)
|
||||
|
||||
if schedulers.DataRetention != nil && (*utils.Cfg.DataRetentionSettings.EnableMessageDeletion || *utils.Cfg.DataRetentionSettings.EnableFileDeletion) {
|
||||
schedulers.DataRetention.Stop()
|
||||
}
|
||||
|
||||
if schedulers.ElasticsearchAggregation != nil && *utils.Cfg.ElasticsearchSettings.EnableIndexing {
|
||||
schedulers.ElasticsearchAggregation.Stop()
|
||||
}
|
||||
|
||||
if schedulers.LdapSync != nil && *utils.Cfg.LdapSettings.Enable {
|
||||
schedulers.LdapSync.Stop()
|
||||
}
|
||||
|
||||
l4g.Info("Stopped schedulers")
|
||||
|
||||
l4g.Info("Stopping schedulers.")
|
||||
close(schedulers.stop)
|
||||
<-schedulers.stopped
|
||||
return schedulers
|
||||
}
|
||||
|
||||
func (schedulers *Schedulers) setNextRunTime(cfg *model.Config, idx int, now time.Time, pendingJobs bool) {
|
||||
scheduler := schedulers.schedulers[idx]
|
||||
|
||||
if !pendingJobs {
|
||||
if pj, err := CheckForPendingJobsByType(scheduler.JobType()); err != nil {
|
||||
l4g.Error("Failed to set next job run time: " + err.Error())
|
||||
schedulers.nextRunTimes[idx] = nil
|
||||
return
|
||||
} else {
|
||||
pendingJobs = pj
|
||||
}
|
||||
}
|
||||
|
||||
lastSuccessfulJob, err := GetLastSuccessfulJobByType(scheduler.JobType())
|
||||
if err != nil {
|
||||
l4g.Error("Failed to set next job run time: " + err.Error())
|
||||
schedulers.nextRunTimes[idx] = nil
|
||||
return
|
||||
}
|
||||
|
||||
schedulers.nextRunTimes[idx] = scheduler.NextScheduleTime(cfg, now, pendingJobs, lastSuccessfulJob)
|
||||
l4g.Debug("Next run time for scheduler %v: %v", scheduler.Name(), schedulers.nextRunTimes[idx])
|
||||
}
|
||||
|
||||
func (schedulers *Schedulers) scheduleJob(cfg *model.Config, scheduler model.Scheduler) (*model.Job, *model.AppError) {
|
||||
pendingJobs, err := CheckForPendingJobsByType(scheduler.JobType())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
lastSuccessfulJob, err2 := GetLastSuccessfulJobByType(scheduler.JobType())
|
||||
if err2 != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return scheduler.ScheduleJob(cfg, pendingJobs, lastSuccessfulJob)
|
||||
}
|
||||
|
||||
func (schedulers *Schedulers) handleConfigChange(oldConfig *model.Config, newConfig *model.Config) {
|
||||
l4g.Debug("Schedulers received config change.")
|
||||
schedulers.configChanged <- newConfig
|
||||
}
|
||||
|
||||
@@ -1,58 +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"
|
||||
)
|
||||
|
||||
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
|
||||
}
|
||||
Ссылка в новой задаче
Block a user