From 81c18a01bd22437457da04b6cdb8d409beb54446 Mon Sep 17 00:00:00 2001 From: Corey Hulen Date: Mon, 25 Sep 2017 13:30:33 -0700 Subject: PLT-7542 Converting LDAP sync to the job server (#7452) * PLT-7542 Converting LDAP sync to the job server * Fixing minor issues * Fixing build failure * Translate error message * Translate error message * Translate error message * Translate error message * Fixing merge * Fixing bad merge --- jobs/jobs.go | 12 +++++++++++- jobs/jobs_watcher.go | 27 ++++++++++++++++++--------- jobs/schedulers.go | 21 +++++++++++++++++++++ jobs/workers.go | 29 +++++++++++++++++++++++++---- 4 files changed, 75 insertions(+), 14 deletions(-) (limited to 'jobs') diff --git a/jobs/jobs.go b/jobs/jobs.go index 6e1230291..a51780865 100644 --- a/jobs/jobs.go +++ b/jobs/jobs.go @@ -11,6 +11,7 @@ import ( l4g "github.com/alecthomas/log4go" "github.com/mattermost/mattermost-server/model" + "github.com/mattermost/mattermost-server/utils" ) const ( @@ -37,6 +38,14 @@ func CreateJob(jobType string, jobData map[string]string) (*model.Job, *model.Ap return &job, nil } +func GetJob(id string) (*model.Job, *model.AppError) { + if result := <-Srv.Store.Job().Get(id); result.Err != nil { + return nil, result.Err + } else { + return result.Data.(*model.Job), nil + } +} + 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 @@ -73,7 +82,8 @@ func SetJobError(job *model.Job, jobError *model.AppError) *model.AppError { if job.Data == nil { job.Data = make(map[string]string) } - job.Data["error"] = jobError.Error() + jobError.Translate(utils.T) + job.Data["error"] = jobError.Message + " (" + jobError.DetailedError + ")" if result := <-Srv.Store.Job().UpdateOptimistically(job, model.JOB_STATUS_IN_PROGRESS); result.Err != nil { return result.Err diff --git a/jobs/jobs_watcher.go b/jobs/jobs_watcher.go index 8355fc11d..56cf9eb2e 100644 --- a/jobs/jobs_watcher.go +++ b/jobs/jobs_watcher.go @@ -12,21 +12,23 @@ import ( ) const ( - WATCHER_POLLING_INTERVAL = 15000 + DEFAULT_WATCHER_POLLING_INTERVAL = 15000 ) type Watcher struct { workers *Workers - stop chan bool - stopped chan bool + stop chan bool + stopped chan bool + pollingInterval int } -func MakeWatcher(workers *Workers) *Watcher { +func MakeWatcher(workers *Workers, pollingInterval int) *Watcher { return &Watcher{ - stop: make(chan bool, 1), - stopped: make(chan bool, 1), - workers: workers, + stop: make(chan bool, 1), + stopped: make(chan bool, 1), + pollingInterval: pollingInterval, + workers: workers, } } @@ -36,7 +38,7 @@ func (watcher *Watcher) Start() { // 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) + _ = <-time.After(time.Duration(rand.Intn(watcher.pollingInterval)) * time.Millisecond) defer func() { l4g.Debug("Watcher Finished") @@ -48,7 +50,7 @@ func (watcher *Watcher) Start() { case <-watcher.stop: l4g.Debug("Watcher: Received stop signal") return - case <-time.After(WATCHER_POLLING_INTERVAL * time.Millisecond): + case <-time.After(time.Duration(watcher.pollingInterval) * time.Millisecond): watcher.PollAndNotify() } } @@ -88,6 +90,13 @@ func (watcher *Watcher) PollAndNotify() { default: } } + } else if job.Type == model.JOB_TYPE_LDAP_SYNC { + if watcher.workers.LdapSync != nil { + select { + case watcher.workers.LdapSync.JobChannel() <- *job: + default: + } + } } } } diff --git a/jobs/schedulers.go b/jobs/schedulers.go index 28cf6808c..1cb4a6f28 100644 --- a/jobs/schedulers.go +++ b/jobs/schedulers.go @@ -18,6 +18,7 @@ type Schedulers struct { DataRetention model.Scheduler ElasticsearchAggregation model.Scheduler + LdapSync model.Scheduler listenerId string } @@ -33,6 +34,10 @@ func InitSchedulers() *Schedulers { schedulers.ElasticsearchAggregation = elasticsearchAggregatorInterface.MakeScheduler() } + if ldaySyncInterface := ejobs.GetLdapSyncInterface(); ldaySyncInterface != nil { + schedulers.LdapSync = ldaySyncInterface.MakeScheduler() + } + return schedulers } @@ -47,6 +52,10 @@ func (schedulers *Schedulers) Start() *Schedulers { 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) @@ -70,6 +79,14 @@ func (schedulers *Schedulers) handleConfigChange(oldConfig *model.Config, newCon 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 { @@ -83,6 +100,10 @@ func (schedulers *Schedulers) Stop() *Schedulers { schedulers.ElasticsearchAggregation.Stop() } + if schedulers.LdapSync != nil && *utils.Cfg.LdapSettings.Enable { + schedulers.LdapSync.Stop() + } + l4g.Info("Stopped schedulers") return schedulers diff --git a/jobs/workers.go b/jobs/workers.go index b473ba092..9f85adaf5 100644 --- a/jobs/workers.go +++ b/jobs/workers.go @@ -14,18 +14,19 @@ import ( type Workers struct { startOnce sync.Once - watcher *Watcher + Watcher *Watcher DataRetention model.Worker ElasticsearchIndexing model.Worker ElasticsearchAggregation model.Worker + LdapSync model.Worker listenerId string } func InitWorkers() *Workers { workers := &Workers{} - workers.watcher = MakeWatcher(workers) + workers.Watcher = MakeWatcher(workers, DEFAULT_WATCHER_POLLING_INTERVAL) if dataRetentionInterface := ejobs.GetDataRetentionInterface(); dataRetentionInterface != nil { workers.DataRetention = dataRetentionInterface.MakeWorker() @@ -39,6 +40,10 @@ func InitWorkers() *Workers { workers.ElasticsearchAggregation = elasticsearchAggregatorInterface.MakeWorker() } + if ldapSyncInterface := ejobs.GetLdapSyncInterface(); ldapSyncInterface != nil { + workers.LdapSync = ldapSyncInterface.MakeWorker() + } + return workers } @@ -58,7 +63,11 @@ func (workers *Workers) Start() *Workers { go workers.ElasticsearchAggregation.Run() } - go workers.watcher.Start() + if workers.LdapSync != nil && *utils.Cfg.LdapSettings.Enable { + go workers.LdapSync.Run() + } + + go workers.Watcher.Start() }) workers.listenerId = utils.AddConfigListener(workers.handleConfigChange) @@ -90,12 +99,20 @@ func (workers *Workers) handleConfigChange(oldConfig *model.Config, newConfig *m workers.ElasticsearchAggregation.Stop() } } + + if workers.LdapSync != nil { + if !*oldConfig.LdapSettings.Enable && *newConfig.LdapSettings.Enable { + go workers.LdapSync.Run() + } else if *oldConfig.LdapSettings.Enable && !*newConfig.LdapSettings.Enable { + workers.LdapSync.Stop() + } + } } func (workers *Workers) Stop() *Workers { utils.RemoveConfigListener(workers.listenerId) - workers.watcher.Stop() + workers.Watcher.Stop() if workers.DataRetention != nil && (*utils.Cfg.DataRetentionSettings.EnableMessageDeletion || *utils.Cfg.DataRetentionSettings.EnableFileDeletion) { workers.DataRetention.Stop() @@ -109,6 +126,10 @@ func (workers *Workers) Stop() *Workers { workers.ElasticsearchAggregation.Stop() } + if workers.LdapSync != nil && *utils.Cfg.LdapSettings.Enable { + workers.LdapSync.Stop() + } + l4g.Info("Stopped workers") return workers -- cgit v1.2.3-1-g7c22