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

Move cluster, webhub and store out of Server (#20899)

Просмотр исходного кода
Этот коммит содержится в:
Ibrahim Serdar Acikgoz
2022-10-06 11:04:21 +03:00
коммит произвёл GitHub
родитель 203df2f537
Коммит 5e69c6b02f
222 изменённых файлов: 9093 добавлений и 7710 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

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

@@ -8,18 +8,15 @@ import (
"context"
"crypto/tls"
"fmt"
"hash/maphash"
"net"
"net/http"
"net/url"
"os"
"os/exec"
"path"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
@@ -59,7 +56,6 @@ import (
"github.com/mattermost/mattermost-server/v6/services/cache"
"github.com/mattermost/mattermost-server/v6/services/httpservice"
"github.com/mattermost/mattermost-server/v6/services/remotecluster"
"github.com/mattermost/mattermost-server/v6/services/searchengine"
"github.com/mattermost/mattermost-server/v6/services/searchengine/bleveengine"
"github.com/mattermost/mattermost-server/v6/services/searchengine/bleveengine/indexer"
"github.com/mattermost/mattermost-server/v6/services/sharedchannel"
@@ -73,11 +69,6 @@ import (
"github.com/mattermost/mattermost-server/v6/shared/mlog"
"github.com/mattermost/mattermost-server/v6/shared/templates"
"github.com/mattermost/mattermost-server/v6/store"
"github.com/mattermost/mattermost-server/v6/store/localcachelayer"
"github.com/mattermost/mattermost-server/v6/store/retrylayer"
"github.com/mattermost/mattermost-server/v6/store/searchlayer"
"github.com/mattermost/mattermost-server/v6/store/sqlstore"
"github.com/mattermost/mattermost-server/v6/store/timerlayer"
"github.com/mattermost/mattermost-server/v6/utils"
)
@@ -109,10 +100,6 @@ const (
)
type Server struct {
sqlStore *sqlstore.SqlStore
Store store.Store
WebSocketRouter *WebSocketRouter
// RootRouter is the starting point for all HTTP requests to the server.
RootRouter *mux.Router
@@ -127,21 +114,13 @@ type Server struct {
Server *http.Server
ListenAddr *net.TCPAddr
RateLimiter *RateLimiter
Busy *Busy
localModeServer *http.Server
didFinishListen chan struct{}
goroutineCount int32
goroutineExitSignal chan struct{}
goroutineBuffered chan struct{}
EmailService email.ServiceInterface
hubs []*Hub
hashSeed maphash.Seed
httpService httpservice.HTTPService
PushNotificationsHub PushNotificationsHub
pushNotificationClient *http.Client // TODO: move this to it's own package
@@ -149,77 +128,65 @@ type Server struct {
runEssentialJobs bool
Jobs *jobs.JobServer
clusterLeaderListeners sync.Map
clusterWrapper *clusterWrapper
licenseValue atomic.Value
clientLicenseValue atomic.Value
licenseListeners map[string]func(*model.License, *model.License)
licenseWrapper *licenseWrapper
licenseWrapper *licenseWrapper
timezones *timezones.Timezones
newStore func() (store.Store, error)
htmlTemplateWatcher *templates.Container
seenPendingPostIdsCache cache.Cache
statusCache cache.Cache
openGraphDataCache cache.Cache
configListenerId string
licenseListenerId string
clusterLeaderListenerId string
searchConfigListenerId string
searchLicenseListenerId string
loggerLicenseListenerId string
filestore filestore.FileBackend
platform *platform.PlatformService
platformOptions []platform.Option
telemetryService *telemetry.TelemetryService
userService *users.UserService
teamService *teams.TeamService
serviceMux sync.RWMutex
remoteClusterService remotecluster.RemoteClusterServiceIFace
sharedChannelService SharedChannelServiceIFace
sharedChannelService SharedChannelServiceIFace // TODO: platform: move to platform package
phase2PermissionsMigrationComplete bool
Audit *audit.Audit
joinCluster bool
startMetrics bool
startSearchEngine bool
skipPostInit bool
joinCluster bool
// startSearchEngine bool
skipPostInit bool
SearchEngine *searchengine.Broker
Cluster einterfaces.ClusterInterface
Cloud einterfaces.CloudInterface
LicenseManager einterfaces.LicenseInterface
CacheProvider cache.Provider
Cloud einterfaces.CloudInterface
tracer *tracing.Tracer
products map[string]Product
}
func (s *Server) Store() store.Store {
if s.platform != nil {
return s.platform.Store
}
return nil
}
func (s *Server) SetStore(st store.Store) {
if s.platform != nil {
s.platform.Store = st
}
}
func NewServer(options ...Option) (*Server, error) {
rootRouter := mux.NewRouter()
localRouter := mux.NewRouter()
s := &Server{
goroutineExitSignal: make(chan struct{}, 1),
goroutineBuffered: make(chan struct{}, runtime.NumCPU()),
RootRouter: rootRouter,
LocalRouter: localRouter,
WebSocketRouter: &WebSocketRouter{
handlers: make(map[string]webSocketHandler),
},
licenseListeners: map[string]func(*model.License, *model.License){},
hashSeed: maphash.MakeSeed(),
timezones: timezones.New(),
products: make(map[string]Product),
RootRouter: rootRouter,
LocalRouter: localRouter,
timezones: timezones.New(),
products: make(map[string]Product),
}
for _, option := range options {
@@ -232,7 +199,7 @@ func NewServer(options ...Option) (*Server, error) {
// performed during server bootup. They are sensitive to order
// and has dependency requirements with the previous step.
//
// Step 1: Config.
// Step 1: Platform.
if s.platform == nil {
innerStore, err := config.NewFileStore("config.json", true)
if err != nil {
@@ -244,23 +211,14 @@ func NewServer(options ...Option) (*Server, error) {
}
platformCfg := platform.ServiceConfig{
ConfigStore: configStore,
StartMetrics: s.startMetrics,
Cluster: s.Cluster,
}
if metricsInterface != nil {
platformCfg.Metrics = metricsInterface(s, *configStore.Get().SqlSettings.DriverName, *configStore.Get().SqlSettings.DataSource)
ConfigStore: configStore,
}
ps, sErr := platform.New(platformCfg)
ps, sErr := platform.New(platformCfg, s.platformOptions...)
if sErr != nil {
return nil, errors.Wrap(sErr, "failed to initialize platform")
}
s.platform = ps
if s.licenseValue.Load() != nil {
ps.SetLicense(s.licenseValue.Load().(*model.License)) // in case license is set in server options
}
}
subpath, err := utils.GetSubpathFromConfig(s.platform.Config())
@@ -269,99 +227,26 @@ func NewServer(options ...Option) (*Server, error) {
}
s.Router = s.RootRouter.PathPrefix(subpath).Subrouter()
// This is called after initLogging() to avoid a race condition.
mlog.Info("Server is initializing...", mlog.String("go_version", runtime.Version()))
s.httpService = httpservice.MakeHTTPService(s.platform)
// Step 3: Search Engine
// Depends on Step 1 (config).
searchEngine := searchengine.NewBroker(s.platform.Config())
bleveEngine := bleveengine.NewBleveEngine(s.platform.Config())
if err := bleveEngine.Start(); err != nil {
return nil, err
}
searchEngine.RegisterBleveEngine(bleveEngine)
s.SearchEngine = searchEngine
// Step 4: Init Enterprise
// Depends on step 3 (s.SearchEngine must be non-nil)
// Step 2: Init Enterprise
// Depends on step 1 (s.Platform must be non-nil)
s.initEnterprise()
// Step 5: Cache provider.
// At the moment we only have this implementation
// in the future the cache provider will be built based on the loaded config
s.CacheProvider = cache.NewProvider()
if err2 := s.CacheProvider.Connect(); err2 != nil {
return nil, errors.Wrapf(err2, "Unable to connect to cache provider")
}
// Step 6: Store.
// Depends on Step 1 (config), 4 (metrics, cluster) and 5 (cacheProvider).
if s.newStore == nil {
s.newStore = func() (store.Store, error) {
s.sqlStore = sqlstore.New(s.platform.Config().SqlSettings, s.GetMetrics())
lcl, err2 := localcachelayer.NewLocalCacheLayer(
retrylayer.New(s.sqlStore),
s.GetMetrics(),
s.Cluster,
s.CacheProvider,
)
if err2 != nil {
return nil, errors.Wrap(err2, "cannot create local cache layer")
}
searchStore := searchlayer.NewSearchLayer(
lcl,
s.SearchEngine,
s.platform.Config(),
)
s.platform.AddConfigListener(func(prevCfg, cfg *model.Config) {
searchStore.UpdateConfig(cfg)
})
s.sqlStore.UpdateLicense(s.License())
s.AddLicenseListener(func(oldLicense, newLicense *model.License) {
s.sqlStore.UpdateLicense(newLicense)
})
return timerlayer.New(
searchStore,
s.GetMetrics(),
), nil
}
}
s.Store, err = s.newStore()
if err != nil {
return nil, errors.Wrap(err, "cannot create store")
}
// Needed to run before loading license.
s.userService, err = users.New(users.ServiceConfig{
UserStore: s.Store.User(),
SessionStore: s.Store.Session(),
OAuthStore: s.Store.OAuth(),
UserStore: s.Store().User(),
SessionStore: s.Store().Session(),
OAuthStore: s.Store().OAuth(),
ConfigFn: s.platform.Config,
Metrics: s.GetMetrics(),
Cluster: s.Cluster,
Cluster: s.platform.Cluster(),
LicenseFn: s.License,
})
if err != nil {
return nil, errors.Wrapf(err, "unable to create users service")
}
// Needed before loading license
if s.statusCache, err = s.CacheProvider.NewCache(&cache.CacheOptions{
Size: model.StatusCacheSize,
Striped: true,
StripedBuckets: maxInt(runtime.NumCPU()-1, 1),
}); err != nil {
return nil, errors.Wrap(err, "Unable to create status cache")
}
if model.BuildEnterpriseReady == "true" {
// Dependent on user service
s.LoadLicense()
@@ -369,7 +254,7 @@ func NewServer(options ...Option) (*Server, error) {
license := s.License()
insecure := s.platform.Config().ServiceSettings.EnableInsecureOutgoingConnections
// Step 7: Initialize filestore
// Step 3: Initialize filestore
backend, err := filestore.NewFileBackend(s.platform.Config().FileSettings.ToFileBackendSettings(license != nil && *license.Features.Compliance, insecure != nil && *insecure))
if err != nil {
return nil, errors.Wrap(err, "failed to initialize filebackend")
@@ -380,16 +265,12 @@ func NewServer(options ...Option) (*Server, error) {
srv: s,
}
s.clusterWrapper = &clusterWrapper{
srv: s,
}
s.teamService, err = teams.New(teams.ServiceConfig{
TeamStore: s.Store.Team(),
ChannelStore: s.Store.Channel(),
GroupStore: s.Store.Group(),
TeamStore: s.Store().Team(),
ChannelStore: s.Store().Channel(),
GroupStore: s.Store().Group(),
Users: s.userService,
WebHub: s,
WebHub: s.platform,
ConfigFn: s.platform.Config,
LicenseFn: s.License,
})
@@ -406,16 +287,16 @@ func NewServer(options ...Option) (*Server, error) {
LicenseKey: s.licenseWrapper,
FilestoreKey: s.filestore,
FileInfoStoreKey: &fileInfoWrapper{srv: s},
ClusterKey: s.clusterWrapper,
ClusterKey: s.platform,
UserKey: New(ServerConnector(s.Channels())),
LogKey: s.Log(),
LogKey: s.platform.Log(),
CloudKey: &cloudWrapper{cloud: s.Cloud},
KVStoreKey: &kvStoreWrapper{srv: s},
StoreKey: store.NewStoreServiceAdapter(s.Store),
KVStoreKey: s.platform,
StoreKey: store.NewStoreServiceAdapter(s.Store()),
SystemKey: &systemServiceAdapter{server: s},
}
// Step 8: Initialize products.
// Step 4: Initialize products.
// Depends on s.httpService.
err = s.initializeProducts(products, serviceMap)
if err != nil {
@@ -424,8 +305,8 @@ func NewServer(options ...Option) (*Server, error) {
// It is important to initialize the hub only after the global logger is set
// to avoid race conditions while logging from inside the hub.
// Step 9: Hub depends on s.Channels() (step 8)
s.HubStart()
// Step 5: Hub depends on s.Channels() (step 8)
s.platform.HubStart(New(ServerConnector(s.Channels())))
// -------------------------------------------------------------------------
// Everything below this is not order sensitive and safe to be moved around.
@@ -475,12 +356,12 @@ func NewServer(options ...Option) (*Server, error) {
}
model.AppErrorInit(i18n.T)
if s.seenPendingPostIdsCache, err = s.CacheProvider.NewCache(&cache.CacheOptions{
if s.seenPendingPostIdsCache, err = s.platform.CacheProvider().NewCache(&cache.CacheOptions{
Size: PendingPostIDsCacheSize,
}); err != nil {
return nil, errors.Wrap(err, "Unable to create pending post ids cache")
}
if s.openGraphDataCache, err = s.CacheProvider.NewCache(&cache.CacheOptions{
if s.openGraphDataCache, err = s.platform.CacheProvider().NewCache(&cache.CacheOptions{
Size: openGraphMetadataCacheSize,
}); err != nil {
return nil, errors.Wrap(err, "Unable to create opengraphdata cache")
@@ -507,35 +388,7 @@ func NewServer(options ...Option) (*Server, error) {
})
s.htmlTemplateWatcher = htmlTemplateWatcher
s.configListenerId = s.platform.AddConfigListener(func(_, _ *model.Config) {
ch := s.Channels()
ch.regenerateClientConfig()
message := model.NewWebSocketEvent(model.WebsocketEventConfigChanged, "", "", "", nil, "")
appInstance := New(ServerConnector(ch))
message.Add("config", appInstance.ClientConfigWithComputed())
s.Go(func() {
s.Publish(message)
})
if err = s.platform.ReconfigureLogger(); err != nil {
mlog.Error("Error re-configuring logging after config change", mlog.Err(err))
return
}
})
s.licenseListenerId = s.AddLicenseListener(func(oldLicense, newLicense *model.License) {
s.Channels().regenerateClientConfig()
message := model.NewWebSocketEvent(model.WebsocketEventLicenseChanged, "", "", "", nil, "")
message.Add("license", s.GetSanitizedClientLicense())
s.Go(func() {
s.Publish(message)
})
})
s.telemetryService = telemetry.New(New(ServerConnector(s.Channels())), s.Store, s.SearchEngine, s.Log())
s.telemetryService = telemetry.New(New(ServerConnector(s.Channels())), s.Store(), s.platform.SearchEngine, s.Log())
s.platform.SetTelemetryId(s.TelemetryId()) // TODO: move this into platform once telemetry service moved to platform.
emailService, err := email.NewService(email.ServiceConfig{
@@ -617,36 +470,6 @@ func NewServer(options ...Option) (*Server, error) {
s.platform.EnableLoggingMetrics()
})
// Enable developer settings if this is a "dev" build
if model.BuildNumber == "dev" {
s.platform.UpdateConfig(func(cfg *model.Config) { *cfg.ServiceSettings.EnableDeveloper = true })
}
if s.startMetrics {
if err := s.platform.RestartMetrics(); err != nil {
return nil, errors.Wrap(err, "failed to start metrics")
}
}
s.AddLicenseListener(func(oldLicense, newLicense *model.License) {
if (oldLicense == nil && newLicense == nil) || !s.startMetrics {
return
}
if oldLicense != nil && newLicense != nil && *oldLicense.Features.Metrics == *newLicense.Features.Metrics {
return
}
if err := s.platform.RestartMetrics(); err != nil {
s.Log().Error("Failed to reset metrics server", mlog.Err(err))
}
})
s.SearchEngine.UpdateConfig(s.platform.Config())
searchConfigListenerId, searchLicenseListenerId := s.StartSearchEngine()
s.searchConfigListenerId = searchConfigListenerId
s.searchLicenseListenerId = searchLicenseListenerId
// if enabled - perform initial product notices fetch
if *s.platform.Config().AnnouncementSettings.AdminNoticesEnabled || *s.platform.Config().AnnouncementSettings.UserNoticesEnabled {
go func() {
@@ -708,13 +531,6 @@ func NewServer(options ...Option) (*Server, error) {
return s, nil
}
func maxInt(a, b int) int {
if a > b {
return a
}
return b
}
func (s *Server) runJobs() {
s.Go(func() {
runSecurityJob(s)
@@ -778,7 +594,7 @@ func (s *Server) Channels() *Channels {
// Return Database type (postgres or mysql) and current version of the schema
func (s *Server) DatabaseTypeAndSchemaVersion() (string, string) {
schemaVersion, _ := s.Store.GetDBSchemaVersion()
schemaVersion, _ := s.Store().GetDBSchemaVersion()
return *s.platform.Config().SqlSettings.DriverName, strconv.Itoa(schemaVersion)
}
@@ -836,6 +652,7 @@ func (s *Server) startInterClusterServices(license *model.License) error {
if err != nil {
return err
}
s.platform.SetSharedChannelService(scs)
if err = scs.Start(); err != nil {
return err
@@ -877,8 +694,6 @@ func (s *Server) Shutdown() {
defer sentry.Flush(2 * time.Second)
s.HubStop()
s.RemoveLicenseListener(s.licenseListenerId)
s.RemoveLicenseListener(s.loggerLicenseListenerId)
s.RemoveClusterLeaderChangedListener(s.clusterLeaderListenerId)
@@ -915,8 +730,7 @@ func (s *Server) Shutdown() {
s.WaitForGoroutines()
s.platform.RemoveConfigListener(s.configListenerId)
s.stopSearchEngine()
s.platform.StopSearchEngine()
s.Audit.Shutdown()
@@ -926,8 +740,8 @@ func (s *Server) Shutdown() {
s.Log().Warn("Failed to shut down config store", mlog.Err(err))
}
if s.Cluster != nil {
s.Cluster.StopInterNodeCommunication()
if s.platform.Cluster() != nil {
s.platform.Cluster().StopInterNodeCommunication()
}
if err = s.platform.ShutdownMetrics(); err != nil {
@@ -956,14 +770,8 @@ func (s *Server) Shutdown() {
}
}
if s.Store != nil {
s.Store.Close()
}
if s.CacheProvider != nil {
if err = s.CacheProvider.Close(); err != nil {
s.Log().Warn("Unable to cleanly shutdown cache", mlog.Err(err))
}
if err = s.platform.Shutdown(); err != nil {
s.Log().Warn("Failed to stop platform", mlog.Err(err))
}
s.Log().Info("Server stopped")
@@ -999,14 +807,6 @@ func (s *Server) Restart() error {
return syscall.Exec(argv0, os.Args, os.Environ())
}
func (s *Server) isUpgradedFromTE() bool {
val, err := s.Store.System().GetByName(model.SystemUpgradedFromTeId)
if err != nil {
return false
}
return val.Value == "true"
}
func (s *Server) CanIUpgradeToE0() error {
return upgrader.CanIUpgradeToE0()
}
@@ -1016,7 +816,7 @@ func (s *Server) UpgradeToE0() error {
return err
}
upgradedFromTE := &model.System{Name: model.SystemUpgradedFromTeId, Value: "true"}
s.Store.System().Save(upgradedFromTE)
s.Store().System().Save(upgradedFromTE)
return nil
}
@@ -1027,44 +827,18 @@ func (s *Server) UpgradeToE0Status() (int64, error) {
// 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()) {
atomic.AddInt32(&s.goroutineCount, 1)
go func() {
f()
atomic.AddInt32(&s.goroutineCount, -1)
select {
case s.goroutineExitSignal <- struct{}{}:
default:
}
}()
s.platform.Go(f)
}
// GoBuffered acts like a semaphore which creates a goroutine, but maintains a record of it
// to ensure that execution completes before the server is shutdown.
func (s *Server) GoBuffered(f func()) {
s.goroutineBuffered <- struct{}{}
atomic.AddInt32(&s.goroutineCount, 1)
go func() {
f()
atomic.AddInt32(&s.goroutineCount, -1)
select {
case s.goroutineExitSignal <- struct{}{}:
default:
}
<-s.goroutineBuffered
}()
s.platform.GoBuffered(f)
}
// WaitForGoroutines blocks until all goroutines created by App.Go exit.
func (s *Server) WaitForGoroutines() {
for atomic.LoadInt32(&s.goroutineCount) != 0 {
<-s.goroutineExitSignal
}
s.platform.WaitForGoroutines()
}
var corsAllowedMethods = []string{
@@ -1112,9 +886,9 @@ func (s *Server) Start() error {
}
}
if s.joinCluster && s.Cluster != nil {
if s.joinCluster && s.platform.Cluster() != nil {
s.registerClusterHandlers()
s.Cluster.StartInterNodeCommunication()
s.platform.Cluster().StartInterNodeCommunication()
}
if err := s.ensureInstallationDate(); err != nil {
@@ -1125,7 +899,7 @@ func (s *Server) Start() error {
return errors.Wrapf(err, "unable to ensure first run timestamp")
}
if err := s.Store.Status().ResetAll(); err != nil {
if err := s.Store().Status().ResetAll(); err != nil {
mlog.Error("Error to reset the server status.", mlog.Err(err))
}
@@ -1193,7 +967,6 @@ func (s *Server) Start() error {
s.RateLimiter = rateLimiter
handler = rateLimiter.RateLimitHandler(handler)
}
s.Busy = NewBusy(s.Cluster)
// Creating a logger for logging errors from http.Server at error level
errStdLog := s.Log().With(mlog.String("source", "httpserver")).StdLogger(mlog.LvlError)
@@ -1471,7 +1244,7 @@ func runReportToAWSMeterJob(s *Server) {
}
func doReportUsageToAWSMeteringService(s *Server) {
awsMeter := awsmeter.New(s.Store, s.platform.Config())
awsMeter := awsmeter.New(s.Store(), s.platform.Config())
if awsMeter == nil {
mlog.Error("Cannot obtain instance of AWS Metering Service.")
return
@@ -1491,11 +1264,11 @@ func doTokenCleanup(s *Server) {
mlog.Debug("Cleaning up token store.")
s.Store.Token().Cleanup(expiry)
s.Store().Token().Cleanup(expiry)
}
func doCommandWebhookCleanup(s *Server) {
s.Store.CommandWebhook().Cleanup()
s.Store().CommandWebhook().Cleanup()
}
const (
@@ -1505,7 +1278,7 @@ const (
func doSessionCleanup(s *Server) {
mlog.Debug("Cleaning up session store.")
err := s.Store.Session().Cleanup(model.GetMillis(), sessionsCleanupBatchSize)
err := s.Store().Session().Cleanup(model.GetMillis(), sessionsCleanupBatchSize)
if err != nil {
mlog.Warn("Error while cleaning up sessions", mlog.Err(err))
}
@@ -1519,7 +1292,7 @@ func doJobsCleanup(s *Server) {
dur := time.Duration(*s.platform.Config().JobSettings.CleanupJobsThresholdDays) * time.Hour * 24
expiry := model.GetMillisForTime(time.Now().Add(-dur))
err := s.Store.Job().Cleanup(expiry, jobsCleanupBatchSize)
err := s.Store().Job().Cleanup(expiry, jobsCleanupBatchSize)
if err != nil {
mlog.Warn("Error while cleaning up jobs", mlog.Err(err))
}
@@ -1542,7 +1315,7 @@ func (s *Server) HandleMetrics(route string, h http.Handler) {
func (s *Server) sendLicenseUpForRenewalEmail(users map[string]*model.User, license *model.License) *model.AppError {
key := model.LicenseUpForRenewalEmailSent + license.Id
if _, err := s.Store.System().GetByName(key); err == nil {
if _, err := s.Store().System().GetByName(key); err == nil {
// return early because the key already exists and that means we already executed the code below to send email successfully
return nil
}
@@ -1578,7 +1351,7 @@ func (s *Server) sendLicenseUpForRenewalEmail(users map[string]*model.User, lice
Value: "true",
}
if err := s.Store.System().Save(&system); err != nil {
if err := s.Store().System().Save(&system); err != nil {
mlog.Debug("Failed to mark license up for renewal email sending as completed.", mlog.Err(err))
}
@@ -1608,7 +1381,7 @@ func (s *Server) doLicenseExpirationCheck() {
return
}
users, err := s.Store.User().GetSystemAdminProfiles()
users, err := s.Store().User().GetSystemAdminProfiles()
if err != nil {
mlog.Error("Failed to get system admins for license expired message from Mattermost.")
return
@@ -1664,109 +1437,24 @@ func (s *Server) SendRemoveExpiredLicenseEmail(email string, renewalLink, locale
return nil
}
func (s *Server) StartSearchEngine() (string, string) {
if s.SearchEngine.ElasticsearchEngine != nil && s.SearchEngine.ElasticsearchEngine.IsActive() {
s.Go(func() {
if err := s.SearchEngine.ElasticsearchEngine.Start(); err != nil {
s.Log().Error(err.Error())
}
})
}
configListenerId := s.platform.AddConfigListener(func(oldConfig *model.Config, newConfig *model.Config) {
if s.SearchEngine == nil {
return
}
s.SearchEngine.UpdateConfig(newConfig)
if s.SearchEngine.ElasticsearchEngine != nil && !*oldConfig.ElasticsearchSettings.EnableIndexing && *newConfig.ElasticsearchSettings.EnableIndexing {
s.Go(func() {
if err := s.SearchEngine.ElasticsearchEngine.Start(); err != nil {
mlog.Error(err.Error())
}
})
} else if s.SearchEngine.ElasticsearchEngine != nil && *oldConfig.ElasticsearchSettings.EnableIndexing && !*newConfig.ElasticsearchSettings.EnableIndexing {
s.Go(func() {
if err := s.SearchEngine.ElasticsearchEngine.Stop(); err != nil {
mlog.Error(err.Error())
}
})
} else if s.SearchEngine.ElasticsearchEngine != nil && *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.SearchEngine.ElasticsearchEngine.Stop(); err != nil {
mlog.Error(err.Error())
}
if err := s.SearchEngine.ElasticsearchEngine.Start(); err != nil {
mlog.Error(err.Error())
}
}
})
}
})
licenseListenerId := s.AddLicenseListener(func(oldLicense, newLicense *model.License) {
if s.SearchEngine == nil {
return
}
if oldLicense == nil && newLicense != nil {
if s.SearchEngine.ElasticsearchEngine != nil && s.SearchEngine.ElasticsearchEngine.IsActive() {
s.Go(func() {
if err := s.SearchEngine.ElasticsearchEngine.Start(); err != nil {
mlog.Error(err.Error())
}
})
}
} else if oldLicense != nil && newLicense == nil {
if s.SearchEngine.ElasticsearchEngine != nil {
s.Go(func() {
if err := s.SearchEngine.ElasticsearchEngine.Stop(); err != nil {
mlog.Error(err.Error())
}
})
}
}
})
return configListenerId, licenseListenerId
}
func (s *Server) stopSearchEngine() {
s.platform.RemoveConfigListener(s.searchConfigListenerId)
s.RemoveLicenseListener(s.searchLicenseListenerId)
if s.SearchEngine != nil && s.SearchEngine.ElasticsearchEngine != nil && s.SearchEngine.ElasticsearchEngine.IsActive() {
s.SearchEngine.ElasticsearchEngine.Stop()
}
if s.SearchEngine != nil && s.SearchEngine.BleveEngine != nil && s.SearchEngine.BleveEngine.IsActive() {
s.SearchEngine.BleveEngine.Stop()
}
}
func (s *Server) FileBackend() filestore.FileBackend {
return s.filestore
}
func (s *Server) TotalWebsocketConnections() int {
// This method is only called after the hub is initialized.
// Therefore, no mutex is needed to protect s.hubs.
count := int64(0)
for _, hub := range s.hubs {
count = count + atomic.LoadInt64(&hub.connectionCount)
}
return int(count)
return s.Platform().TotalWebsocketConnections()
}
func (s *Server) ClusterHealthScore() int {
return s.Cluster.HealthScore()
return s.platform.Cluster().HealthScore()
}
func (ch *Channels) ClientConfigHash() string {
return ch.clientConfigHash.Load().(string)
return ch.srv.Platform().ClientConfigHash()
}
func (s *Server) initJobs() {
s.Jobs = jobs.NewJobServer(s.platform, s.Store, s.GetMetrics())
s.Jobs = jobs.NewJobServer(s.platform, s.Store(), s.GetMetrics())
if jobsDataRetentionJobInterface != nil {
builder := jobsDataRetentionJobInterface(s)
@@ -1795,14 +1483,14 @@ func (s *Server) initJobs() {
s.Jobs.RegisterJobType(
model.JobTypeBlevePostIndexing,
indexer.MakeWorker(s.Jobs, s.SearchEngine.BleveEngine.(*bleveengine.BleveEngine)),
indexer.MakeWorker(s.Jobs, s.platform.SearchEngine.BleveEngine.(*bleveengine.BleveEngine)),
nil,
)
s.Jobs.RegisterJobType(
model.JobTypeMigrations,
migrations.MakeWorker(s.Jobs, s.Store),
migrations.MakeScheduler(s.Jobs, s.Store),
migrations.MakeWorker(s.Jobs, s.Store()),
migrations.MakeScheduler(s.Jobs, s.Store()),
)
s.Jobs.RegisterJobType(
@@ -1831,7 +1519,7 @@ func (s *Server) initJobs() {
s.Jobs.RegisterJobType(
model.JobTypeImportDelete,
import_delete.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())), s.Store),
import_delete.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())), s.Store()),
import_delete.MakeScheduler(s.Jobs),
)
@@ -1849,19 +1537,19 @@ func (s *Server) initJobs() {
s.Jobs.RegisterJobType(
model.JobTypeActiveUsers,
active_users.MakeWorker(s.Jobs, s.Store, func() einterfaces.MetricsInterface { return s.GetMetrics() }),
active_users.MakeWorker(s.Jobs, s.Store(), func() einterfaces.MetricsInterface { return s.GetMetrics() }),
active_users.MakeScheduler(s.Jobs),
)
s.Jobs.RegisterJobType(
model.JobTypeResendInvitationEmail,
resend_invitation_email.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())), s.Store, s.telemetryService),
resend_invitation_email.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())), s.Store(), s.telemetryService),
nil,
)
s.Jobs.RegisterJobType(
model.JobTypeExtractContent,
extract_content.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())), s.Store),
extract_content.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())), s.Store()),
nil,
)
@@ -1888,6 +1576,8 @@ func (s *Server) initJobs() {
notify_admin.MakeTrialNotifyWorker(s.Jobs, s.License(), New(ServerConnector(s.Channels()))),
notify_admin.MakeScheduler(s.Jobs, s.License(), model.JobTypeTrialNotifyAdmin),
)
s.platform.Jobs = s.Jobs
}
func (s *Server) TelemetryId() string {
@@ -1904,7 +1594,7 @@ func (s *Server) HTTPService() httpservice.HTTPService {
// GetStore returns the server's Store. Exposing via a method
// allows interfaces to be created with subsets of server APIs.
func (s *Server) GetStore() store.Store {
return s.Store
return s.Store()
}
// GetRemoteClusterService returns the `RemoteClusterService` instantiated by the server.
@@ -1946,6 +1636,7 @@ func (s *Server) SetSharedChannelSyncService(sharedChannelService SharedChannelS
s.serviceMux.Lock()
defer s.serviceMux.Unlock()
s.sharedChannelService = sharedChannelService
s.platform.SetSharedChannelService(sharedChannelService)
}
func (s *Server) GetProfileImage(user *model.User) ([]byte, bool, *model.AppError) {
@@ -2053,7 +1744,7 @@ func runPostReminderJob(a *App) {
}
func (a *App) GetAppliedSchemaMigrations() ([]model.AppliedMigration, *model.AppError) {
table, err := a.Srv().Store.GetAppliedMigrations()
table, err := a.Srv().Store().GetAppliedMigrations()
if err != nil {
return nil, model.NewAppError("GetDBSchemaTable", "api.file.read_file.app_error", nil, "", http.StatusInternalServerError).Wrap(err)
}