summaryrefslogtreecommitdiffstats
path: root/jobs/workers.go
diff options
context:
space:
mode:
Diffstat (limited to 'jobs/workers.go')
-rw-r--r--jobs/workers.go29
1 files changed, 24 insertions, 5 deletions
diff --git a/jobs/workers.go b/jobs/workers.go
index 592c001fb..fe38641e7 100644
--- a/jobs/workers.go
+++ b/jobs/workers.go
@@ -16,16 +16,15 @@ type Workers struct {
startOnce sync.Once
watcher *Watcher
- DataRetention model.Worker
- ElasticsearchIndexing model.Worker
+ DataRetention model.Worker
+ ElasticsearchIndexing model.Worker
+ ElasticsearchAggregation model.Worker
listenerId string
}
func InitWorkers() *Workers {
- workers := &Workers{
- // SearchIndexing: MakeTestJob(s, "SearchIndexing"),
- }
+ workers := &Workers{}
workers.watcher = MakeWatcher(workers)
if dataRetentionInterface := ejobs.GetDataRetentionInterface(); dataRetentionInterface != nil {
@@ -36,6 +35,10 @@ func InitWorkers() *Workers {
workers.ElasticsearchIndexing = elasticsearchIndexerInterface.MakeWorker()
}
+ if elasticsearchAggregatorInterface := ejobs.GetElasticsearchAggregatorInterface(); elasticsearchAggregatorInterface != nil {
+ workers.ElasticsearchAggregation = elasticsearchAggregatorInterface.MakeWorker()
+ }
+
return workers
}
@@ -51,6 +54,10 @@ func (workers *Workers) Start() *Workers {
go workers.ElasticsearchIndexing.Run()
}
+ if workers.ElasticsearchAggregation != nil && *utils.Cfg.ElasticsearchSettings.EnableIndexing {
+ go workers.ElasticsearchAggregation.Run()
+ }
+
go workers.watcher.Start()
})
@@ -75,6 +82,14 @@ func (workers *Workers) handleConfigChange(oldConfig *model.Config, newConfig *m
workers.ElasticsearchIndexing.Stop()
}
}
+
+ if workers.ElasticsearchAggregation != nil {
+ if !*oldConfig.ElasticsearchSettings.EnableIndexing && *newConfig.ElasticsearchSettings.EnableIndexing {
+ go workers.ElasticsearchAggregation.Run()
+ } else if *oldConfig.ElasticsearchSettings.EnableIndexing && !*newConfig.ElasticsearchSettings.EnableIndexing {
+ workers.ElasticsearchAggregation.Stop()
+ }
+ }
}
func (workers *Workers) Stop() *Workers {
@@ -90,6 +105,10 @@ func (workers *Workers) Stop() *Workers {
workers.ElasticsearchIndexing.Stop()
}
+ if workers.ElasticsearchAggregation != nil && *utils.Cfg.ElasticsearchSettings.EnableIndexing {
+ workers.ElasticsearchAggregation.Stop()
+ }
+
l4g.Info("Stopped workers")
return workers