mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

Cleanup related to context refactor (#9988)

Просмотр исходного кода
Этот коммит содержится в:
Christopher Speller
2018-12-17 08:51:46 -08:00
коммит произвёл GitHub
родитель 829f183bb0
Коммит 8429add371
55 изменённых файлов: 1027 добавлений и 949 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

438
app/server.go
Просмотреть файл

@@ -14,7 +14,6 @@ import (
"net/http"
"net/url"
"os"
"path"
"strings"
"sync"
"sync/atomic"
@@ -33,10 +32,10 @@ import (
"github.com/mattermost/mattermost-server/model"
"github.com/mattermost/mattermost-server/plugin"
"github.com/mattermost/mattermost-server/services/httpservice"
"github.com/mattermost/mattermost-server/services/mailservice"
"github.com/mattermost/mattermost-server/services/timezones"
"github.com/mattermost/mattermost-server/store"
"github.com/mattermost/mattermost-server/store/sqlstore"
"github.com/mattermost/mattermost-server/utils"
"github.com/mattermost/mattermost-server/utils/fileutils"
)
var MaxNotificationsPerChannelDefault int64 = 1000000
@@ -73,7 +72,8 @@ type Server struct {
PushNotificationsHub PushNotificationsHub
Jobs *jobs.JobServer
runjobs bool
Jobs *jobs.JobServer
config atomic.Value
envConfig map[string]interface{}
@@ -85,7 +85,7 @@ type Server struct {
clientLicenseValue atomic.Value
licenseListeners map[string]func()
timezones atomic.Value
timezones *timezones.Timezones
newStore func() store.Store
@@ -124,150 +124,6 @@ type Server struct {
Saml einterfaces.SamlInterface
}
// This is a bridge between the old and new initalization for the context refactor.
// It calls app layer initalization code that then turns around and acts on the server.
// Don't add anything new here, new initilization should be done in the server and
// performed in the NewServer function.
func (s *Server) RunOldAppInitalization() error {
a := s.FakeApp()
a.CreatePushNotificationsHub()
a.StartPushNotificationsHubWorkers()
if utils.T == nil {
if err := utils.TranslationsPreInit(); err != nil {
return errors.Wrapf(err, "unable to load Mattermost translation files")
}
}
model.AppErrorInit(utils.T)
a.LoadTimezones()
if err := utils.InitTranslations(a.Config().LocalizationSettings); err != nil {
return errors.Wrapf(err, "unable to load Mattermost translation files")
}
a.Srv.configListenerId = a.AddConfigListener(func(_, _ *model.Config) {
a.configOrLicenseListener()
message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_CONFIG_CHANGED, "", "", "", nil)
message.Add("config", a.ClientConfigWithComputed())
a.Srv.Go(func() {
a.Publish(message)
})
})
a.Srv.licenseListenerId = a.AddLicenseListener(func() {
a.configOrLicenseListener()
message := model.NewWebSocketEvent(model.WEBSOCKET_EVENT_LICENSE_CHANGED, "", "", "", nil)
message.Add("license", a.GetSanitizedClientLicense())
a.Srv.Go(func() {
a.Publish(message)
})
})
if err := a.SetupInviteEmailRateLimiting(); err != nil {
return err
}
mlog.Info("Server is initializing...")
s.initEnterprise()
if a.Srv.newStore == nil {
a.Srv.newStore = func() store.Store {
return store.NewLayeredStore(sqlstore.NewSqlSupplier(a.Config().SqlSettings, a.Metrics), a.Metrics, a.Cluster)
}
}
if htmlTemplateWatcher, err := utils.NewHTMLTemplateWatcher("templates"); err != nil {
mlog.Error(fmt.Sprintf("Failed to parse server templates %v", err))
} else {
a.Srv.htmlTemplateWatcher = htmlTemplateWatcher
}
a.Srv.Store = a.Srv.newStore()
if err := a.ensureAsymmetricSigningKey(); err != nil {
return errors.Wrapf(err, "unable to ensure asymmetric signing key")
}
if err := a.ensureInstallationDate(); err != nil {
return errors.Wrapf(err, "unable to ensure installation date")
}
a.EnsureDiagnosticId()
a.regenerateClientConfig()
s.initJobs()
a.AddLicenseListener(func() {
s.initJobs()
})
a.Srv.clusterLeaderListenerId = a.AddClusterLeaderChangedListener(func() {
mlog.Info("Cluster leader changed. Determining if job schedulers should be running:", mlog.Bool("isLeader", a.IsLeader()))
a.Srv.Jobs.Schedulers.HandleClusterLeaderChange(a.IsLeader())
})
subpath, err := utils.GetSubpathFromConfig(a.Config())
if err != nil {
return errors.Wrap(err, "failed to parse SiteURL subpath")
}
a.Srv.Router = a.Srv.RootRouter.PathPrefix(subpath).Subrouter()
a.Srv.Router.HandleFunc("/plugins/{plugin_id:[A-Za-z0-9\\_\\-\\.]+}", a.ServePluginRequest)
a.Srv.Router.HandleFunc("/plugins/{plugin_id:[A-Za-z0-9\\_\\-\\.]+}/{anything:.*}", a.ServePluginRequest)
// If configured with a subpath, redirect 404s at the root back into the subpath.
if subpath != "/" {
a.Srv.RootRouter.NotFoundHandler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
r.URL.Path = path.Join(subpath, r.URL.Path)
http.Redirect(w, r, r.URL.String(), http.StatusFound)
})
}
a.Srv.Router.NotFoundHandler = http.HandlerFunc(a.Handle404)
a.Srv.WebSocketRouter = &WebSocketRouter{
app: a,
handlers: make(map[string]webSocketHandler),
}
mailservice.TestConnection(a.Config())
if _, err := url.ParseRequestURI(*a.Config().ServiceSettings.SiteURL); err != nil {
mlog.Error("SiteURL must be set. Some features will operate incorrectly if the SiteURL is not set. See documentation for details: http://about.mattermost.com/default-site-url")
}
backend, appErr := a.FileBackend()
if appErr == nil {
appErr = backend.TestConnection()
}
if appErr != nil {
mlog.Error("Problem with file storage settings: " + appErr.Error())
}
if model.BuildEnterpriseReady == "true" {
a.LoadLicense()
}
a.DoAdvancedPermissionsMigration()
a.DoEmojisPermissionsMigration()
a.InitPostMetadata()
a.InitPlugins(*a.Config().PluginSettings.Directory, *a.Config().PluginSettings.ClientDirectory)
a.AddConfigListener(func(prevCfg, cfg *model.Config) {
if *cfg.PluginSettings.Enable {
a.InitPlugins(*cfg.PluginSettings.Directory, *a.Config().PluginSettings.ClientDirectory)
} else {
a.ShutDownPlugins()
}
})
return nil
}
func NewServer(options ...Option) (*Server, error) {
rootRouter := mux.NewRouter()
@@ -305,11 +161,21 @@ func NewServer(options ...Option) (*Server, error) {
s.HTTPService = httpservice.MakeHTTPService(s.FakeApp())
if utils.T == nil {
if err := utils.TranslationsPreInit(); err != nil {
return nil, errors.Wrapf(err, "unable to load Mattermost translation files")
}
}
err := s.RunOldAppInitalization()
if err != nil {
return nil, err
}
model.AppErrorInit(utils.T)
s.timezones = timezones.New("")
// Start email batching because it's not like the other jobs
s.InitEmailBatching()
s.AddConfigListener(func(_, _ *model.Config) {
@@ -320,7 +186,7 @@ func NewServer(options ...Option) (*Server, error) {
mlog.Info(fmt.Sprintf("Enterprise Enabled: %v", model.BuildEnterpriseReady))
pwd, _ := os.Getwd()
mlog.Info(fmt.Sprintf("Current working directory is %v", pwd))
mlog.Info(fmt.Sprintf("Loaded config file from %v", utils.FindConfigFile(s.configFile)))
mlog.Info(fmt.Sprintf("Loaded config file from %v", fileutils.FindConfigFile(s.configFile)))
license := s.License()
@@ -348,6 +214,50 @@ func NewServer(options ...Option) (*Server, error) {
mlog.Error(fmt.Sprint("Error to reset the server status.", result.Err.Error()))
}
if s.Cluster != nil {
s.FakeApp().RegisterAllClusterMessageHandlers()
s.Cluster.StartInterNodeCommunication()
}
if s.Metrics != nil {
s.Metrics.StartServer()
}
if s.Elasticsearch != nil {
s.StartElasticsearch()
}
s.initJobs()
if s.runjobs {
s.Go(func() {
runSecurityJob(s)
})
s.Go(func() {
runDiagnosticsJob(s)
})
s.Go(func() {
runSessionCleanupJob(s)
})
s.Go(func() {
runTokenCleanupJob(s)
})
s.Go(func() {
runCommandWebhookCleanupJob(s)
})
if complianceI := s.Compliance; complianceI != nil {
complianceI.StartComplianceDailyJob()
}
if *s.Config().JobSettings.RunJobs && s.Jobs != nil {
s.Jobs.StartWorkers()
}
if *s.Config().JobSettings.RunScheduler && s.Jobs != nil {
s.Jobs.StartSchedulers()
}
}
return s, nil
}
@@ -358,19 +268,6 @@ func (s *Server) AppOptions() []AppOption {
}
}
// A temporary bridge to deal with cases where the code is so tighly coupled that
// this is easier as a temporary solution
func (s *Server) FakeApp() *App {
a := New(
ServerConnector(s),
)
return a
}
func (s *Server) StartServer() error {
return s.FakeApp().StartServer()
}
const TIME_TO_WAIT_FOR_CONNECTIONS_TO_CLOSE_ON_SERVER_SHUTDOWN = time.Second
func (s *Server) StopHTTPServer() {
@@ -395,15 +292,6 @@ func (s *Server) StopHTTPServer() {
}
}
func (s *Server) RunOldAppShutdown() {
a := s.FakeApp()
a.HubStop()
a.StopPushNotificationsHubWorkers()
a.ShutDownPlugins()
a.RemoveLicenseListener(s.licenseListenerId)
a.RemoveClusterLeaderChangedListener(s.clusterLeaderListenerId)
}
func (s *Server) Shutdown() error {
mlog.Info("Stopping Server...")
@@ -425,15 +313,23 @@ func (s *Server) Shutdown() error {
s.DisableConfigWatch()
if s.Cluster != nil {
s.Cluster.StopInterNodeCommunication()
}
if s.Metrics != nil {
s.Metrics.StopServer()
}
if s.Jobs != nil && s.runjobs {
s.Jobs.StopWorkers()
s.Jobs.StopSchedulers()
}
mlog.Info("Server stopped")
return nil
}
func (s *Server) License() *model.License {
license, _ := s.licenseValue.Load().(*model.License)
return license
}
// Go creates a goroutine, but maintains a record of it to ensure that execution completes before
// the server is shutdown.
func (s *Server) Go(f func()) {
@@ -493,14 +389,14 @@ func stripPort(hostport string) string {
return net.JoinHostPort(host, "443")
}
func (a *App) StartServer() error {
func (s *Server) Start() error {
mlog.Info("Starting Server...")
var handler http.Handler = a.Srv.RootRouter
if allowedOrigins := *a.Config().ServiceSettings.AllowCorsFrom; allowedOrigins != "" {
exposedCorsHeaders := *a.Config().ServiceSettings.CorsExposedHeaders
allowCredentials := *a.Config().ServiceSettings.CorsAllowCredentials
debug := *a.Config().ServiceSettings.CorsDebug
var handler http.Handler = s.RootRouter
if allowedOrigins := *s.Config().ServiceSettings.AllowCorsFrom; allowedOrigins != "" {
exposedCorsHeaders := *s.Config().ServiceSettings.CorsExposedHeaders
allowCredentials := *s.Config().ServiceSettings.CorsAllowCredentials
debug := *s.Config().ServiceSettings.CorsDebug
corsWrapper := cors.New(cors.Options{
AllowedOrigins: strings.Fields(allowedOrigins),
AllowedMethods: corsAllowedMethods,
@@ -513,34 +409,34 @@ func (a *App) StartServer() error {
// If we have debugging of CORS turned on then forward messages to logs
if debug {
corsWrapper.Log = a.Log.StdLog(mlog.String("source", "cors"))
corsWrapper.Log = s.Log.StdLog(mlog.String("source", "cors"))
}
handler = corsWrapper.Handler(handler)
}
if *a.Config().RateLimitSettings.Enable {
if *s.Config().RateLimitSettings.Enable {
mlog.Info("RateLimiter is enabled")
rateLimiter, err := NewRateLimiter(&a.Config().RateLimitSettings)
rateLimiter, err := NewRateLimiter(&s.Config().RateLimitSettings)
if err != nil {
return err
}
a.Srv.RateLimiter = rateLimiter
s.RateLimiter = rateLimiter
handler = rateLimiter.RateLimitHandler(handler)
}
a.Srv.Server = &http.Server{
s.Server = &http.Server{
Handler: handlers.RecoveryHandler(handlers.RecoveryLogger(&RecoveryLogger{}), handlers.PrintRecoveryStack(true))(handler),
ReadTimeout: time.Duration(*a.Config().ServiceSettings.ReadTimeout) * time.Second,
WriteTimeout: time.Duration(*a.Config().ServiceSettings.WriteTimeout) * time.Second,
ErrorLog: a.Log.StdLog(mlog.String("source", "httpserver")),
ReadTimeout: time.Duration(*s.Config().ServiceSettings.ReadTimeout) * time.Second,
WriteTimeout: time.Duration(*s.Config().ServiceSettings.WriteTimeout) * time.Second,
ErrorLog: s.Log.StdLog(mlog.String("source", "httpserver")),
}
addr := *a.Config().ServiceSettings.ListenAddress
addr := *s.Config().ServiceSettings.ListenAddress
if addr == "" {
if *a.Config().ServiceSettings.ConnectionSecurity == model.CONN_SECURITY_TLS {
if *s.Config().ServiceSettings.ConnectionSecurity == model.CONN_SECURITY_TLS {
addr = ":https"
} else {
addr = ":http"
@@ -552,23 +448,23 @@ func (a *App) StartServer() error {
errors.Wrapf(err, utils.T("api.server.start_server.starting.critical"), err)
return err
}
a.Srv.ListenAddr = listener.Addr().(*net.TCPAddr)
s.ListenAddr = listener.Addr().(*net.TCPAddr)
mlog.Info(fmt.Sprintf("Server is listening on %v", listener.Addr().String()))
// Migration from old let's encrypt library
if *a.Config().ServiceSettings.UseLetsEncrypt {
if stat, err := os.Stat(*a.Config().ServiceSettings.LetsEncryptCertificateCacheFile); err == nil && !stat.IsDir() {
os.Remove(*a.Config().ServiceSettings.LetsEncryptCertificateCacheFile)
if *s.Config().ServiceSettings.UseLetsEncrypt {
if stat, err := os.Stat(*s.Config().ServiceSettings.LetsEncryptCertificateCacheFile); err == nil && !stat.IsDir() {
os.Remove(*s.Config().ServiceSettings.LetsEncryptCertificateCacheFile)
}
}
m := &autocert.Manager{
Cache: autocert.DirCache(*a.Config().ServiceSettings.LetsEncryptCertificateCacheFile),
Cache: autocert.DirCache(*s.Config().ServiceSettings.LetsEncryptCertificateCacheFile),
Prompt: autocert.AcceptTOS,
}
if *a.Config().ServiceSettings.Forward80To443 {
if *s.Config().ServiceSettings.Forward80To443 {
if host, port, err := net.SplitHostPort(addr); err != nil {
mlog.Error("Unable to setup forwarding: " + err.Error())
} else if port != "443" {
@@ -576,11 +472,11 @@ func (a *App) StartServer() error {
} else {
httpListenAddress := net.JoinHostPort(host, "http")
if *a.Config().ServiceSettings.UseLetsEncrypt {
if *s.Config().ServiceSettings.UseLetsEncrypt {
server := &http.Server{
Addr: httpListenAddress,
Handler: m.HTTPHandler(nil),
ErrorLog: a.Log.StdLog(mlog.String("source", "le_forwarder_server")),
ErrorLog: s.Log.StdLog(mlog.String("source", "le_forwarder_server")),
}
go server.ListenAndServe()
} else {
@@ -594,27 +490,27 @@ func (a *App) StartServer() error {
server := &http.Server{
Handler: http.HandlerFunc(handleHTTPRedirect),
ErrorLog: a.Log.StdLog(mlog.String("source", "forwarder_server")),
ErrorLog: s.Log.StdLog(mlog.String("source", "forwarder_server")),
}
server.Serve(redirectListener)
}()
}
}
} else if *a.Config().ServiceSettings.UseLetsEncrypt {
} else if *s.Config().ServiceSettings.UseLetsEncrypt {
return errors.New(utils.T("api.server.start_server.forward80to443.disabled_while_using_lets_encrypt"))
}
a.Srv.didFinishListen = make(chan struct{})
s.didFinishListen = make(chan struct{})
go func() {
var err error
if *a.Config().ServiceSettings.ConnectionSecurity == model.CONN_SECURITY_TLS {
if *s.Config().ServiceSettings.ConnectionSecurity == model.CONN_SECURITY_TLS {
tlsConfig := &tls.Config{
PreferServerCipherSuites: true,
CurvePreferences: []tls.CurveID{tls.CurveP521, tls.CurveP384, tls.CurveP256},
}
switch *a.Config().ServiceSettings.TLSMinVer {
switch *s.Config().ServiceSettings.TLSMinVer {
case "1.0":
tlsConfig.MinVersion = tls.VersionTLS10
case "1.1":
@@ -632,11 +528,11 @@ func (a *App) StartServer() error {
tls.TLS_RSA_WITH_AES_256_GCM_SHA384,
}
if len(a.Config().ServiceSettings.TLSOverwriteCiphers) == 0 {
if len(s.Config().ServiceSettings.TLSOverwriteCiphers) == 0 {
tlsConfig.CipherSuites = defaultCiphers
} else {
var cipherSuites []uint16
for _, cipher := range a.Config().ServiceSettings.TLSOverwriteCiphers {
for _, cipher := range s.Config().ServiceSettings.TLSOverwriteCiphers {
value, ok := model.ServerTLSSupportedCiphers[cipher]
if !ok {
@@ -658,18 +554,18 @@ func (a *App) StartServer() error {
certFile := ""
keyFile := ""
if *a.Config().ServiceSettings.UseLetsEncrypt {
if *s.Config().ServiceSettings.UseLetsEncrypt {
tlsConfig.GetCertificate = m.GetCertificate
tlsConfig.NextProtos = append(tlsConfig.NextProtos, "h2")
} else {
certFile = *a.Config().ServiceSettings.TLSCertFile
keyFile = *a.Config().ServiceSettings.TLSKeyFile
certFile = *s.Config().ServiceSettings.TLSCertFile
keyFile = *s.Config().ServiceSettings.TLSKeyFile
}
a.Srv.Server.TLSConfig = tlsConfig
err = a.Srv.Server.ServeTLS(listener, certFile, keyFile)
s.Server.TLSConfig = tlsConfig
err = s.Server.ServeTLS(listener, certFile, keyFile)
} else {
err = a.Srv.Server.Serve(listener)
err = s.Server.Serve(listener)
}
if err != nil && err != http.ErrServerClosed {
@@ -677,7 +573,7 @@ func (a *App) StartServer() error {
time.Sleep(time.Second)
}
close(a.Srv.didFinishListen)
close(s.didFinishListen)
}()
return nil
@@ -705,3 +601,115 @@ func consumeAndClose(r *http.Response) {
r.Body.Close()
}
}
func runSecurityJob(s *Server) {
doSecurity(s)
model.CreateRecurringTask("Security", func() {
doSecurity(s)
}, time.Hour*4)
}
func runDiagnosticsJob(s *Server) {
doDiagnostics(s)
model.CreateRecurringTask("Diagnostics", func() {
doDiagnostics(s)
}, time.Hour*24)
}
func runTokenCleanupJob(s *Server) {
doTokenCleanup(s)
model.CreateRecurringTask("Token Cleanup", func() {
doTokenCleanup(s)
}, time.Hour*1)
}
func runCommandWebhookCleanupJob(s *Server) {
doCommandWebhookCleanup(s)
model.CreateRecurringTask("Command Hook Cleanup", func() {
doCommandWebhookCleanup(s)
}, time.Hour*1)
}
func runSessionCleanupJob(s *Server) {
doSessionCleanup(s)
model.CreateRecurringTask("Session Cleanup", func() {
doSessionCleanup(s)
}, time.Hour*24)
}
func doSecurity(s *Server) {
s.DoSecurityUpdateCheck()
}
func doDiagnostics(s *Server) {
if *s.Config().LogSettings.EnableDiagnostics {
s.FakeApp().SendDailyDiagnostics()
}
}
func doTokenCleanup(s *Server) {
s.Store.Token().Cleanup()
}
func doCommandWebhookCleanup(s *Server) {
s.Store.CommandWebhook().Cleanup()
}
const (
SESSIONS_CLEANUP_BATCH_SIZE = 1000
)
func doSessionCleanup(s *Server) {
s.Store.Session().Cleanup(model.GetMillis(), SESSIONS_CLEANUP_BATCH_SIZE)
}
func (s *Server) StartElasticsearch() {
s.Go(func() {
if err := s.Elasticsearch.Start(); err != nil {
s.Log.Error(err.Error())
}
})
s.AddConfigListener(func(oldConfig *model.Config, newConfig *model.Config) {
if !*oldConfig.ElasticsearchSettings.EnableIndexing && *newConfig.ElasticsearchSettings.EnableIndexing {
s.Go(func() {
if err := s.Elasticsearch.Start(); err != nil {
mlog.Error(err.Error())
}
})
} else if *oldConfig.ElasticsearchSettings.EnableIndexing && !*newConfig.ElasticsearchSettings.EnableIndexing {
s.Go(func() {
if err := s.Elasticsearch.Stop(); err != nil {
mlog.Error(err.Error())
}
})
} else if *oldConfig.ElasticsearchSettings.Password != *newConfig.ElasticsearchSettings.Password || *oldConfig.ElasticsearchSettings.Username != *newConfig.ElasticsearchSettings.Username || *oldConfig.ElasticsearchSettings.ConnectionUrl != *newConfig.ElasticsearchSettings.ConnectionUrl || *oldConfig.ElasticsearchSettings.Sniff != *newConfig.ElasticsearchSettings.Sniff {
s.Go(func() {
if *oldConfig.ElasticsearchSettings.EnableIndexing {
if err := s.Elasticsearch.Stop(); err != nil {
mlog.Error(err.Error())
}
if err := s.Elasticsearch.Start(); err != nil {
mlog.Error(err.Error())
}
}
})
}
})
s.AddLicenseListener(func() {
if s.License() != nil {
s.Go(func() {
if err := s.Elasticsearch.Start(); err != nil {
mlog.Error(err.Error())
}
})
} else {
s.Go(func() {
if err := s.Elasticsearch.Stop(); err != nil {
mlog.Error(err.Error())
}
})
}
})
}