diff options
author | Corey Hulen <corey@hulen.com> | 2017-09-25 13:30:33 -0700 |
---|---|---|
committer | GitHub <noreply@github.com> | 2017-09-25 13:30:33 -0700 |
commit | 81c18a01bd22437457da04b6cdb8d409beb54446 (patch) | |
tree | 1c1a28dc72fe8685259870e53643de9f558dad79 | |
parent | 1bf2a2e8d5a552d23e16379e73c2d4aaefc9368f (diff) | |
download | chat-81c18a01bd22437457da04b6cdb8d409beb54446.tar.gz chat-81c18a01bd22437457da04b6cdb8d409beb54446.tar.bz2 chat-81c18a01bd22437457da04b6cdb8d409beb54446.zip |
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
-rw-r--r-- | app/app.go | 2 | ||||
-rw-r--r-- | app/ldap.go | 2 | ||||
-rw-r--r-- | cmd/platform/ldap.go | 6 | ||||
-rw-r--r-- | einterfaces/jobs/ldap_sync.go | 23 | ||||
-rw-r--r-- | einterfaces/ldap.go | 9 | ||||
-rw-r--r-- | i18n/en.json | 8 | ||||
-rw-r--r-- | jobs/jobs.go | 12 | ||||
-rw-r--r-- | jobs/jobs_watcher.go | 27 | ||||
-rw-r--r-- | jobs/schedulers.go | 21 | ||||
-rw-r--r-- | jobs/workers.go | 29 | ||||
-rw-r--r-- | model/job.go | 2 |
11 files changed, 119 insertions, 22 deletions
diff --git a/app/app.go b/app/app.go index 1289e4a6d..eaaf9acee 100644 --- a/app/app.go +++ b/app/app.go @@ -102,8 +102,6 @@ func (a *App) initEnterprise() { if err := utils.ValidateLdapFilter(cfg, a.Ldap); err != nil { panic(utils.T(err.Id)) } - - a.Ldap.StartLdapSyncJob() }) } if metricsInterface != nil { diff --git a/app/ldap.go b/app/ldap.go index 279ea3708..96791168e 100644 --- a/app/ldap.go +++ b/app/ldap.go @@ -15,7 +15,7 @@ func (a *App) SyncLdap() { go func() { if utils.IsLicensed() && *utils.License().Features.LDAP && *utils.Cfg.LdapSettings.Enable { if ldapI := a.Ldap; ldapI != nil { - ldapI.SyncNow() + ldapI.StartSynchronizeJob(false) } else { l4g.Error("%v", model.NewAppError("SyncLdap", "ent.ldap.disabled.app_error", nil, "", http.StatusNotImplemented).Error()) } diff --git a/cmd/platform/ldap.go b/cmd/platform/ldap.go index 16ad429bd..1bbcaa2f5 100644 --- a/cmd/platform/ldap.go +++ b/cmd/platform/ldap.go @@ -3,6 +3,7 @@ package main import ( + "github.com/mattermost/mattermost-server/model" "github.com/spf13/cobra" ) @@ -32,8 +33,9 @@ func ldapSyncCmdF(cmd *cobra.Command, args []string) error { } if ldapI := a.Ldap; ldapI != nil { - if err := ldapI.Syncronize(); err != nil { - CommandPrintErrorln("ERROR: AD/LDAP Synchronization Failed") + job, err := ldapI.StartSynchronizeJob(true) + if err != nil || job.Status == model.JOB_STATUS_ERROR || job.Status == model.JOB_STATUS_CANCELED { + CommandPrintErrorln("ERROR: AD/LDAP Synchronization please check the server logs") } else { CommandPrettyPrintln("SUCCESS: AD/LDAP Synchronization Complete") } diff --git a/einterfaces/jobs/ldap_sync.go b/einterfaces/jobs/ldap_sync.go new file mode 100644 index 000000000..97055bfcc --- /dev/null +++ b/einterfaces/jobs/ldap_sync.go @@ -0,0 +1,23 @@ +// Copyright (c) 2017-present Mattermost, Inc. All Rights Reserved. +// See License.txt for license information. + +package jobs + +import ( + "github.com/mattermost/mattermost-server/model" +) + +type LdapSyncInterface interface { + MakeWorker() model.Worker + MakeScheduler() model.Scheduler +} + +var theLdapSyncInterface LdapSyncInterface + +func RegisterLdapSyncInterface(newInterface LdapSyncInterface) { + theLdapSyncInterface = newInterface +} + +func GetLdapSyncInterface() LdapSyncInterface { + return theLdapSyncInterface +} diff --git a/einterfaces/ldap.go b/einterfaces/ldap.go index a69f9cbe1..1505a41ce 100644 --- a/einterfaces/ldap.go +++ b/einterfaces/ldap.go @@ -4,6 +4,8 @@ package einterfaces import ( + "github.com/go-ldap/ldap" + "github.com/mattermost/mattermost-server/model" ) @@ -14,9 +16,10 @@ type LdapInterface interface { CheckPassword(id string, password string) *model.AppError SwitchToLdap(userId, ldapId, ldapPassword string) *model.AppError ValidateFilter(filter string) *model.AppError - Syncronize() *model.AppError - StartLdapSyncJob() - SyncNow() + StartSynchronizeJob(waitForJobToFinish bool) (*model.Job, *model.AppError) RunTest() *model.AppError GetAllLdapUsers() ([]*model.User, *model.AppError) + UserFromLdapUser(ldapUser *ldap.Entry) *model.User + UserHasUpdateFromLdap(existingUser *model.User, currentLdapUser *model.User) bool + UpdateLdapUser(existingUser *model.User, currentLdapUser *model.User) *model.User } diff --git a/i18n/en.json b/i18n/en.json index 1d9040821..f08716721 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -3688,6 +3688,14 @@ "translation": "Compliance export started for job '{{.JobName}}' at '{{.FilePath}}'" }, { + "id": "ent.ldap.sync_worker.create_index_job.error", + "translation": "LDAP sync worker failed to create the sync job" + }, + { + "id": "ent.ldap.sync.index_job_failed.error", + "translation": "LDAP sync worker failed due to the sync job failing" + }, + { "id": "ent.elasticsearch.aggregator_worker.create_index_job.error", "translation": "Elasticsearch aggregator worker failed to create the indexing job" }, 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 diff --git a/model/job.go b/model/job.go index 258fa2bd3..85dfbec92 100644 --- a/model/job.go +++ b/model/job.go @@ -13,6 +13,7 @@ const ( JOB_TYPE_DATA_RETENTION = "data_retention" JOB_TYPE_ELASTICSEARCH_POST_INDEXING = "elasticsearch_post_indexing" JOB_TYPE_ELASTICSEARCH_POST_AGGREGATION = "elasticsearch_post_aggregation" + JOB_TYPE_LDAP_SYNC = "ldap_sync" JOB_STATUS_PENDING = "pending" JOB_STATUS_IN_PROGRESS = "in_progress" @@ -47,6 +48,7 @@ func (j *Job) IsValid() *AppError { case JOB_TYPE_DATA_RETENTION: case JOB_TYPE_ELASTICSEARCH_POST_INDEXING: case JOB_TYPE_ELASTICSEARCH_POST_AGGREGATION: + case JOB_TYPE_LDAP_SYNC: default: return NewAppError("Job.IsValid", "model.job.is_valid.type.app_error", nil, "id="+j.Id, http.StatusBadRequest) } |