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

4 files changed, 75 insertions, 14 deletions

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