From d949bd16389bbae405cc477eeff9ce3c5ca45980 Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Mon, 18 Oct 2021 12:36:13 +0530 Subject: [PATCH] Remove app initialization from WebHub (#18687) - Make it a Server method. - Pass Server instead of App. - Remove global app instance from NewServer, rather create local instances whenever needed. - Remove App from Websocket router, and create dynamically on every request. - Remove HubStart and HubStop from App methods. - Explicitly using s.Log instead of the global logger to indicate dependency on Server. We could have passed the logger explicitly but it doesn't look ideal. ```release-note NONE ``` --- app/app_iface.go | 5 --- app/opentracing/opentracing_layer.go | 47 ---------------------------- app/server.go | 25 ++++++++------- app/web_hub.go | 42 ++++++++++++------------- app/web_hub_test.go | 10 +++--- app/webhub_fuzz.go | 2 +- app/websocket_router.go | 21 ++++++------- 7 files changed, 50 insertions(+), 102 deletions(-) diff --git a/app/app_iface.go b/app/app_iface.go index 52f9ee703c..fa2a19c24f 100644 --- a/app/app_iface.go +++ b/app/app_iface.go @@ -212,8 +212,6 @@ type AppIface interface { HasRemote(channelID string, remoteID string) (bool, error) // HubRegister registers a connection to a hub. HubRegister(webConn *WebConn) - // HubStart starts all the hubs. - HubStart() // HubUnregister unregisters a connection from a hub. HubUnregister(webConn *WebConn) // InstallMarketplacePlugin installs a plugin listed in the marketplace server. It will get the plugin bundle @@ -244,8 +242,6 @@ type AppIface interface { MoveChannel(c *request.Context, team *model.Team, channel *model.Channel, user *model.User) *model.AppError // NewWebConn returns a new WebConn instance. NewWebConn(cfg *WebConnConfig) *WebConn - // NewWebHub creates a new Hub. - NewWebHub() *Hub // NotifySessionsExpired is called periodically from the job server to notify any mobile sessions that have expired. NotifySessionsExpired() *model.AppError // OverrideIconURLIfEmoji changes the post icon override URL prop, if it has an emoji icon, @@ -805,7 +801,6 @@ type AppIface interface { HasPermissionToTeam(askingUserId string, teamID string, permission *model.Permission) bool HasPermissionToUser(askingUserId string, userID string) bool HasSharedChannel(channelID string) (bool, error) - HubStop() ImageProxy() *imageproxy.ImageProxy ImageProxyAdder() func(string) string ImageProxyRemover() (f func(string) string) diff --git a/app/opentracing/opentracing_layer.go b/app/opentracing/opentracing_layer.go index d3f6521982..1ac8f1866f 100644 --- a/app/opentracing/opentracing_layer.go +++ b/app/opentracing/opentracing_layer.go @@ -10449,36 +10449,6 @@ func (a *OpenTracingAppLayer) HubRegister(webConn *app.WebConn) { a.app.HubRegister(webConn) } -func (a *OpenTracingAppLayer) HubStart() { - origCtx := a.ctx - span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.HubStart") - - a.ctx = newCtx - a.app.Srv().Store.SetContext(newCtx) - defer func() { - a.app.Srv().Store.SetContext(origCtx) - a.ctx = origCtx - }() - - defer span.Finish() - a.app.HubStart() -} - -func (a *OpenTracingAppLayer) HubStop() { - origCtx := a.ctx - span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.HubStop") - - a.ctx = newCtx - a.app.Srv().Store.SetContext(newCtx) - defer func() { - a.app.Srv().Store.SetContext(origCtx) - a.ctx = origCtx - }() - - defer span.Finish() - a.app.HubStop() -} - func (a *OpenTracingAppLayer) HubUnregister(webConn *app.WebConn) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.HubUnregister") @@ -11530,23 +11500,6 @@ func (a *OpenTracingAppLayer) NewWebConn(cfg *app.WebConnConfig) *app.WebConn { return resultVar0 } -func (a *OpenTracingAppLayer) NewWebHub() *app.Hub { - origCtx := a.ctx - span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.NewWebHub") - - a.ctx = newCtx - a.app.Srv().Store.SetContext(newCtx) - defer func() { - a.app.Srv().Store.SetContext(origCtx) - a.ctx = origCtx - }() - - defer span.Finish() - resultVar0 := a.app.NewWebHub() - - return resultVar0 -} - func (a *OpenTracingAppLayer) NotifyAndSetWarnMetricAck(warnMetricId string, sender *model.User, forceAck bool, isBot bool) *model.AppError { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.NotifyAndSetWarnMetricAck") diff --git a/app/server.go b/app/server.go index 6c65e0c367..cedc3f9e6c 100644 --- a/app/server.go +++ b/app/server.go @@ -280,8 +280,7 @@ 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. - app := New(ServerConnector(s.Channels())) - app.HubStart() + s.HubStart() if *s.Config().LogSettings.EnableDiagnostics && *s.Config().LogSettings.EnableSentry { if strings.Contains(SentryDSN, "placeholder") { @@ -543,7 +542,6 @@ func NewServer(options ...Option) (*Server, error) { s.WebSocketRouter = &WebSocketRouter{ handlers: make(map[string]webSocketHandler), - app: app, } mailConfig := s.MailServiceConfig() @@ -666,7 +664,8 @@ func NewServer(options ...Option) (*Server, error) { // if enabled - perform initial product notices fetch if *s.Config().AnnouncementSettings.AdminNoticesEnabled || *s.Config().AnnouncementSettings.UserNoticesEnabled { go func() { - if err := app.UpdateProductNotices(); err != nil { + appInstance := New(ServerConnector(s.Channels())) + if err := appInstance.UpdateProductNotices(); err != nil { mlog.Warn("Failied to perform initial product notices fetch", mlog.Err(err)) } }() @@ -678,8 +677,9 @@ func NewServer(options ...Option) (*Server, error) { c := request.EmptyContext() s.AddConfigListener(func(oldConfig *model.Config, newConfig *model.Config) { + appInstance := New(ServerConnector(s.Channels())) if *oldConfig.GuestAccountsSettings.Enable && !*newConfig.GuestAccountsSettings.Enable { - if appErr := app.DeactivateGuests(c); appErr != nil { + if appErr := appInstance.DeactivateGuests(c); appErr != nil { mlog.Error("Unable to deactivate guest accounts", mlog.Err(appErr)) } } @@ -687,16 +687,18 @@ func NewServer(options ...Option) (*Server, error) { // Disable active guest accounts on first run if guest accounts are disabled if !*s.Config().GuestAccountsSettings.Enable { - if appErr := app.DeactivateGuests(c); appErr != nil { + appInstance := New(ServerConnector(s.Channels())) + if appErr := appInstance.DeactivateGuests(c); appErr != nil { mlog.Error("Unable to deactivate guest accounts", mlog.Err(appErr)) } } if s.runEssentialJobs { s.Go(func() { + appInstance := New(ServerConnector(s.Channels())) s.runLicenseExpirationCheckJob() - runCheckAdminSupportStatusJob(app, c) - runDNDStatusExpireJob(app) + runCheckAdminSupportStatusJob(appInstance, c) + runDNDStatusExpireJob(appInstance) }) s.runJobs() } @@ -907,7 +909,7 @@ func (s *Server) removeUnlicensedLogTargets(license *model.License) { }) } -func (s *Server) startInterClusterServices(license *model.License, app *App) error { +func (s *Server) startInterClusterServices(license *model.License) error { if license == nil { mlog.Debug("No license provided; Remote Cluster services disabled") return nil @@ -956,7 +958,8 @@ func (s *Server) startInterClusterServices(license *model.License, app *App) err return nil } - scs, err := sharedchannel.NewSharedChannelService(s, app) + appInstance := New(ServerConnector(s.Channels())) + scs, err := sharedchannel.NewSharedChannelService(s, appInstance) if err != nil { return err } @@ -1417,7 +1420,7 @@ func (s *Server) Start() error { } } - if err := s.startInterClusterServices(s.License(), s.WebSocketRouter.app); err != nil { + if err := s.startInterClusterServices(s.License()); err != nil { mlog.Error("Error starting inter-cluster services", mlog.Err(err)) } diff --git a/app/web_hub.go b/app/web_hub.go index 868e380f73..94d8c8b0dc 100644 --- a/app/web_hub.go +++ b/app/web_hub.go @@ -50,7 +50,7 @@ type Hub struct { // connectionCount should be kept first. // See https://github.com/mattermost/mattermost-server/pull/7281 connectionCount int64 - app *App + srv *Server connectionIndex int register chan *WebConn unregister chan *WebConn @@ -65,10 +65,10 @@ type Hub struct { checkConn chan *webConnCheckMessage } -// NewWebHub creates a new Hub. -func (a *App) NewWebHub() *Hub { +// newWebHub creates a new Hub. +func newWebHub(s *Server) *Hub { return &Hub{ - app: a, + srv: s, register: make(chan *WebConn), unregister: make(chan *WebConn), broadcast: make(chan *model.WebSocketEvent, broadcastQueueSize), @@ -87,21 +87,21 @@ func (a *App) TotalWebsocketConnections() int { } // HubStart starts all the hubs. -func (a *App) HubStart() { +func (s *Server) HubStart() { // Total number of hubs is twice the number of CPUs. numberOfHubs := runtime.NumCPU() * 2 - mlog.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs)) + s.Log.Info("Starting websocket hubs", mlog.Int("number_of_hubs", numberOfHubs)) hubs := make([]*Hub, numberOfHubs) for i := 0; i < numberOfHubs; i++ { - hubs[i] = a.NewWebHub() + hubs[i] = newWebHub(s) hubs[i].connectionIndex = i hubs[i].Start() } // Assigning to the hubs slice without any mutex is fine because it is only assigned once // during the start of the program and always read from after that. - a.ch.srv.hubs = hubs + s.hubs = hubs } func (a *App) invalidateCacheForWebhook(webhookID string) { @@ -117,10 +117,6 @@ func (s *Server) HubStop() { } } -func (a *App) HubStop() { - a.Srv().HubStop() -} - // GetHubForUserId returns the hub for a given user id. func (s *Server) GetHubForUserId(userID string) *Hub { // TODO: check if caching the userID -> hub mapping @@ -356,7 +352,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.app.Metrics(); metrics != nil { + if metrics := h.srv.Metrics; metrics != nil { metrics.IncrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1) } select { @@ -416,6 +412,8 @@ func (h *Hub) Start() { ticker := time.NewTicker(inactiveConnReaperInterval) defer ticker.Stop() + appInstance := New(ServerConnector(h.srv.Channels())) + connIndex := newHubConnectionIndex(inactiveConnReaperInterval) for { @@ -449,7 +447,7 @@ func (h *Hub) Start() { connIndex.RemoveInactiveConnections() case webConn := <-h.register: var oldConn *WebConn - if *h.app.Config().ServiceSettings.EnableReliableWebSockets { + if *h.srv.Config().ServiceSettings.EnableReliableWebSockets { // Delete the old conn from connIndex if it exists. oldConn = connIndex.RemoveInactiveByConnectionID( webConn.GetSession().UserId, @@ -474,7 +472,7 @@ func (h *Hub) Start() { case webConn := <-h.unregister: // If already removed (via queue full), then removing again becomes a noop. // But if not removed, mark inactive. - if *h.app.Config().ServiceSettings.EnableReliableWebSockets { + if *h.srv.Config().ServiceSettings.EnableReliableWebSockets { webConn.active = false } else { connIndex.Remove(webConn) @@ -488,8 +486,8 @@ func (h *Hub) Start() { conns := connIndex.ForUser(webConn.UserId) if len(conns) == 0 || areAllInactive(conns) { - h.app.Srv().Go(func() { - h.app.SetStatusOffline(webConn.UserId, false) + h.srv.Go(func() { + appInstance.SetStatusOffline(webConn.UserId, false) }) continue } @@ -503,9 +501,9 @@ func (h *Hub) Start() { } } - if h.app.IsUserAway(latestActivity) { - h.app.Srv().Go(func() { - h.app.SetStatusLastActivityAt(webConn.UserId, latestActivity) + if appInstance.IsUserAway(latestActivity) { + h.srv.Go(func() { + appInstance.SetStatusLastActivityAt(webConn.UserId, latestActivity) }) } case userID := <-h.invalidateUser: @@ -533,7 +531,7 @@ func (h *Hub) Start() { connIndex.Remove(directMsg.conn) } case msg := <-h.broadcast: - if metrics := h.app.Metrics(); metrics != nil { + if metrics := h.srv.Metrics; metrics != nil { metrics.DecrementWebSocketBroadcastBufferSize(strconv.Itoa(h.connectionIndex), 1) } msg = msg.PrecomputeJSON() @@ -565,7 +563,7 @@ func (h *Hub) Start() { case <-h.stop: for webConn := range connIndex.All() { webConn.Close() - h.app.SetStatusOffline(webConn.UserId, false) + appInstance.SetStatusOffline(webConn.UserId, false) } h.explicitStop = true diff --git a/app/web_hub_test.go b/app/web_hub_test.go index 294c8a4e9a..6866a7fd6f 100644 --- a/app/web_hub_test.go +++ b/app/web_hub_test.go @@ -66,7 +66,7 @@ func TestHubStopWithMultipleConnections(t *testing.T) { s := httptest.NewServer(dummyWebsocketHandler(t)) defer s.Close() - th.App.HubStart() + th.Server.HubStart() wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) wc2 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) wc3 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) @@ -84,12 +84,12 @@ func TestHubStopRaceCondition(t *testing.T) { // So we just use this quick hack for the test. s := httptest.NewServer(dummyWebsocketHandler(t)) - th.App.HubStart() + th.Server.HubStart() wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) defer wc1.Close() hub := th.App.Srv().hubs[0] - th.App.HubStop() + th.Server.HubStop() done := make(chan bool) go func() { @@ -347,7 +347,7 @@ func TestHubIsRegistered(t *testing.T) { s := httptest.NewServer(dummyWebsocketHandler(t)) defer s.Close() - th.App.HubStart() + th.Server.HubStart() wc1 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) wc2 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) wc3 := registerDummyWebConn(t, th.App, s.Listener.Addr(), th.BasicUser.Id) @@ -418,7 +418,7 @@ func BenchmarkGetHubForUserId(b *testing.B) { th := Setup(b).InitBasic() defer th.TearDown() - th.App.HubStart() + th.Server.HubStart() b.ResetTimer() for i := 0; i < b.N; i++ { diff --git a/app/webhub_fuzz.go b/app/webhub_fuzz.go index ea5a1a5f26..00ae47fc80 100644 --- a/app/webhub_fuzz.go +++ b/app/webhub_fuzz.go @@ -153,7 +153,7 @@ func Fuzz(data []byte) int { s := httptest.NewServer(dummyWebsocketHandler()) - th.App.HubStart() + th.Server.HubStart() u1 := th.CreateUser() u2 := th.CreateUser() diff --git a/app/websocket_router.go b/app/websocket_router.go index 8a662d5731..97e9cd9547 100644 --- a/app/websocket_router.go +++ b/app/websocket_router.go @@ -16,7 +16,6 @@ type webSocketHandler interface { } type WebSocketRouter struct { - app *App handlers map[string]webSocketHandler } @@ -27,13 +26,13 @@ func (wr *WebSocketRouter) Handle(action string, handler webSocketHandler) { func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketRequest) { if r.Action == "" { err := model.NewAppError("ServeWebSocket", "api.web_socket_router.no_action.app_error", nil, "", http.StatusBadRequest) - returnWebSocketError(wr.app, conn, r, err) + returnWebSocketError(conn.App, conn, r, err) return } if r.Seq <= 0 { err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_seq.app_error", nil, "", http.StatusBadRequest) - returnWebSocketError(wr.app, conn, r, err) + returnWebSocketError(conn.App, conn, r, err) return } @@ -48,7 +47,7 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque return } - session, err := wr.app.GetSession(token) + session, err := conn.App.GetSession(token) if err != nil { conn.WebSocket.Close() return @@ -59,15 +58,15 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque // TODO: Same logic to reconnect queue as api4/websocket.go - wr.app.HubRegister(conn) + conn.App.HubRegister(conn) - wr.app.Srv().Go(func() { - wr.app.SetStatusOnline(session.UserId, false) - wr.app.UpdateLastActivityAtIfNeeded(*session) + conn.App.Srv().Go(func() { + conn.App.SetStatusOnline(session.UserId, false) + conn.App.UpdateLastActivityAtIfNeeded(*session) }) resp := model.NewWebSocketResponse(model.StatusOk, r.Seq, nil) - hub := wr.app.GetHubForUserId(conn.UserId) + hub := conn.App.GetHubForUserId(conn.UserId) if hub == nil { return } @@ -78,14 +77,14 @@ func (wr *WebSocketRouter) ServeWebSocket(conn *WebConn, r *model.WebSocketReque if !conn.IsAuthenticated() { err := model.NewAppError("ServeWebSocket", "api.web_socket_router.not_authenticated.app_error", nil, "", http.StatusUnauthorized) - returnWebSocketError(wr.app, conn, r, err) + returnWebSocketError(conn.App, conn, r, err) return } handler, ok := wr.handlers[r.Action] if !ok { err := model.NewAppError("ServeWebSocket", "api.web_socket_router.bad_action.app_error", nil, "", http.StatusInternalServerError) - returnWebSocketError(wr.app, conn, r, err) + returnWebSocketError(conn.App, conn, r, err) return }