summaryrefslogtreecommitdiffstats
path: root/jobs
diff options
context:
space:
mode:
authorJonathan <jonfritz@gmail.com>2017-11-30 09:07:04 -0500
committerGitHub <noreply@github.com>2017-11-30 09:07:04 -0500
commit375c0632fab03e3fb54865e320585888499c076d (patch)
treece6cba679337a82370d5a730c428bef21a9964bf /jobs
parentd0d9ba4a7e43301697d1c9f495930e0c0179fdc3 (diff)
downloadchat-375c0632fab03e3fb54865e320585888499c076d.tar.gz
chat-375c0632fab03e3fb54865e320585888499c076d.tar.bz2
chat-375c0632fab03e3fb54865e320585888499c076d.zip
PLT-7503: Create Message Export Scheduled Task and CLI Command (#7612)
* Created message export scheduled task * Added CLI command to immediately kick off an export job * Added email addresses for users joining and leaving the channel to the export * Added support for both MySQL and PostgreSQL * Fixing gofmt error * Added a new ChannelMemberHistory store and associated tests * Updating the ChannelMemberHistory channel as users create/join/leave channels * Added user email to the message export object so it can be included in the actiance export xml * Don't fail to log a leave event if a corresponding join event wasn't logged * Adding copyright notices * Adding message export settings to daily diagnostics report * Added System Console integration for message export * Cleaned up TODOs * Made batch size configurable * Added export from timestamp to CLI command * Made ChannelMemberHistory table updates best effort * Added a context-based timeout option to the message export CLI * Minor PR updates/improvements * Removed unnecessary fields from MessageExport object to reduce query overhead * Removed JSON functions from the message export query in an effort to optimize performance * Changed the way that channel member history queries and purges work to better account for edge cases * Fixing a test I missed with the last refactor * Added file copy functionality to file backend, improved config validation, added default config values * Fixed file copy tests * More concise use of the testing libraries * Fixed context leak error * Changed default export path to correctly place an 'export' directory under the 'data' directory * Can't delete records from a read replica * Fixed copy file tests * Start job workers when license is applied, if configured to do so * Suggestions from the PR * Moar unit tests * Fixed test imports
Diffstat (limited to 'jobs')
-rw-r--r--jobs/jobs_watcher.go7
-rw-r--r--jobs/schedulers.go4
-rw-r--r--jobs/server.go1
-rw-r--r--jobs/workers.go21
4 files changed, 33 insertions, 0 deletions
diff --git a/jobs/jobs_watcher.go b/jobs/jobs_watcher.go
index b36a99051..f519e7cca 100644
--- a/jobs/jobs_watcher.go
+++ b/jobs/jobs_watcher.go
@@ -78,6 +78,13 @@ func (watcher *Watcher) PollAndNotify() {
default:
}
}
+ } else if job.Type == model.JOB_TYPE_MESSAGE_EXPORT {
+ if watcher.workers.MessageExport != nil {
+ select {
+ case watcher.workers.MessageExport.JobChannel() <- *job:
+ default:
+ }
+ }
} else if job.Type == model.JOB_TYPE_ELASTICSEARCH_POST_INDEXING {
if watcher.workers.ElasticsearchIndexing != nil {
select {
diff --git a/jobs/schedulers.go b/jobs/schedulers.go
index cbe5f1749..bec53a49b 100644
--- a/jobs/schedulers.go
+++ b/jobs/schedulers.go
@@ -39,6 +39,10 @@ func (srv *JobServer) InitSchedulers() *Schedulers {
schedulers.schedulers = append(schedulers.schedulers, srv.DataRetentionJob.MakeScheduler())
}
+ if srv.MessageExportJob != nil {
+ schedulers.schedulers = append(schedulers.schedulers, srv.MessageExportJob.MakeScheduler())
+ }
+
if elasticsearchAggregatorInterface := srv.ElasticsearchAggregator; elasticsearchAggregatorInterface != nil {
schedulers.schedulers = append(schedulers.schedulers, elasticsearchAggregatorInterface.MakeScheduler())
}
diff --git a/jobs/server.go b/jobs/server.go
index 40cfb1f64..777b02a26 100644
--- a/jobs/server.go
+++ b/jobs/server.go
@@ -19,6 +19,7 @@ type JobServer struct {
Schedulers *Schedulers
DataRetentionJob ejobs.DataRetentionJobInterface
+ MessageExportJob ejobs.MessageExportJobInterface
ElasticsearchAggregator ejobs.ElasticsearchAggregatorInterface
ElasticsearchIndexer ejobs.ElasticsearchIndexerInterface
LdapSync ejobs.LdapSyncInterface
diff --git a/jobs/workers.go b/jobs/workers.go
index b1d275658..3abd7131c 100644
--- a/jobs/workers.go
+++ b/jobs/workers.go
@@ -17,6 +17,7 @@ type Workers struct {
Watcher *Watcher
DataRetention model.Worker
+ MessageExport model.Worker
ElasticsearchIndexing model.Worker
ElasticsearchAggregation model.Worker
LdapSync model.Worker
@@ -34,6 +35,10 @@ func (srv *JobServer) InitWorkers() *Workers {
workers.DataRetention = srv.DataRetentionJob.MakeWorker()
}
+ if srv.MessageExportJob != nil {
+ workers.MessageExport = srv.MessageExportJob.MakeWorker()
+ }
+
if elasticsearchIndexerInterface := srv.ElasticsearchIndexer; elasticsearchIndexerInterface != nil {
workers.ElasticsearchIndexing = elasticsearchIndexerInterface.MakeWorker()
}
@@ -57,6 +62,10 @@ func (workers *Workers) Start() *Workers {
go workers.DataRetention.Run()
}
+ if workers.MessageExport != nil && *workers.Config().MessageExportSettings.EnableExport {
+ go workers.MessageExport.Run()
+ }
+
if workers.ElasticsearchIndexing != nil && *workers.Config().ElasticsearchSettings.EnableIndexing {
go workers.ElasticsearchIndexing.Run()
}
@@ -86,6 +95,14 @@ func (workers *Workers) handleConfigChange(oldConfig *model.Config, newConfig *m
}
}
+ if workers.MessageExport != nil {
+ if !*oldConfig.MessageExportSettings.EnableExport && *newConfig.MessageExportSettings.EnableExport {
+ go workers.MessageExport.Run()
+ } else if *oldConfig.MessageExportSettings.EnableExport && !*newConfig.MessageExportSettings.EnableExport {
+ workers.MessageExport.Stop()
+ }
+ }
+
if workers.ElasticsearchIndexing != nil {
if !*oldConfig.ElasticsearchSettings.EnableIndexing && *newConfig.ElasticsearchSettings.EnableIndexing {
go workers.ElasticsearchIndexing.Run()
@@ -120,6 +137,10 @@ func (workers *Workers) Stop() *Workers {
workers.DataRetention.Stop()
}
+ if workers.MessageExport != nil && *workers.Config().MessageExportSettings.EnableExport {
+ workers.MessageExport.Stop()
+ }
+
if workers.ElasticsearchIndexing != nil && *workers.Config().ElasticsearchSettings.EnableIndexing {
workers.ElasticsearchIndexing.Stop()
}