PLT-7302: Aggregate Elasticsearch indexes over a certain age. (#7224)

* PLT-7302: Aggregate Elasticsearch indexes over a certain age.

This is done by a scheduled daily job, in order to keep the shard count
to a sensible level in Elasticsearch.

* Use map[string]string instead of StringMap

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