Files
mattermost/app/app.go
Jonathan 375c0632fa 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
2017-11-30 09:07:04 -05:00

420 lines
11 KiB
Go

// Copyright (c) 2016-present Mattermost, Inc. All Rights Reserved.
// See License.txt for license information.
package app
import (
"html/template"
"net"
"net/http"
"runtime/debug"
"strings"
"sync/atomic"
l4g "github.com/alecthomas/log4go"
"github.com/gorilla/mux"
"github.com/mattermost/mattermost-server/einterfaces"
ejobs "github.com/mattermost/mattermost-server/einterfaces/jobs"
"github.com/mattermost/mattermost-server/jobs"
"github.com/mattermost/mattermost-server/model"
"github.com/mattermost/mattermost-server/plugin/pluginenv"
"github.com/mattermost/mattermost-server/store"
"github.com/mattermost/mattermost-server/store/sqlstore"
"github.com/mattermost/mattermost-server/utils"
)
type App struct {
goroutineCount int32
goroutineExitSignal chan struct{}
Srv *Server
PluginEnv *pluginenv.Environment
PluginConfigListenerId string
EmailBatching *EmailBatchingJob
Hubs []*Hub
HubsStopCheckingForDeadlock chan bool
Jobs *jobs.JobServer
AccountMigration einterfaces.AccountMigrationInterface
Brand einterfaces.BrandInterface
Cluster einterfaces.ClusterInterface
Compliance einterfaces.ComplianceInterface
DataRetention einterfaces.DataRetentionInterface
Elasticsearch einterfaces.ElasticsearchInterface
Emoji einterfaces.EmojiInterface
Ldap einterfaces.LdapInterface
MessageExport einterfaces.MessageExportInterface
Metrics einterfaces.MetricsInterface
Mfa einterfaces.MfaInterface
Saml einterfaces.SamlInterface
configFile string
newStore func() store.Store
htmlTemplateWatcher *utils.HTMLTemplateWatcher
sessionCache *utils.Cache
roles map[string]*model.Role
configListenerId string
}
var appCount = 0
// New creates a new App. You must call Shutdown when you're done with it.
// XXX: For now, only one at a time is allowed as some resources are still shared.
func New(options ...Option) *App {
appCount++
if appCount > 1 {
panic("Only one App should exist at a time. Did you forget to call Shutdown()?")
}
app := &App{
goroutineExitSignal: make(chan struct{}, 1),
Srv: &Server{
Router: mux.NewRouter(),
},
sessionCache: utils.NewLru(model.SESSION_CACHE_SIZE),
configFile: "config.json",
}
for _, option := range options {
option(app)
}
if utils.T == nil {
utils.TranslationsPreInit()
}
utils.LoadGlobalConfig(app.configFile)
utils.InitTranslations(utils.Cfg.LocalizationSettings)
app.configListenerId = utils.AddConfigListener(func(_, cfg *model.Config) {
app.SetDefaultRolesBasedOnConfig()
})
app.SetDefaultRolesBasedOnConfig()
l4g.Info(utils.T("api.server.new_server.init.info"))
app.initEnterprise()
if app.newStore == nil {
app.newStore = func() store.Store {
return store.NewLayeredStore(sqlstore.NewSqlSupplier(app.Config().SqlSettings, app.Metrics), app.Metrics, app.Cluster)
}
}
if htmlTemplateWatcher, err := utils.NewHTMLTemplateWatcher("templates"); err != nil {
l4g.Error(utils.T("api.api.init.parsing_templates.error"), err)
} else {
app.htmlTemplateWatcher = htmlTemplateWatcher
}
app.Srv.Store = app.newStore()
app.initJobs()
app.initBuiltInPlugins()
app.Srv.Router.HandleFunc("/plugins/{plugin_id:[A-Za-z0-9\\_\\-\\.]+}", app.ServePluginRequest)
app.Srv.Router.HandleFunc("/plugins/{plugin_id:[A-Za-z0-9\\_\\-\\.]+}/{anything:.*}", app.ServePluginRequest)
app.Srv.Router.NotFoundHandler = http.HandlerFunc(app.Handle404)
app.Srv.WebSocketRouter = &WebSocketRouter{
app: app,
handlers: make(map[string]webSocketHandler),
}
return app
}
func (a *App) Shutdown() {
appCount--
l4g.Info(utils.T("api.server.stop_server.stopping.info"))
a.StopServer()
a.HubStop()
a.ShutDownPlugins()
a.WaitForGoroutines()
a.Srv.Store.Close()
a.Srv = nil
if a.htmlTemplateWatcher != nil {
a.htmlTemplateWatcher.Close()
}
utils.RemoveConfigListener(a.configListenerId)
l4g.Info(utils.T("api.server.stop_server.stopped.info"))
}
var accountMigrationInterface func(*App) einterfaces.AccountMigrationInterface
func RegisterAccountMigrationInterface(f func(*App) einterfaces.AccountMigrationInterface) {
accountMigrationInterface = f
}
var brandInterface func(*App) einterfaces.BrandInterface
func RegisterBrandInterface(f func(*App) einterfaces.BrandInterface) {
brandInterface = f
}
var clusterInterface func(*App) einterfaces.ClusterInterface
func RegisterClusterInterface(f func(*App) einterfaces.ClusterInterface) {
clusterInterface = f
}
var complianceInterface func(*App) einterfaces.ComplianceInterface
func RegisterComplianceInterface(f func(*App) einterfaces.ComplianceInterface) {
complianceInterface = f
}
var dataRetentionInterface func(*App) einterfaces.DataRetentionInterface
func RegisterDataRetentionInterface(f func(*App) einterfaces.DataRetentionInterface) {
dataRetentionInterface = f
}
var elasticsearchInterface func(*App) einterfaces.ElasticsearchInterface
func RegisterElasticsearchInterface(f func(*App) einterfaces.ElasticsearchInterface) {
elasticsearchInterface = f
}
var emojiInterface func(*App) einterfaces.EmojiInterface
func RegisterEmojiInterface(f func(*App) einterfaces.EmojiInterface) {
emojiInterface = f
}
var jobsDataRetentionJobInterface func(*App) ejobs.DataRetentionJobInterface
func RegisterJobsDataRetentionJobInterface(f func(*App) ejobs.DataRetentionJobInterface) {
jobsDataRetentionJobInterface = f
}
var jobsMessageExportJobInterface func(*App) ejobs.MessageExportJobInterface
func RegisterJobsMessageExportJobInterface(f func(*App) ejobs.MessageExportJobInterface) {
jobsMessageExportJobInterface = f
}
var jobsElasticsearchAggregatorInterface func(*App) ejobs.ElasticsearchAggregatorInterface
func RegisterJobsElasticsearchAggregatorInterface(f func(*App) ejobs.ElasticsearchAggregatorInterface) {
jobsElasticsearchAggregatorInterface = f
}
var jobsElasticsearchIndexerInterface func(*App) ejobs.ElasticsearchIndexerInterface
func RegisterJobsElasticsearchIndexerInterface(f func(*App) ejobs.ElasticsearchIndexerInterface) {
jobsElasticsearchIndexerInterface = f
}
var jobsLdapSyncInterface func(*App) ejobs.LdapSyncInterface
func RegisterJobsLdapSyncInterface(f func(*App) ejobs.LdapSyncInterface) {
jobsLdapSyncInterface = f
}
var ldapInterface func(*App) einterfaces.LdapInterface
func RegisterLdapInterface(f func(*App) einterfaces.LdapInterface) {
ldapInterface = f
}
var messageExportInterface func(*App) einterfaces.MessageExportInterface
func RegisterMessageExportInterface(f func(*App) einterfaces.MessageExportInterface) {
messageExportInterface = f
}
var metricsInterface func(*App) einterfaces.MetricsInterface
func RegisterMetricsInterface(f func(*App) einterfaces.MetricsInterface) {
metricsInterface = f
}
var mfaInterface func(*App) einterfaces.MfaInterface
func RegisterMfaInterface(f func(*App) einterfaces.MfaInterface) {
mfaInterface = f
}
var samlInterface func(*App) einterfaces.SamlInterface
func RegisterSamlInterface(f func(*App) einterfaces.SamlInterface) {
samlInterface = f
}
func (a *App) initEnterprise() {
if accountMigrationInterface != nil {
a.AccountMigration = accountMigrationInterface(a)
}
if brandInterface != nil {
a.Brand = brandInterface(a)
}
if clusterInterface != nil {
a.Cluster = clusterInterface(a)
}
if complianceInterface != nil {
a.Compliance = complianceInterface(a)
}
if elasticsearchInterface != nil {
a.Elasticsearch = elasticsearchInterface(a)
}
if emojiInterface != nil {
a.Emoji = emojiInterface(a)
}
if ldapInterface != nil {
a.Ldap = ldapInterface(a)
utils.AddConfigListener(func(_, cfg *model.Config) {
if err := utils.ValidateLdapFilter(cfg, a.Ldap); err != nil {
panic(utils.T(err.Id))
}
})
}
if messageExportInterface != nil {
a.MessageExport = messageExportInterface(a)
}
if metricsInterface != nil {
a.Metrics = metricsInterface(a)
}
if mfaInterface != nil {
a.Mfa = mfaInterface(a)
}
if samlInterface != nil {
a.Saml = samlInterface(a)
utils.AddConfigListener(func(_, cfg *model.Config) {
a.Saml.ConfigureSP()
})
}
if dataRetentionInterface != nil {
a.DataRetention = dataRetentionInterface(a)
}
}
func (a *App) initJobs() {
a.Jobs = jobs.NewJobServer(a.Config, a.Srv.Store)
if jobsDataRetentionJobInterface != nil {
a.Jobs.DataRetentionJob = jobsDataRetentionJobInterface(a)
}
if jobsMessageExportJobInterface != nil {
a.Jobs.MessageExportJob = jobsMessageExportJobInterface(a)
}
if jobsElasticsearchAggregatorInterface != nil {
a.Jobs.ElasticsearchAggregator = jobsElasticsearchAggregatorInterface(a)
}
if jobsElasticsearchIndexerInterface != nil {
a.Jobs.ElasticsearchIndexer = jobsElasticsearchIndexerInterface(a)
}
if jobsLdapSyncInterface != nil {
a.Jobs.LdapSync = jobsLdapSyncInterface(a)
}
}
func (a *App) Config() *model.Config {
return utils.Cfg
}
func (a *App) UpdateConfig(f func(*model.Config)) {
old := utils.Cfg.Clone()
f(utils.Cfg)
utils.InvokeGlobalConfigListeners(old, utils.Cfg)
}
func (a *App) PersistConfig() {
utils.SaveConfig(a.ConfigFileName(), a.Config())
}
func (a *App) ReloadConfig() {
debug.FreeOSMemory()
utils.LoadGlobalConfig(a.ConfigFileName())
// start/restart email batching job if necessary
a.InitEmailBatching()
}
func (a *App) ConfigFileName() string {
return utils.CfgFileName
}
// Go creates a goroutine, but maintains a record of it to ensure that execution completes before
// the app is destroyed.
func (a *App) Go(f func()) {
atomic.AddInt32(&a.goroutineCount, 1)
go func() {
f()
atomic.AddInt32(&a.goroutineCount, -1)
select {
case a.goroutineExitSignal <- struct{}{}:
default:
}
}()
}
// WaitForGoroutines blocks until all goroutines created by App.Go exit.
func (a *App) WaitForGoroutines() {
for atomic.LoadInt32(&a.goroutineCount) != 0 {
<-a.goroutineExitSignal
}
}
func (a *App) HTMLTemplates() *template.Template {
return a.htmlTemplateWatcher.Templates()
}
func (a *App) HTTPClient(trustURLs bool) *http.Client {
insecure := a.Config().ServiceSettings.EnableInsecureOutgoingConnections != nil && *a.Config().ServiceSettings.EnableInsecureOutgoingConnections
if trustURLs {
return utils.NewHTTPClient(insecure, nil, nil)
}
allowHost := func(host string) bool {
if a.Config().ServiceSettings.AllowedUntrustedInternalConnections == nil {
return false
}
for _, allowed := range strings.Fields(*a.Config().ServiceSettings.AllowedUntrustedInternalConnections) {
if host == allowed {
return true
}
}
return false
}
allowIP := func(ip net.IP) bool {
if !utils.IsReservedIP(ip) {
return true
}
if a.Config().ServiceSettings.AllowedUntrustedInternalConnections == nil {
return false
}
for _, allowed := range strings.Fields(*a.Config().ServiceSettings.AllowedUntrustedInternalConnections) {
if _, ipRange, err := net.ParseCIDR(allowed); err == nil && ipRange.Contains(ip) {
return true
}
}
return false
}
return utils.NewHTTPClient(insecure, allowHost, allowIP)
}
func (a *App) Handle404(w http.ResponseWriter, r *http.Request) {
err := model.NewAppError("Handle404", "api.context.404.app_error", nil, "", http.StatusNotFound)
err.Translate(utils.T)
l4g.Debug("%v: code=404 ip=%v", r.URL.Path, utils.GetIpAddress(r))
utils.RenderWebError(err, w, r)
}