Move metrics under platform service (continued) (#20732)
* move metrics into platform Co-authored-by: Agniva De Sarker <agnivade@yahoo.co.in>
Этот коммит содержится в:
коммит произвёл
GitHub
родитель
276594608c
Коммит
ac79a887a2
@@ -117,7 +117,7 @@ func setupTestHelper(dbStore store.Store, searchEngine *searchengine.Broker, ent
|
||||
if includeCache {
|
||||
// Adds the cache layer to the test store
|
||||
options = append(options, app.StoreOverride(func(s *app.Server) store.Store {
|
||||
lcl, err2 := localcachelayer.NewLocalCacheLayer(dbStore, s.Metrics, s.Cluster, s.CacheProvider)
|
||||
lcl, err2 := localcachelayer.NewLocalCacheLayer(dbStore, s.GetMetrics(), s.Cluster, s.CacheProvider)
|
||||
if err2 != nil {
|
||||
panic(err2)
|
||||
}
|
||||
@@ -218,7 +218,7 @@ func setupTestHelper(dbStore store.Store, searchEngine *searchengine.Broker, ent
|
||||
return th
|
||||
}
|
||||
|
||||
func SetupEnterprise(tb testing.TB) *TestHelper {
|
||||
func SetupEnterprise(tb testing.TB, options ...app.Option) *TestHelper {
|
||||
if testing.Short() {
|
||||
tb.SkipNow()
|
||||
}
|
||||
@@ -232,7 +232,7 @@ func SetupEnterprise(tb testing.TB) *TestHelper {
|
||||
dbStore.MarkSystemRanUnitTests()
|
||||
mainHelper.PreloadMigrations()
|
||||
searchEngine := mainHelper.GetSearchEngine()
|
||||
th := setupTestHelper(dbStore, searchEngine, true, true, nil, nil)
|
||||
th := setupTestHelper(dbStore, searchEngine, true, true, nil, options)
|
||||
th.InitLogin()
|
||||
return th
|
||||
}
|
||||
@@ -322,8 +322,8 @@ func SetupWithStoreMock(tb testing.TB) *TestHelper {
|
||||
return th
|
||||
}
|
||||
|
||||
func SetupEnterpriseWithStoreMock(tb testing.TB) *TestHelper {
|
||||
th := setupTestHelper(testlib.GetMockStoreForSetupFunctions(), nil, true, false, nil, nil)
|
||||
func SetupEnterpriseWithStoreMock(tb testing.TB, options ...app.Option) *TestHelper {
|
||||
th := setupTestHelper(testlib.GetMockStoreForSetupFunctions(), nil, true, false, nil, options)
|
||||
statusMock := mocks.StatusStore{}
|
||||
statusMock.On("UpdateExpiredDNDStatuses").Return([]*model.Status{}, nil)
|
||||
statusMock.On("Get", "user1").Return(&model.Status{UserId: "user1", Status: model.StatusOnline}, nil)
|
||||
|
||||
@@ -117,7 +117,7 @@ func (a *App) MessageExport() einterfaces.MessageExportInterface {
|
||||
return a.ch.MessageExport
|
||||
}
|
||||
func (a *App) Metrics() einterfaces.MetricsInterface {
|
||||
return a.ch.srv.Metrics
|
||||
return a.ch.srv.GetMetrics()
|
||||
}
|
||||
func (a *App) Notification() einterfaces.NotificationInterface {
|
||||
return a.ch.Notification
|
||||
|
||||
@@ -79,12 +79,12 @@ func (w *configWrapper) SaveConfig(newCfg *model.Config, sendConfigChangeCluster
|
||||
}
|
||||
|
||||
if w.srv.startMetrics && *w.Config().MetricsSettings.Enable {
|
||||
if w.srv.Metrics != nil {
|
||||
w.srv.Metrics.Register()
|
||||
if w.srv.GetMetrics() != nil {
|
||||
w.srv.GetMetrics().Register()
|
||||
}
|
||||
w.srv.SetupMetricsServer()
|
||||
w.srv.platform.RestartMetrics() // TODO: remove when this moved to the platform service
|
||||
} else {
|
||||
w.srv.StopMetricsServer()
|
||||
w.srv.platform.ShutdownMetrics() // TODO: remove when this moved to the platform service
|
||||
}
|
||||
|
||||
if w.srv.Cluster != nil {
|
||||
|
||||
@@ -112,10 +112,6 @@ func RegisterLicenseInterface(f func(*Server) einterfaces.LicenseInterface) {
|
||||
}
|
||||
|
||||
func (s *Server) initEnterprise() {
|
||||
if metricsInterface != nil {
|
||||
s.Metrics = metricsInterface(s)
|
||||
}
|
||||
|
||||
if clusterInterface != nil && s.Cluster == nil {
|
||||
s.Cluster = clusterInterface(s)
|
||||
}
|
||||
|
||||
@@ -70,7 +70,7 @@ func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer boo
|
||||
if includeCacheLayer {
|
||||
// Adds the cache layer to the test store
|
||||
options = append(options, StoreOverride(func(s *Server) store.Store {
|
||||
lcl, err2 := localcachelayer.NewLocalCacheLayer(dbStore, s.Metrics, s.Cluster, s.CacheProvider)
|
||||
lcl, err2 := localcachelayer.NewLocalCacheLayer(dbStore, s.GetMetrics(), s.Cluster, s.CacheProvider)
|
||||
if err2 != nil {
|
||||
panic(err2)
|
||||
}
|
||||
|
||||
@@ -3,14 +3,34 @@
|
||||
|
||||
package platform
|
||||
|
||||
import (
|
||||
"errors"
|
||||
|
||||
"github.com/mattermost/mattermost-server/v6/config"
|
||||
"github.com/mattermost/mattermost-server/v6/einterfaces"
|
||||
"github.com/mattermost/mattermost-server/v6/shared/mlog"
|
||||
)
|
||||
|
||||
// ServiceConfig is used to initialize the PlatformService.
|
||||
// The mandatory fields will be checked during the initialization of the service.
|
||||
type ServiceConfig struct {
|
||||
// Mandatory fields
|
||||
ConfigStore *config.Store
|
||||
Logger *mlog.Logger
|
||||
StartMetrics bool // TODO: find an elegant way to start/stop metrics server by default
|
||||
// Optional fields
|
||||
Metrics einterfaces.MetricsInterface
|
||||
Cluster einterfaces.ClusterInterface
|
||||
}
|
||||
|
||||
func (c *ServiceConfig) validate() error {
|
||||
// Mandatory fields need to be checked here
|
||||
if c.ConfigStore == nil {
|
||||
return errors.New("ConfigStore is required")
|
||||
}
|
||||
|
||||
if c.Logger == nil {
|
||||
return errors.New("Logger is required")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
184
app/platform/metrics.go
Обычный файл
184
app/platform/metrics.go
Обычный файл
@@ -0,0 +1,184 @@
|
||||
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
||||
// See LICENSE.txt for license information.
|
||||
|
||||
package platform
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/pprof"
|
||||
"runtime"
|
||||
"sync"
|
||||
"text/template"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/handlers"
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/mattermost/mattermost-server/v6/einterfaces"
|
||||
"github.com/mattermost/mattermost-server/v6/model"
|
||||
"github.com/mattermost/mattermost-server/v6/shared/mlog"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
const TimeToWaitForConnectionsToCloseOnServerShutdown = time.Second
|
||||
|
||||
type platformMetrics struct {
|
||||
server *http.Server
|
||||
router *mux.Router
|
||||
lock sync.Mutex
|
||||
logger *mlog.Logger
|
||||
|
||||
metricsImpl einterfaces.MetricsInterface
|
||||
|
||||
cfgFn func() *model.Config
|
||||
}
|
||||
|
||||
// resetMetrics resets the metrics server. Clears the metrics if the metrics are disabled by the config.
|
||||
func (ps *PlatformService) resetMetrics(metricsImpl einterfaces.MetricsInterface, cfgFn func() *model.Config) error {
|
||||
if !*cfgFn().MetricsSettings.Enable {
|
||||
if ps.metrics != nil {
|
||||
return ps.metrics.stopMetricsServer()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
if ps.metrics != nil {
|
||||
if err := ps.metrics.stopMetricsServer(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
ps.metrics = &platformMetrics{
|
||||
cfgFn: cfgFn,
|
||||
metricsImpl: metricsImpl,
|
||||
logger: ps.logger,
|
||||
}
|
||||
|
||||
if err := ps.metrics.initMetricsRouter(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if metricsImpl != nil {
|
||||
metricsImpl.Register()
|
||||
}
|
||||
|
||||
return ps.metrics.startMetricsServer()
|
||||
}
|
||||
|
||||
func (pm *platformMetrics) stopMetricsServer() error {
|
||||
pm.lock.Lock()
|
||||
defer pm.lock.Unlock()
|
||||
|
||||
if pm.server != nil {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), TimeToWaitForConnectionsToCloseOnServerShutdown)
|
||||
defer cancel()
|
||||
|
||||
if err := pm.server.Shutdown(ctx); err != nil {
|
||||
return fmt.Errorf("could not shutdown metrics server: %v", err)
|
||||
}
|
||||
|
||||
pm.logger.Info("Metrics and profiling server is stopped")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pm *platformMetrics) startMetricsServer() error {
|
||||
var notify chan struct{}
|
||||
pm.lock.Lock()
|
||||
defer func() {
|
||||
if notify != nil {
|
||||
<-notify
|
||||
}
|
||||
pm.lock.Unlock()
|
||||
}()
|
||||
|
||||
l, err := net.Listen("tcp", *pm.cfgFn().MetricsSettings.ListenAddress)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
notify = make(chan struct{})
|
||||
pm.server = &http.Server{
|
||||
Handler: handlers.RecoveryHandler(handlers.PrintRecoveryStack(true))(pm.router),
|
||||
ReadTimeout: time.Duration(*pm.cfgFn().ServiceSettings.ReadTimeout) * time.Second,
|
||||
WriteTimeout: time.Duration(*pm.cfgFn().ServiceSettings.WriteTimeout) * time.Second,
|
||||
}
|
||||
|
||||
go func() {
|
||||
close(notify)
|
||||
if err := pm.server.Serve(l); err != nil && err != http.ErrServerClosed {
|
||||
pm.logger.Critical(err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
pm.logger.Info("Metrics and profiling server is started", mlog.String("address", l.Addr().String()))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pm *platformMetrics) initMetricsRouter() error {
|
||||
pm.router = mux.NewRouter()
|
||||
runtime.SetBlockProfileRate(*pm.cfgFn().MetricsSettings.BlockProfileRate)
|
||||
|
||||
metricsPage := `
|
||||
<html>
|
||||
<body>{{if .}}
|
||||
<div><a href="/metrics">Metrics</a></div>{{end}}
|
||||
<div><a href="/debug/pprof/">Profiling Root</a></div>
|
||||
<div><a href="/debug/pprof/cmdline">Profiling Command Line</a></div>
|
||||
<div><a href="/debug/pprof/symbol">Profiling Symbols</a></div>
|
||||
<div><a href="/debug/pprof/goroutine">Profiling Goroutines</a></div>
|
||||
<div><a href="/debug/pprof/heap">Profiling Heap</a></div>
|
||||
<div><a href="/debug/pprof/threadcreate">Profiling Threads</a></div>
|
||||
<div><a href="/debug/pprof/block">Profiling Blocking</a></div>
|
||||
<div><a href="/debug/pprof/trace">Profiling Execution Trace</a></div>
|
||||
<div><a href="/debug/pprof/profile">Profiling CPU</a></div>
|
||||
</body>
|
||||
</html>
|
||||
`
|
||||
metricsPageTmpl, err := template.New("page").Parse(metricsPage)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create template")
|
||||
}
|
||||
|
||||
rootHandler := func(w http.ResponseWriter, r *http.Request) {
|
||||
metricsPageTmpl.Execute(w, pm.metricsImpl != nil)
|
||||
}
|
||||
|
||||
pm.router.HandleFunc("/", rootHandler)
|
||||
pm.router.StrictSlash(true)
|
||||
|
||||
pm.router.Handle("/debug", http.RedirectHandler("/", http.StatusMovedPermanently))
|
||||
pm.router.HandleFunc("/debug/pprof/", pprof.Index)
|
||||
pm.router.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
|
||||
pm.router.HandleFunc("/debug/pprof/profile", pprof.Profile)
|
||||
pm.router.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
|
||||
pm.router.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
||||
|
||||
// Manually add support for paths linked to by index page at /debug/pprof/
|
||||
pm.router.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine"))
|
||||
pm.router.Handle("/debug/pprof/heap", pprof.Handler("heap"))
|
||||
pm.router.Handle("/debug/pprof/threadcreate", pprof.Handler("threadcreate"))
|
||||
pm.router.Handle("/debug/pprof/block", pprof.Handler("block"))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ps *PlatformService) HandleMetrics(route string, h http.Handler) {
|
||||
if ps.metrics != nil {
|
||||
ps.metrics.router.Handle(route, h)
|
||||
}
|
||||
}
|
||||
|
||||
func (ps *PlatformService) RestartMetrics() error {
|
||||
return ps.resetMetrics(ps.serviceConfig.Metrics, ps.serviceConfig.ConfigStore.Get)
|
||||
}
|
||||
|
||||
func (ps *PlatformService) Metrics() einterfaces.MetricsInterface {
|
||||
if ps.metrics == nil {
|
||||
return nil
|
||||
}
|
||||
return ps.metrics.metricsImpl
|
||||
}
|
||||
@@ -3,17 +3,49 @@
|
||||
|
||||
package platform
|
||||
|
||||
import (
|
||||
"github.com/mattermost/mattermost-server/v6/config"
|
||||
"github.com/mattermost/mattermost-server/v6/einterfaces"
|
||||
"github.com/mattermost/mattermost-server/v6/shared/mlog"
|
||||
)
|
||||
|
||||
// PlatformService is the service for the platform related tasks. It is
|
||||
// responsible for non-entity related functionalities that are required
|
||||
// by a product such as database access, configuration access, licensing etc.
|
||||
type PlatformService struct {
|
||||
serviceConfig ServiceConfig
|
||||
configStore *config.Store
|
||||
logger *mlog.Logger
|
||||
|
||||
metrics *platformMetrics
|
||||
|
||||
cluster einterfaces.ClusterInterface
|
||||
}
|
||||
|
||||
// New creates a new PlatformService.
|
||||
func New(c ServiceConfig) (*PlatformService, error) {
|
||||
if err := c.validate(); err != nil {
|
||||
func New(sc ServiceConfig) (*PlatformService, error) {
|
||||
if err := sc.validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &PlatformService{}, nil
|
||||
ps := &PlatformService{
|
||||
serviceConfig: sc,
|
||||
configStore: sc.ConfigStore,
|
||||
logger: sc.Logger,
|
||||
cluster: sc.Cluster,
|
||||
}
|
||||
|
||||
if err := ps.resetMetrics(sc.Metrics, ps.configStore.Get); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return ps, nil
|
||||
}
|
||||
|
||||
func (ps *PlatformService) ShutdownMetrics() error {
|
||||
if ps.metrics != nil {
|
||||
return ps.metrics.stopMetricsServer()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -226,7 +226,7 @@ func (ch *Channels) initPlugins(c *request.Context, pluginDir, webappPluginDir s
|
||||
return New(ServerConnector(ch)).NewPluginAPI(c, manifest)
|
||||
}
|
||||
|
||||
env, err := plugin.NewEnvironment(newAPIFunc, NewDriverImpl(ch.srv), pluginDir, webappPluginDir, ch.srv.Log, ch.srv.Metrics)
|
||||
env, err := plugin.NewEnvironment(newAPIFunc, NewDriverImpl(ch.srv), pluginDir, webappPluginDir, ch.srv.Log, ch.srv.GetMetrics())
|
||||
if err != nil {
|
||||
mlog.Error("Failed to start up plugins", mlog.Err(err))
|
||||
return
|
||||
|
||||
195
app/server.go
195
app/server.go
@@ -9,10 +9,8 @@ import (
|
||||
"crypto/tls"
|
||||
"fmt"
|
||||
"hash/maphash"
|
||||
"html/template"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/pprof"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/exec"
|
||||
@@ -27,7 +25,6 @@ import (
|
||||
|
||||
"github.com/getsentry/sentry-go"
|
||||
sentryhttp "github.com/getsentry/sentry-go/http"
|
||||
"github.com/gorilla/handlers"
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/rs/cors"
|
||||
@@ -35,6 +32,7 @@ import (
|
||||
|
||||
"github.com/mattermost/mattermost-server/v6/app/email"
|
||||
"github.com/mattermost/mattermost-server/v6/app/featureflag"
|
||||
"github.com/mattermost/mattermost-server/v6/app/platform"
|
||||
"github.com/mattermost/mattermost-server/v6/app/request"
|
||||
"github.com/mattermost/mattermost-server/v6/app/teams"
|
||||
"github.com/mattermost/mattermost-server/v6/app/users"
|
||||
@@ -131,10 +129,6 @@ type Server struct {
|
||||
|
||||
localModeServer *http.Server
|
||||
|
||||
metricsServer *http.Server
|
||||
metricsRouter *mux.Router
|
||||
metricsLock sync.Mutex
|
||||
|
||||
didFinishListen chan struct{}
|
||||
|
||||
goroutineCount int32
|
||||
@@ -177,6 +171,7 @@ type Server struct {
|
||||
configStore *configWrapper
|
||||
filestore filestore.FileBackend
|
||||
|
||||
platform *platform.PlatformService
|
||||
telemetryService *telemetry.TelemetryService
|
||||
userService *users.UserService
|
||||
teamService *teams.TeamService
|
||||
@@ -200,7 +195,6 @@ type Server struct {
|
||||
|
||||
Cluster einterfaces.ClusterInterface
|
||||
Cloud einterfaces.CloudInterface
|
||||
Metrics einterfaces.MetricsInterface
|
||||
LicenseManager einterfaces.LicenseInterface
|
||||
|
||||
CacheProvider cache.Provider
|
||||
@@ -286,6 +280,22 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
// Depends on step 3 (s.SearchEngine must be non-nil)
|
||||
s.initEnterprise()
|
||||
|
||||
platformCfg := platform.ServiceConfig{
|
||||
ConfigStore: s.configStore.Store,
|
||||
Logger: s.Log,
|
||||
StartMetrics: s.startMetrics,
|
||||
Cluster: s.Cluster,
|
||||
}
|
||||
if metricsInterface != nil {
|
||||
platformCfg.Metrics = metricsInterface(s)
|
||||
}
|
||||
|
||||
ps, sErr := platform.New(platformCfg)
|
||||
if sErr != nil {
|
||||
return nil, errors.Wrap(sErr, "failed to initialize platform")
|
||||
}
|
||||
s.platform = ps
|
||||
|
||||
// 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
|
||||
@@ -298,11 +308,11 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
// 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.Config().SqlSettings, s.Metrics)
|
||||
s.sqlStore = sqlstore.New(s.Config().SqlSettings, s.GetMetrics())
|
||||
|
||||
lcl, err2 := localcachelayer.NewLocalCacheLayer(
|
||||
retrylayer.New(s.sqlStore),
|
||||
s.Metrics,
|
||||
s.GetMetrics(),
|
||||
s.Cluster,
|
||||
s.CacheProvider,
|
||||
)
|
||||
@@ -327,7 +337,7 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
|
||||
return timerlayer.New(
|
||||
searchStore,
|
||||
s.Metrics,
|
||||
s.GetMetrics(),
|
||||
), nil
|
||||
}
|
||||
}
|
||||
@@ -343,7 +353,7 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
SessionStore: s.Store.Session(),
|
||||
OAuthStore: s.Store.OAuth(),
|
||||
ConfigFn: s.Config,
|
||||
Metrics: s.Metrics,
|
||||
Metrics: s.GetMetrics(),
|
||||
Cluster: s.Cluster,
|
||||
LicenseFn: s.License,
|
||||
})
|
||||
@@ -620,7 +630,9 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
}
|
||||
|
||||
if s.startMetrics {
|
||||
s.SetupMetricsServer()
|
||||
if err := s.platform.RestartMetrics(); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to start metrics")
|
||||
}
|
||||
}
|
||||
|
||||
s.AddLicenseListener(func(oldLicense, newLicense *model.License) {
|
||||
@@ -632,7 +644,9 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
return
|
||||
}
|
||||
|
||||
s.SetupMetricsServer()
|
||||
if err := s.platform.RestartMetrics(); err != nil {
|
||||
s.Log.Error("Failed to reset metrics server", mlog.Err(err))
|
||||
}
|
||||
})
|
||||
|
||||
s.SearchEngine.UpdateConfig(s.Config())
|
||||
@@ -701,24 +715,6 @@ func NewServer(options ...Option) (*Server, error) {
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *Server) SetupMetricsServer() {
|
||||
if !*s.Config().MetricsSettings.Enable {
|
||||
return
|
||||
}
|
||||
|
||||
s.StopMetricsServer()
|
||||
|
||||
if err := s.InitMetricsRouter(); err != nil {
|
||||
mlog.Error("Error initiating metrics router.", mlog.Err(err))
|
||||
}
|
||||
|
||||
if s.Metrics != nil {
|
||||
s.Metrics.Register()
|
||||
}
|
||||
|
||||
s.startMetricsServer()
|
||||
}
|
||||
|
||||
func maxInt(a, b int) int {
|
||||
if a > b {
|
||||
return a
|
||||
@@ -951,11 +947,11 @@ func (s *Server) startInterClusterServices(license *model.License) error {
|
||||
}
|
||||
|
||||
func (s *Server) enableLoggingMetrics() {
|
||||
if s.Metrics == nil {
|
||||
if s.GetMetrics() == nil {
|
||||
return
|
||||
}
|
||||
|
||||
s.Log.SetMetricsCollector(s.Metrics.GetLoggerMetricsCollector(), mlog.DefaultMetricsUpdateFreqMillis)
|
||||
s.Log.SetMetricsCollector(s.GetMetrics().GetLoggerMetricsCollector(), mlog.DefaultMetricsUpdateFreqMillis)
|
||||
|
||||
// logging config needs to be reloaded when metrics collector is added or changed.
|
||||
if err := s.initLogging(); err != nil {
|
||||
@@ -991,7 +987,7 @@ func (s *Server) StopHTTPServer() {
|
||||
}
|
||||
|
||||
func (s *Server) Shutdown() {
|
||||
mlog.Info("Stopping Server...")
|
||||
s.Log.Info("Stopping Server...")
|
||||
|
||||
defer sentry.Flush(2 * time.Second)
|
||||
|
||||
@@ -1002,24 +998,24 @@ func (s *Server) Shutdown() {
|
||||
|
||||
if s.tracer != nil {
|
||||
if err := s.tracer.Close(); err != nil {
|
||||
mlog.Warn("Unable to cleanly shutdown opentracing client", mlog.Err(err))
|
||||
s.Log.Warn("Unable to cleanly shutdown opentracing client", mlog.Err(err))
|
||||
}
|
||||
}
|
||||
|
||||
err := s.telemetryService.Shutdown()
|
||||
if err != nil {
|
||||
mlog.Warn("Unable to cleanly shutdown telemetry client", mlog.Err(err))
|
||||
s.Log.Warn("Unable to cleanly shutdown telemetry client", mlog.Err(err))
|
||||
}
|
||||
|
||||
s.serviceMux.RLock()
|
||||
if s.sharedChannelService != nil {
|
||||
if err = s.sharedChannelService.Shutdown(); err != nil {
|
||||
mlog.Error("Error shutting down shared channel services", mlog.Err(err))
|
||||
s.Log.Error("Error shutting down shared channel services", mlog.Err(err))
|
||||
}
|
||||
}
|
||||
if s.remoteClusterService != nil {
|
||||
if err = s.remoteClusterService.Shutdown(); err != nil {
|
||||
mlog.Error("Error shutting down intercluster services", mlog.Err(err))
|
||||
s.Log.Error("Error shutting down intercluster services", mlog.Err(err))
|
||||
}
|
||||
}
|
||||
s.serviceMux.RUnlock()
|
||||
@@ -1046,7 +1042,9 @@ func (s *Server) Shutdown() {
|
||||
s.Cluster.StopInterNodeCommunication()
|
||||
}
|
||||
|
||||
s.StopMetricsServer()
|
||||
if err = s.platform.ShutdownMetrics(); err != nil {
|
||||
s.Log.Warn("Failed to stop metrics server", mlog.Err(err))
|
||||
}
|
||||
|
||||
// This must be done after the cluster is stopped.
|
||||
if s.Jobs != nil {
|
||||
@@ -1054,10 +1052,10 @@ func (s *Server) Shutdown() {
|
||||
// before stopping them as both calls essentially become no-ops
|
||||
// if nothing is running.
|
||||
if err = s.Jobs.StopWorkers(); err != nil && !errors.Is(err, jobs.ErrWorkersNotRunning) {
|
||||
mlog.Warn("Failed to stop job server workers", mlog.Err(err))
|
||||
s.Log.Warn("Failed to stop job server workers", mlog.Err(err))
|
||||
}
|
||||
if err = s.Jobs.StopSchedulers(); err != nil && !errors.Is(err, jobs.ErrSchedulersNotRunning) {
|
||||
mlog.Warn("Failed to stop job server schedulers", mlog.Err(err))
|
||||
s.Log.Warn("Failed to stop job server schedulers", mlog.Err(err))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1066,7 +1064,7 @@ func (s *Server) Shutdown() {
|
||||
// on parent services.
|
||||
for name, product := range s.products {
|
||||
if err2 := product.Stop(); err2 != nil {
|
||||
mlog.Warn("Unable to cleanly stop product", mlog.String("name", name), mlog.Err(err2))
|
||||
s.Log.Warn("Unable to cleanly stop product", mlog.String("name", name), mlog.Err(err2))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1076,11 +1074,11 @@ func (s *Server) Shutdown() {
|
||||
|
||||
if s.CacheProvider != nil {
|
||||
if err = s.CacheProvider.Close(); err != nil {
|
||||
mlog.Warn("Unable to cleanly shutdown cache", mlog.Err(err))
|
||||
s.Log.Warn("Unable to cleanly shutdown cache", mlog.Err(err))
|
||||
}
|
||||
}
|
||||
|
||||
mlog.Info("Server stopped")
|
||||
s.Log.Info("Server stopped")
|
||||
|
||||
// shutdown main and notification loggers which will flush any remaining log records.
|
||||
timeoutCtx, timeoutCancel := context.WithTimeout(context.Background(), time.Second*15)
|
||||
@@ -1630,104 +1628,8 @@ func doConfigCleanup(s *Server) {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) StopMetricsServer() {
|
||||
s.metricsLock.Lock()
|
||||
defer s.metricsLock.Unlock()
|
||||
|
||||
if s.metricsServer != nil {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), TimeToWaitForConnectionsToCloseOnServerShutdown)
|
||||
defer cancel()
|
||||
|
||||
s.metricsServer.Shutdown(ctx)
|
||||
s.Log.Info("Metrics and profiling server is stopping")
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) HandleMetrics(route string, h http.Handler) {
|
||||
if s.metricsRouter != nil {
|
||||
s.metricsRouter.Handle(route, h)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) InitMetricsRouter() error {
|
||||
s.metricsRouter = mux.NewRouter()
|
||||
runtime.SetBlockProfileRate(*s.Config().MetricsSettings.BlockProfileRate)
|
||||
|
||||
metricsPage := `
|
||||
<html>
|
||||
<body>{{if .}}
|
||||
<div><a href="/metrics">Metrics</a></div>{{end}}
|
||||
<div><a href="/debug/pprof/">Profiling Root</a></div>
|
||||
<div><a href="/debug/pprof/cmdline">Profiling Command Line</a></div>
|
||||
<div><a href="/debug/pprof/symbol">Profiling Symbols</a></div>
|
||||
<div><a href="/debug/pprof/goroutine">Profiling Goroutines</a></div>
|
||||
<div><a href="/debug/pprof/heap">Profiling Heap</a></div>
|
||||
<div><a href="/debug/pprof/threadcreate">Profiling Threads</a></div>
|
||||
<div><a href="/debug/pprof/block">Profiling Blocking</a></div>
|
||||
<div><a href="/debug/pprof/trace">Profiling Execution Trace</a></div>
|
||||
<div><a href="/debug/pprof/profile">Profiling CPU</a></div>
|
||||
</body>
|
||||
</html>
|
||||
`
|
||||
metricsPageTmpl, err := template.New("page").Parse(metricsPage)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "failed to create template")
|
||||
}
|
||||
|
||||
rootHandler := func(w http.ResponseWriter, r *http.Request) {
|
||||
metricsPageTmpl.Execute(w, s.Metrics != nil)
|
||||
}
|
||||
|
||||
s.metricsRouter.HandleFunc("/", rootHandler)
|
||||
s.metricsRouter.StrictSlash(true)
|
||||
|
||||
s.metricsRouter.Handle("/debug", http.RedirectHandler("/", http.StatusMovedPermanently))
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/", pprof.Index)
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/profile", pprof.Profile)
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
|
||||
s.metricsRouter.HandleFunc("/debug/pprof/trace", pprof.Trace)
|
||||
|
||||
// Manually add support for paths linked to by index page at /debug/pprof/
|
||||
s.metricsRouter.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine"))
|
||||
s.metricsRouter.Handle("/debug/pprof/heap", pprof.Handler("heap"))
|
||||
s.metricsRouter.Handle("/debug/pprof/threadcreate", pprof.Handler("threadcreate"))
|
||||
s.metricsRouter.Handle("/debug/pprof/block", pprof.Handler("block"))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) startMetricsServer() {
|
||||
var notify chan struct{}
|
||||
s.metricsLock.Lock()
|
||||
defer func() {
|
||||
if notify != nil {
|
||||
<-notify
|
||||
}
|
||||
s.metricsLock.Unlock()
|
||||
}()
|
||||
|
||||
l, err := net.Listen("tcp", *s.Config().MetricsSettings.ListenAddress)
|
||||
if err != nil {
|
||||
mlog.Error(err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
notify = make(chan struct{})
|
||||
s.metricsServer = &http.Server{
|
||||
Handler: handlers.RecoveryHandler(handlers.PrintRecoveryStack(true))(s.metricsRouter),
|
||||
ReadTimeout: time.Duration(*s.Config().ServiceSettings.ReadTimeout) * time.Second,
|
||||
WriteTimeout: time.Duration(*s.Config().ServiceSettings.WriteTimeout) * time.Second,
|
||||
}
|
||||
|
||||
go func() {
|
||||
close(notify)
|
||||
if err := s.metricsServer.Serve(l); err != nil && err != http.ErrServerClosed {
|
||||
mlog.Critical(err.Error())
|
||||
}
|
||||
}()
|
||||
|
||||
s.Log.Info("Metrics and profiling server is started", mlog.String("address", l.Addr().String()))
|
||||
s.platform.HandleMetrics(route, h)
|
||||
}
|
||||
|
||||
func (s *Server) sendLicenseUpForRenewalEmail(users map[string]*model.User, license *model.License) *model.AppError {
|
||||
@@ -1956,7 +1858,7 @@ func (ch *Channels) ClientConfigHash() string {
|
||||
}
|
||||
|
||||
func (s *Server) initJobs() {
|
||||
s.Jobs = jobs.NewJobServer(s, s.Store, s.Metrics)
|
||||
s.Jobs = jobs.NewJobServer(s, s.Store, s.GetMetrics())
|
||||
|
||||
if jobsDataRetentionJobInterface != nil {
|
||||
builder := jobsDataRetentionJobInterface(s)
|
||||
@@ -2039,7 +1941,7 @@ func (s *Server) initJobs() {
|
||||
|
||||
s.Jobs.RegisterJobType(
|
||||
model.JobTypeActiveUsers,
|
||||
active_users.MakeWorker(s.Jobs, s.Store, func() einterfaces.MetricsInterface { return s.Metrics }),
|
||||
active_users.MakeWorker(s.Jobs, s.Store, func() einterfaces.MetricsInterface { return s.GetMetrics() }),
|
||||
active_users.MakeScheduler(s.Jobs),
|
||||
)
|
||||
|
||||
@@ -2106,7 +2008,10 @@ func (s *Server) GetSharedChannelSyncService() SharedChannelServiceIFace {
|
||||
// GetMetrics returns the server's Metrics interface. Exposing via a method
|
||||
// allows interfaces to be created with subsets of server APIs.
|
||||
func (s *Server) GetMetrics() einterfaces.MetricsInterface {
|
||||
return s.Metrics
|
||||
if s.platform == nil {
|
||||
return nil
|
||||
}
|
||||
return s.platform.Metrics()
|
||||
}
|
||||
|
||||
// SetRemoteClusterService sets the `RemoteClusterService` to be used by the server.
|
||||
|
||||
@@ -65,7 +65,7 @@ func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer boo
|
||||
options = append(options, app.ConfigStore(memoryStore))
|
||||
if includeCacheLayer {
|
||||
options = append(options, app.StoreOverride(func(s *app.Server) store.Store {
|
||||
lcl, err2 := localcachelayer.NewLocalCacheLayer(dbStore, s.Metrics, s.Cluster, s.CacheProvider)
|
||||
lcl, err2 := localcachelayer.NewLocalCacheLayer(dbStore, s.GetMetrics(), s.Cluster, s.CacheProvider)
|
||||
if err2 != nil {
|
||||
panic(err2)
|
||||
}
|
||||
|
||||
@@ -157,8 +157,8 @@ func (a *App) HubUnregister(webConn *WebConn) {
|
||||
}
|
||||
|
||||
func (s *Server) Publish(message *model.WebSocketEvent) {
|
||||
if s.Metrics != nil {
|
||||
s.Metrics.IncrementWebsocketEvent(message.EventType())
|
||||
if s.GetMetrics() != nil {
|
||||
s.GetMetrics().IncrementWebsocketEvent(message.EventType())
|
||||
}
|
||||
|
||||
s.PublishSkipClusterSend(message)
|
||||
@@ -357,7 +357,7 @@ func (h *Hub) Broadcast(message *model.WebSocketEvent) {
|
||||
// And possibly, we can look into doing the hub initialization inside
|
||||
// NewServer itself.
|
||||
if h != nil && message != nil {
|
||||
if metrics := h.srv.Metrics; metrics != nil {
|
||||
if metrics := h.srv.GetMetrics(); metrics != nil {
|
||||
metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
|
||||
}
|
||||
select {
|
||||
@@ -525,7 +525,7 @@ func (h *Hub) Start() {
|
||||
connIndex.Remove(directMsg.conn)
|
||||
}
|
||||
case msg := <-h.broadcast:
|
||||
if metrics := h.srv.Metrics; metrics != nil {
|
||||
if metrics := h.srv.GetMetrics(); metrics != nil {
|
||||
metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1)
|
||||
}
|
||||
msg = msg.PrecomputeJSON()
|
||||
|
||||
@@ -96,7 +96,7 @@ func setupTestHelper(tb testing.TB, includeCacheLayer bool) *TestHelper {
|
||||
}
|
||||
if includeCacheLayer {
|
||||
// Adds the cache layer to the test store
|
||||
s.Store, err = localcachelayer.NewLocalCacheLayer(s.Store, s.Metrics, s.Cluster, s.CacheProvider)
|
||||
s.Store, err = localcachelayer.NewLocalCacheLayer(s.Store, s.GetMetrics(), s.Cluster, s.CacheProvider)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
Ссылка в новой задаче
Block a user