diff --git a/app/platform/cluster.go b/app/platform/cluster.go index 47722ba007..6575ab7daa 100644 --- a/app/platform/cluster.go +++ b/app/platform/cluster.go @@ -180,8 +180,8 @@ func (ps *PlatformService) InvokeClusterLeaderChangedListeners() { } func (ps *PlatformService) Publish(message *model.WebSocketEvent) { - if ps.metricsImpl() != nil { - ps.metricsImpl().IncrementWebsocketEvent(message.EventType()) + if ps.metricsIFace != nil { + ps.metricsIFace.IncrementWebsocketEvent(message.EventType()) } ps.PublishSkipClusterSend(message) diff --git a/app/platform/config.go b/app/platform/config.go index 5b0218dcd6..fd5bb05f76 100644 --- a/app/platform/config.go +++ b/app/platform/config.go @@ -32,7 +32,6 @@ type ServiceConfig struct { ConfigStore *config.Store Store store.Store // Optional fields - Metrics einterfaces.MetricsInterface Cluster einterfaces.ClusterInterface } diff --git a/app/platform/enterprise.go b/app/platform/enterprise.go index 37849fb3f3..cf7cdf0399 100644 --- a/app/platform/enterprise.go +++ b/app/platform/enterprise.go @@ -26,8 +26,8 @@ func RegisterLicenseInterface(f func(*PlatformService) einterfaces.LicenseInterf licenseInterface = f } -var metricsInterface func(*PlatformService, string, string) einterfaces.MetricsInterface +var metricsInterfaceFn func(*PlatformService, string, string) einterfaces.MetricsInterface func RegisterMetricsInterface(f func(*PlatformService, string, string) einterfaces.MetricsInterface) { - metricsInterface = f + metricsInterfaceFn = f } diff --git a/app/platform/helper_test.go b/app/platform/helper_test.go index 17e40627a3..b8d2da4594 100644 --- a/app/platform/helper_test.go +++ b/app/platform/helper_test.go @@ -55,7 +55,7 @@ func (ms *mockSuite) UserCanSeeOtherUser(userID string, otherUserId string) (boo return true, nil } -func Setup(tb testing.TB) *TestHelper { +func Setup(tb testing.TB, options ...Option) *TestHelper { if testing.Short() { tb.SkipNow() } @@ -64,7 +64,7 @@ func Setup(tb testing.TB) *TestHelper { dbStore.MarkSystemRanUnitTests() mainHelper.PreloadMigrations() - return setupTestHelper(dbStore, false, true, tb) + return setupTestHelper(dbStore, false, true, tb, options...) } func (th *TestHelper) InitBasic() *TestHelper { @@ -96,9 +96,9 @@ func (th *TestHelper) InitBasic() *TestHelper { return th } -func SetupWithStoreMock(tb testing.TB) *TestHelper { +func SetupWithStoreMock(tb testing.TB, options ...Option) *TestHelper { mockStore := testlib.GetMockStoreForSetupFunctions() - th := setupTestHelper(mockStore, false, false, tb) + th := setupTestHelper(mockStore, false, false, tb, options...) statusMock := mocks.StatusStore{} statusMock.On("UpdateExpiredDNDStatuses").Return([]*model.Status{}, nil) statusMock.On("Get", "user1").Return(&model.Status{UserId: "user1", Status: model.StatusOnline}, nil) @@ -126,7 +126,7 @@ func SetupWithCluster(tb testing.TB, cluster einterfaces.ClusterInterface) *Test return th } -func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer bool, tb testing.TB) *TestHelper { +func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer bool, tb testing.TB, options ...Option) *TestHelper { tempWorkspace, err := ioutil.TempDir("", "apptest") if err != nil { panic(err) @@ -149,7 +149,7 @@ func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer boo ps, err := New(ServiceConfig{ ConfigStore: configStore, Store: dbStore, - }) + }, options...) if err != nil { panic(err) } @@ -181,7 +181,10 @@ func setupTestHelper(dbStore store.Store, enterprise bool, includeCacheLayer boo th.Service.SetLicense(nil) } - th.Service.Start(th.Suite) + err = th.Service.Start(th.Suite) + if err != nil { + panic(err) + } return th } diff --git a/app/platform/log.go b/app/platform/log.go index 1d2cc8cf33..42ad85b54e 100644 --- a/app/platform/log.go +++ b/app/platform/log.go @@ -89,11 +89,11 @@ func (ps *PlatformService) NotificationsLogger() *mlog.Logger { } func (ps *PlatformService) EnableLoggingMetrics() { - if ps.metrics == nil || ps.metricsImpl() == nil { + if ps.metrics == nil || ps.metricsIFace == nil { return } - ps.logger.SetMetricsCollector(ps.metricsImpl().GetLoggerMetricsCollector(), mlog.DefaultMetricsUpdateFreqMillis) + ps.logger.SetMetricsCollector(ps.metricsIFace.GetLoggerMetricsCollector(), mlog.DefaultMetricsUpdateFreqMillis) // logging config needs to be reloaded when metrics collector is added or changed. if err := ps.initLogging(); err != nil { diff --git a/app/platform/metrics.go b/app/platform/metrics.go index f486f85cbc..2253594011 100644 --- a/app/platform/metrics.go +++ b/app/platform/metrics.go @@ -32,20 +32,13 @@ type platformMetrics struct { metricsImpl einterfaces.MetricsInterface - cfgFn func() *model.Config -} - -func (ps *PlatformService) metricsImpl() einterfaces.MetricsInterface { - if ps.metrics == nil { - return nil - } - - return ps.metrics.metricsImpl + cfgFn func() *model.Config + listenAddr string } // 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 { +func (ps *PlatformService) resetMetrics() error { + if !*ps.Config().MetricsSettings.Enable { if ps.metrics != nil { return ps.metrics.stopMetricsServer() } @@ -59,8 +52,8 @@ func (ps *PlatformService) resetMetrics(metricsImpl einterfaces.MetricsInterface } ps.metrics = &platformMetrics{ - cfgFn: cfgFn, - metricsImpl: metricsImpl, + cfgFn: ps.Config, + metricsImpl: ps.metricsIFace, logger: ps.logger, } @@ -68,8 +61,8 @@ func (ps *PlatformService) resetMetrics(metricsImpl einterfaces.MetricsInterface return err } - if metricsImpl != nil { - metricsImpl.Register() + if ps.metricsIFace != nil { + ps.metricsIFace.Register() } return ps.metrics.startMetricsServer() @@ -122,7 +115,8 @@ func (pm *platformMetrics) startMetricsServer() error { } }() - pm.logger.Info("Metrics and profiling server is started", mlog.String("address", l.Addr().String())) + pm.listenAddr = l.Addr().String() + pm.logger.Info("Metrics and profiling server is started", mlog.String("address", pm.listenAddr)) return nil } @@ -181,7 +175,7 @@ func (ps *PlatformService) HandleMetrics(route string, h http.Handler) { } func (ps *PlatformService) RestartMetrics() error { - return ps.resetMetrics(ps.serviceConfig.Metrics, ps.configStore.Get) + return ps.resetMetrics() } func (ps *PlatformService) Metrics() einterfaces.MetricsInterface { @@ -189,5 +183,5 @@ func (ps *PlatformService) Metrics() einterfaces.MetricsInterface { return nil } - return ps.metricsImpl() + return ps.metricsIFace } diff --git a/app/platform/options.go b/app/platform/options.go index 99d63d8f26..072914e104 100644 --- a/app/platform/options.go +++ b/app/platform/options.go @@ -46,7 +46,7 @@ func StoreOverride(override any) Option { func StoreOverrideWithCache(override store.Store) Option { return func(ps *PlatformService) error { ps.newStore = func() (store.Store, error) { - lcl, err := localcachelayer.NewLocalCacheLayer(override, ps.metricsImpl(), ps.clusterIFace, ps.cacheProvider) + lcl, err := localcachelayer.NewLocalCacheLayer(override, ps.metricsIFace, ps.clusterIFace, ps.cacheProvider) if err != nil { return nil, err } diff --git a/app/platform/service.go b/app/platform/service.go index b5a506ae50..d2104fb2bb 100644 --- a/app/platform/service.go +++ b/app/platform/service.go @@ -39,8 +39,7 @@ type PlatformService struct { WebSocketRouter *WebSocketRouter - serviceConfig *ServiceConfig - configStore *config.Store + configStore *config.Store cacheProvider cache.Provider statusCache cache.Cache @@ -57,6 +56,7 @@ type PlatformService struct { startMetrics bool metrics *platformMetrics + metricsIFace einterfaces.MetricsInterface featureFlagSynchronizerMutex sync.Mutex featureFlagSynchronizer *featureflag.Synchronizer @@ -100,7 +100,6 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) { // Step 0: Create the PlatformService. // ConfigStore is and should be handled on a upper level. ps := &PlatformService{ - serviceConfig: &sc, Store: sc.Store, configStore: sc.ConfigStore, clusterIFace: sc.Cluster, @@ -170,15 +169,20 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) { // Depends on step 3 (s.SearchEngine must be non-nil) ps.initEnterprise() - // Step 5: Store. - // Depends on Step 1 (config), 4 (metrics, cluster) and 5 (cacheProvider). + // Step 5: Init Metrics + if metricsInterfaceFn != nil { + ps.metricsIFace = metricsInterfaceFn(ps, *ps.configStore.Get().SqlSettings.DriverName, *ps.configStore.Get().SqlSettings.DataSource) + } + + // Step 6: Store. + // Depends on Step 0 (config), 1 (cacheProvider), 3 (search engine), 5 (metrics) and cluster. if ps.newStore == nil { ps.newStore = func() (store.Store, error) { - ps.sqlStore = sqlstore.New(ps.Config().SqlSettings, ps.Metrics()) + ps.sqlStore = sqlstore.New(ps.Config().SqlSettings, ps.metricsIFace) lcl, err2 := localcachelayer.NewLocalCacheLayer( retrylayer.New(ps.sqlStore), - ps.Metrics(), + ps.metricsIFace, ps.clusterIFace, ps.cacheProvider, ) @@ -204,7 +208,7 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) { return timerlayer.New( searchStore, - ps.Metrics(), + ps.metricsIFace, ), nil } } @@ -234,20 +238,19 @@ func New(sc ServiceConfig, options ...Option) (*PlatformService, error) { return nil, fmt.Errorf("could not create session cache: %w", err) } + // Step 7: Init License if model.BuildEnterpriseReady == "true" { ps.LoadLicense() } - if metricsInterface != nil { - sc.Metrics = metricsInterface(ps, *ps.configStore.Get().SqlSettings.DriverName, *ps.configStore.Get().SqlSettings.DataSource) - } - + // Step 8: Init Metrics Server depends on step 6 (store) and 7 (license) if ps.startMetrics { - if err = ps.resetMetrics(sc.Metrics, ps.configStore.Get); err != nil { - return nil, err + if mErr := ps.resetMetrics(); mErr != nil { + return nil, mErr } } + // Step 9: Init AsymmetricSigningKey depends on step 6 (store) if err = ps.EnsureAsymmetricSigningKey(); err != nil { return nil, fmt.Errorf("unable to ensure asymmetric signing key: %w", err) } @@ -299,6 +302,7 @@ func (ps *PlatformService) Start(suite SuiteIFace) error { return } }) + ps.licenseListenerId = ps.AddLicenseListener(func(oldLicense, newLicense *model.License) { ps.regenerateClientConfig() diff --git a/app/platform/service_test.go b/app/platform/service_test.go index e53afaae98..b2b7fa1e89 100644 --- a/app/platform/service_test.go +++ b/app/platform/service_test.go @@ -4,12 +4,16 @@ package platform import ( + "net/http" "os" + "strings" "testing" "github.com/mattermost/mattermost-server/v6/config" + "github.com/mattermost/mattermost-server/v6/einterfaces/mocks" "github.com/mattermost/mattermost-server/v6/model" "github.com/mattermost/mattermost-server/v6/store/storetest" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" ) @@ -83,3 +87,69 @@ func TestReadReplicaDisabledBasedOnLicense(t *testing.T) { require.Len(t, ps.Config().SqlSettings.DataSourceSearchReplicas, 1) }) } + +func TestMetrics(t *testing.T) { + t.Run("ensure the metrics server is not started by default", func(t *testing.T) { + th := Setup(t) + defer th.TearDown() + + require.Nil(t, th.Service.metrics) + }) + + t.Run("ensure the metrics server is started", func(t *testing.T) { + th := Setup(t, StartMetrics()) + defer th.TearDown() + + // there is no config listener for the metrics + // we handle it on config save step + th.Service.UpdateConfig(func(c *model.Config) { + c.MetricsSettings.Enable = model.NewBool(true) + }) + th.Service.SaveConfig(th.Service.Config(), false) + + require.NotNil(t, th.Service.metrics) + metricsAddr := strings.Replace(th.Service.metrics.listenAddr, "[::]", "http://localhost", 1) + + resp, err := http.Get(metricsAddr) + require.NoError(t, err) + require.Equal(t, http.StatusOK, resp.StatusCode) + + th.Service.UpdateConfig(func(c *model.Config) { + c.MetricsSettings.Enable = model.NewBool(false) + }) + th.Service.SaveConfig(th.Service.Config(), false) + + _, err = http.Get(metricsAddr) + require.Error(t, err) + }) + + t.Run("ensure the metrics server is started with advanced metrics", func(t *testing.T) { + th := Setup(t, StartMetrics()) + defer th.TearDown() + + mockMetricsImpl := &mocks.MetricsInterface{} + mockMetricsImpl.On("Register").Return() + + th.Service.metricsIFace = mockMetricsImpl + err := th.Service.resetMetrics() + require.NoError(t, err) + + mockMetricsImpl.AssertExpectations(t) + }) + + t.Run("ensure advanced metrics have database metrics", func(t *testing.T) { + mockMetricsImpl := &mocks.MetricsInterface{} + mockMetricsImpl.On("Register").Return() + mockMetricsImpl.On("ObserveStoreMethodDuration", mock.Anything, mock.Anything, mock.Anything).Return() + + th := Setup(t, StartMetrics(), func(ps *PlatformService) error { + ps.metricsIFace = mockMetricsImpl + return nil + }) + defer th.TearDown() + + _ = th.CreateUserOrGuest(false) + + mockMetricsImpl.AssertExpectations(t) + }) +} diff --git a/app/platform/session.go b/app/platform/session.go index 1b085f7dd1..f6da9aec77 100644 --- a/app/platform/session.go +++ b/app/platform/session.go @@ -59,7 +59,7 @@ func (ps *PlatformService) ClearUserSessionCacheLocal(userID string) { if err := ps.sessionCache.Get(key, &session); err == nil { if session.UserId == userID { ps.sessionCache.Remove(key) - if m := ps.metricsImpl(); m != nil { + if m := ps.metricsIFace; m != nil { m.IncrementMemCacheInvalidationCounterSession() } } @@ -100,11 +100,11 @@ func (ps *PlatformService) ClearAllUsersSessionCache() { func (ps *PlatformService) GetSession(token string) (*model.Session, error) { var session = ps.sessionPool.Get().(*model.Session) if err := ps.sessionCache.Get(token, session); err == nil { - if m := ps.metricsImpl(); m != nil { + if m := ps.metricsIFace; m != nil { m.IncrementMemCacheHitCounterSession() } } else { - if m := ps.metricsImpl(); m != nil { + if m := ps.metricsIFace; m != nil { m.IncrementMemCacheMissCounterSession() } } diff --git a/app/platform/web_conn.go b/app/platform/web_conn.go index 2107333428..4e4fc75062 100644 --- a/app/platform/web_conn.go +++ b/app/platform/web_conn.go @@ -409,7 +409,7 @@ func (wc *WebConn) writePump() { wc.logSocketErr("websocket.drainDeadQueue", err) return } - if m := wc.Platform.metricsImpl(); m != nil { + if m := wc.Platform.metricsIFace; m != nil { m.IncrementWebsocketReconnectEvent(reconnectFound) } } else if wc.hasMsgLoss() { @@ -427,11 +427,11 @@ func (wc *WebConn) writePump() { wc.logSocketErr("websocket.sendHello", err) return } - if m := wc.Platform.metricsImpl(); m != nil { + if m := wc.Platform.metricsIFace; m != nil { m.IncrementWebsocketReconnectEvent(reconnectNotFound) } } else { - if m := wc.Platform.metricsImpl(); m != nil { + if m := wc.Platform.metricsIFace; m != nil { m.IncrementWebsocketReconnectEvent(reconnectLossless) } } @@ -488,7 +488,7 @@ func (wc *WebConn) writePump() { return } - if m := wc.Platform.metricsImpl(); m != nil { + if m := wc.Platform.metricsIFace; m != nil { m.IncrementWebSocketBroadcast(msg.EventType()) } case <-ticker.C: diff --git a/app/platform/web_hub.go b/app/platform/web_hub.go index 3a0d5fd6bb..dc83d574ce 100644 --- a/app/platform/web_hub.go +++ b/app/platform/web_hub.go @@ -142,7 +142,7 @@ func (ps *PlatformService) GetHubForUserId(userID string) *Hub { func (ps *PlatformService) HubRegister(webConn *WebConn) { hub := ps.GetHubForUserId(webConn.UserId) if hub != nil { - if metrics := ps.metricsImpl(); metrics != nil { + if metrics := ps.metricsIFace; metrics != nil { metrics.IncrementWebSocketBroadcastUsersRegistered(strconv.Itoa(hub.connectionIndex), 1) } hub.Register(webConn) @@ -153,7 +153,7 @@ func (ps *PlatformService) HubRegister(webConn *WebConn) { func (ps *PlatformService) HubUnregister(webConn *WebConn) { hub := ps.GetHubForUserId(webConn.UserId) if hub != nil { - if metrics := ps.metricsImpl(); metrics != nil { + if metrics := ps.metricsIFace; metrics != nil { metrics.DecrementWebSocketBroadcastUsersRegistered(strconv.Itoa(hub.connectionIndex), 1) } hub.Unregister(webConn) @@ -317,7 +317,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.platform.metricsImpl(); metrics != nil { + if metrics := h.platform.metricsIFace; metrics != nil { metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1) } select { @@ -483,7 +483,7 @@ func (h *Hub) Start(suite SuiteIFace) { connIndex.Remove(directMsg.conn) } case msg := <-h.broadcast: - if metrics := h.platform.metricsImpl(); metrics != nil { + if metrics := h.platform.metricsIFace; metrics != nil { metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1) } msg = msg.PrecomputeJSON()