[MM-47468] Fix store metrics initialization (#21374)

* initialize metrics before store

* add tests

* reflect review comments

app/platform/cluster.go

@@ -180,8 +180,8 @@ func (ps *PlatformService) InvokeClusterLeaderChangedListeners() {
 }
 
 func (ps *PlatformService) Publish(message *model.WebSocketEvent) {
-	if ps.metricsImpl() != nil {
+	if ps.metricsIFace != nil {
-		ps.metricsImpl().IncrementWebsocketEvent(message.EventType())
+		ps.metricsIFace.IncrementWebsocketEvent(message.EventType())
 	}
 
 	ps.PublishSkipClusterSend(message)

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
 }
 

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
 }

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
 }

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 {

app/platform/metrics.go

@@ -32,20 +32,13 @@ type platformMetrics struct {
 
 	metricsImpl einterfaces.MetricsInterface
 
-	cfgFn func() *model.Config
+	cfgFn      func() *model.Config
-}
+	listenAddr string
-
-func (ps *PlatformService) metricsImpl() einterfaces.MetricsInterface {
-	if ps.metrics == nil {
-		return nil
-	}
-
-	return ps.metrics.metricsImpl
 }
 
 // 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 {
+func (ps *PlatformService) resetMetrics() error {
-	if !*cfgFn().MetricsSettings.Enable {
+	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,
+		cfgFn:       ps.Config,
-		metricsImpl: metricsImpl,
+		metricsImpl: ps.metricsIFace,
 		logger:      ps.logger,
 	}
 
@@ -68,8 +61,8 @@ func (ps *PlatformService) resetMetrics(metricsImpl einterfaces.MetricsInterface
 		return err
 	}
 
-	if metricsImpl != nil {
+	if ps.metricsIFace != nil {
-		metricsImpl.Register()
+		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
 }

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
 			}

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.
+	// Step 5: Init Metrics
-	// Depends on Step 1 (config), 4 (metrics, cluster) and 5 (cacheProvider).
+	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 {
+	// Step 8: Init Metrics Server depends on step 6 (store) and 7 (license)
-		sc.Metrics = metricsInterface(ps, *ps.configStore.Get().SqlSettings.DriverName, *ps.configStore.Get().SqlSettings.DataSource)
-	}
-
 	if ps.startMetrics {
-		if err = ps.resetMetrics(sc.Metrics, ps.configStore.Get); err != nil {
+		if mErr := ps.resetMetrics(); mErr != nil {
-			return nil, err
+			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()
 

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)
+	})
+}

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()
 		}
 	}

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:

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()